From e0c86a8e835d98fa3b94c248ad314f11bf3605f0 Mon Sep 17 00:00:00 2001 From: "Colin P. Mccabe" Date: Thu, 13 May 2021 12:41:39 -0700 Subject: [PATCH] KAFKA-12778: Fix QuorumController request timeouts and electLeaders Not all RPC requests to the quorum controller include a timeout, but we should honor the timeouts that do exist. For electLeaders, attempt to trigger a leader election for all partitions when the request specifies null for the topics argument. Add some unit tests for the above. --- checkstyle/suppressions.xml | 2 +- .../scala/kafka/server/ControllerApis.scala | 15 ++- .../test/java/kafka/test/MockController.java | 8 +- .../apache/kafka/controller/Controller.java | 23 +++- .../kafka/controller/QuorumController.java | 87 ++++++++++---- .../controller/ReplicationControlManager.java | 44 +++++-- .../controller/QuorumControllerTest.java | 110 ++++++++++++++++++ 7 files changed, 241 insertions(+), 48 deletions(-) diff --git a/checkstyle/suppressions.xml b/checkstyle/suppressions.xml index 03ba807e6d025..21cd5f4cc5ebc 100644 --- a/checkstyle/suppressions.xml +++ b/checkstyle/suppressions.xml @@ -270,7 +270,7 @@ + files="(QuorumControllerTest|ReplicationControlManager|ReplicationControlManagerTest).java"/> + controller.findTopicNames(deadlineNs, providedIds).thenCompose { topicNames => topicNames.forEach { (id, nameOrError) => if (nameOrError.isError) { appendResponse(null, id, nameOrError.error()) @@ -291,7 +293,7 @@ class ControllerApis(val requestChannel: RequestChannel, } // For each topic that was provided by name, check if authentication failed. // If so, create an error response for it. Otherwise, add it to the idToName map. - controller.findTopicIds(providedNames).thenCompose { topicIds => + controller.findTopicIds(deadlineNs, providedNames).thenCompose { topicIds => topicIds.forEach { (name, idOrError) => if (!describeable.contains(name)) { appendResponse(name, ZERO_UUID, new ApiError(TOPIC_AUTHORIZATION_FAILED)) @@ -315,7 +317,7 @@ class ControllerApis(val requestChannel: RequestChannel, } // Finally, the idToName map contains all the topics that we are authorized to delete. // Perform the deletion and create responses for each one. - controller.deleteTopics(idToName.keySet).thenApply { idToError => + controller.deleteTopics(deadlineNs, idToName.keySet).thenApply { idToError => idToError.forEach { (id, error) => appendResponse(idToName.get(id), id, error) } @@ -706,6 +708,7 @@ class ControllerApis(val requestChannel: RequestChannel, hasClusterAuth: Boolean, getCreatableTopics: Iterable[String] => Set[String]) : CompletableFuture[util.List[CreatePartitionsTopicResult]] = { + val deadlineNs = time.nanoseconds() + NANOSECONDS.convert(request.timeoutMs, MILLISECONDS); val responses = new util.ArrayList[CreatePartitionsTopicResult]() val duplicateTopicNames = new util.HashSet[String]() val topicNames = new util.HashSet[String]() @@ -739,7 +742,7 @@ class ControllerApis(val requestChannel: RequestChannel, setErrorCode(TOPIC_AUTHORIZATION_FAILED.code)) } } - controller.createPartitions(topics).thenApply { results => + controller.createPartitions(deadlineNs, topics).thenApply { results => results.forEach(response => responses.add(response)) responses } @@ -750,7 +753,7 @@ class ControllerApis(val requestChannel: RequestChannel, authHelper.authorizeClusterOperation(request, ALTER) val response = controller.alterPartitionReassignments(alterRequest.data()).get() requestHelper.sendResponseMaybeThrottle(request, requestThrottleMs => - new AlterPartitionReassignmentsResponse(response)) + new AlterPartitionReassignmentsResponse(response.setThrottleTimeMs(requestThrottleMs))) } def handleListPartitionReassignments(request: RequestChannel.Request): Unit = { @@ -758,6 +761,6 @@ class ControllerApis(val requestChannel: RequestChannel, authHelper.authorizeClusterOperation(request, DESCRIBE) val response = controller.listPartitionReassignments(listRequest.data()).get() requestHelper.sendResponseMaybeThrottle(request, requestThrottleMs => - new ListPartitionReassignmentsResponse(response)) + new ListPartitionReassignmentsResponse(response.setThrottleTimeMs(requestThrottleMs))) } } diff --git a/core/src/test/java/kafka/test/MockController.java b/core/src/test/java/kafka/test/MockController.java index fc14145e28037..1fba2952165b0 100644 --- a/core/src/test/java/kafka/test/MockController.java +++ b/core/src/test/java/kafka/test/MockController.java @@ -144,7 +144,7 @@ static class MockTopic { @Override synchronized public CompletableFuture>> - findTopicIds(Collection topicNames) { + findTopicIds(long deadlineNs, Collection topicNames) { Map> results = new HashMap<>(); for (String topicName : topicNames) { if (!topicNameToId.containsKey(topicName)) { @@ -158,7 +158,7 @@ static class MockTopic { @Override synchronized public CompletableFuture>> - findTopicNames(Collection topicIds) { + findTopicNames(long deadlineNs, Collection topicIds) { Map> results = new HashMap<>(); for (Uuid topicId : topicIds) { MockTopic topic = topics.get(topicId); @@ -173,7 +173,7 @@ static class MockTopic { @Override synchronized public CompletableFuture> - deleteTopics(Collection topicIds) { + deleteTopics(long deadlineNs, Collection topicIds) { if (!active) { CompletableFuture> future = new CompletableFuture<>(); future.completeExceptionally(NOT_CONTROLLER_EXCEPTION); @@ -303,7 +303,7 @@ public CompletableFuture waitForReadyBrokers(int minBrokers) { @Override synchronized public CompletableFuture> - createPartitions(List topicList) { + createPartitions(long deadlineNs, List topicList) { if (!active) { CompletableFuture> future = new CompletableFuture<>(); future.completeExceptionally(NOT_CONTROLLER_EXCEPTION); diff --git a/metadata/src/main/java/org/apache/kafka/controller/Controller.java b/metadata/src/main/java/org/apache/kafka/controller/Controller.java index 071076474dfea..a34b084ea1302 100644 --- a/metadata/src/main/java/org/apache/kafka/controller/Controller.java +++ b/metadata/src/main/java/org/apache/kafka/controller/Controller.java @@ -80,27 +80,36 @@ public interface Controller extends AutoCloseable { /** * Find the ids for topic names. * + * @param deadlineNs The time by which this operation needs to be complete, before + * we will complete this operation with a timeout. * @param topicNames The topic names to resolve. * @return A future yielding a map from topic name to id. */ - CompletableFuture>> findTopicIds(Collection topicNames); + CompletableFuture>> findTopicIds(long deadlineNs, + Collection topicNames); /** * Find the names for topic ids. * + * @param deadlineNs The time by which this operation needs to be complete, before + * we will complete this operation with a timeout. * @param topicIds The topic ids to resolve. * @return A future yielding a map from topic id to name. */ - CompletableFuture>> findTopicNames(Collection topicIds); + CompletableFuture>> findTopicNames(long deadlineNs, + Collection topicIds); /** * Delete a batch of topics. * + * @param deadlineNs The time by which this operation needs to be complete, before + * we will complete this operation with a timeout. * @param topicIds The IDs of the topics to delete. * * @return A future yielding the response. */ - CompletableFuture> deleteTopics(Collection topicIds); + CompletableFuture> deleteTopics(long deadlineNs, + Collection topicIds); /** * Describe the current configuration of various resources. @@ -225,11 +234,13 @@ CompletableFuture> alterClientQuotas( /** * Create partitions on certain topics. * - * @param topics The list of topics to create partitions for. - * @return A future yielding per-topic results. + * @param deadlineNs The time by which this operation needs to be complete, before + * we will complete this operation with a timeout. + * @param topics The list of topics to create partitions for. + * @return A future yielding per-topic results. */ CompletableFuture> - createPartitions(List topics); + createPartitions(long deadlineNs, List topics); /** * Begin shutting down, but don't block. You must still call close to clean up all diff --git a/metadata/src/main/java/org/apache/kafka/controller/QuorumController.java b/metadata/src/main/java/org/apache/kafka/controller/QuorumController.java index 7995b4b5134ea..5cb5efc20f345 100644 --- a/metadata/src/main/java/org/apache/kafka/controller/QuorumController.java +++ b/metadata/src/main/java/org/apache/kafka/controller/QuorumController.java @@ -27,6 +27,7 @@ import java.util.OptionalLong; import java.util.Random; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CountDownLatch; import java.util.concurrent.RejectedExecutionException; import java.util.concurrent.TimeUnit; import java.util.function.Function; @@ -89,6 +90,7 @@ import org.slf4j.Logger; import static java.util.concurrent.TimeUnit.MICROSECONDS; +import static java.util.concurrent.TimeUnit.MILLISECONDS; import static java.util.concurrent.TimeUnit.NANOSECONDS; @@ -475,6 +477,12 @@ CompletableFuture appendReadEvent(String name, Supplier handler) { return event.future(); } + CompletableFuture appendReadEvent(String name, long deadlineNs, Supplier handler) { + ControllerReadEvent event = new ControllerReadEvent(name, handler); + queue.appendWithDeadline(deadlineNs, event); + return event.future(); + } + interface ControllerWriteOperation { /** * Generate the metadata records needed to implement this controller write @@ -602,11 +610,10 @@ public String toString() { } private CompletableFuture appendWriteEvent(String name, - long timeoutMs, + long deadlineNs, ControllerWriteOperation op) { ControllerWriteEvent event = new ControllerWriteEvent<>(name, op); - queue.appendWithDeadline(time.nanoseconds() + - NANOSECONDS.convert(timeoutMs, TimeUnit.MILLISECONDS), event); + queue.appendWithDeadline(deadlineNs, event); return event.future(); } @@ -961,8 +968,9 @@ public CompletableFuture alterIsr(AlterIsrRequestData requ if (request.topics().isEmpty()) { return CompletableFuture.completedFuture(new CreateTopicsResponseData()); } - return appendWriteEvent("createTopics", () -> - replicationControl.createTopics(request)); + return appendWriteEvent("createTopics", + time.nanoseconds() + NANOSECONDS.convert(request.timeoutMs(), MILLISECONDS), + () -> replicationControl.createTopics(request)); } @Override @@ -972,23 +980,26 @@ public CompletableFuture unregisterBroker(int brokerId) { } @Override - public CompletableFuture>> findTopicIds(Collection names) { + public CompletableFuture>> findTopicIds(long deadlineNs, + Collection names) { if (names.isEmpty()) return CompletableFuture.completedFuture(Collections.emptyMap()); - return appendReadEvent("findTopicIds", + return appendReadEvent("findTopicIds", deadlineNs, () -> replicationControl.findTopicIds(lastCommittedOffset, names)); } @Override - public CompletableFuture>> findTopicNames(Collection ids) { + public CompletableFuture>> findTopicNames(long deadlineNs, + Collection ids) { if (ids.isEmpty()) return CompletableFuture.completedFuture(Collections.emptyMap()); - return appendReadEvent("findTopicNames", + return appendReadEvent("findTopicNames", deadlineNs, () -> replicationControl.findTopicNames(lastCommittedOffset, ids)); } @Override - public CompletableFuture> deleteTopics(Collection ids) { + public CompletableFuture> deleteTopics(long deadlineNs, + Collection ids) { if (ids.isEmpty()) return CompletableFuture.completedFuture(Collections.emptyMap()); - return appendWriteEvent("deleteTopics", + return appendWriteEvent("deleteTopics", deadlineNs, () -> replicationControl.deleteTopics(ids)); } @@ -1002,7 +1013,13 @@ public CompletableFuture> deleteTopics(Collection ids) @Override public CompletableFuture electLeaders(ElectLeadersRequestData request) { - return appendWriteEvent("electLeaders", request.timeoutMs(), + // If topicPartitions is null, we will try to trigger a new leader election on + // all partitions (!). But if it's empty, there is nothing to do. + if (request.topicPartitions() != null && request.topicPartitions().isEmpty()) { + return CompletableFuture.completedFuture(new ElectLeadersResponseData()); + } + return appendWriteEvent("electLeaders", + time.nanoseconds() + NANOSECONDS.convert(request.timeoutMs(), MILLISECONDS), () -> replicationControl.electLeaders(request)); } @@ -1016,6 +1033,9 @@ public CompletableFuture finalizedFeatures() { public CompletableFuture> incrementalAlterConfigs( Map>> configChanges, boolean validateOnly) { + if (configChanges.isEmpty()) { + return CompletableFuture.completedFuture(Collections.emptyMap()); + } return appendWriteEvent("incrementalAlterConfigs", () -> { ControllerResult> result = configurationControl.incrementalAlterConfigs(configChanges); @@ -1030,17 +1050,24 @@ public CompletableFuture> incrementalAlterConfigs( @Override public CompletableFuture alterPartitionReassignments(AlterPartitionReassignmentsRequestData request) { - CompletableFuture future = new CompletableFuture<>(); - future.completeExceptionally(new UnsupportedOperationException()); - return future; + if (request.topics().isEmpty()) { + return CompletableFuture.completedFuture(new AlterPartitionReassignmentsResponseData()); + } + return appendWriteEvent("alterPartitionReassignments", + time.nanoseconds() + NANOSECONDS.convert(request.timeoutMs(), MILLISECONDS), + () -> { + throw new UnsupportedOperationException(); + }); } @Override public CompletableFuture listPartitionReassignments(ListPartitionReassignmentsRequestData request) { - CompletableFuture future = new CompletableFuture<>(); - future.completeExceptionally(new UnsupportedOperationException()); - return future; + return appendReadEvent("listPartitionReassignments", + time.nanoseconds() + NANOSECONDS.convert(request.timeoutMs(), MILLISECONDS), + () -> { + throw new UnsupportedOperationException(); + }); } @Override @@ -1118,12 +1145,12 @@ public CompletableFuture> alterClientQuotas( @Override public CompletableFuture> - createPartitions(List topics) { + createPartitions(long deadlineNs, List topics) { if (topics.isEmpty()) { return CompletableFuture.completedFuture(Collections.emptyList()); } - return appendWriteEvent("createPartitions", () -> - replicationControl.createPartitions(topics)); + return appendWriteEvent("createPartitions", deadlineNs, + () -> replicationControl.createPartitions(topics)); } @Override @@ -1165,4 +1192,22 @@ public long curClaimEpoch() { public void close() throws InterruptedException { queue.close(); } + + // VisibleForTesting + CountDownLatch pause() { + final CountDownLatch latch = new CountDownLatch(1); + appendControlEvent("pause", () -> { + try { + latch.await(); + } catch (InterruptedException e) { + log.info("Interrupted while waiting for unpause.", e); + } + }); + return latch; + } + + // VisibleForTesting + Time time() { + return time; + } } diff --git a/metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java b/metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java index 48ab1c92a44a5..cba9ae5c01b6a 100644 --- a/metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java +++ b/metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java @@ -800,16 +800,40 @@ ControllerResult electLeaders(ElectLeadersRequestData boolean uncleanOk = electionTypeIsUnclean(request.electionType()); List records = new ArrayList<>(); ElectLeadersResponseData response = new ElectLeadersResponseData(); - for (TopicPartitions topic : request.topicPartitions()) { - ReplicaElectionResult topicResults = - new ReplicaElectionResult().setTopic(topic.topic()); - response.replicaElectionResults().add(topicResults); - for (int partitionId : topic.partitions()) { - ApiError error = electLeader(topic.topic(), partitionId, uncleanOk, records); - topicResults.partitionResult().add(new PartitionResult(). - setPartitionId(partitionId). - setErrorCode(error.error().code()). - setErrorMessage(error.message())); + if (request.topicPartitions() == null) { + // If topicPartitions is null, we try to elect a new leader for every partition. There + // are some obvious issues with this wire protocol. For example, what if we have too + // many partitions to fit the results in a single RPC? This behavior should probably be + // removed from the protocol. For now, however, we have to implement this for + // compatibility with the old controller. + for (Entry topicEntry : topicsByName.entrySet()) { + String topicName = topicEntry.getKey(); + ReplicaElectionResult topicResults = + new ReplicaElectionResult().setTopic(topicName); + response.replicaElectionResults().add(topicResults); + TopicControlInfo topic = topics.get(topicEntry.getValue()); + if (topic != null) { + for (int partitionId : topic.parts.keySet()) { + ApiError error = electLeader(topicName, partitionId, uncleanOk, records); + topicResults.partitionResult().add(new PartitionResult(). + setPartitionId(partitionId). + setErrorCode(error.error().code()). + setErrorMessage(error.message())); + } + } + } + } else { + for (TopicPartitions topic : request.topicPartitions()) { + ReplicaElectionResult topicResults = + new ReplicaElectionResult().setTopic(topic.topic()); + response.replicaElectionResults().add(topicResults); + for (int partitionId : topic.partitions()) { + ApiError error = electLeader(topic.topic(), partitionId, uncleanOk, records); + topicResults.partitionResult().add(new PartitionResult(). + setPartitionId(partitionId). + setErrorCode(error.error().code()). + setErrorMessage(error.message())); + } } } return ControllerResult.of(records, response); diff --git a/metadata/src/test/java/org/apache/kafka/controller/QuorumControllerTest.java b/metadata/src/test/java/org/apache/kafka/controller/QuorumControllerTest.java index b6eea362c9024..5e5ca01707ddd 100644 --- a/metadata/src/test/java/org/apache/kafka/controller/QuorumControllerTest.java +++ b/metadata/src/test/java/org/apache/kafka/controller/QuorumControllerTest.java @@ -24,21 +24,34 @@ import java.util.List; import java.util.Map; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.Future; import java.util.concurrent.LinkedBlockingDeque; import java.util.function.Function; import org.apache.kafka.common.Uuid; import org.apache.kafka.common.config.ConfigResource; +import org.apache.kafka.common.errors.TimeoutException; +import org.apache.kafka.common.message.AlterPartitionReassignmentsRequestData; +import org.apache.kafka.common.message.AlterPartitionReassignmentsRequestData.ReassignableTopic; +import org.apache.kafka.common.message.AlterPartitionReassignmentsResponseData; import org.apache.kafka.common.message.BrokerHeartbeatRequestData; import org.apache.kafka.common.message.BrokerRegistrationRequestData.Listener; import org.apache.kafka.common.message.BrokerRegistrationRequestData.ListenerCollection; import org.apache.kafka.common.message.BrokerRegistrationRequestData; +import org.apache.kafka.common.message.CreatePartitionsRequestData.CreatePartitionsTopic; +import org.apache.kafka.common.message.CreatePartitionsResponseData.CreatePartitionsTopicResult; import org.apache.kafka.common.message.CreateTopicsRequestData.CreatableReplicaAssignment; import org.apache.kafka.common.message.CreateTopicsRequestData.CreatableReplicaAssignmentCollection; import org.apache.kafka.common.message.CreateTopicsRequestData.CreatableTopic; import org.apache.kafka.common.message.CreateTopicsRequestData.CreatableTopicCollection; import org.apache.kafka.common.message.CreateTopicsRequestData; import org.apache.kafka.common.message.CreateTopicsResponseData; +import org.apache.kafka.common.message.ElectLeadersRequestData; +import org.apache.kafka.common.message.ElectLeadersResponseData; +import org.apache.kafka.common.message.ListPartitionReassignmentsRequestData; +import org.apache.kafka.common.message.ListPartitionReassignmentsResponseData; import org.apache.kafka.common.metadata.PartitionRecord; import org.apache.kafka.common.metadata.RegisterBrokerRecord; import org.apache.kafka.common.metadata.RegisterBrokerRecord.BrokerEndpoint; @@ -57,12 +70,14 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import static java.util.concurrent.TimeUnit.HOURS; import static org.apache.kafka.clients.admin.AlterConfigOp.OpType.SET; import static org.apache.kafka.controller.ConfigurationControlManagerTest.BROKER0; import static org.apache.kafka.controller.ConfigurationControlManagerTest.CONFIGS; import static org.apache.kafka.controller.ConfigurationControlManagerTest.entry; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; @@ -323,4 +338,99 @@ private void checkSnapshotContents(Uuid fooId, setRack(null), (short) 0))), iterator); } + + /** + * Test that certain controller operations time out if they stay on the controller + * queue for too long. + */ + @Test + public void testTimeouts() throws Throwable { + try (LocalLogManagerTestEnv logEnv = new LocalLogManagerTestEnv(1)) { + try (QuorumControllerTestEnv controlEnv = + new QuorumControllerTestEnv(logEnv, b -> b.setConfigDefs(CONFIGS))) { + QuorumController controller = controlEnv.activeController(); + CountDownLatch countDownLatch = controller.pause(); + CompletableFuture createFuture = + controller.createTopics(new CreateTopicsRequestData().setTimeoutMs(0). + setTopics(new CreatableTopicCollection(Collections.singleton( + new CreatableTopic().setName("foo")).iterator()))); + long now = controller.time().nanoseconds(); + CompletableFuture> deleteFuture = + controller.deleteTopics(now, Collections.singletonList(Uuid.ZERO_UUID)); + CompletableFuture>> findTopicIdsFuture = + controller.findTopicIds(now, Collections.singletonList("foo")); + CompletableFuture>> findTopicNamesFuture = + controller.findTopicNames(now, Collections.singletonList(Uuid.ZERO_UUID)); + CompletableFuture> createPartitionsFuture = + controller.createPartitions(now, Collections.singletonList( + new CreatePartitionsTopic())); + CompletableFuture electLeadersFuture = + controller.electLeaders(new ElectLeadersRequestData().setTimeoutMs(0). + setTopicPartitions(null)); + CompletableFuture alterReassignmentsFuture = + controller.alterPartitionReassignments( + new AlterPartitionReassignmentsRequestData().setTimeoutMs(0). + setTopics(Collections.singletonList(new ReassignableTopic()))); + CompletableFuture listReassignmentsFuture = + controller.listPartitionReassignments( + new ListPartitionReassignmentsRequestData().setTimeoutMs(0)); + while (controller.time().nanoseconds() == now) { + Thread.sleep(0, 10); + } + countDownLatch.countDown(); + assertYieldsTimeout(createFuture); + assertYieldsTimeout(deleteFuture); + assertYieldsTimeout(findTopicIdsFuture); + assertYieldsTimeout(findTopicNamesFuture); + assertYieldsTimeout(createPartitionsFuture); + assertYieldsTimeout(electLeadersFuture); + assertYieldsTimeout(alterReassignmentsFuture); + assertYieldsTimeout(listReassignmentsFuture); + } + } + } + + private static void assertYieldsTimeout(Future future) { + assertEquals(TimeoutException.class, assertThrows(ExecutionException.class, + () -> future.get()).getCause().getClass()); + } + + /** + * Test that certain controller operations finish immediately without putting an event + * on the controller queue, if there is nothing to do. + */ + @Test + public void testEarlyControllerResults() throws Throwable { + try (LocalLogManagerTestEnv logEnv = new LocalLogManagerTestEnv(1)) { + try (QuorumControllerTestEnv controlEnv = + new QuorumControllerTestEnv(logEnv, b -> b.setConfigDefs(CONFIGS))) { + QuorumController controller = controlEnv.activeController(); + CountDownLatch countDownLatch = controller.pause(); + CompletableFuture createFuture = + controller.createTopics(new CreateTopicsRequestData().setTimeoutMs(120000)); + long deadlineMs = controller.time().nanoseconds() + HOURS.toNanos(1); + CompletableFuture> deleteFuture = + controller.deleteTopics(deadlineMs, Collections.emptyList()); + CompletableFuture>> findTopicIdsFuture = + controller.findTopicIds(deadlineMs, Collections.emptyList()); + CompletableFuture>> findTopicNamesFuture = + controller.findTopicNames(deadlineMs, Collections.emptyList()); + CompletableFuture> createPartitionsFuture = + controller.createPartitions(deadlineMs, Collections.emptyList()); + CompletableFuture electLeadersFuture = + controller.electLeaders(new ElectLeadersRequestData().setTimeoutMs(120000)); + CompletableFuture alterReassignmentsFuture = + controller.alterPartitionReassignments( + new AlterPartitionReassignmentsRequestData().setTimeoutMs(12000)); + createFuture.get(); + deleteFuture.get(); + findTopicIdsFuture.get(); + findTopicNamesFuture.get(); + createPartitionsFuture.get(); + electLeadersFuture.get(); + alterReassignmentsFuture.get(); + countDownLatch.countDown(); + } + } + } }