Add an opt-in controller-copy METADATA push for same-name segment refreshes

[KKcorps]

TL;DR

BaseSingleSegmentConversionExecutor gets an opt-in hook, isCopyToDeepStoreForMetadataPush(),
default false. A task that returns true gets a METADATA push that is safe for refreshing a
segment under its own name: the tar is staged under a name that can never be a live segment path,
the push keeps the TAR path's If-Match and REFRESH_ONLY guards, the controller copies the bytes
into the segment's existing deep store location, and the staged tar is deleted afterwards. Nothing
changes for PurgeTask, RefreshSegmentTask or UpsertCompactionTask unless they opt in.

The problem

#17632 added a METADATA mode to this executor. It stages the converted tar at
<outputDir>/<segment>.tar.gz and registers that URI as the segment's download URL, with
COPY_SEGMENT_TO_DEEP_STORE=false. That is the right shape for a new segment. For a task that
refreshes an existing segment under the same name it has three gaps:

1. The If-Match (CRC) and REFRESH_ONLY headers that the TAR path sends are dropped. A push
   from a stale task can overwrite a segment refreshed after the task was planned, and a push for a
   segment deleted in the meantime recreates it.
2. <table>/<segment>.tar.gz is the live file for a segment that was ingested by a metadata-push
   ingestion task. With overwriteOutput=false the refresh fails. With true it overwrites live
   bytes before ZooKeeper is updated, and the failure cleanup then deletes the live file.
3. Because the download URL moves to the staged path, the staged tar becomes the live file and can
   never be deleted, the previous deep store object is orphaned, and the second refresh of the same
   segment fails with "already exists".

flowchart LR
  subgraph Before["Existing METADATA push (unchanged default)"]
    A[stage at table/segment.tar.gz] --> B[register staged URI as download URL]
    B --> C[old object orphaned, next refresh collides]
  end
  subgraph After["Opt-in controller copy"]
    D[stage at table/segment.taskId.tar.gz] --> E[push metadata + If-Match + REFRESH_ONLY + copy flag]
    E --> F[controller copies into table/segment]
    F --> G[download URL unchanged, staged tar deleted]
  end

Loading

The approach

1. isCopyToDeepStoreForMetadataPush() returns false by default. The existing METADATA method is
   kept byte for byte and stays the default branch. The TAR path is untouched.
2. When the hook returns true, the executor builds the metadata tar (metadata.properties and
   creation.meta) from the local converted segment before the segment directories are deleted.
   The old path downloaded the staged tar back from the output filesystem to extract those two files.
3. If the conversion left the segment unchanged (same CRC) and the task carries a download URL, the
   executor skips tarring and staging and only re-registers the metadata against that URL.
4. Otherwise the tar is staged at <outputDir>/<segment>.<taskId>.tar.gz. Live segments are
   stored as <segment> or <segment>.tar.gz, so the name can never collide. It is stable across
   retries of the same task, so a retry overwrites what an interrupted attempt left, and unique
   across tasks. A plain-path output dir (local deep store) is sent as a file URI, because the
   controller picks the filesystem by scheme.
5. The push reuses the TAR path's header list and adds DOWNLOAD_URI, UPLOAD_TYPE=METADATA and
   COPY_SEGMENT_TO_DEEP_STORE=true. The controller runs its existing CRC check and parallel-push
   lock, copies the staged tar server-side into <dataDir>/<table>/<segment>, updates ZooKeeper
   and sends the refresh message. That is the same order the TAR path already uses.
6. The staged tar is deleted in a finally. Nothing references it after the controller copied it,
   so this is safe on success, failure and timeout.

Key components

Class / file	Role
BaseSingleSegmentConversionExecutor	New hook, uploadSegmentMetadataWithControllerCopy, staging name, header assembly. Existing uploadSegmentWithMetadata restored as the default branch.
BaseTaskExecutor	moveSegmentToOutputPinotFS(configs, file, outputFileName, overwrite) overload (the 2-arg version delegates to it), deleteFromOutputPinotFS, createSegmentMetadataTarFile.
SegmentConversionUtils	uploadSegmentMetadata, sharing the retry, round-robin and HOST header logic with uploadSegment through one private uploadWithRetry.

Flow (opt-in path)

sequenceDiagram
  participant M as Minion
  participant O as Output PinotFS
  participant C as Controller
  participant D as Deep store
  participant Z as ZooKeeper
  M->>M: convert, build metadata tar from local segment
  alt segment unchanged (same CRC)
    M->>C: POST metadata, DOWNLOAD_URI = current download URL, copy=false
    C->>Z: update custom map / refresh time
  else segment changed
    M->>O: put table/segment.taskId.tar.gz (overwrite ok)
    M->>C: POST metadata, If-Match, REFRESH_ONLY, DOWNLOAD_URI = staged, copy=true
    C->>C: CRC check, parallel-push lock
    C->>D: server-side copy staged -> table/segment
    C->>Z: update segment metadata, send refresh message
    C-->>M: 200
    M->>O: delete staged tar (always, in finally)
  end

Loading

Compatibility

• No behavior change for any existing task. The default branch is the pre-existing method and the
  test that pinned its failure cleanup is kept.
• No new config keys. The opt-in is a protected method, not a task config, so it cannot be turned
  on by accident from a table config.
