This is an automated email from the ASF dual-hosted git repository.
showuon pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new ec151c82783 KAFKA-16563: retry pollEvent in KRaftMigrationDriver for
retriable errors (#15732)
ec151c82783 is described below
commit ec151c82783f75de3ee755c80ba9ccfbe338c512
Author: Luke Chen <show...@gmail.com>
AuthorDate: Mon Apr 29 18:44:47 2024 +0900
KAFKA-16563: retry pollEvent in KRaftMigrationDriver for retriable errors
(#15732)
When running ZK migrating to KRaft process, we encountered an issue that
the migrating is hanging and the ZkMigrationState cannot move to MIGRATION
state. And it is because the pollEvent didn't retry with the retriable
MigrationClientException (ZK client retriable errors) while it should. This PR
fixes it and add test. And because of this, the poll event will not poll
anymore, which causes the KRaftMigrationDriver hanging.
Reviewers: Luke Chen <show...@gmail.com>, Igor Soarez<soa...@apple.com>,
Akhilesh C <akhilesh...@users.noreply.github.com>
---
.../metadata/migration/KRaftMigrationDriver.java | 36 +++++----
.../migration/KRaftMigrationDriverTest.java | 94 ++++++++++++++++++----
2 files changed, 100 insertions(+), 30 deletions(-)
diff --git
a/metadata/src/main/java/org/apache/kafka/metadata/migration/KRaftMigrationDriver.java
b/metadata/src/main/java/org/apache/kafka/metadata/migration/KRaftMigrationDriver.java
index 4c796f9eade..7cf82b8762c 100644
---
a/metadata/src/main/java/org/apache/kafka/metadata/migration/KRaftMigrationDriver.java
+++
b/metadata/src/main/java/org/apache/kafka/metadata/migration/KRaftMigrationDriver.java
@@ -165,19 +165,6 @@ public class KRaftMigrationDriver implements
MetadataPublisher {
return stateFuture;
}
- private void recoverMigrationStateFromZK() {
- applyMigrationOperation("Recovering migration state from ZK",
zkMigrationClient::getOrCreateMigrationRecoveryState);
- String maybeDone =
migrationLeadershipState.initialZkMigrationComplete() ? "done" : "not done";
- log.info("Initial migration of ZK metadata is {}.", maybeDone);
-
- // Once we've recovered the migration state from ZK, install this
class as a metadata publisher
- // by calling the initialZkLoadHandler.
- initialZkLoadHandler.accept(this);
-
- // Transition to INACTIVE state and wait for leadership events.
- transitionTo(MigrationDriverState.INACTIVE);
- }
-
private boolean isControllerQuorumReadyForMigration() {
Optional<String> notReadyMsg =
this.quorumFeatures.reasonAllControllersZkMigrationNotReady(
image.features().metadataVersion(),
image.cluster().controllers());
@@ -414,7 +401,7 @@ public class KRaftMigrationDriver implements
MetadataPublisher {
/**
* An event generated by a call to {@link
MetadataPublisher#onControllerChange}. This will not be called until
* this class is registered with {@link
org.apache.kafka.image.loader.MetadataLoader}. The registration happens
- * after the migration state is loaded from ZooKeeper in {@link
#recoverMigrationStateFromZK}.
+ * after the migration state is loaded from ZooKeeper in {@link
RecoverMigrationStateFromZKEvent}.
*/
class KRaftLeaderEvent extends MigrationEvent {
private final LeaderAndEpoch leaderAndEpoch;
@@ -786,12 +773,31 @@ public class KRaftMigrationDriver implements
MetadataPublisher {
}
}
+ class RecoverMigrationStateFromZKEvent extends MigrationEvent {
+ @Override
+ public void run() throws Exception {
+ if (checkDriverState(MigrationDriverState.UNINITIALIZED, this)) {
+ applyMigrationOperation("Recovering migration state from ZK",
zkMigrationClient::getOrCreateMigrationRecoveryState);
+ String maybeDone =
migrationLeadershipState.initialZkMigrationComplete() ? "done" : "not done";
+ log.info("Initial migration of ZK metadata is {}.", maybeDone);
+
+ // Once we've recovered the migration state from ZK, install
this class as a metadata publisher
+ // by calling the initialZkLoadHandler.
+ initialZkLoadHandler.accept(KRaftMigrationDriver.this);
+
+ // Transition to INACTIVE state and wait for leadership events.
+ transitionTo(MigrationDriverState.INACTIVE);
+ }
+ }
+ }
+
class PollEvent extends MigrationEvent {
+
@Override
public void run() throws Exception {
switch (migrationState) {
case UNINITIALIZED:
- recoverMigrationStateFromZK();
+ eventQueue.append(new RecoverMigrationStateFromZKEvent());
break;
case INACTIVE:
// Nothing to do when the driver is inactive. We must wait
until a KRaftLeaderEvent
diff --git
a/metadata/src/test/java/org/apache/kafka/metadata/migration/KRaftMigrationDriverTest.java
b/metadata/src/test/java/org/apache/kafka/metadata/migration/KRaftMigrationDriverTest.java
index dea5e62db96..14f347ca1e1 100644
---
a/metadata/src/test/java/org/apache/kafka/metadata/migration/KRaftMigrationDriverTest.java
+++
b/metadata/src/test/java/org/apache/kafka/metadata/migration/KRaftMigrationDriverTest.java
@@ -253,7 +253,7 @@ public class KRaftMigrationDriverTest {
MetadataImage image = MetadataImage.EMPTY;
MetadataDelta delta = new MetadataDelta(image);
- driver.start();
+ startAndWaitForRecoveringMigrationStateFromZK(driver);
setupDeltaForMigration(delta, registerControllers);
delta.replay(ZkMigrationState.PRE_MIGRATION.toRecord().message());
delta.replay(zkBrokerRecord(1));
@@ -338,7 +338,7 @@ public class KRaftMigrationDriverTest {
MetadataDelta delta = new MetadataDelta(image);
setupDeltaForMigration(delta, true);
- driver.start();
+ startAndWaitForRecoveringMigrationStateFromZK(driver);
delta.replay(ZkMigrationState.PRE_MIGRATION.toRecord().message());
delta.replay(zkBrokerRecord(1));
delta.replay(zkBrokerRecord(2));
@@ -363,6 +363,62 @@ public class KRaftMigrationDriverTest {
}
}
+ @Test
+ public void testMigrationWithClientExceptionWhileMigratingZnodeCreation()
throws Exception {
+ CountingMetadataPropagator metadataPropagator = new
CountingMetadataPropagator();
+ // suppose the ZNode creation failed 3 times
+ CountDownLatch createZnodeAttempts = new CountDownLatch(3);
+ CapturingMigrationClient migrationClient = new
CapturingMigrationClient(new HashSet<>(Arrays.asList(1, 2, 3)),
+ new CapturingTopicMigrationClient(),
+ new CapturingConfigMigrationClient(),
+ new CapturingAclMigrationClient(),
+ new CapturingDelegationTokenMigrationClient(),
+ CapturingMigrationClient.EMPTY_BATCH_SUPPLIER) {
+ @Override
+ public ZkMigrationLeadershipState
getOrCreateMigrationRecoveryState(ZkMigrationLeadershipState initialState) {
+ if (createZnodeAttempts.getCount() == 0) {
+ this.setMigrationRecoveryState(initialState);
+ return initialState;
+ } else {
+ createZnodeAttempts.countDown();
+ throw new MigrationClientException("Some kind of ZK
error!");
+ }
+ }
+ };
+ MockFaultHandler faultHandler = new
MockFaultHandler("testMigrationClientExpiration");
+ KRaftMigrationDriver.Builder builder = defaultTestBuilder()
+ .setZkMigrationClient(migrationClient)
+ .setFaultHandler(faultHandler)
+ .setPropagator(metadataPropagator);
+ try (KRaftMigrationDriver driver = builder.build()) {
+ MetadataImage image = MetadataImage.EMPTY;
+ MetadataDelta delta = new MetadataDelta(image);
+ setupDeltaForMigration(delta, true);
+
+ startAndWaitForRecoveringMigrationStateFromZK(driver);
+
+ delta.replay(ZkMigrationState.PRE_MIGRATION.toRecord().message());
+ delta.replay(zkBrokerRecord(1));
+ delta.replay(zkBrokerRecord(2));
+ delta.replay(zkBrokerRecord(3));
+ MetadataProvenance provenance = new MetadataProvenance(100, 1, 1);
+ image = delta.apply(provenance);
+ // Before leadership claiming, the
getOrCreateMigrationRecoveryState should be able to get correct state
+ assertTrue(createZnodeAttempts.await(1, TimeUnit.MINUTES));
+
+ // Notify the driver that it is the leader
+ driver.onControllerChange(new LeaderAndEpoch(OptionalInt.of(3000),
1));
+ // Publish metadata of all the ZK brokers being ready
+ driver.onMetadataUpdate(delta, image,
logDeltaManifestBuilder(provenance,
+ new LeaderAndEpoch(OptionalInt.of(3000), 1)).build());
+
+ TestUtils.waitForCondition(() -> driver.migrationState().get(1,
TimeUnit.MINUTES).equals(MigrationDriverState.DUAL_WRITE),
+ "Waiting for KRaftMigrationDriver to enter DUAL_WRITE
state");
+
+ Assertions.assertNull(faultHandler.firstException());
+ }
+ }
+
private void setupDeltaForMigration(
MetadataDelta delta,
boolean registerControllers
@@ -413,7 +469,7 @@ public class KRaftMigrationDriverTest {
MetadataImage image = MetadataImage.EMPTY;
MetadataDelta delta = new MetadataDelta(image);
- driver.start();
+ startAndWaitForRecoveringMigrationStateFromZK(driver);
if (allNodePresent) {
setupDeltaWithControllerRegistrations(delta, Arrays.asList(4,
5, 6), Arrays.asList());
} else {
@@ -469,7 +525,7 @@ public class KRaftMigrationDriverTest {
migrationClient.setMigrationRecoveryState(
ZkMigrationLeadershipState.EMPTY.withKRaftMetadataOffsetAndEpoch(100, 1));
- driver.start();
+ startAndWaitForRecoveringMigrationStateFromZK(driver);
delta.replay(ZkMigrationState.PRE_MIGRATION.toRecord().message());
delta.replay(zkBrokerRecord(1));
delta.replay(zkBrokerRecord(2));
@@ -483,7 +539,7 @@ public class KRaftMigrationDriverTest {
new LeaderAndEpoch(OptionalInt.of(3000), 1)).build());
TestUtils.waitForCondition(() -> driver.migrationState().get(1,
TimeUnit.MINUTES).equals(MigrationDriverState.DUAL_WRITE),
- "Waiting for KRaftMigrationDriver to enter ZK_MIGRATION
state");
+ "Waiting for KRaftMigrationDriver to enter DUAL_WRITE state");
}
}
@@ -546,7 +602,7 @@ public class KRaftMigrationDriverTest {
DelegationTokenImage.EMPTY);
MetadataDelta delta = new MetadataDelta(image);
- driver.start();
+ startAndWaitForRecoveringMigrationStateFromZK(driver);
setupDeltaForMigration(delta, true);
delta.replay(ZkMigrationState.PRE_MIGRATION.toRecord().message());
delta.replay(zkBrokerRecord(0));
@@ -565,7 +621,7 @@ public class KRaftMigrationDriverTest {
// Wait for migration
TestUtils.waitForCondition(() -> driver.migrationState().get(1,
TimeUnit.MINUTES).equals(MigrationDriverState.DUAL_WRITE),
- "Waiting for KRaftMigrationDriver to enter ZK_MIGRATION
state");
+ "Waiting for KRaftMigrationDriver to enter DUAL_WRITE state");
// Modify topics in a KRaft snapshot -- delete foo, modify bar,
add baz, add new foo, add bam, delete bam
provenance = new MetadataProvenance(200, 1, 1);
@@ -601,7 +657,7 @@ public class KRaftMigrationDriverTest {
DelegationTokenImage.EMPTY);
MetadataDelta delta = new MetadataDelta(image);
- driver.start();
+ startAndWaitForRecoveringMigrationStateFromZK(driver);
setupDeltaForMigration(delta, true);
delta.replay(ZkMigrationState.PRE_MIGRATION.toRecord().message());
delta.replay(zkBrokerRecord(0));
@@ -656,7 +712,7 @@ public class KRaftMigrationDriverTest {
DelegationTokenImage.EMPTY);
MetadataDelta delta = new MetadataDelta(image);
- driver.start();
+ startAndWaitForRecoveringMigrationStateFromZK(driver);
setupDeltaForMigration(delta, true);
delta.replay(ZkMigrationState.PRE_MIGRATION.toRecord().message());
delta.replay(zkBrokerRecord(0));
@@ -673,7 +729,7 @@ public class KRaftMigrationDriverTest {
driver.onControllerChange(newLeader);
TestUtils.waitForCondition(() -> driver.migrationState().get(1,
TimeUnit.MINUTES).equals(MigrationDriverState.WAIT_FOR_CONTROLLER_QUORUM),
- "Waiting for KRaftMigrationDriver to enter DUAL_WRITE state");
+ "Waiting for KRaftMigrationDriver to enter
WAIT_FOR_CONTROLLER_QUORUM state");
driver.onMetadataUpdate(delta, image,
logDeltaManifestBuilder(provenance, newLeader).build());
@@ -711,7 +767,7 @@ public class KRaftMigrationDriverTest {
DelegationTokenImage.EMPTY);
MetadataDelta delta = new MetadataDelta(image);
- driver.start();
+ startAndWaitForRecoveringMigrationStateFromZK(driver);
setupDeltaForMigration(delta, true);
delta.replay(ZkMigrationState.PRE_MIGRATION.toRecord().message());
delta.replay(zkBrokerRecord(0));
@@ -778,7 +834,7 @@ public class KRaftMigrationDriverTest {
MetadataImage image = MetadataImage.EMPTY;
MetadataDelta delta = new MetadataDelta(image);
- driver.start();
+ startAndWaitForRecoveringMigrationStateFromZK(driver);
setupDeltaForMigration(delta, true);
delta.replay(ZkMigrationState.PRE_MIGRATION.toRecord().message());
delta.replay(zkBrokerRecord(1));
@@ -798,7 +854,7 @@ public class KRaftMigrationDriverTest {
new LeaderAndEpoch(OptionalInt.of(3000), 1)).build());
TestUtils.waitForCondition(() -> driver.migrationState().get(1,
TimeUnit.MINUTES).equals(MigrationDriverState.DUAL_WRITE),
- "Waiting for KRaftMigrationDriver to enter ZK_MIGRATION
state");
+ "Waiting for KRaftMigrationDriver to enter DUAL_WRITE
state");
assertEquals(1, migrationBeginCalls.get());
}
}
@@ -864,7 +920,7 @@ public class KRaftMigrationDriverTest {
MetadataImage image = MetadataImage.EMPTY;
MetadataDelta delta = new MetadataDelta(image);
- driver.start();
+ startAndWaitForRecoveringMigrationStateFromZK(driver);
setupDeltaForMigration(delta, true);
delta.replay(ZkMigrationState.PRE_MIGRATION.toRecord().message());
delta.replay(zkBrokerRecord(1));
@@ -881,10 +937,18 @@ public class KRaftMigrationDriverTest {
new LeaderAndEpoch(OptionalInt.of(3000), 1)).build());
TestUtils.waitForCondition(() -> driver.migrationState().get(1,
TimeUnit.MINUTES).equals(MigrationDriverState.DUAL_WRITE),
- "Waiting for KRaftMigrationDriver to enter ZK_MIGRATION
state");
+ "Waiting for KRaftMigrationDriver to enter DUAL_WRITE
state");
assertEquals(expectedBatchCount, batchesPassedToController.size());
assertEquals(expectedRecordCount,
batchesPassedToController.stream().mapToInt(List::size).sum());
}
}
+
+ // Wait until the driver has recovered MigrationState From ZK. This is to
simulate the driver needs to be installed as the metadata publisher
+ // so that it can receive onControllerChange (KRaftLeaderEvent) and
onMetadataUpdate (MetadataChangeEvent) events.
+ private void
startAndWaitForRecoveringMigrationStateFromZK(KRaftMigrationDriver driver)
throws InterruptedException {
+ driver.start();
+ TestUtils.waitForCondition(() -> driver.migrationState().get(1,
TimeUnit.MINUTES).equals(MigrationDriverState.INACTIVE),
+ "Waiting for KRaftMigrationDriver to enter INACTIVE state");
+ }
}
