Jackie-Jiang commented on code in PR #19400:
URL: https://github.com/apache/pinot/pull/19400#discussion_r4186744598
##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/PinotFSSegmentUploader.java:
##########
@@ -69,47 +84,207 @@ public URI uploadSegment(File segmentFile, LLCSegmentName
segmentName, int timeo
return null;
}
final String rawTableName =
TableNameBuilder.extractRawTableName(segmentName.getTableName());
- Callable<URI> uploadTask = () -> {
- URI destUri = new URI(StringUtil.join(File.separator,
_segmentStoreUriStr, segmentName.getTableName(),
-
SegmentCompletionUtils.generateTmpSegmentFileName(segmentName.getSegmentName())));
- long startTime = System.currentTimeMillis();
- try {
- PinotFS pinotFS = PinotFSFactory.create(new
URI(_segmentStoreUriStr).getScheme());
- // Check and delete any existing segment file.
- if (pinotFS.exists(destUri)) {
- pinotFS.delete(destUri, true);
- }
- pinotFS.copyFromLocalFile(segmentFile, destUri);
- return destUri;
- } catch (Exception e) {
- LOGGER.warn("Failed copy segment tar file {} to segment store {}: {}",
segmentFile.getName(), destUri, e);
- } finally {
- long duration = System.currentTimeMillis() - startTime;
- _serverMetrics.addTimedTableValue(rawTableName,
ServerTimer.SEGMENT_UPLOAD_TIME_MS, duration,
- TimeUnit.MILLISECONDS);
- }
- return null;
- };
+ AtomicReference<AttemptState> state = new
AtomicReference<>(AttemptState.IN_FLIGHT);
+ AtomicReference<URI> destUriRef = new AtomicReference<>();
Review Comment:
Could we generate the unique destination URI before submitting the upload
task and capture it as an immutable per-attempt value? That would remove
`AtomicReference<URI>`, the special read-after-publication handling, and the
hooks/tests needed only for a delayed URI assignment. We would still need the
atomic attempt state to arbitrate publication versus abandonment. Please
preserve the existing failure handling for URI construction.
##########
pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManagerTest.java:
##########
@@ -1250,12 +1251,15 @@ public void
testSegmentAlreadyThereAndExtraneousFilesDeleted()
String otherSegmentName = new LLCSegmentName(RAW_TABLE_NAME, 1, 0,
CURRENT_TIME_MS).getSegmentName();
String segmentFileName =
SegmentCompletionUtils.generateTmpSegmentFileName(segmentName);
String extraSegmentFileName =
SegmentCompletionUtils.generateTmpSegmentFileName(segmentName);
+ String extraInstanceIdFileName = segmentName + ".tmp.Server_host-a_8098";
Review Comment:
This instance-ID temp-file fixture appears to be left over from the earlier
naming approach. The current change retains UUID-per-attempt names, and this
test exercises the existing commit-time prefix cleanup. Could we remove this
fixture and its associated write/assertion to keep the tests focused on the
final change?
##########
pinot-core/src/test/java/org/apache/pinot/core/data/manager/realtime/PinotFSSegmentUploaderTest.java:
##########
@@ -52,37 +63,274 @@ public void setUp()
Map<String, Object> properties = new HashMap<>();
properties.put("class.hdfs",
"org.apache.pinot.core.data.manager.realtime.PinotFSSegmentUploaderTest$AlwaysSucceedPinotFS");
- properties.put("class.timeout",
-
"org.apache.pinot.core.data.manager.realtime.PinotFSSegmentUploaderTest$AlwaysTimeoutPinotFS");
- properties.put("class.existing",
-
"org.apache.pinot.core.data.manager.realtime.PinotFSSegmentUploaderTest$AlwaysExistPinotFS");
+ properties.put("class.record",
+
"org.apache.pinot.core.data.manager.realtime.PinotFSSegmentUploaderTest$RecordingPinotFS");
PinotFSFactory.init(new PinotConfiguration(properties));
_file = FileUtils.getFile(FileUtils.getTempDirectory(),
UUID.randomUUID().toString());
_file.deleteOnExit();
_llcSegmentName = new LLCSegmentName("test_REALTIME", 1, 0,
System.currentTimeMillis());
}
+ @BeforeMethod
+ public void resetRecordedUploads() {
+ RecordingPinotFS.reset();
+ PinotFSSegmentUploader._testHooks = null;
+ Mockito.clearInvocations(_serverMetrics);
+ }
+
@Test
public void testSuccessfulUpload() {
- SegmentUploader segmentUploader = new
PinotFSSegmentUploader("hdfs://root", TIMEOUT_IN_MS, _serverMetrics);
- URI segmentURI = segmentUploader.uploadSegment(_file, _llcSegmentName);
- Assert.assertTrue(segmentURI.toString().startsWith(StringUtil
- .join(File.separator, "hdfs://root", _llcSegmentName.getTableName(),
_llcSegmentName.getSegmentName())));
+ SegmentUploader segmentUploader = new
PinotFSSegmentUploader("record://root", TIMEOUT_IN_MS, _serverMetrics);
+ URI first = segmentUploader.uploadSegment(_file, _llcSegmentName);
+ URI second = segmentUploader.uploadSegment(_file, _llcSegmentName);
+ Assert.assertNotNull(first);
Review Comment:
Please use static imports for the TestNG `Assert` methods throughout this
test class, e.g. `assertNotNull(first)` rather than
`Assert.assertNotNull(first)`.
##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/PinotFSSegmentUploader.java:
##########
@@ -69,47 +84,207 @@ public URI uploadSegment(File segmentFile, LLCSegmentName
segmentName, int timeo
return null;
}
final String rawTableName =
TableNameBuilder.extractRawTableName(segmentName.getTableName());
- Callable<URI> uploadTask = () -> {
- URI destUri = new URI(StringUtil.join(File.separator,
_segmentStoreUriStr, segmentName.getTableName(),
-
SegmentCompletionUtils.generateTmpSegmentFileName(segmentName.getSegmentName())));
- long startTime = System.currentTimeMillis();
- try {
- PinotFS pinotFS = PinotFSFactory.create(new
URI(_segmentStoreUriStr).getScheme());
- // Check and delete any existing segment file.
- if (pinotFS.exists(destUri)) {
- pinotFS.delete(destUri, true);
- }
- pinotFS.copyFromLocalFile(segmentFile, destUri);
- return destUri;
- } catch (Exception e) {
- LOGGER.warn("Failed copy segment tar file {} to segment store {}: {}",
segmentFile.getName(), destUri, e);
- } finally {
- long duration = System.currentTimeMillis() - startTime;
- _serverMetrics.addTimedTableValue(rawTableName,
ServerTimer.SEGMENT_UPLOAD_TIME_MS, duration,
- TimeUnit.MILLISECONDS);
- }
- return null;
- };
+ AtomicReference<AttemptState> state = new
AtomicReference<>(AttemptState.IN_FLIGHT);
+ AtomicReference<URI> destUriRef = new AtomicReference<>();
+ Callable<URI> uploadTask = () -> copySegment(segmentFile, segmentName,
rawTableName, state, destUriRef);
Future<URI> future = _executorService.submit(uploadTask);
try {
URI segmentLocation = future.get(timeoutInMillis, TimeUnit.MILLISECONDS);
- LOGGER.info("Successfully upload segment {} to {}.", segmentName,
segmentLocation);
- _serverMetrics.addMeteredTableValue(rawTableName,
- segmentLocation == null ? ServerMeter.SEGMENT_UPLOAD_FAILURE :
ServerMeter.SEGMENT_UPLOAD_SUCCESS, 1);
- return segmentLocation;
+ if (segmentLocation != null) {
+ LOGGER.info("Successfully upload segment {} to {}.", segmentName,
segmentLocation);
+ _serverMetrics.addMeteredTableValue(rawTableName,
ServerMeter.SEGMENT_UPLOAD_SUCCESS, 1);
+ return segmentLocation;
+ }
} catch (InterruptedException e) {
LOGGER.info("Interrupted while waiting for segment upload of {} to {}.",
segmentName, _segmentStoreUriStr);
+ // future.get already cleared the interrupt flag. Leave it clear until a
published object is deleted.
+ // S3 and HDFS abort a delete when the calling thread is interrupted,
and the worker will not delete a URI
+ // this thread already published. Restore the flag after that delete.
+ if (!state.compareAndSet(AttemptState.IN_FLIGHT, AttemptState.ABANDONED)
+ && state.get() == AttemptState.PUBLISHED) {
+ deleteBestEffort(destUriRef.get());
+ }
Thread.currentThread().interrupt();
} catch (TimeoutException e) {
- // Emit a separate metric for timeout since this is relatively more
common than other errors.
+ URI publishedUri = resolveTimedOutAttempt(state, destUriRef);
+ if (publishedUri != null) {
+ LOGGER.info("Successfully upload segment {} to {}.", segmentName,
publishedUri);
+ _serverMetrics.addMeteredTableValue(rawTableName,
ServerMeter.SEGMENT_UPLOAD_SUCCESS, 1);
+ return publishedUri;
+ }
_serverMetrics.addMeteredTableValue(rawTableName,
ServerMeter.SEGMENT_UPLOAD_TIMEOUT, 1);
LOGGER.warn("Timed out waiting to upload segment: {} for table: {}",
segmentName.getSegmentName(), rawTableName);
} catch (Exception e) {
LOGGER.warn("Failed to upload file {} of segment {} for table {}",
- segmentFile.getAbsolutePath(), segmentName, rawTableName, e);
+ segmentFile.getAbsolutePath(), segmentName, rawTableName, e);
}
_serverMetrics.addMeteredTableValue(rawTableName,
ServerMeter.SEGMENT_UPLOAD_FAILURE, 1);
+ return null;
+ }
+ /// Caller side of a timed-out `future.get`. Returns the published URI when
the copy won the race with the deadline.
+ /// The URI is read after that CAS loses. The worker stores it before its
own publish CAS, so a timeout that fired
+ /// while the ref was still empty still observes the URI when the worker has
published. Do not pass a URI captured
+ /// before the CAS.
+ /// The caller must not delete. A returned URI is handed to the controller.
An abandoned attempt is deleted by the
+ /// worker after the copy finishes, including a late success. An early
delete can race a copy that recreates the
+ /// object. Does not cancel the future. An interrupt mid-upload can leave
multipart parts this code cannot abort.
+ static URI resolveTimedOutAttempt(AtomicReference<AttemptState> state,
AtomicReference<URI> destUriRef) {
+ TestHooks hooks = _testHooks;
+ if (hooks != null && hooks._beforeTimeoutCas != null) {
+ hooks._beforeTimeoutCas.run();
+ }
+ if (state.compareAndSet(AttemptState.IN_FLIGHT, AttemptState.ABANDONED)) {
+ return null;
+ }
+ if (hooks != null && hooks._afterTimeoutCasLost != null) {
+ hooks._afterTimeoutCasLost.run();
+ }
+ if (state.get() == AttemptState.PUBLISHED) {
+ return destUriRef.get();
+ }
return null;
}
+
+ private URI copySegment(File segmentFile, LLCSegmentName segmentName, String
rawTableName,
+ AtomicReference<AttemptState> state, AtomicReference<URI> destUriRef) {
+ URI destUri;
+ try {
+ destUri = new URI(StringUtil.join(File.separator, _segmentStoreUriStr,
segmentName.getTableName(),
+
SegmentCompletionUtils.generateTmpSegmentFileName(segmentName.getSegmentName())));
+ } catch (Exception e) {
+ state.compareAndSet(AttemptState.IN_FLIGHT, AttemptState.FAILED);
+ LOGGER.warn("Failed copy segment tar file {} to segment store {}",
segmentFile.getName(), _segmentStoreUriStr,
+ e);
+ return null;
+ }
+ // Minted inside the task. future.get throws the URI away on timeout, so
the caller cannot rebuild it.
+ // The store happens before the publish CAS. A timeout that loses that CAS
must read the ref afterwards.
+ awaitBeforeDestUriStoreForTest();
+ destUriRef.set(destUri);
+ long startTime = System.currentTimeMillis();
+ try {
+ PinotFS pinotFS = openSegmentStore(segmentFile, destUri, state);
+ if (pinotFS == null) {
+ return null;
+ }
+ try {
+ pinotFS.copyFromLocalFile(segmentFile, destUri);
+ } catch (Exception e) {
+ // The SDK may have stored the object and then thrown. Delete that
object, and do not let the delete
+ // replace the copy failure.
+ deleteBestEffort(pinotFS, destUri);
+ state.compareAndSet(AttemptState.IN_FLIGHT, AttemptState.FAILED);
+ LOGGER.warn("Failed copy segment tar file {} to segment store {}",
segmentFile.getName(), destUri, e);
+ return null;
+ }
+ if (state.compareAndSet(AttemptState.IN_FLIGHT, AttemptState.PUBLISHED))
{
+ pauseAfterPublishForTest();
+ return destUri;
+ }
+ // The caller already abandoned this attempt and will not return the URI.
+ deleteBestEffort(pinotFS, destUri);
+ return null;
+ } finally {
+ long duration = System.currentTimeMillis() - startTime;
+ _serverMetrics.addTimedTableValue(rawTableName,
ServerTimer.SEGMENT_UPLOAD_TIME_MS, duration,
+ TimeUnit.MILLISECONDS);
+ }
+ }
+
+ /// Returns null when the filesystem cannot be opened. Nothing was stored,
so there is nothing to delete.
+ private PinotFS openSegmentStore(File segmentFile, URI destUri,
AtomicReference<AttemptState> state) {
+ try {
+ return PinotFSFactory.create(new URI(_segmentStoreUriStr).getScheme());
+ } catch (Exception e) {
+ state.compareAndSet(AttemptState.IN_FLIGHT, AttemptState.FAILED);
+ LOGGER.warn("Failed copy segment tar file {} to segment store {}",
segmentFile.getName(), destUri, e);
+ return null;
+ }
+ }
+
+ /// Blocks before `destUriRef` is stored so a test can time out while the
ref is still empty.
+ private static void awaitBeforeDestUriStoreForTest() {
Review Comment:
Could we reduce the production test scaffolding while preserving the
timeout/interrupt race coverage? The global `TestHooks`, multiple callbacks,
and latch-handling methods add substantial code to this uploader. Precomputing
the URI would eliminate the delayed-URI hooks; for the remaining handoff races,
consider a small per-attempt helper or a controllable executor so the tests can
exercise the transitions with less global mutable state.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java:
##########
@@ -2428,21 +2429,31 @@ public int deleteTmpSegments(String realtimeTableName,
List<SegmentZKMetadata> s
throws IOException {
Preconditions.checkState(!_isStopping, "Segment manager is stopping");
- // NOTE: Do not delete the file if it is used as download URL. This could
happen when user uses temporary file to
- // backfill segment.
- Set<String> downloadUrls =
Sets.newHashSetWithExpectedSize(segmentsZKMetadata.size());
- for (SegmentZKMetadata segmentZKMetadata : segmentsZKMetadata) {
- if (segmentZKMetadata.getStatus() == Status.DONE) {
- downloadUrls.add(segmentZKMetadata.getDownloadUrl());
- }
- }
-
String rawTableName =
TableNameBuilder.extractRawTableName(realtimeTableName);
URI tableDirURI = URIUtils.getUri(_controllerConf.getDataDir(),
rawTableName);
PinotFS pinotFS = PinotFSFactory.create(tableDirURI.getScheme());
- int numDeletedTmpSegments = 0;
+ List<String> tmpPaths = new ArrayList<>();
for (String filePath : pinotFS.listFiles(tableDirURI, false)) {
- if (isTmpAndCanDelete(filePath, downloadUrls, pinotFS)) {
+ if (SegmentCompletionUtils.isTmpFile(filePath)) {
+ tmpPaths.add(filePath);
+ }
+ }
+ if (tmpPaths.isEmpty()) {
+ return 0;
+ }
+
+ // A completed segment can legally be named like a temp upload. A
file-scheme data dir stores an HTTP download
+ // URL, so the permanent path has to be kept as well as that URL.
IN_PROGRESS and COMMITTING are not permanent
+ // files. Offline metadata is loaded only for names whose permanent path
is one of the listed temp files.
+ // getTableConfig on the helix resource manager returns null when the
offline table is missing.
+ // PinotLLCRealtimeSegmentManager.getTableConfig throws in that case, so
this method does not call it.
+ Set<String> protectedPaths = new HashSet<>();
+ collectCompletedSegmentPaths(rawTableName, segmentsZKMetadata,
protectedPaths);
+ protectCompletedOfflineSegments(rawTableName, tmpPaths, protectedPaths);
Review Comment:
Could we move the periodic-cleaner expansion and its tests into a separate
PR? Restoring the existing UUID-only matcher already removes the broad-match
regression from the earlier version. Protecting permanent segment names that
themselves end in `.tmp.<uuid>` is useful hardening for a pre-existing edge
case, but the offline metadata lookup and path normalization are independent of
the failed/abandoned upload cleanup in this PR.
##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/PinotFSSegmentUploader.java:
##########
@@ -38,18 +40,31 @@
import org.slf4j.LoggerFactory;
-/// A segment uploader which does segment upload to a segment store (with
store root dir configured as
-/// \_segmentStoreUriStr) using PinotFS within a configurable timeout period.
The final segment location would be in the
-/// URI \_segmentStoreUriStr/\_tableNameWithType/segmentName+random_uuid if
successful.
+/// Uploads a segment to the configured segment store with PinotFS, bounded by
a timeout.
+/// Each attempt uses `{segment}.tmp.{uuid}`. The object is deleted only when
that attempt's URI is not returned.
+/// A timed-out copy that is still in flight is abandoned. The worker deletes
it after the copy finishes, including a
+/// late success. A copy that published before the timeout is returned and
kept.
public class PinotFSSegmentUploader implements SegmentUploader {
private static final Logger LOGGER =
LoggerFactory.getLogger(PinotFSSegmentUploader.class);
public static final int DEFAULT_SEGMENT_UPLOAD_TIMEOUT_MILLIS = 10 * 1000;
+ /// Latches and callbacks used by tests to force the timeout and interrupt
races. Production leaves this null.
+ static volatile TestHooks _testHooks;
Review Comment:
If `_testHooks` is retained, please annotate it with `@VisibleForTesting`
and move it below the final fields. Does it need to be `volatile`? The current
tests install fully initialized hooks before `ExecutorService.submit` /
`Thread.start`, which already provide the happens-before relationship for
initial publication. `volatile` is relevant if resets or replacements can
overlap worker reads, as can happen during test cleanup; it can be removed if
the lifecycle guarantees that all workers finish before the hooks are changed.
Capturing the hooks per attempt would also avoid workers observing another
test's hooks.
--
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]
