xiangfu0 commented on code in PR #19776:
URL: https://github.com/apache/pinot/pull/19776#discussion_r4278281818
##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/SplitSegmentCommitter.java:
##########
@@ -100,4 +124,50 @@ protected String uploadSegment(File segmentTarFile,
SegmentUploader segmentUploa
}
return null;
}
+
+ // A peer:// location means the segment is NOT in deep store, so there is
nothing for a sidecar to describe.
+ // Mirrors the check in PinotLLCRealtimeSegmentManager.commitSegmentFile().
+ private static boolean isPeerSegmentLocation(String segmentLocation) {
+ return segmentLocation.regionMatches(true, 0,
CommonConstants.Segment.PEER_SEGMENT_DOWNLOAD_SCHEME, 0,
+ CommonConstants.Segment.PEER_SEGMENT_DOWNLOAD_SCHEME.length());
+ }
+
+ /// Best-effort: builds a tar containing one directory with
metadata.properties and creation.meta, and uploads it
+ /// under a tmp name next to the segment. Returns the tmp location for the
controller to move at commit end, or
+ /// null on any problem. Never fails or throws; a sidecar problem must not
affect the segment commit.
+ @Nullable
+ private String uploadMetadataTarQuietly(@Nullable Map<String, File>
metadataFiles, String segmentName) {
+ if (metadataFiles == null || metadataFiles.isEmpty()) {
+ return null;
+ }
+ String rawTableName = null;
+ File stagingDir = null;
+ File metadataTarFile = null;
+ try {
+ LLCSegmentName llcSegmentName = new LLCSegmentName(segmentName);
+ rawTableName =
TableNameBuilder.extractRawTableName(llcSegmentName.getTableName());
+ stagingDir = new File(FileUtils.getTempDirectory(), segmentName +
"_meta_" + UUID.randomUUID());
+ FileUtils.forceMkdir(stagingDir);
+ for (Map.Entry<String, File> entry : metadataFiles.entrySet()) {
+ // Byte-for-byte copy: creation.meta is binary.
+ FileUtils.copyFile(entry.getValue(), new File(stagingDir,
entry.getKey()));
+ }
+ metadataTarFile = new File(FileUtils.getTempDirectory(),
+ segmentName + "_" + UUID.randomUUID() +
Constants.METADATA_TAR_GZ_FILE_EXT);
+ // Tar the DIRECTORY so that the first archive entry is a directory, as
DefaultMetadataExtractor expects.
+ TarCompressionUtils.createCompressedTarFile(stagingDir, metadataTarFile);
+ URI metadataTarLocation =
+ _segmentUploader.uploadMetadataTar(metadataTarFile, llcSegmentName,
METADATA_TAR_UPLOAD_TIMEOUT_MS);
+ return metadataTarLocation == null ? null :
metadataTarLocation.toString();
+ } catch (Exception e) {
+ _segmentLogger.warn("Failed to upload metadata tar for segment: {};
continuing with commit", segmentName, e);
+ if (rawTableName != null) {
+ _serverMetrics.addMeteredTableValue(rawTableName,
ServerMeter.METADATA_TAR_UPLOAD_FAILURE, 1);
+ }
+ return null;
+ } finally {
+ FileUtils.deleteQuietly(stagingDir);
+ FileUtils.deleteQuietly(metadataTarFile);
Review Comment:
Minor: on an upload timeout `uploadMetadataTar` returns after cancelling the
future, but the underlying `copyFromLocalFile` may still be reading
`metadataTarFile` in the executor thread when this `finally` deletes it. It's
benign here (only a tmp object can be affected, and the controller writes the
final sidecar only on reported success), but worth a brief comment so a future
reader doesn't assume the local tar outlives the copy.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/SegmentDeletionManager.java:
##########
@@ -369,6 +370,33 @@ private void moveSegmentsToDeletedDir(String segmentId,
Long deletedSegmentsRete
LOGGER.warn("Could not move segment {} from {} to {}", segmentId,
fileToDeleteURI.toString(),
deletedSegmentMoveDestURI.toString(), e);
}
+ // Independent of whether the segment object existed, so no orphan sidecar
is left in the live directory.
+ moveSegmentMetadataToDeletedDir(pinotFS, fileToDeleteURI, segmentId,
rawTableName, retentionSuffix);
+ }
+
+ private void moveSegmentMetadataToDeletedDir(PinotFS pinotFS, URI
segmentFileUri, String segmentId,
+ String rawTableName, String retentionSuffix) {
+ // Live sidecar is <segment>.metadata.tar.gz; in Deleted_Segments it is
renamed to <segment>.metadata<suffix> so
+ // that, like the segment object, it carries no .tar.gz extension and the
retention suffix stays last.
+ try {
+ URI segmentMetadataUri =
SegmentPushUtils.generateSegmentMetadataURI(segmentFileUri.toString(),
segmentId);
+ if (!pinotFS.exists(segmentMetadataUri)) {
+ return;
+ }
+ URI deletedMetadataUri = URIUtils.getUri(_dataDir, DELETED_SEGMENTS,
rawTableName,
+ URIUtils.encode(segmentId) + DELETED_METADATA_FILE_SUFFIX +
retentionSuffix);
+ if (pinotFS.move(segmentMetadataUri, deletedMetadataUri, true)) {
+ pinotFS.touch(deletedMetadataUri);
Review Comment:
Nice touch using `touch()` so the retention-null sidecar expires from now
rather than inheriting a stale lastModified. The test covers this well. No
change needed.
##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/PinotFSSegmentUploader.java:
##########
@@ -112,4 +115,81 @@ public URI uploadSegment(File segmentFile, LLCSegmentName
segmentName, int timeo
return null;
}
+
+ @Override
+ public boolean isMetadataTarUploadSupported() {
+ return true;
+ }
+
+ @Override
+ public URI uploadMetadataTar(File metadataTarFile, LLCSegmentName
segmentName, int timeoutInMillis) {
+ if (_segmentStoreUriStr == null || _segmentStoreUriStr.isEmpty()) {
+ LOGGER.error("Missing segment store uri. Failed to upload metadata tar
{} for {}.", metadataTarFile.getName(),
+ segmentName.getSegmentName());
+ return null;
+ }
+ final String rawTableName =
TableNameBuilder.extractRawTableName(segmentName.getTableName());
+ // Like the segment file, upload to a unique tmp name. The controller
moves it to <segment>.metadata.tar.gz at
+ // commit end. The final object is never written or deleted here, so a
failed or timed-out upload cannot destroy a
+ // sidecar left by a previous attempt.
+ final URI destUri;
+ try {
+ destUri = new URI(StringUtil.join(File.separator, _segmentStoreUriStr,
segmentName.getTableName(),
+ SegmentCompletionUtils.generateTmpSegmentFileName(
+ segmentName.getSegmentName() +
Constants.METADATA_TAR_GZ_FILE_EXT)));
+ } catch (URISyntaxException e) {
+ LOGGER.warn("Invalid segment store uri {} for metadata tar of segment
{}", _segmentStoreUriStr, segmentName, e);
+ _serverMetrics.addMeteredTableValue(rawTableName,
ServerMeter.METADATA_TAR_UPLOAD_FAILURE, 1);
+ return null;
+ }
+ Callable<URI> uploadTask = () -> {
+ long startTime = System.currentTimeMillis();
+ try {
+ PinotFS pinotFS = PinotFSFactory.create(new
URI(_segmentStoreUriStr).getScheme());
+ pinotFS.copyFromLocalFile(metadataTarFile, destUri);
+ return destUri;
+ } catch (Exception e) {
+ LOGGER.warn("Failed copy metadata tar file {} to segment store {}:
{}", metadataTarFile.getName(), destUri, e);
+ } finally {
+ long duration = System.currentTimeMillis() - startTime;
+ _serverMetrics.addTimedTableValue(rawTableName,
ServerTimer.METADATA_TAR_UPLOAD_TIME_MS, duration,
+ TimeUnit.MILLISECONDS);
+ }
+ return null;
+ };
+ Future<URI> future = _executorService.submit(uploadTask);
+ try {
+ URI metadataTarLocation = future.get(timeoutInMillis,
TimeUnit.MILLISECONDS);
+ if (metadataTarLocation != null) {
+ LOGGER.info("Successfully uploaded metadata tar of segment {} to {}.",
segmentName, metadataTarLocation);
+ _serverMetrics.addMeteredTableValue(rawTableName,
ServerMeter.METADATA_TAR_UPLOAD_SUCCESS, 1);
+ return metadataTarLocation;
+ }
+ } catch (InterruptedException e) {
+ LOGGER.info("Interrupted while waiting for metadata tar upload of {} to
{}.", segmentName, _segmentStoreUriStr);
+ future.cancel(true);
+ Thread.currentThread().interrupt();
+ } catch (TimeoutException e) {
+ future.cancel(true);
+ _serverMetrics.addMeteredTableValue(rawTableName,
ServerMeter.METADATA_TAR_UPLOAD_TIMEOUT, 1);
+ LOGGER.warn("Timed out waiting to upload metadata tar of segment: {} for
table: {}",
+ segmentName.getSegmentName(), rawTableName);
+ } catch (Exception e) {
+ LOGGER.warn("Failed to upload metadata tar {} of segment {} for table
{}", metadataTarFile.getAbsolutePath(),
+ segmentName, rawTableName, e);
+ }
+ // Fire and forget: the store may be slow or hung (that is why we got
here) and the commit must not wait for it.
+ // A copy that outlives the timeout can still leave a tmp object; the
controller's tmp file cleanup removes it.
+ _executorService.submit(() -> deleteTmpQuietly(destUri));
Review Comment:
Minor: this fire-and-forget delete can run before the still-in-flight copy
finishes, so it may miss a tmp object that materializes slightly later. The
comment already notes the controller's tmp cleanup is the backstop, so this is
fine — just flagging that the delete here is best-effort, not a guarantee.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java:
##########
@@ -622,6 +628,31 @@ private boolean isPeerUrl(String segmentLocation) {
CommonConstants.Segment.PEER_SEGMENT_DOWNLOAD_SCHEME.length());
}
+ /// Moves the metadata sidecar the server uploaded under a tmp name to
`<segment>.metadata.tar.gz`, like
+ /// [#moveSegmentFile] does for the segment file. Best effort: a failure is
logged and counted, and never fails the
+ /// commit.
+ private void moveMetadataTarFile(String realtimeTableName, String
rawTableName, String segmentName,
+ @Nullable String metadataTarLocation, PinotFS pinotFS) {
+ if (metadataTarLocation == null) {
+ return;
+ }
+ URI uriToMoveTo = createSegmentPath(rawTableName, segmentName +
METADATA_TAR_GZ_FILE_EXT);
Review Comment:
Note: the final sidecar name here is `encode(segmentName +
".metadata.tar.gz")`, while the deletion path in `SegmentDeletionManager`
locates it via `generateSegmentMetadataURI(..., segmentId)`, which uses the raw
(unencoded) name. For LLC segment names these are identical (nothing to
encode), so it works, but the two sides disagree on encoding — a segment name
containing an encodable character would make deletion miss the sidecar.
Harmless today; worth keeping consistent if that ever changes.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
