shuturmurgh commented on code in PR #20111:
URL: https://github.com/apache/pinot/pull/20111#discussion_r4432645735
##########
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,
Review Comment:
Good catch, fixed in e5f97ae160. The wait now gets B's `ThreadInfo` from
`ManagementFactory.getThreadMXBean()`. It checks that
`getLockInfo().getIdentityHashCode()` matches
`System.identityHashCode(lineageLock)`, not just that B is BLOCKED. A stall on
another monitor before the existence check no longer satisfies the wait, so A
can't finish first and turn this into a "does not exist" failure.
--
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]
