xiangfu0 commented on code in PR #20111:
URL: https://github.com/apache/pinot/pull/20111#discussion_r4432583968
##########
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:
Nit (possible flake): `Thread.State.BLOCKED` doesn't prove B is blocked on
*this* lock. A short stall on another monitor, such as a ZK packet or class
loading, before the existence check would let A finish first. B would then fail
with "does not exist" and the message assertion would fail. To make it
deterministic, compare
`ManagementFactory.getThreadMXBean().getThreadInfo(id).getLockInfo().getIdentityHashCode()`
with `System.identityHashCode(lock)`.
##########
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:
Nit: this case is rejected by the existing pre-lock existence check, because
`s1` is already filtered out as replaced. It never reaches the new in-lock
branch. Consider asserting the "does not exist" message so the test documents
which guard fires.
--
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]
