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,14 @@
 
 package org.apache.kafka.clients.admin;
 
+import java.time.Duration;
+import java.util.Collection;
+import java.util.HashSet;
+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 +34,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.
  *
@@ -839,35 +841,75 @@ public DeleteConsumerGroupsResult deleteConsumerGroups(Collection<String> groupI
     }
 
     /**
-     * Elect the preferred broker of the given {@code partitions} as leader, or
-     * elect the preferred broker for all partitions as leader if the argument to {@code partitions} is null.
+     * Elect the preferred replica as leader for topic partitions.
      *
-     * This is a convenience method for {@link #electPreferredLeaders(Collection, ElectPreferredLeadersOptions)}
-     * with default options.
-     * See the overload for more details.
+     * This is a convenience method for {@link #electLeaders(ElectionType, Set, ElectLeadersOptions)}
+     * with preferred election type and default options.
+     *
+     * This operation is supported by brokers with version 2.2.0 or higher.
      *
      * @param partitions      The partitions for which the preferred leader should be elected.
      * @return                The ElectPreferredLeadersResult.
+     * @deprecated            Since 2.4.0. Use {@link #electLeaders(ElectionType, Set)}.
      */
+    @Deprecated
     public ElectPreferredLeadersResult electPreferredLeaders(Collection<TopicPartition> partitions) {
         return electPreferredLeaders(partitions, new ElectPreferredLeadersOptions());
     }
 
     /**
-     * Elect the preferred broker of the given {@code partitions} as leader, or
-     * elect the preferred broker for all partitions as leader if the argument to {@code partitions} is null.
+     * Elect the preferred replica as leader for topic partitions.
+     *
+     * This is a convenience method for {@link #electLeaders(ElectionType, Set, ElectLeadersOptions)}
+     * with preferred election type.
+     *
+     * This operation is supported by brokers with version 2.2.0 or higher.
+     *
+     * @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 2.4.0. Use {@link #electLeaders(ElectionType, Set, ElectLeadersOptions)}.
+     */
+    @Deprecated
+    public ElectPreferredLeadersResult electPreferredLeaders(Collection<TopicPartition> partitions,
+                                                             ElectPreferredLeadersOptions options) {
+        final ElectLeadersOptions newOptions = new ElectLeadersOptions();
+        newOptions.timeoutMs(options.timeoutMs());
+        final Set<TopicPartition> topicPartitions = partitions == null ? null : new HashSet<>(partitions);
+
+        return new ElectPreferredLeadersResult(electLeaders(ElectionType.PREFERRED, topicPartitions, newOptions));
+    }
+
+    /**
+     * Elect a replica as leader for topic partitions.
+     *
+     * This is a convenience method for {@link #electLeaders(ElectionType, Set, ElectLeadersOptions)}
+     * with default options.
+     *
+     * @param electionType            The type of election to conduct.
+     * @param partitions              The topics and partitions for which to conduct elections.
+     * @return                        The ElectLeadersResult.
+     */
+    public ElectLeadersResult electLeaders(ElectionType electionType, Set<TopicPartition> partitions) {
+        return electLeaders(electionType, partitions, new ElectLeadersOptions());
+    }
+
+    /**
+     * Elect a replica as leader for the given {@code partitions}, or for all partitions if the argumentl
+     * to {@code partitions} is null.
      *
      * This operation is not transactional so it may succeed for some partitions while fail for others.
      *
-     * It may take several seconds after this method returns
-     * success for all the brokers in the cluster to become aware that the partitions have new leaders.
-     * During this time, {@link AdminClient#describeTopics(Collection)}
-     * may not return information about the partitions' new leaders.
+     * It may take several seconds after this method returns success for all the brokers in the cluster
+     * to become aware that the partitions have new leaders. During this time,
+     * {@link AdminClient#describeTopics(Collection)} may not return information about the partitions'
+     * new leaders.
      *
-     * This operation is supported by brokers with version 2.2.0 or higher.
+     * This operation is supported by brokers with version 2.2.0 or later if preferred eleciton is use;
+     * otherwise the brokers most be 2.4.0 or higher.
      *
-     * <p>The following exceptions can be anticipated when calling {@code get()} on the futures obtained from
-     * the returned {@code ElectPreferredLeadersResult}:</p>
+     * <p>The following exceptions can be anticipated when calling {@code get()} on the future obtained
+     * from the returned {@code ElectLeadersResult}:</p>
      * <ul>
      *   <li>{@link org.apache.kafka.common.errors.ClusterAuthorizationException}
      *   if the authenticated user didn't have alter access to the cluster.</li>
@@ -883,12 +925,15 @@ public ElectPreferredLeadersResult electPreferredLeaders(Collection<TopicPartiti
      *   if the preferred leader was not alive or not in the ISR.</li>
      * </ul>
      *
-     * @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.
+     * @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 ElectPreferredLeadersResult electPreferredLeaders(Collection<TopicPartition> partitions,
-                                                                      ElectPreferredLeadersOptions options);
+    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,29 @@
+/*
+ * 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;
+
+/**
+ * Options for {@link AdminClient#electLeaders(ElectionType, Set, 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,76 @@
+/*
+ * 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.Map;
+import java.util.Optional;
+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;
+
+/**
+ * The result of {@link AdminClient#electLeaders(ElectionType, Set, ElectLeadersOptions)}
+ *
+ * The API of this class is evolving, see {@link AdminClient} for details.
+ */
+@InterfaceStability.Evolving
+final public class ElectLeadersResult {
+    private final KafkaFutureImpl<Map<TopicPartition, Optional<Throwable>>> electionFuture;
+
+    ElectLeadersResult(KafkaFutureImpl<Map<TopicPartition, Optional<Throwable>>> electionFuture) {
+        this.electionFuture = electionFuture;
+    }
+
+    /**
+     * <p>Get a future for the topic partitions for which a leader election was attempted.
+     * If the election succeeded then the value for a topic partition will be the empty Optional.
+     * Otherwise the election failed and the Optional will be set with the error.</p>
+     */
+    public KafkaFuture<Map<TopicPartition, Optional<Throwable>>> partitions() {
+        return electionFuture;
+    }
+
+    /**
+     * Return a future which succeeds if all the topic elections succeed.
+     */
+    public KafkaFuture<Void> all() {
+        final KafkaFutureImpl<Void> result = new KafkaFutureImpl<>();
+
+        partitions().whenComplete(
+                new KafkaFuture.BiConsumer<Map<TopicPartition, Optional<Throwable>>, Throwable>() {
+                    @Override
+                    public void accept(Map<TopicPartition, Optional<Throwable>> topicPartitions, Throwable throwable) {
+                        if (throwable != null) {
+                            result.completeExceptionally(throwable);
+                        } else {
+                            for (Optional<Throwable> exception : topicPartitions.values()) {
+                                if (exception.isPresent()) {
+                                    result.completeExceptionally(exception.get());
+                                    return;
+                                }
+                            }
+                            result.complete(null);
+                        }
+                    }
+                });
+
+        return result;
+    }
+}

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

@@ -18,14 +18,16 @@
 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.
+ *
+ * @deprecated Since 2.4.0. Use {@link AdminClient#electLeaders(ElectionType, Set, ElectLeadersOptions)}.
  */
 @InterfaceStability.Evolving
+@Deprecated
 public class ElectPreferredLeadersOptions extends AbstractOptions<ElectPreferredLeadersOptions> {
 }