From f34bd4ba0e61a910e77ba0a6c9152974737412ca Mon Sep 17 00:00:00 2001
From: Tom Bentley
Date: Wed, 6 Sep 2017 15:39:24 +0100
Subject: [PATCH 01/13] KAFKA-5692: Change
PreferredReplicaLeaderElectionCommand to use AdminClient
See also KIP-183.
This implements the following algorithm:
1. AdminClient sends ElectPreferredLeadersRequest.
2. KafakApis receives ElectPreferredLeadersRequest and delegates to
ReplicaManager.electPreferredLeaders()
3. ReplicaManager delegates to KafkaController.electPreferredLeaders()
4. KafkaController adds a PreferredReplicaLeaderElection to the EventManager,
5. ReplicaManager.electPreferredLeaders()'s callback uses the
delayedElectPreferredReplicasPurgatory to wait for the results of the
election to appear in the metadata cache. If there are no results
because of errors, or because the preferred leaders are already leading
the partitions then a response is returned immediately.
In the EventManager work thread the preferred leader is elected as follows:
1. The EventManager runs PreferredReplicaLeaderElection.process()
2. process() calls KafkaController.onPreferredReplicaElectionWithResults()
3. KafkaController.onPreferredReplicaElectionWithResults()
calls the PartitionStateMachine.handleStateChangesWithResults() to
perform the election (asynchronously the PSM will send LeaderAndIsrRequest
to the new and old leaders and UpdateMetadataRequest to all brokers)
then invokes the callback.
Note: the change in parameter type for CollectionUtils.groupDataByTopic().
This makes sense because the AdminClient APIs use Collection consistently,
rather than List or Set. If binary compatiblity is a consideration the old
version should be kept, delegating to the new version.
I had to add PartitionStateMachine.handleStateChangesWithResults()
in order to be able to process a set of state changes in the
PartitionStateMachine *and get back individual results*.
At the same time I noticed that all callers of existing handleStateChange()
were destructuring a TopicAndPartition that they already had in order
to call handleStateChange(), and that handleStateChange() immediately
instantiated a new TopicAndPartition. Since TopicAndPartition is immutable
this is pointless, so I refactored it. handleStateChange() also now returns
any exception it caught, which is necessary for handleStateChangesWithResults()
---
.../kafka/clients/admin/AdminClient.java | 27 ++
.../admin/ElectPreferredLeadersOptions.java | 31 ++
.../admin/ElectPreferredLeadersResult.java | 116 +++++++
.../kafka/clients/admin/KafkaAdminClient.java | 68 ++++
.../apache/kafka/common/protocol/ApiKeys.java | 6 +-
.../common/requests/AbstractRequest.java | 2 +
.../common/requests/AbstractResponse.java | 2 +
.../ElectPreferredLeadersRequest.java | 159 +++++++++
.../ElectPreferredLeadersResponse.java | 120 +++++++
.../kafka/common/utils/CollectionUtils.java | 3 +-
.../clients/admin/KafkaAdminClientTest.java | 35 ++
.../common/requests/RequestResponseTest.java | 22 ++
...referredReplicaLeaderElectionCommand.scala | 200 +++++++++---
.../main/scala/kafka/cluster/Partition.scala | 1 +
.../kafka/controller/KafkaController.scala | 64 +++-
.../controller/PartitionStateMachine.scala | 49 ++-
.../server/DelayedElectPreferredLeader.scala | 96 ++++++
.../main/scala/kafka/server/KafkaApis.scala | 26 ++
.../scala/kafka/server/ReplicaManager.scala | 43 +++
core/src/main/scala/kafka/utils/ZkUtils.scala | 22 ++
.../scala/unit/kafka/admin/AdminTest.scala | 4 +-
...rredReplicaLeaderElectionCommandTest.scala | 307 ++++++++++++++++++
.../kafka/server/ReplicaManagerTest.scala | 4 +-
.../unit/kafka/server/RequestQuotaTest.scala | 4 +
24 files changed, 1336 insertions(+), 75 deletions(-)
create mode 100644 clients/src/main/java/org/apache/kafka/clients/admin/ElectPreferredLeadersOptions.java
create mode 100644 clients/src/main/java/org/apache/kafka/clients/admin/ElectPreferredLeadersResult.java
create mode 100644 clients/src/main/java/org/apache/kafka/common/requests/ElectPreferredLeadersRequest.java
create mode 100644 clients/src/main/java/org/apache/kafka/common/requests/ElectPreferredLeadersResponse.java
create mode 100644 core/src/main/scala/kafka/server/DelayedElectPreferredLeader.scala
create mode 100644 core/src/test/scala/unit/kafka/admin/PreferredReplicaLeaderElectionCommandTest.scala
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/AdminClient.java b/clients/src/main/java/org/apache/kafka/clients/admin/AdminClient.java
index fd695f9dc33d9..e209a20659ec3 100644
--- a/clients/src/main/java/org/apache/kafka/clients/admin/AdminClient.java
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/AdminClient.java
@@ -17,6 +17,7 @@
package org.apache.kafka.clients.admin;
+import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.TopicPartitionReplica;
import org.apache.kafka.common.acl.AclBinding;
import org.apache.kafka.common.acl.AclBindingFilter;
@@ -508,4 +509,30 @@ public CreatePartitionsResult createPartitions(Map newPar
public abstract CreatePartitionsResult createPartitions(Map newPartitions,
CreatePartitionsOptions options);
+ /*
+ * Elect the preferred replica of the given {@code partitions} as leader, or
+ * elect the preferred replica for all partitions as leader if the argument to {@code partitions} is null.
+ *
+ * This is a convenience method for {@link #electPreferredLeaders(Collection, ElectPreferredLeadersOptions)} with default options.
+ * See the overload for more details.
+ *
+ * @param partitions The partitions for which the the preferred leader should be elected.
+ * @return The ElectPreferredLeadersResult.
+ */
+ public ElectPreferredLeadersResult electPreferredLeaders(Collection partitions) {
+ return electPreferredLeaders(partitions, new ElectPreferredLeadersOptions());
+ }
+
+ /**
+ * Elect the preferred replica of the given {@code partitions} as leader, or
+ * elect the preferred replica for all partitions as leader if the argument to {@code partitions} is null.
+ *
+ * This operation is supported by brokers with version 1.0.0 or higher.
+ *
+ * @param partitions The partitions for which the the preferred leader should be elected.
+ * @param options The options to use when electing the preferred leaders.
+ * @return The ElectPreferredLeadersResult.
+ */
+ public abstract ElectPreferredLeadersResult electPreferredLeaders(Collection partitions, ElectPreferredLeadersOptions options);
+
}
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/ElectPreferredLeadersOptions.java b/clients/src/main/java/org/apache/kafka/clients/admin/ElectPreferredLeadersOptions.java
new file mode 100644
index 0000000000000..80b00975f7da9
--- /dev/null
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/ElectPreferredLeadersOptions.java
@@ -0,0 +1,31 @@
+/*
+ * 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.clients.admin;
+
+import org.apache.kafka.common.annotation.InterfaceStability;
+
+import java.util.Collection;
+
+/**
+ * Options for {@link AdminClient#electPreferredLeaders(Collection, ElectPreferredLeadersOptions)}.
+ *
+ * The API of this class is evolving, see {@link AdminClient} for details.
+ */
+@InterfaceStability.Evolving
+public class ElectPreferredLeadersOptions extends AbstractOptions {
+}
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/ElectPreferredLeadersResult.java b/clients/src/main/java/org/apache/kafka/clients/admin/ElectPreferredLeadersResult.java
new file mode 100644
index 0000000000000..e2075b2ef95cb
--- /dev/null
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/ElectPreferredLeadersResult.java
@@ -0,0 +1,116 @@
+/*
+ * 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.clients.admin;
+
+import org.apache.kafka.common.KafkaFuture;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.annotation.InterfaceStability;
+import org.apache.kafka.common.internals.KafkaFutureImpl;
+
+import java.util.Collection;
+import java.util.Map;
+import java.util.concurrent.ExecutionException;
+
+/**
+ * The result of {@link AdminClient#electPreferredLeaders(Collection, ElectPreferredLeadersOptions)}
+ *
+ * The {@code KafkaFuture}s available from instances of this class may be completed
+ * exceptionally due to:
+ *
+ * - {@link org.apache.kafka.common.protocol.Errors#CLUSTER_AUTHORIZATION_FAILED CLUSTER_AUTHORIZATION_FAILED} if the authenticated user didn't have {@code Alter} access to the cluster.
+ * - {@link org.apache.kafka.common.protocol.Errors#UNKNOWN_TOPIC_OR_PARTITION UNKNOWN_TOPIC_OR_PARTITION} if the topic or partition did not exist within the cluster.
+ * - {@link org.apache.kafka.common.protocol.Errors#INVALID_TOPIC_EXCEPTION INVALID_TOPIC_EXCEPTION} if the topic was already queued for deletion.
+ * - {@link org.apache.kafka.common.protocol.Errors#NOT_CONTROLLER NOT_CONTROLLER} if the request was sent to a broker that was not the controller for the cluster.
+ * - {@link org.apache.kafka.common.protocol.Errors#REQUEST_TIMED_OUT REQUEST_TIMED_OUT} if the request timed out before the election was complete.
+ * - {@link org.apache.kafka.common.protocol.Errors#UNKNOWN_SERVER_ERROR UNKNOWN_SERVER_ERROR} if the preferred leader was not alive or not in the ISR.
+ *
+ *
+ * The API of this class is evolving, see {@link AdminClient} for details.
+ */
+@InterfaceStability.Evolving
+public class ElectPreferredLeadersResult {
+
+ private final KafkaFuture extends Map>> futures;
+
+ ElectPreferredLeadersResult(KafkaFuture extends Map>> futures) {
+ this.futures = futures;
+ }
+
+ /** Return a new future that has completed exceptionally */
+ private static KafkaFuture exceptionalFuture(Exception e) {
+ KafkaFutureImpl future = new KafkaFutureImpl<>();
+ future.completeExceptionally(e);
+ return future;
+ }
+
+ /**
+ * Get the result of the election for the given {@code partition}.
+ * If there was not an election triggered for the given {@code partition}, the
+ * returned future will complete with an error.
+ */
+ public KafkaFuture partitionResult(TopicPartition partition) {
+ final Map> map;
+ try {
+ map = futures.get();
+ } catch (InterruptedException | ExecutionException e) {
+ return exceptionalFuture(e);
+ }
+ KafkaFuture result = map.get(partition);
+ if (result == null) {
+ KafkaFutureImpl future = new KafkaFutureImpl<>();
+ future.completeExceptionally(new IllegalArgumentException(
+ "Preferred leader election for partition \"" + partition + "\" was not attempted"));
+ result = future;
+ }
+ return result;
+ }
+
+ /**
+ * Get a future for the topic partitions for which a leader election
+ * was attempted. A partition will be present in this result if
+ * an election was attempted even if the election was not successful.
+ *
+ * This method is provided to discover the partitions attempted when
+ * {@link AdminClient#electPreferredLeaders(Collection)} is called
+ * with a null {@code partitions} argument.
+ */
+ public KafkaFuture> partitions() {
+ final Map> map;
+ try {
+ map = futures.get();
+ } catch (InterruptedException | ExecutionException e) {
+ return exceptionalFuture(e);
+ }
+ KafkaFutureImpl> result = new KafkaFutureImpl<>();
+ result.complete(map.keySet());
+ return result;
+ }
+
+ /**
+ * Return a future which succeeds if all the topic elections succeed.
+ */
+ public KafkaFuture all() {
+ final Map> map;
+ try {
+ map = futures.get();
+ } catch (InterruptedException | ExecutionException e) {
+ return exceptionalFuture(e);
+ }
+ return KafkaFuture.allOf(map.values().toArray(new KafkaFuture[0]));
+ }
+}
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java b/clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java
index ece27caeb405d..0f67ac453584c 100644
--- a/clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java
@@ -86,6 +86,8 @@
import org.apache.kafka.common.requests.DescribeConfigsResponse;
import org.apache.kafka.common.requests.DescribeLogDirsRequest;
import org.apache.kafka.common.requests.DescribeLogDirsResponse;
+import org.apache.kafka.common.requests.ElectPreferredLeadersRequest;
+import org.apache.kafka.common.requests.ElectPreferredLeadersResponse;
import org.apache.kafka.common.requests.MetadataRequest;
import org.apache.kafka.common.requests.MetadataResponse;
import org.apache.kafka.common.requests.Resource;
@@ -1863,6 +1865,7 @@ public void handleResponse(AbstractResponse abstractResponse) {
future.completeExceptionally(result.getValue().exception());
}
}
+
}
@Override
@@ -1873,4 +1876,69 @@ void handleFailure(Throwable throwable) {
return new CreatePartitionsResult(new HashMap>(futures));
}
+ public ElectPreferredLeadersResult electPreferredLeaders(final Collection partitions, ElectPreferredLeadersOptions options) {
+
+ final KafkaFutureImpl
+ *
+ * - {@link org.apache.kafka.common.errors.ClusterAuthorizationException}
+ * if the authenticated user didn't have alter access to the cluster.
+ * - {@link org.apache.kafka.common.errors.UnknownTopicOrPartitionException}
+ * if the topic or partition did not exist within the cluster.
+ * - {@link org.apache.kafka.common.errors.InvalidTopicException}
+ * if the topic was already queued for deletion.
+ * - {@link org.apache.kafka.common.errors.NotControllerException}
+ * if the request was sent to a broker that was not the controller for the cluster.
+ * - {@link org.apache.kafka.common.errors.TimeoutException}
+ * if the request timed out before the election was complete.
+ * - {@link org.apache.kafka.common.errors.LeaderNotAvailableException}
+ * if the preferred leader was not alive or not in the ISR.
+ *
*
* @param partitions The partitions for which the the preferred leader should be elected.
* @param options The options to use when electing the preferred leaders.
* @return The ElectPreferredLeadersResult.
*/
- public abstract ElectPreferredLeadersResult electPreferredLeaders(Collection partitions, ElectPreferredLeadersOptions options);
+ public abstract ElectPreferredLeadersResult electPreferredLeaders(Collection partitions,
+ ElectPreferredLeadersOptions options);
}
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/ElectPreferredLeadersResult.java b/clients/src/main/java/org/apache/kafka/clients/admin/ElectPreferredLeadersResult.java
index 133c335021dfc..350414be5ca1b 100644
--- a/clients/src/main/java/org/apache/kafka/clients/admin/ElectPreferredLeadersResult.java
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/ElectPreferredLeadersResult.java
@@ -24,22 +24,12 @@
import java.util.Collection;
import java.util.Map;
+import java.util.Set;
import java.util.concurrent.ExecutionException;
/**
* The result of {@link AdminClient#electPreferredLeaders(Collection, ElectPreferredLeadersOptions)}
*
- * The {@code KafkaFuture}s available from instances of this class may be completed
- * exceptionally due to:
- *
- * - {@link org.apache.kafka.common.protocol.Errors#CLUSTER_AUTHORIZATION_FAILED CLUSTER_AUTHORIZATION_FAILED} if the authenticated user didn't have {@code Alter} access to the cluster.
- * - {@link org.apache.kafka.common.protocol.Errors#UNKNOWN_TOPIC_OR_PARTITION UNKNOWN_TOPIC_OR_PARTITION} if the topic or partition did not exist within the cluster.
- * - {@link org.apache.kafka.common.protocol.Errors#INVALID_TOPIC_EXCEPTION INVALID_TOPIC_EXCEPTION} if the topic was already queued for deletion.
- * - {@link org.apache.kafka.common.protocol.Errors#NOT_CONTROLLER NOT_CONTROLLER} if the request was sent to a broker that was not the controller for the cluster.
- * - {@link org.apache.kafka.common.protocol.Errors#REQUEST_TIMED_OUT REQUEST_TIMED_OUT} if the request timed out before the election was complete.
- * - {@link org.apache.kafka.common.protocol.Errors#UNKNOWN_SERVER_ERROR UNKNOWN_SERVER_ERROR} if the preferred leader was not alive or not in the ISR.
- *
- *
* The API of this class is evolving, see {@link AdminClient} for details.
*/
@InterfaceStability.Evolving
@@ -89,7 +79,7 @@ public KafkaFuture partitionResult(TopicPartition partition) {
* {@link AdminClient#electPreferredLeaders(Collection)} is called
* with a null {@code partitions} argument.
*/
- public KafkaFuture> partitions() {
+ public KafkaFuture> partitions() {
final Map> map;
try {
map = futures.get();
@@ -98,7 +88,7 @@ public KafkaFuture> partitions() {
} catch (ExecutionException e) {
return exceptionalFuture(e.getCause());
}
- KafkaFutureImpl> result = new KafkaFutureImpl<>();
+ KafkaFutureImpl> result = new KafkaFutureImpl<>();
result.complete(map.keySet());
return result;
}
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java b/clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java
index 0f67ac453584c..21b9761877814 100644
--- a/clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java
@@ -1865,7 +1865,6 @@ public void handleResponse(AbstractResponse abstractResponse) {
future.completeExceptionally(result.getValue().exception());
}
}
-
}
@Override
@@ -1876,7 +1875,9 @@ void handleFailure(Throwable throwable) {
return new CreatePartitionsResult(new HashMap>(futures));
}
- public ElectPreferredLeadersResult electPreferredLeaders(final Collection partitions, ElectPreferredLeadersOptions options) {
+ @Override
+ public ElectPreferredLeadersResult electPreferredLeaders(final Collection partitions,
+ ElectPreferredLeadersOptions options) {
final KafkaFutureImpl