Filter segments near retention expiry from MergeRollup and UpsertCompactMerge task generators

[jineshparakh]

Summary

Add a retention-aware filter to MergeRollupTaskGenerator and UpsertCompactMergeTaskGenerator that excludes segments nearing their retention expiry from task generation. This closes a race condition where RetentionManager could delete segments between the time a task is generated (controller-side) and when the executor downloads them (minion-side), causing NoSuchKeyException failures and misleading alerts.

Design Decisions

Generator-side filtering over executor-side skipping

For merge/rollup tasks, we chose generator-side filtering rather than executor-side skipping because:

• Executor-side skipping would break segment lineage: if segment B is skipped mid-merge of {A, B, C} into D, the resulting merged segment D would be incomplete, and the lineage
  record would incorrectly claim B was replaced.
• Executor-side skipping would also break CRC validation and ValidDocIds fetching for upsert compaction, requiring significant refactoring.
• Generator-side filtering is safe because it happens before the task config (including lineage entries and segment lists) is built, so no lineage or CRC issues can arise.

Configurable buffer via retentionExpiryBufferPeriod

• A new optional task config key retentionExpiryBufferPeriod (e.g. "1h", "30m") allows operators to exclude segments earlier than the exact retention boundary, providing a safety
  margin against clock skew between controller and RetentionManager.
• The effective retention becomes retentionMs - bufferMs. If the buffer >= retention, the filter fails open (returns all segments) with a WARN log.
• The config is read from the per-task-type config block (e.g. task.MergeRollupTask.retentionExpiryBufferPeriod), so different tasks on the same table can have different buffer values.
  It is table-level within each task (not per-merge-level for MergeRollupTask) because retention itself is table-level.

Single source of truth for retention logic

• RetentionUtils.isPurgeable() is the single source of truth for the time-comparison and creation-time fallback logic. It supports the
  controller.retentionManager.enableCreationTimeFallback cluster config introduced in Add support for handling scenarios where end time is invalid during RetentionManager run #18148.
• TimeRetentionStrategy delegates to RetentionUtils.isPurgeable() after its completion-status check, ensuring the controller's RetentionManager and the minion task generators
  always use identical retention logic — the invariant is enforced by construction, not by discipline.
• Both MergeRollupTaskGenerator and UpsertCompactMergeTaskGenerator read the creation-time fallback flag from Helix cluster config via
  MinionTaskUtils.isCreationTimeFallbackEnabled() — the same authoritative source that RetentionManager reacts to via onChange(). The read is hoisted before the per-table loop to
  avoid redundant ZK calls and to provide a consistent config snapshot across all tables in a single scheduling pass.

Fail-open on all error paths

• Missing retention config: return all segments (no filtering)
• Malformed retention unit/value: WARN log, return all segments
• Non-positive retention: WARN log, return all segments
• Malformed buffer period: WARN log, use 0 (no buffer)
• Buffer >= retention: WARN log, return all segments
• Invalid end time with fallback disabled: WARN log, not purgeable
• Invalid end time and invalid creation time with fallback enabled: WARN log, not purgeable

Watermark impact (MergeRollupTask)

If all segments in an early time bucket are filtered out, the watermark will advance past them permanently. This is expected since those segments would be purged by RetentionManager
regardless. However, if caused by a misconfigured buffer, correcting the config will not recover already-skipped buckets. This is documented in the Javadoc.

currentTimeMs as a parameter

filterSegmentsPastRetention accepts currentTimeMs as an explicit parameter rather than calling System.currentTimeMillis() internally. This makes the method deterministic and testable
so boundary tests can use exact values without clock-drift workarounds.

Single constant definition

RETENTION_EXPIRY_BUFFER_PERIOD_KEY is defined once as a public static final in MinionTaskUtils (the utility that reads it). It is not duplicated in MinionConstants.MergeTask or
MinionConstants.UpsertCompactMergeTask to avoid triple-definition drift.

Test Plan

• RetentionUtilsTest: boundary tests, creation-time fallback (enabled/disabled), invalid times, priority of end time over creation time
• MinionTaskUtilsTest: all edge cases for filterSegmentsPastRetention (expired, retained, invalid end time, malformed config, buffer, boundary)
• MinionTaskUtilsTest: creation-time fallback integration through filterSegmentsPastRetention (fallback enabled filters old creation time, fallback disabled keeps it, fallback
  enabled keeps recent creation time)
• MinionTaskUtilsTest: isCreationTimeFallbackEnabled helper (null/absent key, "true", "false", non-boolean string)
• UpsertCompactMergeTaskGeneratorTest.testRetentionExpiryBufferFiltersCandidates: pipeline integration with getCandidateSegments
• MergeRollupTaskGeneratorTest.testRetentionExpiryBufferFiltersSegments: full generateTasks flow with mocked ClusterInfoAccessor
• TimeRetentionStrategyTest: existing tests continue to pass (now exercise RetentionUtils delegation path)
• No flaky tests: all time-sensitive tests pass controlled currentTimeMs

[jineshparakh]

@KKcorps @shounakmk219 requesting review

[shounakmk219]

lgtm

[codecov-commenter]

Codecov Report

❌ Patch coverage is 88.31169% with 9 lines in your changes missing coverage. Please review.
✅ Project coverage is 63.39%. Comparing base (ab9ac39) to head (87c337d).
⚠️ Report is 16 commits behind head on master.

Files with missing lines	Patch %	Lines
...che/pinot/plugin/minion/tasks/MinionTaskUtils.java	89.79%	3 Missing and 2 partials ⚠️
...tcompactmerge/UpsertCompactMergeTaskGenerator.java	20.00%	4 Missing ⚠️

Additional details and impacted files

@@             Coverage Diff              @@
##             master   #18285      +/-   ##
============================================
- Coverage     63.61%   63.39%   -0.23%     
- Complexity     1659     1679      +20     
============================================
  Files          3246     3253       +7     
  Lines        197514   198721    +1207     
  Branches      30578    30780     +202     
============================================
+ Hits         125656   125982     +326     
- Misses        61813    62668     +855     
- Partials      10045    10071      +26     

Flag	Coverage Δ
custom-integration1	100.00% <ø> (ø)
integration	100.00% <ø> (ø)
integration1	100.00% <ø> (ø)
integration2	0.00% <ø> (ø)
java-11	?
java-21	63.39% <88.31%> (-0.19%)	⬇️
temurin	63.39% <88.31%> (-0.23%)	⬇️
unittests	63.39% <88.31%> (-0.23%)	⬇️
unittests1	55.35% <100.00%> (-0.24%)	⬇️
unittests2	34.93% <88.31%> (-0.12%)	⬇️

Flags with carried forward coverage won't be shown. Click here to find out more.

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[jineshparakh]

@xiangfu0 @noob-se7en can any one of you please re-trigger the failed tests?
The failures seem unrelated.

[jineshparakh]

@xiangfu0 can you please re-review this? I've made the changes

[xiangfu0]

Found one high-signal issue; see inline comment.