-
[OPIK-3056] [BE] Add cleanup job for dataset exports (PR#6) (#4743)
frostbyte_neo 发布于
2026-01-16 10:07:55 +00:00 -
OPIK-3056: Add database schema and API models for dataset export jobs
-
Revision 2: Address PR comments - fix migration number, author, and field sizes
-
Revision 3: Change file_path to TEXT for unlimited length
-
Revision 4: Add ON UPDATE CURRENT_TIMESTAMP to last_updated_at
-
Revision 5: Add last_updated_by field for audit trail consistency
-
Revision 9: Rename migration file from 000042 to 000043 to avoid conflict
-
OPIK-3056: Add streaming upload method to FileService
-
[OPIK-3056] [BE] Add dataset export job management with Redis stream integration
-
Remove files
-
Revision 2: Address PR comments - add workspace security, fix listener error handling, extract constants, rename polling
-
Fix config issues
-
Revision 5: Remove manual last_updated_at handling (now automatic via DB)
-
Revision 6: Rename DatasetExportJobListener to DatasetExportJobSubscriber and add toggle
-
Revision 7: Split update methods into updateToCompleted and updateToFailed
-
Revision 8: Add unit tests for DatasetExportJobSubscriber lifecycle gating
-
Revision 8: Add last_updated_by field for audit trail
-
Revision 4: Address PR comments - add workspace security, feature toggle, and tests
- Add workspace_id filtering to DatasetExportJobDAO.deleteByIds to prevent cross-tenant deletion
- Add isEnabled() check in CsvDatasetExportService.startExport to prevent jobs when export is disabled
- Add test for disabled export feature
- Update existing tests to mock isEnabled() returning true
- Verify last_updated_at is handled by DB default (ON UPDATE CURRENT_TIMESTAMP)
- Existing DatasetExportJobSubscriberTest already covers lifecycle gating
- Revision 5: Fix DatasetExportJobDAO injection issue
- Remove direct DAO injection from DatasetExportJobService
- Update test to match new constructor signature
- DAO is correctly attached via handle.attach() in transactions
- Fixes Guice injection error on application startup
- All tests pass successfully
- Revision 6: Refactor duplicate NotFoundException tests into parameterized test
- Merge updateJobToCompleted and updateJobToFailed NotFoundException tests
- Use @MethodSource instead of hardcoded test data
- Generate dynamic file paths and error messages
- Reduces test duplication while maintaining coverage
- [OPIK-3056] Implement CSV generation and export processing
- Add CsvDatasetExportProcessor for generating CSV from dataset items
- Implement DatasetExportJobSubscriber to consume export jobs from Redis
- Add DatasetItemDAO.getColumns() method for column discovery
- Update DatasetExportJobService to use TransactionTemplate correctly
- Add comprehensive unit tests for CSV processor and subscriber
- All tests passing (7/7)
-
Fix toggle
-
OPIK-3056: Implement CSV generation with S3 multipart upload streaming
-
Revision 2: Add error message truncation utility and infrastructure updates
-
Fix config
-
Revision 3: Address all PR review comments
- Fix empty dataset handling: Add defaultIfEmpty to getColumns for header-only CSV
- Fix NPE: Handle null error messages in DatasetExportJobSubscriber
- Fix data loss: Refactor buffer loop to process all rows without early return
- Fix exception type: Replace RuntimeException with IllegalStateException
- Fix pagination: Implement proper cursor-based pagination in streamAllItems
- Implement abort: Add abortMultipartUpload to FileService for cleanup
- Fix test config: Enable datasetExport feature in config-test.yml
- Add S3 validation: Clamp minPartSize to S3 minimum (5MB)
- Revision 4: Use StringUtils.defaultIfBlank for NPE handling
Replace manual null check with Apache Commons StringUtils.defaultIfBlank utility method,
following project conventions for handling null strings.- Revision 5: Fix CsvDatasetExportProcessorImplTest for multipart upload
- Update all test verifications to check multipart upload methods (createMultipartUpload,
uploadPart, completeMultipartUpload) instead of the old single upload method - Handle special case for empty datasets which abort multipart and fall back to regular upload
- All 5 tests now pass successfully
- Revision 6: Address new PR review comments
-
Validate maxPartSize against minPartSize to prevent S3 EntityTooSmall errors
- Clamp maxPartSize to Math.max(configuredMaxPartSize, minPartSize)
- Log warning when clamping occurs
- Ensures non-final parts always meet S3's 5MB minimum requirement
-
Extract duplicate NotFoundException check into verifyJobExistsOrThrow helper
- Reduces code duplication in updateJobToCompleted and updateJobToFailed
- Centralizes job existence validation logic
- Improves maintainability
- Revision 7: Simplify minPartSize/maxPartSize validation
Remove intermediate variables and warning logs since S3 has a hard limit of 5MB.
The Math.max() clamping is sufficient - no need to track original config values.- Revision 8: Extract duplicate CSV writing plumbing into helper method
Extract repeated ByteArrayOutputStream + OutputStreamWriter + CSVPrinter setup
into a reusable writeCsv() helper method with ThrowingConsumer interface.Benefits:
- Removes code duplication between createCsvHeader and convertItemToCsvRow
- Centralizes CSV writer creation, flushing, and exception handling
- Simplifies both methods to focus on record data logic only
- Maintains same charset (UTF-8) and CSVFormat.DEFAULT behavior
All tests pass - behavior unchanged.
- OPIK-3056: Add download URL generation and cleanup job for dataset exports
- Add download_url field to DatasetExportJob model
- Generate presigned URLs for completed exports with configurable TTL
- Add viewed_at field to track when users view export jobs
- Implement DatasetExportCleanupJob for hourly cleanup of:
- Expired completed jobs (after TTL expires)
- Viewed failed jobs (after user has seen the error)
- Add distributed lock to cleanup job to prevent concurrent execution
- Update state machine to properly handle:
- PROCESSING → COMPLETED: Set download_url, expires_at, clear viewed_at
- PENDING/PROCESSING → FAILED: Preserve file_path for cleanup
- Add overloaded presignDownloadUrl method with custom TTL
- Add comprehensive unit and integration tests for cleanup job
- Fix Mono.zip() reactive execution issue in concurrent cleanup
- Add database migrations for download_url and viewed_at columns
- Add configurable cleanup timeout and lock wait time
-
OPIK-3056: Address PR comments - use StringUtils, add idempotency, validate TTL
-
OPIK-3056: Refactor idempotency checks to eliminate duplication
-
Revision 6: Address PR comments on config validation and performance
- Updated config.yml comments to clarify runtime clamping behavior
- Removed unnecessary String allocations in convertItemToCsvRow
- Changed convertItemToCsvRow to return byte[] instead of String
- Updated appendToBuffer to work with byte[] directly
- Preserve database column order (removed sorting)
- Updated test expectations to match database column order
-
Revision 5: Address PR review comments - exception handling, scheduler, idempotency docs
-
OPIK-3056: Remove downloadUrl from dataset export - downloads now use proxy endpoint
-
Revision 2: Address PR review comments
- Fix Liquibase changeset ID mismatch (000045_add_viewed_at_to_dataset_export_jobs)
- Rename updateStatus to markPendingJobAsProcessing in DAO for clarity
- Rename deleteExpiredJobs to deleteJobsByIds in DAO for clarity
- Simplify verifyJobUpdatedToStatus (removed special terminal state handling)
- Move logging into verifyJobUpdatedToStatus to avoid duplication
- Improve exception handling in DatasetExportCleanupJob.doJob
-
Revision 3: Fix flaky integration tests and remove debug logging
-
Revision 5: Add MinIO container to cleanup integration tests
-
[OPIK-3056] [BE] Address Daniel's PR review comments
- Remove status parameter from markPendingJobAsProcessing (hardcode PROCESSING)
- Remove status parameter from updateToFailed (hardcode FAILED)
- Remove debug log from PreSignerService
- Merge Liquibase migrations 00045 and 00046 into single file
- Make CLEANUP_BATCH_SIZE configurable via DatasetExportConfig
- Update all test mocks and verifications to match new method signatures
- [OPIK-3056] [BE] Add cleanupBatchSize to config files
- Add cleanupBatchSize configuration to config.yml with environment variable support
- Add cleanupBatchSize to test config file
下载附件
-