KAFKA-8286; Leader Election Admin RPC (KIP-460)

bin/kafka-leader-election.sh

@@ -0,0 +1,17 @@
+#!/bin/bash
+# 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.
+
+exec $(dirname $0)/kafka-run-class.sh kafka.admin.LeaderElectionCommand "$@"

bin/windows/kafka-leader-election.bat

@@ -0,0 +1,17 @@
+@echo off
+rem Licensed to the Apache Software Foundation (ASF) under one or more
+rem contributor license agreements.  See the NOTICE file distributed with
+rem this work for additional information regarding copyright ownership.
+rem The ASF licenses this file to You under the Apache License, Version 2.0
+rem (the "License"); you may not use this file except in compliance with
+rem the License.  You may obtain a copy of the License at
+rem
+rem     http://www.apache.org/licenses/LICENSE-2.0
+rem
+rem Unless required by applicable law or agreed to in writing, software
+rem distributed under the License is distributed on an "AS IS" BASIS,
+rem WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+rem See the License for the specific language governing permissions and
+rem limitations under the License.
+
+"%~dp0kafka-run-class.bat" kafka.admin.LeaderElectionCommand %*

clients/src/main/java/org/apache/kafka/clients/admin/AdminClient.java

@@ -17,6 +17,13 @@
 
 package org.apache.kafka.clients.admin;
 
+import java.time.Duration;
+import java.util.Collection;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import org.apache.kafka.common.ElectionType;
 import org.apache.kafka.common.Metric;
 import org.apache.kafka.common.MetricName;
 import org.apache.kafka.common.TopicPartition;
@@ -26,12 +33,6 @@
 import org.apache.kafka.common.annotation.InterfaceStability;
 import org.apache.kafka.common.config.ConfigResource;
 
-import java.time.Duration;
-import java.util.Collection;
-import java.util.Map;
-import java.util.Properties;
-import java.util.concurrent.TimeUnit;
-
 /**
  * The administrative client for Kafka, which supports managing and inspecting topics, brokers, configurations and ACLs.
  *
@@ -848,7 +849,9 @@ public DeleteConsumerGroupsResult deleteConsumerGroups(Collection<String> groupI
      *
      * @param partitions      The partitions for which the preferred leader should be elected.
      * @return                The ElectPreferredLeadersResult.
+     * @deprecated            Since TBD. Use {@link #electLeaders}.
      */
+    @Deprecated
     public ElectPreferredLeadersResult electPreferredLeaders(Collection<TopicPartition> partitions) {
         return electPreferredLeaders(partitions, new ElectPreferredLeadersOptions());
     }
