shuturmurgh commented on code in PR #20111:
URL: https://github.com/apache/pinot/pull/20111#discussion_r4432647741
##########
pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManagerStatelessTest.java:
##########
@@ -1509,6 +1514,83 @@ public void testSegmentReplacementRegular()
assertNull(segmentLineage);
}
+ /// Run A completes its replacement after run B's `startReplaceSegments`
passed the 'segmentsFrom' existence check
+ /// but before B takes the lineage lock. B must be rejected instead of
adding a second 'COMPLETED' entry for the same
+ /// 'segmentsFrom', even with 'forceCleanup = true'.
+ @Test
+ public void
testStartReplaceSegmentsForceCleanupRejectsConcurrentlyCompletedEntry()
+ throws Exception {
+ addDummySchema(RAW_TABLE_NAME);
+ TableConfig tableConfig =
+ new
TableConfigBuilder(TableType.OFFLINE).setTableName(RAW_TABLE_NAME).setBrokerTenant(BROKER_TENANT_NAME)
+ .setServerTenant(SERVER_TENANT_NAME).build();
+ waitForEVToDisappear(tableConfig.getTableName());
+ _helixResourceManager.addTable(tableConfig);
+ for (int i = 0; i < 3; i++) {
+ _helixResourceManager.addNewSegment(OFFLINE_TABLE_NAME,
+ SegmentMetadataMockUtils.mockSegmentMetadata(OFFLINE_TABLE_NAME, "s"
+ i),
+ getDownloadURL(_controllerDataDir, RAW_TABLE_NAME, "s" + i));
+ }
+
+ // Run A starts replacing (s0, s1) -> (a0) and uploads its data, but has
not ended yet
+ List<String> segmentsFrom = Arrays.asList("s0", "s1");
+ String lineageEntryIdA =
+ _helixResourceManager.startReplaceSegments(OFFLINE_TABLE_NAME,
segmentsFrom, List.of("a0"), true, null);
+ _helixResourceManager.addNewSegment(OFFLINE_TABLE_NAME,
+ SegmentMetadataMockUtils.mockSegmentMetadata(OFFLINE_TABLE_NAME, "a0"),
+ getDownloadURL(_controllerDataDir, RAW_TABLE_NAME, "a0"));
+
+ ExecutorService executor = Executors.newSingleThreadExecutor();
+ try {
+ Future<String> startB;
+ // Hold the lineage lock so run B passes the existence check (A is still
'IN_PROGRESS') and then blocks on the
+ // lock. 'synchronized' is reentrant, so this thread can still end run A
while holding it.
+ synchronized
(_helixResourceManager.getLineageUpdaterLock(OFFLINE_TABLE_NAME)) {
+ AtomicReference<Thread> threadB = new AtomicReference<>();
+ startB = executor.submit(() -> {
+ threadB.set(Thread.currentThread());
+ return
_helixResourceManager.startReplaceSegments(OFFLINE_TABLE_NAME, segmentsFrom,
List.of("b0"), true,
+ null);
+ });
+ TestUtils.waitForCondition(
+ aVoid -> threadB.get() != null && threadB.get().getState() ==
Thread.State.BLOCKED, 10_000L,
+ "Run B did not reach the lineage lock");
+ _helixResourceManager.endReplaceSegments(OFFLINE_TABLE_NAME,
lineageEntryIdA, null);
+ }
+
+ Throwable failure = null;
+ try {
+ startB.get(30, TimeUnit.SECONDS);
+ } catch (ExecutionException e) {
+ failure = e.getCause();
+ }
+ assertNotNull(failure, "Run B should be rejected because (s0, s1) were
already replaced by run A");
+ while (failure.getCause() != null) {
+ failure = failure.getCause();
+ }
+ assertTrue(failure instanceof IllegalStateException);
+ assertTrue(failure.getMessage().contains("replacing a replaced
segment"), failure.getMessage());
+ } finally {
+ executor.shutdownNow();
+ }
+
+ // Only A's entry exists, and nothing from run B is left behind
+ SegmentLineage segmentLineage =
SegmentLineageAccessHelper.getSegmentLineage(_propertyStore,
OFFLINE_TABLE_NAME);
+ assertEquals(segmentLineage.getLineageEntryIds(), Set.of(lineageEntryIdA));
+ assertEquals(segmentLineage.getLineageEntry(lineageEntryIdA).getState(),
LineageEntryState.COMPLETED);
+
+ // A partially overlapping retry is rejected too, while a disjoint one is
still accepted
+ assertThrows(RuntimeException.class,
Review Comment:
Agreed, s1 is already filtered out as replaced, so the check before the lock
rejects it. Updated in e5f97ae160: the case now unwraps to the root
`IllegalStateException` and asserts `"does not exist"`. The comment also says
which check fires. The new in-lock check is covered only by the interleaving
case above it.
--
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]