• Everything the opt-in path sends already exists on the controller side: If-Match,
  REFRESH_ONLY, the custom-map header, UPLOAD_TYPE=METADATA, DOWNLOAD_URI and
  COPY_SEGMENT_TO_DEEP_STORE, including the deep store copy on an existing-segment refresh. A new
  minion works against an old controller.
• uploadSegment keeps its signature. Only additions to BaseTaskExecutor.

Failure handling

Controller dies after the deep store copy, before the ZooKeeper update. The deep store then holds the new
bytes while ZooKeeper still has the old CRC and the segment's upload lock. This window is not new: the TAR path
copies into the deep store before the ZooKeeper write in the same method (copyFromSegmentFileToDeepStore). It
is closed by rerunning the task, which is what Helix does on failure:

• The minion's own retries first hit the held lock (409, controller.segment.upload.timeoutInMillis, default
  10 minutes), then the task attempt fails and Helix reruns it. The rerun passes the CRC pre-check (ZooKeeper still
  has the original CRC), downloads the already-converted bytes from the canonical location, converts them again
  (idempotent), stages, and pushes with the same If-Match. Once the lock has expired the controller copies again
  and commits. ZooKeeper and the deep store converge on the new CRC.
• If the controller had already committed ZooKeeper and only the response was lost, the retry gets 412 from
  If-Match, the attempt fails, and the rerun stops at the CRC pre-check: the segment is already refreshed.
• A server that downloads during the window loads the new bytes; when the commit lands its local CRC already
  matches, so the refresh message costs it nothing.

Minion dies after the upload, before deleting the staged tar. One <segment>.<taskId>.tar.gz is left in the
table's deep store directory. A retry of the same task reuses the name and overwrites it, then deletes it. If the
task never runs again, the file is an untracked segment for RetentionManager's deep store sweep
(controller.retentionManager.untrackedSegmentDeletionEnabled, default off in OSS, retention
controller.retentionManager.untrackedSegmentsRetentionTimeInDays, default 3), which moves it to
Deleted_Segments and then purges it. That sweep is non-recursive, which is why the staged tar stays flat in the
table directory instead of a subdirectory. Deleting a staged tar is always safe because the controller never points
a download URL at it.

Testing

BaseSingleSegmentConversionExecutorTest:

• Default path: still registers the staged URI through SegmentPushUtils.sendSegmentUriAndMetadata
  and never calls the new helper; still deletes the staged tar when the push fails (existing test).
• Opt-in path: staging name and its deletion after success and after failure; the header set
  (If-Match, REFRESH_ONLY, custom-map, DOWNLOAD_URI, UPLOAD_TYPE, copy flag); the metadata
  tar contains exactly metadata.properties and creation.meta and describes the converted segment;
  a leftover staged tar from an earlier attempt is overwritten; a scheme-less output dir is sent as a
  file URI; an unchanged segment is not staged and is re-registered against its download URL.

PurgeMinionClusterIntegrationTest passes, including testFirstRunMetadataPushPurge on the
unchanged default path. The opt-in path was exercised end to end on a local-FS cluster from the
StarTree purge task that overrides the hook.

Follow-up: #19490 (stacked on this PR) builds the metadata tar from the local segment for every METADATA push, including the default path, so no path downloads the staged tar back.

🤖 Generated with Claude Code

https://claude.ai/code/session_01Y375AgHYsh1YqNSsfvF1a8

[codecov-commenter]

Codecov Report

❌ Patch coverage is 77.14286% with 16 lines in your changes missing coverage. Please review.
✅ Project coverage is 67.73%. Comparing base (5e05b9c) to head (c0a33a7).
⚠️ Report is 2 commits behind head on master.

Files with missing lines	Patch %	Lines
...ot/plugin/minion/tasks/SegmentConversionUtils.java	0.00%	7 Missing ⚠️
...ion/tasks/BaseSingleSegmentConversionExecutor.java	86.66%	2 Missing and 4 partials ⚠️
...he/pinot/plugin/minion/tasks/BaseTaskExecutor.java	83.33%	2 Missing and 1 partial ⚠️

Additional details and impacted files

@@              Coverage Diff              @@
##             master   #19488       +/-   ##
=============================================
+ Coverage     39.44%   67.73%   +28.28%     
- Complexity     1429     1450       +21     
=============================================
  Files          3488     3488               
  Lines        224424   224483       +59     
  Branches      35429    35437        +8     
=============================================
+ Hits          88528   152051    +63523     
+ Misses       127912    60421    -67491     
- Partials       7984    12011     +4027     

Flag	Coverage Δ
integration	100.00% <ø> (ø)
integration1	100.00% <ø> (ø)
integration2	0.00% <ø> (ø)
java-25	67.73% <77.14%> (+28.28%)	⬆️
lane-a	100.00% <ø> (ø)
lane-b	0.00% <ø> (ø)
temurin	67.73% <77.14%> (+28.28%)	⬆️
unittests	67.73% <77.14%> (+28.28%)	⬆️
unittests1	57.78% <ø> (?)
unittests2	39.46% <77.14%> (+0.01%)	⬆️

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

☔ View full report in Codecov by Harness.
📢 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.

[shounakmk219]

LGTM