@@ -882,13 +885,40 @@ public ElectPreferredLeadersResult electPreferredLeaders(Collection<TopicPartiti
      *   <li>{@link org.apache.kafka.common.errors.LeaderNotAvailableException}
      *   if the preferred leader was not alive or not in the ISR.</li>
      * </ul>
+     * TODO: This can be implemeted interms of the method below.
      *
      * @param partitions      The partitions for which the preferred leader should be elected.
      * @param options         The options to use when electing the preferred leaders.
      * @return                The ElectPreferredLeadersResult.
+     * @deprecated            Since TBD. Use {@link #electLeaders}.
      */
+    @Deprecated
     public abstract ElectPreferredLeadersResult electPreferredLeaders(Collection<TopicPartition> partitions,
                                                                       ElectPreferredLeadersOptions options);
+    /**
+     * TODO: Write documentation for this method.
+     *
+     * @param electionType            The type of election to conduct.
+     * @param partitions              The topics and partitions for which to conduct elections.
+     * @param options                 The options to use when electing the leaders.
+     * @return                        The ElectLeadersResult.
+     */
+    public ElectLeadersResult electLeaders(ElectionType electionType, Set<TopicPartition> partitions) {
+        return electLeaders(electionType, partitions, new ElectLeadersOptions());
+    }
+
+    /**
+     * TODO: Write documentation for this method.
+     *
+     * @param electionType            The type of election to conduct.
+     * @param partitions              The topics and partitions for which to conduct elections.
+     * @param options                 The options to use when electing the leaders.
+     * @return                        The ElectLeadersResult.
+     */
+    public abstract ElectLeadersResult electLeaders(
+            ElectionType electionType,
+            Set<TopicPartition> partitions,
+            ElectLeadersOptions options);
 
     /**
      * Get the metrics kept by the adminClient

clients/src/main/java/org/apache/kafka/clients/admin/ElectLeadersOptions.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#electLeaders(ElectionType, Collection, ElectLeadersOptions)}.
+ *
+ * The API of this class is evolving, see {@link AdminClient} for details.
+ */
+@InterfaceStability.Evolving
+final public class ElectLeadersOptions extends AbstractOptions<ElectLeadersOptions> {
+}

clients/src/main/java/org/apache/kafka/clients/admin/ElectLeadersResult.java

@@ -0,0 +1,137 @@
+/*
+ * 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 java.util.Collection;
+import java.util.Map;
+import java.util.Set;
+import org.apache.kafka.common.KafkaFuture;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.annotation.InterfaceStability;
+import org.apache.kafka.common.errors.ApiException;
+import org.apache.kafka.common.errors.UnknownTopicOrPartitionException;
+import org.apache.kafka.common.internals.KafkaFutureImpl;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.requests.ApiError;
+
+/**
+ * The result of {@link AdminClient#electLeaders(ElectionType, Collection, ElectLeadersOptions)}
+ *
+ * The API of this class is evolving, see {@link AdminClient} for details.
+ */
+@InterfaceStability.Evolving
+final public class ElectLeadersResult {
+    private final KafkaFutureImpl<Map<TopicPartition, ApiError>> electionFuture;
+    private final Set<TopicPartition> partitions;
+
+    ElectLeadersResult(KafkaFutureImpl<Map<TopicPartition, ApiError>> electionFuture, Set<TopicPartition> partitions) {
+        this.electionFuture = electionFuture;
+        this.partitions = partitions;
+    }
+
+    /**
+     * 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<Void> partitionResult(final TopicPartition partition) {
+        final KafkaFutureImpl<Void> result = new KafkaFutureImpl<>();
+        electionFuture.whenComplete(new KafkaFuture.BiConsumer<Map<TopicPartition, ApiError>, Throwable>() {
+            @Override
+            public void accept(Map<TopicPartition, ApiError> topicPartitions, Throwable throwable) {
+                if (throwable != null) {
+                    result.completeExceptionally(throwable);
+                } else if (!topicPartitions.containsKey(partition)) {
+                    result.completeExceptionally(new UnknownTopicOrPartitionException(
+                            "Leader election for partition \"" + partition +
+                                    "\" was not attempted"));
+                } else if (partitions == null && topicPartitions.isEmpty()) {
+                    // If partitions is null, we requested information about all partitions.  In
+                    // that case, if topicPartitions is empty, that indicates a
+                    // CLUSTER_AUTHORIZATION_FAILED error.
+                    result.completeExceptionally(Errors.CLUSTER_AUTHORIZATION_FAILED.exception());
+                } else {
+                    ApiException exception = topicPartitions.get(partition).exception();
+                    if (exception == null) {
+                        result.complete(null);
+                    } else {
+                        result.completeExceptionally(exception);
+                    }
+                }
+            }
+        });
+        return result;
+    }
+
+    /**
+     * <p>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.</p>
+     *
+     * <p>This method is provided to discover the partitions attempted when
+     * {@link AdminClient#electLeaders(ElectionType, Collection, ElectLeadersOptions)} is called
+     * with a null {@code partitions} argument.</p>
+     */
+    public KafkaFuture<Set<TopicPartition>> partitions() {
+        if (partitions != null) {
+            return KafkaFutureImpl.completedFuture(this.partitions);
+        } else {
+            final KafkaFutureImpl<Set<TopicPartition>> result = new KafkaFutureImpl<>();
+            electionFuture.whenComplete(new KafkaFuture.BiConsumer<Map<TopicPartition, ApiError>, Throwable>() {
+                @Override
+                public void accept(Map<TopicPartition, ApiError> topicPartitions, Throwable throwable) {
+                    if (throwable != null) {
+                        result.completeExceptionally(throwable);
+                    } else if (topicPartitions.isEmpty()) {
+                        result.completeExceptionally(Errors.CLUSTER_AUTHORIZATION_FAILED.exception());
+                    } else {
+                        for (ApiError apiError : topicPartitions.values()) {
+                            if (apiError.isFailure()) {
+                                result.completeExceptionally(apiError.exception());
+                            }
+                        }
+                        result.complete(topicPartitions.keySet());
+                    }
+                }
+            });
+            return result;
+        }
+    }
+
+    /**
+     * Return a future which succeeds if all the topic elections succeed.
+     */
+    public KafkaFuture<Void> all() {
+        final KafkaFutureImpl<Void> result = new KafkaFutureImpl<>();
+        electionFuture.thenApply(new KafkaFuture.Function<Map<TopicPartition, ApiError>, Void>() {
+            @Override
+            public Void apply(Map<TopicPartition, ApiError> topicPartitions) {
+                for (ApiError apiError : topicPartitions.values()) {
+                    if (apiError.isFailure()) {
+                        result.completeExceptionally(apiError.exception());
+                        return null;
+                    }
+                }
+                result.complete(null);
+                return null;
+            }
+        });
+        return result;
+    }
+}

clients/src/main/java/org/apache/kafka/clients/admin/ElectPreferredLeadersOptions.java

@@ -25,7 +25,10 @@
  * Options for {@link AdminClient#electPreferredLeaders(Collection, ElectPreferredLeadersOptions)}.
  *
  * The API of this class is evolving, see {@link AdminClient} for details.
+ *
+ * @deprecated Since TBD. Use {@link AdminClient#electLeaders(ElectionType, Collection, ElectLeadersOption)}.
  */
 @InterfaceStability.Evolving
+@Deprecated
 public class ElectPreferredLeadersOptions extends AbstractOptions<ElectPreferredLeadersOptions> {
 }

clients/src/main/java/org/apache/kafka/clients/admin/ElectPreferredLeadersResult.java

@@ -34,9 +34,12 @@
  * The result of {@link AdminClient#electPreferredLeaders(Collection, ElectPreferredLeadersOptions)}
  *
  * The API of this class is evolving, see {@link AdminClient} for details.
+ *
+ * @deprecated Since TDB. Use {@link AdminClient#electLeaders(ElectionType, Collection, ElectLeadersOption)}.
  */
 @InterfaceStability.Evolving
-public class ElectPreferredLeadersResult {
+@Deprecated
+final public class ElectPreferredLeadersResult {
 
     private final KafkaFutureImpl<Map<TopicPartition, ApiError>> electionFuture;
     private final Set<TopicPartition> partitions;