KAFKA-12778: Fix QuorumController request timeouts and electLeaders (#10688)

The QuorumController should honor the timeout for RPC requests
which feature a timeout. For electLeaders, attempt to trigger a leader
election for all partitions when the request specifies null for the topics
argument.

Reviewers: David Arthur <[email protected]>

Author: cmccabe

checkstyle/suppressions.xml

@@ -270,7 +270,7 @@
 
     <!-- metadata -->
     <suppress checks="ClassDataAbstractionCoupling"
-              files="(ReplicationControlManager|ReplicationControlManagerTest).java"/>
+              files="(QuorumControllerTest|ReplicationControlManager|ReplicationControlManagerTest).java"/>
     <suppress checks="ClassFanOutComplexity"
               files="(QuorumController|ReplicationControlManager).java"/>
     <suppress checks="ParameterNumber"

core/src/main/scala/kafka/server/ControllerApis.scala

@@ -21,6 +21,7 @@ import java.util
 import java.util.Collections
 import java.util.Map.Entry
 import java.util.concurrent.{CompletableFuture, ExecutionException}
+import java.util.concurrent.TimeUnit.{MILLISECONDS, NANOSECONDS}
 import kafka.network.RequestChannel
 import kafka.raft.RaftManager
 import kafka.server.QuotaFactory.QuotaManagers
@@ -204,6 +205,7 @@ class ControllerApis(val requestChannel: RequestChannel,
         throw new TopicDeletionDisabledException()
       }
     }
+    val deadlineNs = time.nanoseconds() + NANOSECONDS.convert(request.timeoutMs, MILLISECONDS);
     // The first step is to load up the names and IDs that have been provided by the
     // request.  This is a bit messy because we support multiple ways of referring to
     // topics (both by name and by id) and because we need to check for duplicates or
@@ -256,7 +258,7 @@ class ControllerApis(val requestChannel: RequestChannel,
     val toAuthenticate = new util.HashSet[String]
     toAuthenticate.addAll(providedNames)
     val idToName = new util.HashMap[Uuid, String]
-    controller.findTopicNames(providedIds).thenCompose { topicNames =>
+    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,14 +753,14 @@ 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 = {
     val listRequest = request.body[ListPartitionReassignmentsRequest]
     authHelper.authorizeClusterOperation(request, DESCRIBE)
     val response = controller.listPartitionReassignments(listRequest.data()).get()
     requestHelper.sendResponseMaybeThrottle(request, requestThrottleMs =>
-      new ListPartitionReassignmentsResponse(response))
+      new ListPartitionReassignmentsResponse(response.setThrottleTimeMs(requestThrottleMs)))
   }
 }

core/src/test/java/kafka/test/MockController.java

