From 3f1df0291bb497a558264b54d581c17e835ac6c8 Mon Sep 17 00:00:00 2001 From: Yang Guo Date: Mon, 27 Jul 2026 16:22:04 +0800 Subject: [PATCH 1/2] feat: committed isr boundary --- .../CoordinatorEventProcessor.java | 20 +- .../CoordinatorEventProcessorTest.java | 322 +++++++++++++++--- 2 files changed, 289 insertions(+), 53 deletions(-) diff --git a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessor.java b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessor.java index c5b90f5904..7159939433 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessor.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessor.java @@ -1925,7 +1925,8 @@ private List tryProcessAdjustIsr( // TODO verify leader epoch. List result = new ArrayList<>(); - Map newLeaderAndIsrList = new HashMap<>(); + Map proposedChanges = new HashMap<>(); + Map committedChanges = new HashMap<>(); for (Map.Entry entry : leaderAndIsrList.entrySet()) { TableBucket tableBucket = entry.getKey(); LeaderAndIsr tryAdjustLeaderAndIsr = entry.getValue(); @@ -1959,19 +1960,20 @@ private List tryProcessAdjustIsr( tryAdjustLeaderAndIsr.standbyReplicas(), coordinatorContext.getCoordinatorEpoch(), currentLeaderAndIsr.bucketEpoch() + 1); - newLeaderAndIsrList.put(tableBucket, newLeaderAndIsr); + proposedChanges.put(tableBucket, newLeaderAndIsr); } try { zooKeeperClient.batchUpdateLeaderAndIsr( - newLeaderAndIsrList, coordinatorContext.getCoordinatorZkVersion()); - newLeaderAndIsrList.forEach( + proposedChanges, coordinatorContext.getCoordinatorZkVersion()); + committedChanges.putAll(proposedChanges); + committedChanges.forEach( (tableBucket, newLeaderAndIsr) -> result.add(new AdjustIsrResultForBucket(tableBucket, newLeaderAndIsr))); } catch (Exception batchException) { LOG.error("Error when batch update leader and isr. Try one by one.", batchException); - for (Map.Entry entry : newLeaderAndIsrList.entrySet()) { + for (Map.Entry entry : proposedChanges.entrySet()) { TableBucket tableBucket = entry.getKey(); LeaderAndIsr newLeaderAndIsr = entry.getValue(); try { @@ -1979,21 +1981,21 @@ private List tryProcessAdjustIsr( tableBucket, newLeaderAndIsr, coordinatorContext.getCoordinatorZkVersion()); + committedChanges.put(tableBucket, newLeaderAndIsr); + result.add(new AdjustIsrResultForBucket(tableBucket, newLeaderAndIsr)); } catch (Exception e) { LOG.error("Error when register leader and isr.", e); result.add( new AdjustIsrResultForBucket(tableBucket, ApiError.fromThrowable(e))); } - // Successful return. - result.add(new AdjustIsrResultForBucket(tableBucket, newLeaderAndIsr)); } } // update coordinator leader and isr cache. - newLeaderAndIsrList.forEach(coordinatorContext::putBucketLeaderAndIsr); + committedChanges.forEach(coordinatorContext::putBucketLeaderAndIsr); // First, try to judge whether the bucket is in rebalance task when isr change. - newLeaderAndIsrList.keySet().forEach(this::tryToCompleteRebalanceTaskOnLeaderAndIsrChange); + committedChanges.keySet().forEach(this::tryToCompleteRebalanceTaskOnLeaderAndIsrChange); // TODO update metadata for all alive tablet servers. diff --git a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessorTest.java b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessorTest.java index 5a5a68b3c8..d4251fa081 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessorTest.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessorTest.java @@ -74,6 +74,7 @@ import org.apache.fluss.server.metadata.TableMetadata; import org.apache.fluss.server.metrics.group.TestingMetricGroups; import org.apache.fluss.server.tablet.TestTabletServerGateway; +import org.apache.fluss.server.zk.CuratorFrameworkWithUnhandledErrorListener; import org.apache.fluss.server.zk.NOPErrorHandler; import org.apache.fluss.server.zk.ZkEpoch; import org.apache.fluss.server.zk.ZooKeeperClient; @@ -87,6 +88,7 @@ import org.apache.fluss.server.zk.data.ZkData; import org.apache.fluss.server.zk.data.ZkData.PartitionIdsZNode; import org.apache.fluss.server.zk.data.ZkData.TableIdsZNode; +import org.apache.fluss.shaded.curator5.org.apache.curator.retry.ExponentialBackoffRetry; import org.apache.fluss.testutils.common.AllCallbackWrapper; import org.apache.fluss.testutils.common.ManuallyTriggeredScheduledExecutorService; import org.apache.fluss.types.DataTypes; @@ -109,6 +111,7 @@ import java.util.Arrays; import java.util.Collections; import java.util.HashMap; +import java.util.HashSet; import java.util.List; import java.util.Map; import java.util.Optional; @@ -135,6 +138,9 @@ import static org.apache.fluss.server.utils.ServerRpcMessageUtils.getAdjustIsrResponseData; import static org.apache.fluss.server.utils.ServerRpcMessageUtils.getUpdateMetadataRequestData; import static org.apache.fluss.server.utils.TableAssignmentUtils.generateAssignment; +import static org.apache.fluss.server.zk.ZooKeeperUtils.startZookeeperClient; +import static org.apache.fluss.shaded.curator5.org.apache.curator.framework.CuratorFrameworkFactory.Builder; +import static org.apache.fluss.shaded.curator5.org.apache.curator.framework.CuratorFrameworkFactory.builder; import static org.apache.fluss.testutils.common.CommonTestUtils.retry; import static org.apache.fluss.testutils.common.CommonTestUtils.waitValue; import static org.assertj.core.api.Assertions.assertThat; @@ -177,6 +183,7 @@ class CoordinatorEventProcessorTest { private KvSnapshotLeaseManager kvSnapshotLeaseManager; private Scheduler scheduler; private String remoteDataDir; + private TestingAdjustIsrZooKeeperClient testingAdjustIsrZooKeeperClient; @BeforeAll static void baseBeforeAll() throws Exception { @@ -253,6 +260,9 @@ void afterEach() throws Exception { if (scheduler != null) { scheduler.shutdown(); } + if (testingAdjustIsrZooKeeperClient != null) { + testingAdjustIsrZooKeeperClient.close(); + } metadataManager.dropDatabase(defaultDatabase, false, true); // clear the assignment info for all tables; ZOO_KEEPER_EXTENSION_WRAPPER.getCustomExtension().cleanupPath(TableIdsZNode.path()); @@ -1222,55 +1232,112 @@ void testNotifyOffsetsWithShrinkISR(@TempDir Path tempDir) throws Exception { @Test void testProcessAdjustIsr() throws Exception { - // make sure all request to gateway should be successful - initCoordinatorChannel(); - // create a table, - TablePath t1 = TablePath.of(defaultDatabase, "create_process_adjust_isr"); - int nBuckets = 3; - int replicationFactor = 3; - TableAssignment tableAssignment = - generateAssignment( - nBuckets, - replicationFactor, - new TabletServerInfo[] { - new TabletServerInfo(0, "rack0"), - new TabletServerInfo(1, "rack1"), - new TabletServerInfo(2, "rack2") - }); - long t1Id = - metadataManager.createTable(t1, remoteDataDir, TEST_TABLE, tableAssignment, false); - verifyTableCreated(t1Id, tableAssignment, nBuckets, replicationFactor); - - // get the origin bucket leaderAndIsr Map bucketLeaderAndIsrMap = - new HashMap<>( - waitValue( - () -> fromCtx((ctx) -> Optional.of(ctx.bucketLeaderAndIsr())), - Duration.ofMinutes(1), - "leader not elected")); + createAdjustIsrTestTable("process_adjust_isr"); - // verify AdjustIsrReceivedEvent - CompletableFuture response = new CompletableFuture<>(); - eventProcessor - .getCoordinatorEventManager() - .put(new AdjustIsrReceivedEvent(bucketLeaderAndIsrMap, response)); - - retryVerifyContext( - ctx -> - bucketLeaderAndIsrMap.forEach( - (tableBucket, leaderAndIsr) -> - assertThat(ctx.getBucketLeaderAndIsr(tableBucket)) - .contains( - leaderAndIsr.newLeaderAndIsr( - leaderAndIsr.isr())))); - - // verify the response - AdjustIsrResponse adjustIsrResponse = response.get(); + AdjustIsrResponse adjustIsrResponse = adjustIsr(bucketLeaderAndIsrMap); Map resultForBucketMap = getAdjustIsrResponseData(adjustIsrResponse); + + assertAdjustIsrResponseCount(adjustIsrResponse, bucketLeaderAndIsrMap.size()); assertThat(resultForBucketMap.keySet()) - .containsAnyElementsOf(bucketLeaderAndIsrMap.keySet()); + .containsExactlyInAnyOrderElementsOf(bucketLeaderAndIsrMap.keySet()); assertThat(resultForBucketMap.values()).allMatch(AdjustIsrResultForBucket::succeeded); + assertAdjustedIsrInstalled(bucketLeaderAndIsrMap, bucketLeaderAndIsrMap.keySet()); + } + + @Test + void testProcessAdjustIsrFallsBackToSuccessfulIndividualUpdates() throws Exception { + Map originalLeaderAndIsrs = + createAdjustIsrTestTable("adjust_isr_individual_success"); + useTestingAdjustIsrZooKeeperClient(true, Collections.emptySet()); + + AdjustIsrResponse response = adjustIsr(originalLeaderAndIsrs); + Map results = getAdjustIsrResponseData(response); + + assertAdjustIsrResponseCount(response, originalLeaderAndIsrs.size()); + assertThat(results.values()).allMatch(AdjustIsrResultForBucket::succeeded); + assertThat(testingAdjustIsrZooKeeperClient.getIndividualUpdates().keySet()) + .containsExactlyInAnyOrderElementsOf(originalLeaderAndIsrs.keySet()); + assertAdjustedIsrInstalled(originalLeaderAndIsrs, originalLeaderAndIsrs.keySet()); + } + + @Test + void testProcessAdjustIsrInstallsOnlyCommittedIndividualUpdates() throws Exception { + Map originalLeaderAndIsrs = + createAdjustIsrTestTable("adjust_isr_mixed_individual_results"); + List buckets = + Arrays.asList(originalLeaderAndIsrs.keySet().toArray(new TableBucket[0])); + TableBucket failedBucket = buckets.get(1); + Set committedBuckets = new HashSet<>(originalLeaderAndIsrs.keySet()); + committedBuckets.remove(failedBucket); + useTestingAdjustIsrZooKeeperClient(true, Collections.singleton(failedBucket)); + + AdjustIsrResponse response = adjustIsr(originalLeaderAndIsrs); + Map results = getAdjustIsrResponseData(response); + + assertAdjustIsrResponseCount(response, originalLeaderAndIsrs.size()); + assertThat(results.get(failedBucket).failed()).isTrue(); + assertThat(results.entrySet()) + .filteredOn(entry -> !entry.getKey().equals(failedBucket)) + .allMatch(entry -> entry.getValue().succeeded()); + assertAdjustedIsrInstalled(originalLeaderAndIsrs, committedBuckets); + } + + @Test + void testProcessAdjustIsrInstallsNothingWhenAllIndividualUpdatesFail() throws Exception { + Map originalLeaderAndIsrs = + createAdjustIsrTestTable("adjust_isr_individual_failure"); + useTestingAdjustIsrZooKeeperClient(true, originalLeaderAndIsrs.keySet()); + + AdjustIsrResponse response = adjustIsr(originalLeaderAndIsrs); + Map results = getAdjustIsrResponseData(response); + + assertAdjustIsrResponseCount(response, originalLeaderAndIsrs.size()); + assertThat(results.values()).allMatch(AdjustIsrResultForBucket::failed); + assertAdjustedIsrInstalled(originalLeaderAndIsrs, Collections.emptySet()); + } + + @Test + void testProcessAdjustIsrKeepsValidationAndPersistenceResultsUnique() throws Exception { + Map originalLeaderAndIsrs = + createAdjustIsrTestTable("adjust_isr_validation_and_persistence"); + List buckets = + Arrays.asList(originalLeaderAndIsrs.keySet().toArray(new TableBucket[0])); + TableBucket invalidBucket = buckets.get(0); + TableBucket persistenceFailureBucket = buckets.get(1); + LeaderAndIsr original = originalLeaderAndIsrs.get(invalidBucket); + Map requestedLeaderAndIsrs = + new HashMap<>(originalLeaderAndIsrs); + requestedLeaderAndIsrs.put( + invalidBucket, + new LeaderAndIsr( + original.leader(), + original.leaderEpoch() - 1, + original.isr(), + original.standbyReplicas(), + original.coordinatorEpoch(), + original.bucketEpoch())); + useTestingAdjustIsrZooKeeperClient(true, Collections.singleton(persistenceFailureBucket)); + + AdjustIsrResponse response = adjustIsr(requestedLeaderAndIsrs); + Map results = getAdjustIsrResponseData(response); + + assertAdjustIsrResponseCount(response, requestedLeaderAndIsrs.size()); + assertThat(results.get(invalidBucket).failed()).isTrue(); + assertThat(results.get(persistenceFailureBucket).failed()).isTrue(); + assertThat(testingAdjustIsrZooKeeperClient.getBatchUpdates()) + .doesNotContainKey(invalidBucket); + assertThat(testingAdjustIsrZooKeeperClient.getIndividualUpdates()) + .doesNotContainKey(invalidBucket); + assertAdjustedIsrInstalled( + originalLeaderAndIsrs, + originalLeaderAndIsrs.keySet().stream() + .filter( + bucket -> + !bucket.equals(invalidBucket) + && !bucket.equals(persistenceFailureBucket)) + .collect(Collectors.toSet())); } @Test @@ -2085,12 +2152,106 @@ private void verifyIsr(TableBucket tb, int expectedLeader, List expecte .hasSameElementsAs(expectedIsr); } + private Map createAdjustIsrTestTable(String tableName) + throws Exception { + initCoordinatorChannel(); + TableAssignment tableAssignment = + generateAssignment( + N_BUCKETS, + REPLICATION_FACTOR, + new TabletServerInfo[] { + new TabletServerInfo(0, "rack0"), + new TabletServerInfo(1, "rack1"), + new TabletServerInfo(2, "rack2") + }); + long tableId = + metadataManager.createTable( + TablePath.of(defaultDatabase, tableName), + remoteDataDir, + TEST_TABLE, + tableAssignment, + false); + verifyTableCreated(tableId, tableAssignment, N_BUCKETS, REPLICATION_FACTOR); + + return waitValue( + () -> + fromCtx( + ctx -> { + Map leaderAndIsrs = new HashMap<>(); + ctx.bucketLeaderAndIsr() + .forEach( + (tableBucket, leaderAndIsr) -> { + if (tableBucket.getTableId() == tableId) { + leaderAndIsrs.put( + tableBucket, leaderAndIsr); + } + }); + return leaderAndIsrs.size() == N_BUCKETS + ? Optional.of(leaderAndIsrs) + : Optional.empty(); + }), + Duration.ofMinutes(1), + "leader not elected"); + } + + private void useTestingAdjustIsrZooKeeperClient( + boolean failBatchUpdate, Set failedIndividualUpdates) { + testingAdjustIsrZooKeeperClient = new TestingAdjustIsrZooKeeperClient(remoteDataDir); + CoordinatorEventProcessor previousEventProcessor = eventProcessor; + CoordinatorEventProcessor testingEventProcessor = + buildCoordinatorEventProcessor(testingAdjustIsrZooKeeperClient); + testingEventProcessor.startup(); + previousEventProcessor.shutdown(); + eventProcessor = testingEventProcessor; + testingAdjustIsrZooKeeperClient.enableFailureInjection( + failBatchUpdate, failedIndividualUpdates); + } + + private AdjustIsrResponse adjustIsr(Map leaderAndIsrs) + throws Exception { + CompletableFuture response = new CompletableFuture<>(); + eventProcessor + .getCoordinatorEventManager() + .put(new AdjustIsrReceivedEvent(leaderAndIsrs, response)); + return response.get(); + } + + private void assertAdjustIsrResponseCount(AdjustIsrResponse response, int expectedCount) { + assertThat( + response.getTablesRespsList().stream() + .mapToInt(tableResponse -> tableResponse.getBucketsRespsCount()) + .sum()) + .isEqualTo(expectedCount); + } + + private void assertAdjustedIsrInstalled( + Map originalLeaderAndIsrs, Set committedBuckets) + throws Exception { + for (Map.Entry entry : originalLeaderAndIsrs.entrySet()) { + TableBucket tableBucket = entry.getKey(); + LeaderAndIsr originalLeaderAndIsr = entry.getValue(); + Optional installedLeaderAndIsr = + fromCtx(ctx -> ctx.getBucketLeaderAndIsr(tableBucket)); + assertThat(installedLeaderAndIsr) + .contains( + committedBuckets.contains(tableBucket) + ? originalLeaderAndIsr.newLeaderAndIsr( + originalLeaderAndIsr.isr()) + : originalLeaderAndIsr); + } + } + private CoordinatorEventProcessor buildCoordinatorEventProcessor() { + return buildCoordinatorEventProcessor(zookeeperClient); + } + + private CoordinatorEventProcessor buildCoordinatorEventProcessor( + ZooKeeperClient coordinatorZooKeeperClient) { Configuration conf = new Configuration(); conf.set(ConfigOptions.REMOTE_DATA_DIR, remoteDataDir); conf.set(ConfigOptions.COORDINATOR_OFFLINE_LEADER_RETRY_DELAY, Duration.ofDays(1)); return new CoordinatorEventProcessor( - zookeeperClient, + coordinatorZooKeeperClient, serverMetadataCache, testCoordinatorChannelManager, new CoordinatorContext(zkEpoch), @@ -2106,6 +2267,79 @@ private CoordinatorEventProcessor buildCoordinatorEventProcessor() { SystemClock.getInstance()); } + private static final class TestingAdjustIsrZooKeeperClient extends ZooKeeperClient { + + private final Map batchUpdates = new HashMap<>(); + private final Map individualUpdates = new HashMap<>(); + + private boolean failureInjectionEnabled; + private boolean failBatchUpdate; + private Set failedIndividualUpdates = Collections.emptySet(); + + private TestingAdjustIsrZooKeeperClient(String remoteDataDir) { + super(createCuratorFrameworkWrapper(), createConfiguration(remoteDataDir)); + } + + private void enableFailureInjection( + boolean failBatchUpdate, Set failedIndividualUpdates) { + this.failBatchUpdate = failBatchUpdate; + this.failedIndividualUpdates = new HashSet<>(failedIndividualUpdates); + failureInjectionEnabled = true; + } + + @Override + public void batchUpdateLeaderAndIsr( + Map leaderAndIsrs, int expectedZkVersion) + throws Exception { + if (!failureInjectionEnabled) { + super.batchUpdateLeaderAndIsr(leaderAndIsrs, expectedZkVersion); + return; + } + batchUpdates.putAll(leaderAndIsrs); + if (failBatchUpdate) { + throw new RuntimeException("Expected batch update failure"); + } + super.batchUpdateLeaderAndIsr(leaderAndIsrs, expectedZkVersion); + } + + @Override + public void updateLeaderAndIsr( + TableBucket tableBucket, LeaderAndIsr leaderAndIsr, int expectedZkVersion) + throws Exception { + individualUpdates.put(tableBucket, leaderAndIsr); + if (failedIndividualUpdates.contains(tableBucket)) { + throw new RuntimeException("Expected individual update failure"); + } + super.updateLeaderAndIsr(tableBucket, leaderAndIsr, expectedZkVersion); + } + + private Map getBatchUpdates() { + return batchUpdates; + } + + private Map getIndividualUpdates() { + return individualUpdates; + } + + private static CuratorFrameworkWithUnhandledErrorListener createCuratorFrameworkWrapper() { + Builder builder = + builder() + .connectString( + ZOO_KEEPER_EXTENSION_WRAPPER + .getCustomExtension() + .getConnectString()) + .namespace("fluss") + .retryPolicy(new ExponentialBackoffRetry(1000, 3)); + return startZookeeperClient(builder, NOPErrorHandler.INSTANCE); + } + + private static Configuration createConfiguration(String remoteDataDir) { + Configuration configuration = new Configuration(); + configuration.set(ConfigOptions.REMOTE_DATA_DIR, remoteDataDir); + return configuration; + } + } + private static class RecordingAutoPartitionManager extends AutoPartitionManager { private final CoordinatorMetadataCache metadataCache; From d1de220d3be2fbef6afa5215954b4fd198f9bb5d Mon Sep 17 00:00:00 2001 From: Yang Guo Date: Wed, 29 Jul 2026 10:16:52 +0800 Subject: [PATCH 2/2] add test comment --- .../CoordinatorEventProcessorTest.java | 23 +++++++++++++++++++ 1 file changed, 23 insertions(+) diff --git a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessorTest.java b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessorTest.java index d4251fa081..fdc5f29869 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessorTest.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessorTest.java @@ -1235,6 +1235,7 @@ void testProcessAdjustIsr() throws Exception { Map bucketLeaderAndIsrMap = createAdjustIsrTestTable("process_adjust_isr"); + // The normal batch path should commit every proposed change. AdjustIsrResponse adjustIsrResponse = adjustIsr(bucketLeaderAndIsrMap); Map resultForBucketMap = getAdjustIsrResponseData(adjustIsrResponse); @@ -1250,11 +1251,13 @@ void testProcessAdjustIsr() throws Exception { void testProcessAdjustIsrFallsBackToSuccessfulIndividualUpdates() throws Exception { Map originalLeaderAndIsrs = createAdjustIsrTestTable("adjust_isr_individual_success"); + // Force the batch write to fail, but allow every per-bucket retry to reach ZooKeeper. useTestingAdjustIsrZooKeeperClient(true, Collections.emptySet()); AdjustIsrResponse response = adjustIsr(originalLeaderAndIsrs); Map results = getAdjustIsrResponseData(response); + // Every fallback write is committed, so each bucket succeeds and is installed. assertAdjustIsrResponseCount(response, originalLeaderAndIsrs.size()); assertThat(results.values()).allMatch(AdjustIsrResultForBucket::succeeded); assertThat(testingAdjustIsrZooKeeperClient.getIndividualUpdates().keySet()) @@ -1271,16 +1274,19 @@ void testProcessAdjustIsrInstallsOnlyCommittedIndividualUpdates() throws Excepti TableBucket failedBucket = buckets.get(1); Set committedBuckets = new HashSet<>(originalLeaderAndIsrs.keySet()); committedBuckets.remove(failedBucket); + // Simulate a mixed fallback result: this bucket fails while all other retries succeed. useTestingAdjustIsrZooKeeperClient(true, Collections.singleton(failedBucket)); AdjustIsrResponse response = adjustIsr(originalLeaderAndIsrs); Map results = getAdjustIsrResponseData(response); + // The failed bucket must have one error, not an error followed by a success. assertAdjustIsrResponseCount(response, originalLeaderAndIsrs.size()); assertThat(results.get(failedBucket).failed()).isTrue(); assertThat(results.entrySet()) .filteredOn(entry -> !entry.getKey().equals(failedBucket)) .allMatch(entry -> entry.getValue().succeeded()); + // Only successful ZooKeeper writes may cross the committed-state boundary. assertAdjustedIsrInstalled(originalLeaderAndIsrs, committedBuckets); } @@ -1288,11 +1294,13 @@ void testProcessAdjustIsrInstallsOnlyCommittedIndividualUpdates() throws Excepti void testProcessAdjustIsrInstallsNothingWhenAllIndividualUpdatesFail() throws Exception { Map originalLeaderAndIsrs = createAdjustIsrTestTable("adjust_isr_individual_failure"); + // Fail the batch write and every per-bucket fallback write. useTestingAdjustIsrZooKeeperClient(true, originalLeaderAndIsrs.keySet()); AdjustIsrResponse response = adjustIsr(originalLeaderAndIsrs); Map results = getAdjustIsrResponseData(response); + // No proposed change was committed, so CoordinatorContext must retain all old states. assertAdjustIsrResponseCount(response, originalLeaderAndIsrs.size()); assertThat(results.values()).allMatch(AdjustIsrResultForBucket::failed); assertAdjustedIsrInstalled(originalLeaderAndIsrs, Collections.emptySet()); @@ -1309,6 +1317,7 @@ void testProcessAdjustIsrKeepsValidationAndPersistenceResultsUnique() throws Exc LeaderAndIsr original = originalLeaderAndIsrs.get(invalidBucket); Map requestedLeaderAndIsrs = new HashMap<>(originalLeaderAndIsrs); + // A lower leader epoch makes this request fail validation before any ZooKeeper write. requestedLeaderAndIsrs.put( invalidBucket, new LeaderAndIsr( @@ -1318,14 +1327,17 @@ void testProcessAdjustIsrKeepsValidationAndPersistenceResultsUnique() throws Exc original.standbyReplicas(), original.coordinatorEpoch(), original.bucketEpoch())); + // A different valid bucket reaches the fallback path but fails its individual write. useTestingAdjustIsrZooKeeperClient(true, Collections.singleton(persistenceFailureBucket)); AdjustIsrResponse response = adjustIsr(requestedLeaderAndIsrs); Map results = getAdjustIsrResponseData(response); + // Validation and persistence failures are each reported exactly once. assertAdjustIsrResponseCount(response, requestedLeaderAndIsrs.size()); assertThat(results.get(invalidBucket).failed()).isTrue(); assertThat(results.get(persistenceFailureBucket).failed()).isTrue(); + // Validation failures must never enter either persistence attempt. assertThat(testingAdjustIsrZooKeeperClient.getBatchUpdates()) .doesNotContainKey(invalidBucket); assertThat(testingAdjustIsrZooKeeperClient.getIndividualUpdates()) @@ -2200,6 +2212,8 @@ private void useTestingAdjustIsrZooKeeperClient( CoordinatorEventProcessor previousEventProcessor = eventProcessor; CoordinatorEventProcessor testingEventProcessor = buildCoordinatorEventProcessor(testingAdjustIsrZooKeeperClient); + // Start the replacement first so it can load the existing table state before the original + // processor is shut down. Failure injection is enabled only after startup initialization. testingEventProcessor.startup(); previousEventProcessor.shutdown(); eventProcessor = testingEventProcessor; @@ -2217,6 +2231,8 @@ private AdjustIsrResponse adjustIsr(Map leaderAndIsrs } private void assertAdjustIsrResponseCount(AdjustIsrResponse response, int expectedCount) { + // Inspect the raw protobuf response: converting it to a map would hide duplicate results + // for the same bucket, such as an error followed by an unconditional success. assertThat( response.getTablesRespsList().stream() .mapToInt(tableResponse -> tableResponse.getBucketsRespsCount()) @@ -2232,6 +2248,8 @@ private void assertAdjustedIsrInstalled( LeaderAndIsr originalLeaderAndIsr = entry.getValue(); Optional installedLeaderAndIsr = fromCtx(ctx -> ctx.getBucketLeaderAndIsr(tableBucket)); + // A committed bucket advances its bucket epoch; an uncommitted bucket must remain + // equal to the state observed before the request. assertThat(installedLeaderAndIsr) .contains( committedBuckets.contains(tableBucket) @@ -2284,6 +2302,8 @@ private void enableFailureInjection( boolean failBatchUpdate, Set failedIndividualUpdates) { this.failBatchUpdate = failBatchUpdate; this.failedIndividualUpdates = new HashSet<>(failedIndividualUpdates); + // Startup uses the real ZooKeeper behavior; only the AdjustIsr call under test should + // observe the configured failures. failureInjectionEnabled = true; } @@ -2295,6 +2315,8 @@ public void batchUpdateLeaderAndIsr( super.batchUpdateLeaderAndIsr(leaderAndIsrs, expectedZkVersion); return; } + // Record every proposed change before throwing so tests can verify validation + // filtering. batchUpdates.putAll(leaderAndIsrs); if (failBatchUpdate) { throw new RuntimeException("Expected batch update failure"); @@ -2306,6 +2328,7 @@ public void batchUpdateLeaderAndIsr( public void updateLeaderAndIsr( TableBucket tableBucket, LeaderAndIsr leaderAndIsr, int expectedZkVersion) throws Exception { + // Record all fallback attempts, but persist only the buckets not selected to fail. individualUpdates.put(tableBucket, leaderAndIsr); if (failedIndividualUpdates.contains(tableBucket)) { throw new RuntimeException("Expected individual update failure");