tarun11Mavani commented on code in PR #19776:
URL: https://github.com/apache/pinot/pull/19776#discussion_r4385284307
##########
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:
Agreed, it's best effort; the controller's tmp cleanup is the guarantee.
--
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]
