From 160b20aee8133ab9585e854074df59e0e2e803ef Mon Sep 17 00:00:00 2001 From: David Arthur Date: Thu, 9 Mar 2023 12:15:16 -0500 Subject: [PATCH 01/12] Catch KeeperException in ZkMigrationClient, add unit test --- .../scala/kafka/zk/ZkMigrationClient.scala | 130 +++++++++++------- .../migration/KRaftMigrationDriver.java | 28 ++-- .../metadata/migration/MigrationClient.java | 13 -- .../migration/MigrationClientException.java | 17 +++ .../migration/KRaftMigrationDriverTest.java | 60 ++++++-- 5 files changed, 157 insertions(+), 91 deletions(-) create mode 100644 metadata/src/main/java/org/apache/kafka/metadata/migration/MigrationClientException.java diff --git a/core/src/main/scala/kafka/zk/ZkMigrationClient.scala b/core/src/main/scala/kafka/zk/ZkMigrationClient.scala index e13609d7fb5cf..5e7e3c74be3e4 100644 --- a/core/src/main/scala/kafka/zk/ZkMigrationClient.scala +++ b/core/src/main/scala/kafka/zk/ZkMigrationClient.scala @@ -28,9 +28,8 @@ import org.apache.kafka.common.metadata.ClientQuotaRecord.EntityData import org.apache.kafka.common.metadata._ import org.apache.kafka.common.quota.ClientQuotaEntity import org.apache.kafka.common.{TopicPartition, Uuid} -import org.apache.kafka.image.{MetadataDelta, MetadataImage} import org.apache.kafka.metadata.{LeaderRecoveryState, PartitionRegistration} -import org.apache.kafka.metadata.migration.{MigrationClient, ZkMigrationLeadershipState} +import org.apache.kafka.metadata.migration.{MigrationClient, MigrationClientException, ZkMigrationLeadershipState} import org.apache.kafka.server.common.{ApiMessageAndVersion, ProducerIdsBlock} import org.apache.zookeeper.KeeperException.Code import org.apache.zookeeper.{CreateMode, KeeperException} @@ -47,16 +46,31 @@ import scala.jdk.CollectionConverters._ */ class ZkMigrationClient(zkClient: KafkaZkClient) extends MigrationClient with Logging { - override def getOrCreateMigrationRecoveryState(initialState: ZkMigrationLeadershipState): ZkMigrationLeadershipState = { - zkClient.createTopLevelPaths() - zkClient.getOrCreateMigrationState(initialState) + @throws(classOf[MigrationClientException]) + def wrapZkException[T](fn: => T): T = { + try { + fn + } catch { + case e: KeeperException => throw new MigrationClientException(e) + } } - override def setMigrationRecoveryState(state: ZkMigrationLeadershipState): ZkMigrationLeadershipState = { + override def getOrCreateMigrationRecoveryState( + initialState: ZkMigrationLeadershipState + ): ZkMigrationLeadershipState = wrapZkException { + zkClient.createTopLevelPaths() + zkClient.getOrCreateMigrationState(initialState) + } + + override def setMigrationRecoveryState( + state: ZkMigrationLeadershipState + ): ZkMigrationLeadershipState = wrapZkException { zkClient.updateMigrationState(state) } - override def claimControllerLeadership(state: ZkMigrationLeadershipState): ZkMigrationLeadershipState = { + override def claimControllerLeadership( + state: ZkMigrationLeadershipState + ): ZkMigrationLeadershipState = wrapZkException { zkClient.tryRegisterKRaftControllerAsActiveController(state.kraftControllerId(), state.kraftControllerEpoch()) match { case SuccessfulRegistrationResult(controllerEpoch, controllerEpochZkVersion) => state.withZkController(controllerEpoch, controllerEpochZkVersion) @@ -64,7 +78,9 @@ class ZkMigrationClient(zkClient: KafkaZkClient) extends MigrationClient with Lo } } - override def releaseControllerLeadership(state: ZkMigrationLeadershipState): ZkMigrationLeadershipState = { + override def releaseControllerLeadership( + state: ZkMigrationLeadershipState + ): ZkMigrationLeadershipState = wrapZkException { try { zkClient.deleteController(state.zkControllerEpochZkVersion()) state.withUnknownZkController() @@ -73,7 +89,7 @@ class ZkMigrationClient(zkClient: KafkaZkClient) extends MigrationClient with Lo // If the controller moved, no need to release state.withUnknownZkController() case t: Throwable => - throw new RuntimeException("Could not release controller leadership due to underlying error", t) + throw new MigrationClientException("Could not release controller leadership due to underlying error", t) } } @@ -211,19 +227,21 @@ class ZkMigrationClient(zkClient: KafkaZkClient) extends MigrationClient with Lo } } - override def readAllMetadata(batchConsumer: Consumer[util.List[ApiMessageAndVersion]], - brokerIdConsumer: Consumer[Integer]): Unit = { + override def readAllMetadata( + batchConsumer: Consumer[util.List[ApiMessageAndVersion]], + brokerIdConsumer: Consumer[Integer] + ): Unit = wrapZkException { migrateTopics(batchConsumer, brokerIdConsumer) migrateBrokerConfigs(batchConsumer) migrateClientQuotas(batchConsumer) migrateProducerId(batchConsumer) } - override def readBrokerIds(): util.Set[Integer] = { + override def readBrokerIds(): util.Set[Integer] = wrapZkException { zkClient.getSortedBrokerList.map(Integer.valueOf).toSet.asJava } - override def readBrokerIdsFromTopicAssignments(): util.Set[Integer] = { + override def readBrokerIdsFromTopicAssignments(): util.Set[Integer] = wrapZkException { val topics = zkClient.getAllTopicsInCluster() val replicaAssignmentAndTopicIds = zkClient.getReplicaAssignmentAndTopicIdForTopics(topics) val brokersWithAssignments = new util.HashSet[Integer]() @@ -235,10 +253,12 @@ class ZkMigrationClient(zkClient: KafkaZkClient) extends MigrationClient with Lo brokersWithAssignments } - override def createTopic(topicName: String, - topicId: Uuid, - partitions: util.Map[Integer, PartitionRegistration], - state: ZkMigrationLeadershipState): ZkMigrationLeadershipState = { + override def createTopic( + topicName: String, + topicId: Uuid, + partitions: util.Map[Integer, PartitionRegistration], + state: ZkMigrationLeadershipState + ): ZkMigrationLeadershipState = wrapZkException { val assignments = partitions.asScala.map { case (partitionId, partition) => new TopicPartition(topicName, partitionId) -> ReplicaAssignment(partition.replicas, partition.addingReplicas, partition.removingReplicas) @@ -280,18 +300,22 @@ class ZkMigrationClient(zkClient: KafkaZkClient) extends MigrationClient with Lo state.withMigrationZkVersion(migrationZkVersion) } else { // not ok - throw new RuntimeException(s"Failed to create or update topic $topicName. ZK operation had results $resultCodes") + throw new MigrationClientException(s"Failed to create or update topic $topicName. ZK operation had results $resultCodes") } } - private def createTopicPartition(topicPartition: TopicPartition): CreateRequest = { + private def createTopicPartition( + topicPartition: TopicPartition + ): CreateRequest = wrapZkException { val path = TopicPartitionZNode.path(topicPartition) CreateRequest(path, null, zkClient.defaultAcls(path), CreateMode.PERSISTENT, Some(topicPartition)) } - private def partitionStatePathAndData(topicPartition: TopicPartition, - partitionRegistration: PartitionRegistration, - controllerEpoch: Int): (String, Array[Byte]) = { + private def partitionStatePathAndData( + topicPartition: TopicPartition, + partitionRegistration: PartitionRegistration, + controllerEpoch: Int + ): (String, Array[Byte]) = { val path = TopicPartitionStateZNode.path(topicPartition) val data = TopicPartitionStateZNode.encode(LeaderIsrAndControllerEpoch(new LeaderAndIsr( partitionRegistration.leader, @@ -302,22 +326,28 @@ class ZkMigrationClient(zkClient: KafkaZkClient) extends MigrationClient with Lo (path, data) } - private def createTopicPartitionState(topicPartition: TopicPartition, - partitionRegistration: PartitionRegistration, - controllerEpoch: Int): CreateRequest = { + private def createTopicPartitionState( + topicPartition: TopicPartition, + partitionRegistration: PartitionRegistration, + controllerEpoch: Int + ): CreateRequest = { val (path, data) = partitionStatePathAndData(topicPartition, partitionRegistration, controllerEpoch) CreateRequest(path, data, zkClient.defaultAcls(path), CreateMode.PERSISTENT, Some(topicPartition)) } - private def updateTopicPartitionState(topicPartition: TopicPartition, - partitionRegistration: PartitionRegistration, - controllerEpoch: Int): SetDataRequest = { + private def updateTopicPartitionState( + topicPartition: TopicPartition, + partitionRegistration: PartitionRegistration, + controllerEpoch: Int + ): SetDataRequest = { val (path, data) = partitionStatePathAndData(topicPartition, partitionRegistration, controllerEpoch) SetDataRequest(path, data, ZkVersion.MatchAnyVersion, Some(topicPartition)) } - override def updateTopicPartitions(topicPartitions: util.Map[String, util.Map[Integer, PartitionRegistration]], - state: ZkMigrationLeadershipState): ZkMigrationLeadershipState = { + override def updateTopicPartitions( + topicPartitions: util.Map[String, util.Map[Integer, PartitionRegistration]], + state: ZkMigrationLeadershipState + ): ZkMigrationLeadershipState = wrapZkException { val requests = topicPartitions.asScala.flatMap { case (topicName, partitionRegistrations) => partitionRegistrations.asScala.flatMap { case (partitionId, partitionRegistration) => val topicPartition = new TopicPartition(topicName, partitionId) @@ -332,18 +362,20 @@ class ZkMigrationClient(zkClient: KafkaZkClient) extends MigrationClient with Lo if (resultCodes.forall { case (_, code) => code.equals(Code.OK) } ) { state.withMigrationZkVersion(migrationZkVersion) } else { - throw new RuntimeException(s"Failed to update partition states: $topicPartitions. ZK transaction had results $resultCodes") + throw new MigrationClientException(s"Failed to update partition states: $topicPartitions. ZK transaction had results $resultCodes") } } } // Try to update an entity config and the migration state. If NoNode is encountered, it probably means we // need to recursively create the parent ZNode. In this case, return None. - def tryWriteEntityConfig(entityType: String, - path: String, - props: Properties, - create: Boolean, - state: ZkMigrationLeadershipState): Option[ZkMigrationLeadershipState] = { + def tryWriteEntityConfig( + entityType: String, + path: String, + props: Properties, + create: Boolean, + state: ZkMigrationLeadershipState + ): Option[ZkMigrationLeadershipState] = wrapZkException { val configData = ConfigEntityZNode.encode(props) val requests = if (create) { @@ -366,8 +398,7 @@ class ZkMigrationClient(zkClient: KafkaZkClient) extends MigrationClient with Lo entity: util.Map[String, String], quotas: util.Map[String, java.lang.Double], state: ZkMigrationLeadershipState - ): ZkMigrationLeadershipState = { - + ): ZkMigrationLeadershipState = wrapZkException { val entityMap = entity.asScala val hasUser = entityMap.contains(ClientQuotaEntity.USER) val hasClient = entityMap.contains(ClientQuotaEntity.CLIENT_ID) @@ -409,13 +440,16 @@ class ZkMigrationClient(zkClient: KafkaZkClient) extends MigrationClient with Lo tryWriteEntityConfig(configType.get, path.get, props, create=true, state) match { case Some(newStateSecondTry) => newStateSecondTry - case None => throw new RuntimeException( + case None => throw new MigrationClientException( s"Could not write client quotas for $entity on second attempt when using Create instead of SetData") } } } - override def writeProducerId(nextProducerId: Long, state: ZkMigrationLeadershipState): ZkMigrationLeadershipState = { + override def writeProducerId( + nextProducerId: Long, + state: ZkMigrationLeadershipState + ): ZkMigrationLeadershipState = wrapZkException { val newProducerIdBlockData = ProducerIdBlockZNode.generateProducerIdBlockJson( new ProducerIdsBlock(-1, nextProducerId, ProducerIdsBlock.PRODUCER_ID_BLOCK_SIZE)) @@ -424,9 +458,11 @@ class ZkMigrationClient(zkClient: KafkaZkClient) extends MigrationClient with Lo state.withMigrationZkVersion(migrationZkVersion) } - override def writeConfigs(resource: ConfigResource, - configs: util.Map[String, String], - state: ZkMigrationLeadershipState): ZkMigrationLeadershipState = { + override def writeConfigs( + resource: ConfigResource, + configs: util.Map[String, String], + state: ZkMigrationLeadershipState + ): ZkMigrationLeadershipState = wrapZkException { val configType = resource.`type`() match { case ConfigResource.Type.BROKER => Some(ConfigType.Broker) case ConfigResource.Type.TOPIC => Some(ConfigType.Topic) @@ -447,7 +483,7 @@ class ZkMigrationClient(zkClient: KafkaZkClient) extends MigrationClient with Lo tryWriteEntityConfig(configType.get, configName, props, create=true, state) match { case Some(newStateSecondTry) => newStateSecondTry - case None => throw new RuntimeException( + case None => throw new MigrationClientException( s"Could not write ${configType.get} configs on second attempt when using Create instead of SetData.") } } @@ -456,10 +492,4 @@ class ZkMigrationClient(zkClient: KafkaZkClient) extends MigrationClient with Lo state } } - - override def writeMetadataDeltaToZookeeper(delta: MetadataDelta, - image: MetadataImage, - state: ZkMigrationLeadershipState): ZkMigrationLeadershipState = { - state - } } 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 3394741f278c4..067de375d3e33 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 @@ -266,7 +266,9 @@ public void close() throws Exception { abstract class MigrationEvent implements EventQueue.Event { @Override public void handleException(Throwable e) { - if (e instanceof RejectedExecutionException) { + if (e instanceof MigrationClientException) { + log.warn(String.format("Encountered client error during event %s. Will retry.", this), e.getCause()); + } else if (e instanceof RejectedExecutionException) { log.info("Not processing {} because the event queue is closed.", this); } else { KRaftMigrationDriver.this.faultHandler.handleFault( @@ -376,23 +378,17 @@ public void run() throws Exception { class BecomeZkControllerEvent extends MigrationEvent { @Override public void run() throws Exception { - switch (migrationState) { - case BECOME_CONTROLLER: - // TODO: Handle unhappy path. - apply("BecomeZkLeaderEvent", zkMigrationClient::claimControllerLeadership); - if (migrationLeadershipState.zkControllerEpochZkVersion() == -1) { - // We could not claim leadership, stay in BECOME_CONTROLLER to retry + if (migrationState == MigrationDriverState.BECOME_CONTROLLER) { + apply("BecomeZkLeaderEvent", zkMigrationClient::claimControllerLeadership); + if (migrationLeadershipState.zkControllerEpochZkVersion() == -1) { + log.debug("Unable to claim leadership, will retry until we learn of a different KRaft leader"); + } else { + if (!migrationLeadershipState.zkMigrationComplete()) { + transitionTo(MigrationDriverState.ZK_MIGRATION); } else { - if (!migrationLeadershipState.zkMigrationComplete()) { - transitionTo(MigrationDriverState.ZK_MIGRATION); - } else { - transitionTo(MigrationDriverState.KRAFT_CONTROLLER_TO_BROKER_COMM); - } + transitionTo(MigrationDriverState.KRAFT_CONTROLLER_TO_BROKER_COMM); } - break; - default: - // Ignore the event as we're not trying to become controller anymore. - break; + } } } } diff --git a/metadata/src/main/java/org/apache/kafka/metadata/migration/MigrationClient.java b/metadata/src/main/java/org/apache/kafka/metadata/migration/MigrationClient.java index 11284b399c19d..b026897db4b64 100644 --- a/metadata/src/main/java/org/apache/kafka/metadata/migration/MigrationClient.java +++ b/metadata/src/main/java/org/apache/kafka/metadata/migration/MigrationClient.java @@ -18,8 +18,6 @@ import org.apache.kafka.common.Uuid; import org.apache.kafka.common.config.ConfigResource; -import org.apache.kafka.image.MetadataDelta; -import org.apache.kafka.image.MetadataImage; import org.apache.kafka.metadata.PartitionRegistration; import org.apache.kafka.server.common.ApiMessageAndVersion; @@ -108,15 +106,4 @@ ZkMigrationLeadershipState writeProducerId( Set readBrokerIds(); Set readBrokerIdsFromTopicAssignments(); - - /** - * Convert the Metadata delta to Zookeeper writes and persist the changes. On successful - * write, update the migration state with new metadata offset and epoch. - * @param delta Changes in the cluster metadata - * @param image New metadata after the changes in `delta` are applied - * @param state Current migration state before writing to Zookeeper. - */ - ZkMigrationLeadershipState writeMetadataDeltaToZookeeper(MetadataDelta delta, - MetadataImage image, - ZkMigrationLeadershipState state); } diff --git a/metadata/src/main/java/org/apache/kafka/metadata/migration/MigrationClientException.java b/metadata/src/main/java/org/apache/kafka/metadata/migration/MigrationClientException.java new file mode 100644 index 0000000000000..f0a06a435d691 --- /dev/null +++ b/metadata/src/main/java/org/apache/kafka/metadata/migration/MigrationClientException.java @@ -0,0 +1,17 @@ +package org.apache.kafka.metadata.migration; + +import org.apache.kafka.common.KafkaException; + +public class MigrationClientException extends KafkaException { + public MigrationClientException(String message, Throwable t) { + super(message, t); + } + + public MigrationClientException(Throwable t) { + super(t); + } + + public MigrationClientException(String message) { + super(message); + } +} 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 b25f1a10a765b..da8ef61064bee 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 @@ -45,11 +45,12 @@ import java.util.OptionalInt; import java.util.Set; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import java.util.function.Consumer; public class KRaftMigrationDriverTest { - class NoOpRecordConsumer implements ZkRecordConsumer { + static class NoOpRecordConsumer implements ZkRecordConsumer { @Override public void beginMigration() { @@ -71,7 +72,7 @@ public void abortMigration() { } } - class CapturingMigrationClient implements MigrationClient { + static class CapturingMigrationClient implements MigrationClient { private final Set brokerIds; public final Map> capturedConfigs = new HashMap<>(); @@ -162,18 +163,9 @@ public Set readBrokerIds() { public Set readBrokerIdsFromTopicAssignments() { return brokerIds; } - - @Override - public ZkMigrationLeadershipState writeMetadataDeltaToZookeeper( - MetadataDelta delta, - MetadataImage image, - ZkMigrationLeadershipState state - ) { - return state; - } } - class CountingMetadataPropagator implements LegacyPropagator { + static class CountingMetadataPropagator implements LegacyPropagator { public int deltas = 0; public int images = 0; @@ -315,4 +307,48 @@ public void testOnlySendNeededRPCsToBrokers() throws Exception { driver.close(); } + + @Test + public void testZkSessionExpiration() throws Exception { + CountingMetadataPropagator metadataPropagator = new CountingMetadataPropagator(); + CountDownLatch claimLeaderAttempts = new CountDownLatch(3); + CapturingMigrationClient migrationClient = new CapturingMigrationClient(new HashSet<>(Arrays.asList(1, 2, 3))) { + @Override + public ZkMigrationLeadershipState claimControllerLeadership(ZkMigrationLeadershipState state) { + if (claimLeaderAttempts.getCount() == 0) { + return super.claimControllerLeadership(state); + } else { + claimLeaderAttempts.countDown(); + throw new MigrationClientException("Some kind of ZK error!"); + } + + } + }; + try (KRaftMigrationDriver driver = new KRaftMigrationDriver( + 3000, + new NoOpRecordConsumer(), + migrationClient, + metadataPropagator, + metadataPublisher -> { }, + new MockFaultHandler("test") + )) { + MetadataImage image = MetadataImage.EMPTY; + MetadataDelta delta = new MetadataDelta(image); + + driver.start(); + delta.replay(zkBrokerRecord(1)); + delta.replay(zkBrokerRecord(2)); + delta.replay(zkBrokerRecord(3)); + MetadataProvenance provenance = new MetadataProvenance(100, 1, 1); + image = delta.apply(provenance); + + // Publish a delta with this node (3000) as the leader + driver.publishLogDelta(delta, image, new LogDeltaManifest(provenance, + new LeaderAndEpoch(OptionalInt.of(3000), 1), 1, 100, 42)); + + Assertions.assertTrue(claimLeaderAttempts.await(1, TimeUnit.MINUTES)); + TestUtils.waitForCondition(() -> driver.migrationState().get(1, TimeUnit.MINUTES).equals(MigrationDriverState.ZK_MIGRATION), + "Waiting for KRaftMigrationDriver to enter ZK_MIGRATION state"); + } + } } From d139892b1362f2f4c880a8ba6350a292890a3b60 Mon Sep 17 00:00:00 2001 From: David Arthur Date: Thu, 9 Mar 2023 12:25:24 -0500 Subject: [PATCH 02/12] Improve WAIT_FOR_BROKERS performance --- .../migration/KRaftMigrationDriver.java | 31 +++++++++++++------ 1 file changed, 22 insertions(+), 9 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 067de375d3e33..d7d0e2540d111 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 @@ -132,24 +132,37 @@ private boolean isControllerQuorumReadyForMigration() { return true; } + private boolean imageDoesNotContainAllBrokers(MetadataImage image, Set brokerIds) { + for (BrokerRegistration broker : image.cluster().brokers().values()) { + if (broker.isMigratingZkBroker()) { + brokerIds.remove(broker.id()); + } + } + return !brokerIds.isEmpty(); + } + private boolean areZkBrokersReadyForMigration() { if (image == MetadataImage.EMPTY) { // TODO maybe add WAIT_FOR_INITIAL_METADATA_PUBLISH state to avoid this kind of check? log.info("Waiting for initial metadata publish before checking if Zk brokers are registered."); return false; } - Set zkRegisteredZkBrokers = zkMigrationClient.readBrokerIdsFromTopicAssignments(); - for (BrokerRegistration broker : image.cluster().brokers().values()) { - if (broker.isMigratingZkBroker()) { - zkRegisteredZkBrokers.remove(broker.id()); - } + + // First check the brokers registered in ZK + Set zkBrokerRegistrations = zkMigrationClient.readBrokerIds(); + if (imageDoesNotContainAllBrokers(image, zkBrokerRegistrations)) { + log.info("Still waiting for ZK brokers {} to register with KRaft.", zkBrokerRegistrations); + return false; } - if (zkRegisteredZkBrokers.isEmpty()) { - return true; - } else { - log.info("Still waiting for ZK brokers {} to register with KRaft.", zkRegisteredZkBrokers); + + // Once all of those are found, check the topic assignments. This is much more expensive than listing /brokers + Set zkBrokersWithAssignments = zkMigrationClient.readBrokerIdsFromTopicAssignments(); + if (imageDoesNotContainAllBrokers(image, zkBrokersWithAssignments)) { + log.info("Still waiting for ZK brokers {} to register with KRaft.", zkBrokersWithAssignments); return false; } + + return true; } private void apply(String name, Function stateMutator) { From e8dded83b55ef9747ff77ae77759ead670017433 Mon Sep 17 00:00:00 2001 From: David Arthur Date: Thu, 9 Mar 2023 12:31:30 -0500 Subject: [PATCH 03/12] checkstyle --- .../metadata/migration/KRaftMigrationDriver.java | 6 ++++++ .../migration/MigrationClientException.java | 16 ++++++++++++++++ 2 files changed, 22 insertions(+) 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 d7d0e2540d111..67a798f848f98 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,6 +165,12 @@ private boolean areZkBrokersReadyForMigration() { return true; } + /** + * Apply a function which transforms our internal migration state. + * + * @param name A descriptive name of the function that is being applied + * @param stateMutator A function which performs some migration operations and possibly transforms our internal state + */ private void apply(String name, Function stateMutator) { ZkMigrationLeadershipState beforeState = this.migrationLeadershipState; ZkMigrationLeadershipState afterState = stateMutator.apply(beforeState); diff --git a/metadata/src/main/java/org/apache/kafka/metadata/migration/MigrationClientException.java b/metadata/src/main/java/org/apache/kafka/metadata/migration/MigrationClientException.java index f0a06a435d691..c957a81e35427 100644 --- a/metadata/src/main/java/org/apache/kafka/metadata/migration/MigrationClientException.java +++ b/metadata/src/main/java/org/apache/kafka/metadata/migration/MigrationClientException.java @@ -1,3 +1,19 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ package org.apache.kafka.metadata.migration; import org.apache.kafka.common.KafkaException; From c2316cd47ee02b957c196af0184d72ec4ec46916 Mon Sep 17 00:00:00 2001 From: David Arthur Date: Thu, 9 Mar 2023 13:28:23 -0500 Subject: [PATCH 04/12] Create second exception type --- .../scala/kafka/zk/ZkMigrationClient.scala | 7 +++-- .../migration/KRaftMigrationDriver.java | 19 +++++++++----- .../MigrationClientAuthException.java | 26 +++++++++++++++++++ .../migration/MigrationClientException.java | 5 ++++ .../migration/KRaftMigrationDriverTest.java | 24 +++++++++++++---- 5 files changed, 68 insertions(+), 13 deletions(-) create mode 100644 metadata/src/main/java/org/apache/kafka/metadata/migration/MigrationClientAuthException.java diff --git a/core/src/main/scala/kafka/zk/ZkMigrationClient.scala b/core/src/main/scala/kafka/zk/ZkMigrationClient.scala index 5e7e3c74be3e4..0f25f2a6d2e7a 100644 --- a/core/src/main/scala/kafka/zk/ZkMigrationClient.scala +++ b/core/src/main/scala/kafka/zk/ZkMigrationClient.scala @@ -29,9 +29,9 @@ import org.apache.kafka.common.metadata._ import org.apache.kafka.common.quota.ClientQuotaEntity import org.apache.kafka.common.{TopicPartition, Uuid} import org.apache.kafka.metadata.{LeaderRecoveryState, PartitionRegistration} -import org.apache.kafka.metadata.migration.{MigrationClient, MigrationClientException, ZkMigrationLeadershipState} +import org.apache.kafka.metadata.migration.{MigrationClient, MigrationClientAuthException, MigrationClientException, ZkMigrationLeadershipState} import org.apache.kafka.server.common.{ApiMessageAndVersion, ProducerIdsBlock} -import org.apache.zookeeper.KeeperException.Code +import org.apache.zookeeper.KeeperException.{AuthFailedException, Code, NoAuthException, SessionClosedRequireAuthException} import org.apache.zookeeper.{CreateMode, KeeperException} import java.util @@ -51,6 +51,9 @@ class ZkMigrationClient(zkClient: KafkaZkClient) extends MigrationClient with Lo try { fn } catch { + case e @ (_: AuthFailedException | _: NoAuthException | _: SessionClosedRequireAuthException) => + // We don't expect authentication errors to be recoverable, so treat them differently + throw new MigrationClientAuthException(e) case e: KeeperException => throw new MigrationClientException(e) } } 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 67a798f848f98..729256575510d 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 @@ -93,7 +93,7 @@ public KRaftMigrationDriver( this.log = LoggerFactory.getLogger(KRaftMigrationDriver.class); this.migrationState = MigrationDriverState.UNINITIALIZED; this.migrationLeadershipState = ZkMigrationLeadershipState.EMPTY; - this.eventQueue = new KafkaEventQueue(Time.SYSTEM, new LogContext("KRaftMigrationDriver"), "kraft-migration"); + this.eventQueue = new KafkaEventQueue(Time.SYSTEM, new LogContext("KRaftMigrationDriver "), "kraft-migration"); this.image = MetadataImage.EMPTY; this.leaderAndEpoch = LeaderAndEpoch.UNKNOWN; this.initialZkLoadHandler = initialZkLoadHandler; @@ -283,17 +283,24 @@ public void close() throws Exception { // Events handled by Migration Driver. abstract class MigrationEvent implements EventQueue.Event { + @SuppressWarnings("ThrowableNotThrown") @Override public void handleException(Throwable e) { - if (e instanceof MigrationClientException) { - log.warn(String.format("Encountered client error during event %s. Will retry.", this), e.getCause()); + if (e instanceof MigrationClientAuthException) { + KRaftMigrationDriver.this.faultHandler.handleFault("Encountered client auth error in " + this, e); + } else if (e instanceof MigrationClientException) { + log.debug(String.format("Encountered client error during event %s. Will retry.", this), e.getCause()); } else if (e instanceof RejectedExecutionException) { - log.info("Not processing {} because the event queue is closed.", this); + log.debug("Not processing {} because the event queue is closed.", this); } else { - KRaftMigrationDriver.this.faultHandler.handleFault( - "Unhandled error in " + this.getClass().getSimpleName(), e); + KRaftMigrationDriver.this.faultHandler.handleFault("Unhandled error in " + this, e); } } + + @Override + public String toString() { + return this.getClass().getSimpleName(); + } } class PollEvent extends MigrationEvent { diff --git a/metadata/src/main/java/org/apache/kafka/metadata/migration/MigrationClientAuthException.java b/metadata/src/main/java/org/apache/kafka/metadata/migration/MigrationClientAuthException.java new file mode 100644 index 0000000000000..a10b8ae3f4915 --- /dev/null +++ b/metadata/src/main/java/org/apache/kafka/metadata/migration/MigrationClientAuthException.java @@ -0,0 +1,26 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.kafka.metadata.migration; + +/** + * Wrapped for authentication exceptions in the migration client such as ZooKeeper AuthFailedException + */ +public class MigrationClientAuthException extends MigrationClientException { + public MigrationClientAuthException(Throwable t) { + super(t); + } +} diff --git a/metadata/src/main/java/org/apache/kafka/metadata/migration/MigrationClientException.java b/metadata/src/main/java/org/apache/kafka/metadata/migration/MigrationClientException.java index c957a81e35427..6c8164237617d 100644 --- a/metadata/src/main/java/org/apache/kafka/metadata/migration/MigrationClientException.java +++ b/metadata/src/main/java/org/apache/kafka/metadata/migration/MigrationClientException.java @@ -18,6 +18,11 @@ import org.apache.kafka.common.KafkaException; +/** + * Unchecked exception that can be thrown by the migration client. + * + * Authentication related errors should use {@link MigrationClientAuthException}. + */ public class MigrationClientException extends KafkaException { public MigrationClientException(String message, Throwable t) { super(message, t); 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 da8ef61064bee..5ce90d543b671 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 @@ -36,6 +36,8 @@ import org.apache.kafka.test.TestUtils; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; import java.util.Arrays; import java.util.HashMap; @@ -308,8 +310,9 @@ public void testOnlySendNeededRPCsToBrokers() throws Exception { driver.close(); } - @Test - public void testZkSessionExpiration() throws Exception { + @ParameterizedTest + @ValueSource(booleans = {true, false}) + public void testMigrationWithClientException(boolean authException) throws Exception { CountingMetadataPropagator metadataPropagator = new CountingMetadataPropagator(); CountDownLatch claimLeaderAttempts = new CountDownLatch(3); CapturingMigrationClient migrationClient = new CapturingMigrationClient(new HashSet<>(Arrays.asList(1, 2, 3))) { @@ -319,18 +322,23 @@ public ZkMigrationLeadershipState claimControllerLeadership(ZkMigrationLeadershi return super.claimControllerLeadership(state); } else { claimLeaderAttempts.countDown(); - throw new MigrationClientException("Some kind of ZK error!"); + if (authException) { + throw new MigrationClientAuthException(new RuntimeException("Some kind of ZK auth error!")); + } else { + throw new MigrationClientException("Some kind of ZK error!"); + } } } }; + MockFaultHandler faultHandler = new MockFaultHandler("testMigrationClientExpiration"); try (KRaftMigrationDriver driver = new KRaftMigrationDriver( 3000, new NoOpRecordConsumer(), migrationClient, metadataPropagator, metadataPublisher -> { }, - new MockFaultHandler("test") + faultHandler )) { MetadataImage image = MetadataImage.EMPTY; MetadataDelta delta = new MetadataDelta(image); @@ -349,6 +357,12 @@ public ZkMigrationLeadershipState claimControllerLeadership(ZkMigrationLeadershi Assertions.assertTrue(claimLeaderAttempts.await(1, TimeUnit.MINUTES)); TestUtils.waitForCondition(() -> driver.migrationState().get(1, TimeUnit.MINUTES).equals(MigrationDriverState.ZK_MIGRATION), "Waiting for KRaftMigrationDriver to enter ZK_MIGRATION state"); + + if (authException) { + Assertions.assertEquals(MigrationClientAuthException.class, faultHandler.firstException().getCause().getClass()); + } else { + Assertions.assertNull(faultHandler.firstException()); + } } } -} +} \ No newline at end of file From d5c56315f67ac93b2564c71562ef675a0f6cadfb Mon Sep 17 00:00:00 2001 From: David Arthur Date: Thu, 9 Mar 2023 13:45:45 -0500 Subject: [PATCH 05/12] Use mutable set --- .../apache/kafka/metadata/migration/KRaftMigrationDriver.java | 4 ++-- 1 file changed, 2 insertions(+), 2 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 729256575510d..64b308815ddf4 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 @@ -149,14 +149,14 @@ private boolean areZkBrokersReadyForMigration() { } // First check the brokers registered in ZK - Set zkBrokerRegistrations = zkMigrationClient.readBrokerIds(); + Set zkBrokerRegistrations = new HashSet<>(zkMigrationClient.readBrokerIds()); if (imageDoesNotContainAllBrokers(image, zkBrokerRegistrations)) { log.info("Still waiting for ZK brokers {} to register with KRaft.", zkBrokerRegistrations); return false; } // Once all of those are found, check the topic assignments. This is much more expensive than listing /brokers - Set zkBrokersWithAssignments = zkMigrationClient.readBrokerIdsFromTopicAssignments(); + Set zkBrokersWithAssignments = new HashSet<>(zkMigrationClient.readBrokerIdsFromTopicAssignments()); if (imageDoesNotContainAllBrokers(image, zkBrokersWithAssignments)) { log.info("Still waiting for ZK brokers {} to register with KRaft.", zkBrokersWithAssignments); return false; From 9ffb90847a468edd8cb46681b4c4b5221b768469 Mon Sep 17 00:00:00 2001 From: David Arthur Date: Fri, 10 Mar 2023 12:18:08 -0500 Subject: [PATCH 06/12] fix flaky test failure --- .../scala/integration/kafka/zk/ZkMigrationIntegrationTest.scala | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/core/src/test/scala/integration/kafka/zk/ZkMigrationIntegrationTest.scala b/core/src/test/scala/integration/kafka/zk/ZkMigrationIntegrationTest.scala index 231bf01c91743..38466354de866 100644 --- a/core/src/test/scala/integration/kafka/zk/ZkMigrationIntegrationTest.scala +++ b/core/src/test/scala/integration/kafka/zk/ZkMigrationIntegrationTest.scala @@ -85,7 +85,7 @@ class ZkMigrationIntegrationTest { quotas.add(new ClientQuotaAlteration( new ClientQuotaEntity(Map("ip" -> "8.8.8.8").asJava), List(new ClientQuotaAlteration.Op("connection_creation_rate", 10.0)).asJava)) - admin.alterClientQuotas(quotas) + admin.alterClientQuotas(quotas).all().get(60, TimeUnit.SECONDS) val zkClient = clusterInstance.asInstanceOf[ZkClusterInstance].getUnderlying().zkClient val migrationClient = new ZkMigrationClient(zkClient) From fde17371593dedc4871de6aa845dcfdd463a8e7f Mon Sep 17 00:00:00 2001 From: David Arthur Date: Mon, 13 Mar 2023 13:06:40 -0400 Subject: [PATCH 07/12] Update to new method name --- .../kafka/metadata/migration/KRaftMigrationDriverTest.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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 ce8d081af35af..3e233721162be 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 @@ -352,7 +352,7 @@ public ZkMigrationLeadershipState claimControllerLeadership(ZkMigrationLeadershi image = delta.apply(provenance); // Publish a delta with this node (3000) as the leader - driver.publishLogDelta(delta, image, new LogDeltaManifest(provenance, + driver.onMetadataUpdate(delta, image, new LogDeltaManifest(provenance, new LeaderAndEpoch(OptionalInt.of(3000), 1), 1, 100, 42)); Assertions.assertTrue(claimLeaderAttempts.await(1, TimeUnit.MINUTES)); From 86bb0e4973726a6b815067875f2d201dcc0c832f Mon Sep 17 00:00:00 2001 From: David Arthur Date: Tue, 14 Mar 2023 11:12:56 -0400 Subject: [PATCH 08/12] PR feedback --- .../scala/kafka/zk/ZkMigrationClient.scala | 29 ++++++++++++++----- .../migration/KRaftMigrationDriver.java | 13 +++++---- 2 files changed, 28 insertions(+), 14 deletions(-) diff --git a/core/src/main/scala/kafka/zk/ZkMigrationClient.scala b/core/src/main/scala/kafka/zk/ZkMigrationClient.scala index 0f25f2a6d2e7a..286a04bd21eec 100644 --- a/core/src/main/scala/kafka/zk/ZkMigrationClient.scala +++ b/core/src/main/scala/kafka/zk/ZkMigrationClient.scala @@ -42,15 +42,22 @@ import scala.jdk.CollectionConverters._ /** * Migration client in KRaft controller responsible for handling communication to Zookeeper and - * the ZkBrokers present in the cluster. + * the ZkBrokers present in the cluster. Methods that directly use KafkaZkClient should use the wrapZkException + * wrapper function in order to translate KeeperExceptions into something usable by the caller. */ class ZkMigrationClient(zkClient: KafkaZkClient) extends MigrationClient with Logging { + /** + * Wrap a function such that any KeeperExceptions is captured and converted to a MigrationClientException. + * Any authentication related exception is converted to a MigrationClientAuthException which may be treated + * differently by the caller. + */ @throws(classOf[MigrationClientException]) - def wrapZkException[T](fn: => T): T = { + private def wrapZkException[T](fn: => T): T = { try { fn } catch { + case e @ (_: MigrationClientException | _: MigrationClientAuthException) => throw e case e @ (_: AuthFailedException | _: NoAuthException | _: SessionClosedRequireAuthException) => // We don't expect authentication errors to be recoverable, so treat them differently throw new MigrationClientAuthException(e) @@ -99,7 +106,7 @@ class ZkMigrationClient(zkClient: KafkaZkClient) extends MigrationClient with Lo def migrateTopics( recordConsumer: Consumer[util.List[ApiMessageAndVersion]], brokerIdConsumer: Consumer[Integer] - ): Unit = { + ): Unit = wrapZkException { val topics = zkClient.getAllTopicsInCluster() val topicConfigs = zkClient.getEntitiesConfigs(ConfigType.Topic, topics) val replicaAssignmentAndTopicIds = zkClient.getReplicaAssignmentAndTopicIdForTopics(topics) @@ -153,7 +160,9 @@ class ZkMigrationClient(zkClient: KafkaZkClient) extends MigrationClient with Lo } } - def migrateBrokerConfigs(recordConsumer: Consumer[util.List[ApiMessageAndVersion]]): Unit = { + def migrateBrokerConfigs( + recordConsumer: Consumer[util.List[ApiMessageAndVersion]] + ): Unit = wrapZkException { val brokerEntities = zkClient.getAllEntitiesWithConfig(ConfigType.Broker) val batch = new util.ArrayList[ApiMessageAndVersion]() zkClient.getEntitiesConfigs(ConfigType.Broker, brokerEntities.toSet).foreach { case (broker, props) => @@ -175,7 +184,9 @@ class ZkMigrationClient(zkClient: KafkaZkClient) extends MigrationClient with Lo } } - def migrateClientQuotas(recordConsumer: Consumer[util.List[ApiMessageAndVersion]]): Unit = { + def migrateClientQuotas( + recordConsumer: Consumer[util.List[ApiMessageAndVersion]] + ): Unit = wrapZkException { val adminZkClient = new AdminZkClient(zkClient) def migrateEntityType(entityType: String): Unit = { @@ -217,7 +228,9 @@ class ZkMigrationClient(zkClient: KafkaZkClient) extends MigrationClient with Lo migrateEntityType(ConfigType.Ip) } - def migrateProducerId(recordConsumer: Consumer[util.List[ApiMessageAndVersion]]): Unit = { + def migrateProducerId( + recordConsumer: Consumer[util.List[ApiMessageAndVersion]] + ): Unit = wrapZkException { val (dataOpt, _) = zkClient.getDataAndVersion(ProducerIdBlockZNode.path) dataOpt match { case Some(data) => @@ -233,7 +246,7 @@ class ZkMigrationClient(zkClient: KafkaZkClient) extends MigrationClient with Lo override def readAllMetadata( batchConsumer: Consumer[util.List[ApiMessageAndVersion]], brokerIdConsumer: Consumer[Integer] - ): Unit = wrapZkException { + ): Unit = { migrateTopics(batchConsumer, brokerIdConsumer) migrateBrokerConfigs(batchConsumer) migrateClientQuotas(batchConsumer) @@ -241,7 +254,7 @@ class ZkMigrationClient(zkClient: KafkaZkClient) extends MigrationClient with Lo } override def readBrokerIds(): util.Set[Integer] = wrapZkException { - zkClient.getSortedBrokerList.map(Integer.valueOf).toSet.asJava + zkClient.getSortedBrokerList.map(Integer.valueOf).to(collection.mutable.Set).asJava } override def readBrokerIdsFromTopicAssignments(): util.Set[Integer] = wrapZkException { 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 451126c1ec0af..4ef07ab6789c2 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 @@ -90,10 +90,11 @@ public KRaftMigrationDriver( this.zkMigrationClient = zkMigrationClient; this.propagator = propagator; this.time = Time.SYSTEM; - this.log = LoggerFactory.getLogger(KRaftMigrationDriver.class); + LogContext logContext = new LogContext(String.format("[KRaftMigrationDriver nodeId=%d] ", nodeId)); + this.log = logContext.logger(KRaftMigrationDriver.class); this.migrationState = MigrationDriverState.UNINITIALIZED; this.migrationLeadershipState = ZkMigrationLeadershipState.EMPTY; - this.eventQueue = new KafkaEventQueue(Time.SYSTEM, new LogContext("KRaftMigrationDriver "), "kraft-migration"); + this.eventQueue = new KafkaEventQueue(Time.SYSTEM, logContext, "kraft-migration"); this.image = MetadataImage.EMPTY; this.leaderAndEpoch = LeaderAndEpoch.UNKNOWN; this.initialZkLoadHandler = initialZkLoadHandler; @@ -149,14 +150,14 @@ private boolean areZkBrokersReadyForMigration() { } // First check the brokers registered in ZK - Set zkBrokerRegistrations = new HashSet<>(zkMigrationClient.readBrokerIds()); + Set zkBrokerRegistrations = zkMigrationClient.readBrokerIds(); if (imageDoesNotContainAllBrokers(image, zkBrokerRegistrations)) { log.info("Still waiting for ZK brokers {} to register with KRaft.", zkBrokerRegistrations); return false; } // Once all of those are found, check the topic assignments. This is much more expensive than listing /brokers - Set zkBrokersWithAssignments = new HashSet<>(zkMigrationClient.readBrokerIdsFromTopicAssignments()); + Set zkBrokersWithAssignments = zkMigrationClient.readBrokerIdsFromTopicAssignments(); if (imageDoesNotContainAllBrokers(image, zkBrokersWithAssignments)) { log.info("Still waiting for ZK brokers {} to register with KRaft.", zkBrokersWithAssignments); return false; @@ -292,9 +293,9 @@ abstract class MigrationEvent implements EventQueue.Event { @Override public void handleException(Throwable e) { if (e instanceof MigrationClientAuthException) { - KRaftMigrationDriver.this.faultHandler.handleFault("Encountered client auth error in " + this, e); + KRaftMigrationDriver.this.faultHandler.handleFault("Encountered ZooKeeper authentication in " + this, e); } else if (e instanceof MigrationClientException) { - log.debug(String.format("Encountered client error during event %s. Will retry.", this), e.getCause()); + log.info(String.format("Encountered ZooKeeper error during event %s. Will retry.", this), e.getCause()); } else if (e instanceof RejectedExecutionException) { log.debug("Not processing {} because the event queue is closed.", this); } else { From fa4f10fb04e3a8b1a876881be4ca5562b832c85f Mon Sep 17 00:00:00 2001 From: David Arthur Date: Tue, 14 Mar 2023 11:17:39 -0400 Subject: [PATCH 09/12] checkstyle --- .../apache/kafka/metadata/migration/KRaftMigrationDriver.java | 1 - 1 file changed, 1 deletion(-) 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 4ef07ab6789c2..c6b1c0fc26877 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 @@ -34,7 +34,6 @@ import org.apache.kafka.server.common.ApiMessageAndVersion; import org.apache.kafka.server.fault.FaultHandler; import org.slf4j.Logger; -import org.slf4j.LoggerFactory; import java.util.Collection; import java.util.Collections; From 95d5675b14bce7470a49b4156a1a464cb6962877 Mon Sep 17 00:00:00 2001 From: David Arthur Date: Wed, 15 Mar 2023 09:42:07 -0400 Subject: [PATCH 10/12] Fix Scala 2.12 build --- core/src/main/scala/kafka/zk/ZkMigrationClient.scala | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/core/src/main/scala/kafka/zk/ZkMigrationClient.scala b/core/src/main/scala/kafka/zk/ZkMigrationClient.scala index 286a04bd21eec..6c81d37f68216 100644 --- a/core/src/main/scala/kafka/zk/ZkMigrationClient.scala +++ b/core/src/main/scala/kafka/zk/ZkMigrationClient.scala @@ -254,7 +254,7 @@ class ZkMigrationClient(zkClient: KafkaZkClient) extends MigrationClient with Lo } override def readBrokerIds(): util.Set[Integer] = wrapZkException { - zkClient.getSortedBrokerList.map(Integer.valueOf).to(collection.mutable.Set).asJava + new util.HashSet[Integer](zkClient.getSortedBrokerList.map(Integer.valueOf).toSet.asJava) } override def readBrokerIdsFromTopicAssignments(): util.Set[Integer] = wrapZkException { From 230a9171e733deb945528b431d3abbf40532c5df Mon Sep 17 00:00:00 2001 From: David Arthur Date: Wed, 15 Mar 2023 10:41:42 -0400 Subject: [PATCH 11/12] Add timeout to ZkIntegrationTest --- .../integration/kafka/zk/ZkMigrationIntegrationTest.scala | 3 +++ 1 file changed, 3 insertions(+) diff --git a/core/src/test/scala/integration/kafka/zk/ZkMigrationIntegrationTest.scala b/core/src/test/scala/integration/kafka/zk/ZkMigrationIntegrationTest.scala index 38466354de866..167420a7d9ada 100644 --- a/core/src/test/scala/integration/kafka/zk/ZkMigrationIntegrationTest.scala +++ b/core/src/test/scala/integration/kafka/zk/ZkMigrationIntegrationTest.scala @@ -34,6 +34,7 @@ import org.apache.kafka.metadata.migration.ZkMigrationLeadershipState import org.apache.kafka.raft.RaftConfig import org.apache.kafka.server.common.{ApiMessageAndVersion, MetadataVersion, ProducerIdsBlock} import org.junit.jupiter.api.Assertions.{assertEquals, assertFalse, assertNotNull, assertTrue} +import org.junit.jupiter.api.Timeout import org.junit.jupiter.api.extension.ExtendWith import org.slf4j.LoggerFactory @@ -43,7 +44,9 @@ import java.util.concurrent.TimeUnit import scala.collection.Seq import scala.jdk.CollectionConverters._ + @ExtendWith(value = Array(classOf[ClusterTestExtensions])) +@Timeout(120) class ZkMigrationIntegrationTest { val log = LoggerFactory.getLogger(classOf[ZkMigrationIntegrationTest]) From 1d4f60e90c1f96ca5d639ca41dfb3bd777b51044 Mon Sep 17 00:00:00 2001 From: David Arthur Date: Thu, 16 Mar 2023 09:34:44 -0400 Subject: [PATCH 12/12] Fix KRaftMigrationDriverTest --- .../kafka/metadata/migration/KRaftMigrationDriverTest.java | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) 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 3e233721162be..18645996c3854 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 @@ -351,10 +351,11 @@ public ZkMigrationLeadershipState claimControllerLeadership(ZkMigrationLeadershi MetadataProvenance provenance = new MetadataProvenance(100, 1, 1); image = delta.apply(provenance); - // Publish a delta with this node (3000) as the leader + // 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, new LogDeltaManifest(provenance, new LeaderAndEpoch(OptionalInt.of(3000), 1), 1, 100, 42)); - Assertions.assertTrue(claimLeaderAttempts.await(1, TimeUnit.MINUTES)); TestUtils.waitForCondition(() -> driver.migrationState().get(1, TimeUnit.MINUTES).equals(MigrationDriverState.ZK_MIGRATION), "Waiting for KRaftMigrationDriver to enter ZK_MIGRATION state");