@@ -144,7 +144,7 @@ static class MockTopic {
 
     @Override
     synchronized public CompletableFuture<Map<String, ResultOrError<Uuid>>>
-            findTopicIds(Collection<String> topicNames) {
+            findTopicIds(long deadlineNs, Collection<String> topicNames) {
         Map<String, ResultOrError<Uuid>> results = new HashMap<>();
         for (String topicName : topicNames) {
             if (!topicNameToId.containsKey(topicName)) {
@@ -158,7 +158,7 @@ static class MockTopic {
 
     @Override
     synchronized public CompletableFuture<Map<Uuid, ResultOrError<String>>>
-            findTopicNames(Collection<Uuid> topicIds) {
+            findTopicNames(long deadlineNs, Collection<Uuid> topicIds) {
         Map<Uuid, ResultOrError<String>> results = new HashMap<>();
         for (Uuid topicId : topicIds) {
             MockTopic topic = topics.get(topicId);
@@ -173,7 +173,7 @@ static class MockTopic {
 
     @Override
     synchronized public CompletableFuture<Map<Uuid, ApiError>>
-            deleteTopics(Collection<Uuid> topicIds) {
+            deleteTopics(long deadlineNs, Collection<Uuid> topicIds) {
         if (!active) {
             CompletableFuture<Map<Uuid, ApiError>> future = new CompletableFuture<>();
             future.completeExceptionally(NOT_CONTROLLER_EXCEPTION);
@@ -303,7 +303,7 @@ public CompletableFuture<Void> waitForReadyBrokers(int minBrokers) {
 
     @Override
     synchronized public CompletableFuture<List<CreatePartitionsTopicResult>>
-            createPartitions(List<CreatePartitionsTopic> topicList) {
+            createPartitions(long deadlineNs, List<CreatePartitionsTopic> topicList) {
         if (!active) {
             CompletableFuture<List<CreatePartitionsTopicResult>> future = new CompletableFuture<>();
             future.completeExceptionally(NOT_CONTROLLER_EXCEPTION);

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<Map<String, ResultOrError<Uuid>>> findTopicIds(Collection<String> topicNames);
+    CompletableFuture<Map<String, ResultOrError<Uuid>>> findTopicIds(long deadlineNs,
+                                                                     Collection<String> 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<Map<Uuid, ResultOrError<String>>> findTopicNames(Collection<Uuid> topicIds);
+    CompletableFuture<Map<Uuid, ResultOrError<String>>> findTopicNames(long deadlineNs,
+                                                                       Collection<Uuid> 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<Map<Uuid, ApiError>> deleteTopics(Collection<Uuid> topicIds);
+    CompletableFuture<Map<Uuid, ApiError>> deleteTopics(long deadlineNs,
+                                                        Collection<Uuid> topicIds);
 
     /**
      * Describe the current configuration of various resources.
@@ -225,11 +234,13 @@ CompletableFuture<Map<ClientQuotaEntity, ApiError>> 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<List<CreatePartitionsTopicResult>>
-            createPartitions(List<CreatePartitionsTopic> topics);
+            createPartitions(long deadlineNs, List<CreatePartitionsTopic> topics);
 
     /**
      * Begin shutting down, but don't block.  You must still call close to clean up all

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 @@ <T> CompletableFuture<T> appendReadEvent(String name, Supplier<T> handler) {
         return event.future();
     }
 
+    <T> CompletableFuture<T> appendReadEvent(String name, long deadlineNs, Supplier<T> handler) {
+        ControllerReadEvent<T> event = new ControllerReadEvent<T>(name, handler);
+        queue.appendWithDeadline(deadlineNs, event);
+        return event.future();
+    }
+
     interface ControllerWriteOperation<T> {
         /**
          * Generate the metadata records needed to implement this controller write
@@ -602,11 +610,10 @@ public String toString() {
     }
 
     private <T> CompletableFuture<T> appendWriteEvent(String name,
-                                                      long timeoutMs,
+                                                      long deadlineNs,
                                                       ControllerWriteOperation<T> op) {
         ControllerWriteEvent<T> 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<AlterIsrResponseData> 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<Void> unregisterBroker(int brokerId) {
     }
 
     @Override
-    public CompletableFuture<Map<String, ResultOrError<Uuid>>> findTopicIds(Collection<String> names) {
+    public CompletableFuture<Map<String, ResultOrError<Uuid>>> findTopicIds(long deadlineNs,
+                                                                            Collection<String> names) {
         if (names.isEmpty()) return CompletableFuture.completedFuture(Collections.emptyMap());
-        return appendReadEvent("findTopicIds",
+        return appendReadEvent("findTopicIds", deadlineNs,
             () -> replicationControl.findTopicIds(lastCommittedOffset, names));
     }
 
     @Override
-    public CompletableFuture<Map<Uuid, ResultOrError<String>>> findTopicNames(Collection<Uuid> ids) {
+    public CompletableFuture<Map<Uuid, ResultOrError<String>>> findTopicNames(long deadlineNs,
+                                                                              Collection<Uuid> ids) {
         if (ids.isEmpty()) return CompletableFuture.completedFuture(Collections.emptyMap());
-        return appendReadEvent("findTopicNames",
+        return appendReadEvent("findTopicNames", deadlineNs,
             () -> replicationControl.findTopicNames(lastCommittedOffset, ids));
     }
 
     @Override
-    public CompletableFuture<Map<Uuid, ApiError>> deleteTopics(Collection<Uuid> ids) {
+    public CompletableFuture<Map<Uuid, ApiError>> deleteTopics(long deadlineNs,
+                                                               Collection<Uuid> 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<Map<Uuid, ApiError>> deleteTopics(Collection<Uuid> ids)
     @Override
     public CompletableFuture<ElectLeadersResponseData>
             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<FeatureMapAndEpoch> finalizedFeatures() {
     public CompletableFuture<Map<ConfigResource, ApiError>> incrementalAlterConfigs(
         Map<ConfigResource, Map<String, Entry<OpType, String>>> configChanges,
         boolean validateOnly) {
+        if (configChanges.isEmpty()) {
+            return CompletableFuture.completedFuture(Collections.emptyMap());
+        }
         return appendWriteEvent("incrementalAlterConfigs", () -> {
             ControllerResult<Map<ConfigResource, ApiError>> result =
                 configurationControl.incrementalAlterConfigs(configChanges);
@@ -1030,17 +1050,24 @@ public CompletableFuture<Map<ConfigResource, ApiError>> incrementalAlterConfigs(
     @Override
     public CompletableFuture<AlterPartitionReassignmentsResponseData>
             alterPartitionReassignments(AlterPartitionReassignmentsRequestData request) {
-        CompletableFuture<AlterPartitionReassignmentsResponseData> 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<ListPartitionReassignmentsResponseData>
             listPartitionReassignments(ListPartitionReassignmentsRequestData request) {
-        CompletableFuture<ListPartitionReassignmentsResponseData> 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<Map<ClientQuotaEntity, ApiError>> alterClientQuotas(
 
     @Override
     public CompletableFuture<List<CreatePartitionsTopicResult>>
-            createPartitions(List<CreatePartitionsTopic> topics) {
+            createPartitions(long deadlineNs, List<CreatePartitionsTopic> 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;
+    }
 }