fvaleri commented on code in PR #13204:
URL: https://github.com/apache/kafka/pull/13204#discussion_r1102366936
##########
build.gradle:
##########
@@ -1756,6 +1756,7 @@ project(':tools') {
dependencies {
implementation project(':clients')
+ implementation project(':core')
Review Comment:
The direction from previous migrations is that we don't want to depend on
core. I see there are a few kafka.* dependencies in LeaderElectionCommand
class. Do you think we can also migrate them as part of this PR?
##########
tools/src/main/java/org/apache/kafka/tools/LeaderElectionCommand.java:
##########
@@ -0,0 +1,335 @@
+/*
+ * 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.tools;
+
+import joptsimple.AbstractOptionSpec;
+import joptsimple.ArgumentAcceptingOptionSpec;
+import joptsimple.OptionSpecBuilder;
+import joptsimple.util.EnumConverter;
+import kafka.admin.AdminOperationException;
+import kafka.common.AdminCommandFailedException;
+import kafka.utils.CoreUtils;
+import kafka.utils.Json;
+import kafka.utils.json.DecodeJson;
+import kafka.utils.json.JsonObject;
+import kafka.utils.json.JsonValue;
+import org.apache.kafka.clients.admin.Admin;
+import org.apache.kafka.clients.admin.AdminClientConfig;
+import org.apache.kafka.common.ElectionType;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.errors.ClusterAuthorizationException;
+import org.apache.kafka.common.errors.ElectionNotNeededException;
+import org.apache.kafka.common.errors.TimeoutException;
+import org.apache.kafka.common.utils.Utils;
+import org.apache.kafka.server.util.CommandDefaultOptions;
+import org.apache.kafka.server.util.CommandLineUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import scala.Option;
+import scala.collection.Iterable;
+import scala.collection.Iterator;
+import scala.collection.JavaConverters;
+import scala.collection.mutable.Seq;
+
+import java.io.IOException;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Properties;
+import java.util.Set;
+import java.util.concurrent.ExecutionException;
+import java.util.stream.Collectors;
+
+public class LeaderElectionCommand {
+ private static final Logger log =
LoggerFactory.getLogger(LeaderElectionCommand.class);
+
+ public static void main(String... args) {
+ try {
+ run(Duration.ofSeconds(30), args);
+ } catch (TerseException e) {
+ System.err.println(e.getMessage());
+ } catch (Exception e) {
+ System.err.println(e.getMessage());
+ System.err.println(Utils.stackTrace(e));
+ }
+ }
+
+ static void run(Duration timeout, String... args) throws Exception {
+ LeaderElectionCommandOptions commandOptions = new
LeaderElectionCommandOptions(args);
+
+ CommandLineUtils.maybePrintHelpOrVersion(
+ commandOptions,
+ "This tool attempts to elect a new leader for a set of topic
partitions. The type of elections supported are preferred replicas and unclean
replicas."
+ );
+
+ validate(commandOptions);
+ ElectionType electionType =
commandOptions.options.valueOf(commandOptions.electionType);
Review Comment:
You can create a LeaderElectionCommandOptions wrapper method for every
`commandOptions.` parsing/transformation logic. It would be much more readable.
##########
tools/src/main/java/org/apache/kafka/tools/LeaderElectionCommand.java:
##########
@@ -0,0 +1,335 @@
+/*
+ * 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.tools;
+
+import joptsimple.AbstractOptionSpec;
+import joptsimple.ArgumentAcceptingOptionSpec;
+import joptsimple.OptionSpecBuilder;
+import joptsimple.util.EnumConverter;
+import kafka.admin.AdminOperationException;
+import kafka.common.AdminCommandFailedException;
+import kafka.utils.CoreUtils;
+import kafka.utils.Json;
+import kafka.utils.json.DecodeJson;
+import kafka.utils.json.JsonObject;
+import kafka.utils.json.JsonValue;
+import org.apache.kafka.clients.admin.Admin;
+import org.apache.kafka.clients.admin.AdminClientConfig;
+import org.apache.kafka.common.ElectionType;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.errors.ClusterAuthorizationException;
+import org.apache.kafka.common.errors.ElectionNotNeededException;
+import org.apache.kafka.common.errors.TimeoutException;
+import org.apache.kafka.common.utils.Utils;
+import org.apache.kafka.server.util.CommandDefaultOptions;
+import org.apache.kafka.server.util.CommandLineUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import scala.Option;
+import scala.collection.Iterable;
+import scala.collection.Iterator;
+import scala.collection.JavaConverters;
+import scala.collection.mutable.Seq;
+
+import java.io.IOException;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Properties;
+import java.util.Set;
+import java.util.concurrent.ExecutionException;
+import java.util.stream.Collectors;
+
+public class LeaderElectionCommand {
+ private static final Logger log =
LoggerFactory.getLogger(LeaderElectionCommand.class);
+
+ public static void main(String... args) {
+ try {
+ run(Duration.ofSeconds(30), args);
+ } catch (TerseException e) {
+ System.err.println(e.getMessage());
+ } catch (Exception e) {
+ System.err.println(e.getMessage());
+ System.err.println(Utils.stackTrace(e));
+ }
+ }
+
+ static void run(Duration timeout, String... args) throws Exception {
Review Comment:
I know this is as it was in the original command, but I think we can improve
by passing milliseconds rather than seconds and change the name to timeoutMs.
##########
tools/src/main/java/org/apache/kafka/tools/LeaderElectionCommand.java:
##########
@@ -0,0 +1,335 @@
+/*
+ * 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.tools;
+
+import joptsimple.AbstractOptionSpec;
+import joptsimple.ArgumentAcceptingOptionSpec;
+import joptsimple.OptionSpecBuilder;
+import joptsimple.util.EnumConverter;
+import kafka.admin.AdminOperationException;
+import kafka.common.AdminCommandFailedException;
+import kafka.utils.CoreUtils;
+import kafka.utils.Json;
+import kafka.utils.json.DecodeJson;
+import kafka.utils.json.JsonObject;
+import kafka.utils.json.JsonValue;
+import org.apache.kafka.clients.admin.Admin;
+import org.apache.kafka.clients.admin.AdminClientConfig;
+import org.apache.kafka.common.ElectionType;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.errors.ClusterAuthorizationException;
+import org.apache.kafka.common.errors.ElectionNotNeededException;
+import org.apache.kafka.common.errors.TimeoutException;
+import org.apache.kafka.common.utils.Utils;
+import org.apache.kafka.server.util.CommandDefaultOptions;
+import org.apache.kafka.server.util.CommandLineUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import scala.Option;
+import scala.collection.Iterable;
+import scala.collection.Iterator;
+import scala.collection.JavaConverters;
+import scala.collection.mutable.Seq;
+
+import java.io.IOException;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Properties;
+import java.util.Set;
+import java.util.concurrent.ExecutionException;
+import java.util.stream.Collectors;
+
+public class LeaderElectionCommand {
+ private static final Logger log =
LoggerFactory.getLogger(LeaderElectionCommand.class);
+
+ public static void main(String... args) {
+ try {
+ run(Duration.ofSeconds(30), args);
+ } catch (TerseException e) {
+ System.err.println(e.getMessage());
+ } catch (Exception e) {
+ System.err.println(e.getMessage());
+ System.err.println(Utils.stackTrace(e));
+ }
+ }
+
+ static void run(Duration timeout, String... args) throws Exception {
+ LeaderElectionCommandOptions commandOptions = new
LeaderElectionCommandOptions(args);
+
+ CommandLineUtils.maybePrintHelpOrVersion(
+ commandOptions,
+ "This tool attempts to elect a new leader for a set of topic
partitions. The type of elections supported are preferred replicas and unclean
replicas."
+ );
+
+ validate(commandOptions);
Review Comment:
We can make this method part of LeaderElectionCommandOptions and simply call
commandOptions.validate(), also including the above maybePrintHelpOrVersion.
##########
tools/src/main/java/org/apache/kafka/tools/LeaderElectionCommand.java:
##########
@@ -0,0 +1,335 @@
+/*
+ * 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.tools;
+
+import joptsimple.AbstractOptionSpec;
+import joptsimple.ArgumentAcceptingOptionSpec;
+import joptsimple.OptionSpecBuilder;
+import joptsimple.util.EnumConverter;
+import kafka.admin.AdminOperationException;
+import kafka.common.AdminCommandFailedException;
+import kafka.utils.CoreUtils;
+import kafka.utils.Json;
+import kafka.utils.json.DecodeJson;
+import kafka.utils.json.JsonObject;
+import kafka.utils.json.JsonValue;
+import org.apache.kafka.clients.admin.Admin;
+import org.apache.kafka.clients.admin.AdminClientConfig;
+import org.apache.kafka.common.ElectionType;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.errors.ClusterAuthorizationException;
+import org.apache.kafka.common.errors.ElectionNotNeededException;
+import org.apache.kafka.common.errors.TimeoutException;
+import org.apache.kafka.common.utils.Utils;
+import org.apache.kafka.server.util.CommandDefaultOptions;
+import org.apache.kafka.server.util.CommandLineUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import scala.Option;
+import scala.collection.Iterable;
+import scala.collection.Iterator;
+import scala.collection.JavaConverters;
+import scala.collection.mutable.Seq;
+
+import java.io.IOException;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Properties;
+import java.util.Set;
+import java.util.concurrent.ExecutionException;
+import java.util.stream.Collectors;
+
+public class LeaderElectionCommand {
+ private static final Logger log =
LoggerFactory.getLogger(LeaderElectionCommand.class);
+
+ public static void main(String... args) {
+ try {
+ run(Duration.ofSeconds(30), args);
+ } catch (TerseException e) {
+ System.err.println(e.getMessage());
+ } catch (Exception e) {
+ System.err.println(e.getMessage());
+ System.err.println(Utils.stackTrace(e));
+ }
+ }
+
+ static void run(Duration timeout, String... args) throws Exception {
+ LeaderElectionCommandOptions commandOptions = new
LeaderElectionCommandOptions(args);
+
+ CommandLineUtils.maybePrintHelpOrVersion(
+ commandOptions,
+ "This tool attempts to elect a new leader for a set of topic
partitions. The type of elections supported are preferred replicas and unclean
replicas."
+ );
+
+ validate(commandOptions);
+ ElectionType electionType =
commandOptions.options.valueOf(commandOptions.electionType);
+ Optional<Set<TopicPartition>> jsonFileTopicPartitions =
+
Optional.ofNullable(commandOptions.options.valueOf(commandOptions.pathToJsonFile))
+ .map(path -> parseReplicaElectionData(path));
+
+ Optional<String> topicOption =
Optional.ofNullable(commandOptions.options.valueOf(commandOptions.topic));
+ Optional<Integer> partitionOption =
Optional.ofNullable(commandOptions.options.valueOf(commandOptions.partition));
+ final Optional<Set<TopicPartition>> singleTopicPartition =
+ (topicOption.isPresent() && partitionOption.isPresent()) ?
+ Optional.of(Collections.singleton(new
TopicPartition(topicOption.get(), partitionOption.get()))) :
+ Optional.empty();
+
+ /* Note: No need to look at --all-topic-partitions as we want this to
be None if it is use.
+ * The validate function should be checking that this option is
required if the --topic and --path-to-json-file
+ * are not specified.
+ */
+ Optional<Set<TopicPartition>> topicPartitions =
jsonFileTopicPartitions.map(Optional::of).orElse(singleTopicPartition);
+
+ Properties props = new Properties();
+ if (commandOptions.options.has(commandOptions.adminClientConfig)) {
+
props.putAll(Utils.loadProps(commandOptions.options.valueOf(commandOptions.adminClientConfig)));
+ }
+ props.setProperty(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG,
commandOptions.options.valueOf(commandOptions.bootstrapServer));
+ props.setProperty(AdminClientConfig.DEFAULT_API_TIMEOUT_MS_CONFIG,
Long.toString(timeout.toMillis()));
+ props.setProperty(AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG,
Long.toString(timeout.toMillis() / 2));
+
+ Admin adminClient = Admin.create(props);
+ try {
+ electLeaders(adminClient, electionType, topicPartitions);
+ } finally {
+ adminClient.close();
+ }
+ }
+
+ private static void electLeaders(Admin client, ElectionType electionType,
Optional<Set<TopicPartition>> partitions) {
+ log.debug(String.format("Calling AdminClient.electLeaders(%s, %s)",
electionType, partitions.orElse(null)));
+ Map<TopicPartition, Optional<Throwable>> electionResults;
+ try {
+ electionResults = client.electLeaders(electionType,
partitions.orElse(null)).partitions().get();
+ } catch (ExecutionException e) {
+ if (e.getCause() instanceof TimeoutException) {
+ String message = "Timeout waiting for election results";
+ System.err.println(message);
+ throw new AdminCommandFailedException(message, e.getCause());
+ } else if (e.getCause() instanceof ClusterAuthorizationException) {
+ String message = "Not authorized to perform leader election";
+ System.err.println(message);
+ throw new AdminCommandFailedException(message,
e.getCause().getCause());
+ } else {
+ throw new RuntimeException(e);
+ }
+ } catch (InterruptedException e) {
+ System.err.println("Error while making request");
+ throw new RuntimeException(e);
+ }
+
+ Set<TopicPartition> succeeded = new HashSet<>();
+ Set<TopicPartition> noop = new HashSet<>();
+ Map<TopicPartition, Throwable> failed = new HashMap<>();
+
+ electionResults.entrySet().stream().forEach(entry -> {
+ Optional<Throwable> error = entry.getValue();
+ if (error.isPresent()) {
+ if (error.get() instanceof ElectionNotNeededException) {
+ noop.add(entry.getKey());
+ } else {
+ failed.put(entry.getKey(), error.get());
+ }
+ } else {
+ succeeded.add(entry.getKey());
+ }
+ });
+
+ if (!succeeded.isEmpty()) {
+ String partitionsAsString = succeeded.stream()
+ .map(TopicPartition::toString)
+ .collect(Collectors.joining(", "));
+ System.out.println(String.format("Successfully completed leader
election (%s) for partitions %s",
+ electionType, partitionsAsString));
+ }
+
+ if (!noop.isEmpty()) {
+ String partitionsAsString = noop.stream()
+ .map(TopicPartition::toString)
+ .collect(Collectors.joining(", "));
+ System.out.println(String.format("Valid replica already elected
for partitions %s", partitionsAsString));
+ }
+
+ if (!failed.isEmpty()) {
+ AdminCommandFailedException rootException =
+ new AdminCommandFailedException(String.format("%s replica(s)
could not be elected", failed.size()));
+ failed.entrySet().forEach(entry -> {
+ System.err.println(String.format("Error completing leader
election (%s) for partition: %s: %s",
+ electionType, entry.getKey(), entry.getValue()));
+ rootException.addSuppressed(entry.getValue());
+ });
+ throw rootException;
+ }
+ }
+
+ private static Set<TopicPartition> parseReplicaElectionData(String path) {
+ Optional<JsonValue> jsonFile;
+ try {
+ jsonFile =
Optional.ofNullable(Json.parseFull(Utils.readFileAsString(path)).getOrElse(null));
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ return jsonFile.map(js -> topicPartitions(js))
+ .orElseThrow(() -> new AdminOperationException("Replica election
data is empty"));
+ }
+
+ private static Set<TopicPartition> topicPartitions(JsonValue js) {
+ Option<Seq<TopicPartition>> topicPartitions =
js.asJsonObject().get("partitions").map(partitionsList -> {
+ Iterator<JsonObject> partitionsRaw =
partitionsList.asJsonArray().iterator()
+ .map(jsonValue -> jsonValue.asJsonObject());
+
+ Seq<TopicPartition> partitions = partitionsRaw
+ .map(p -> new TopicPartition(
+ p.get("topic").get().to(DecodeJson.DecodeString$.MODULE$),
+ (int)
p.get("partition").get().to(DecodeJson.DecodeInt$.MODULE$)))
+ .toBuffer();
+
+ Iterable<TopicPartition> duplicatePartitions =
CoreUtils.duplicates(partitions);
+ if (duplicatePartitions.nonEmpty()) {
+ throw new AdminOperationException(String.format(
+ "Replica election data contains duplicate partitions: %s",
duplicatePartitions.mkString(","))
+ );
+ }
+ return partitions;
+ });
+ if (topicPartitions.isEmpty()) {
+ throw new AdminOperationException("Replica election data is
missing \"partitions\" field");
+ } else {
+ return new
HashSet<>(JavaConverters.asJavaCollection(topicPartitions.get()));
+ }
+ }
+ private static void validate(LeaderElectionCommandOptions commandOptions) {
+ // required options: --bootstrap-server and --election-type
+ List<String> missingOptions = new ArrayList<>();
+
+ if (!commandOptions.options.has(commandOptions.bootstrapServer)) {
+
missingOptions.add(commandOptions.bootstrapServer.options().get(0).toString());
+ }
+ if (!commandOptions.options.has(commandOptions.electionType)) {
+ missingOptions.add(commandOptions.electionType.options().get(0));
+ }
+ if (!missingOptions.isEmpty()) {
+ throw new AdminCommandFailedException("Missing required option(s):
" + String.join(", ", missingOptions));
+ }
+
+ // One and only one is required: --topic, --all-topic-partitions or
--path-to-json-file
+ List<AbstractOptionSpec<?>> mutuallyExclusiveOptions = Arrays.asList(
+ commandOptions.topic,
+ commandOptions.allTopicPartitions,
+ commandOptions.pathToJsonFile
+ );
+
+ long mutuallyExclusiveOptionsCount = mutuallyExclusiveOptions.stream()
+ .filter(abstractOptionSpec ->
commandOptions.options.has(abstractOptionSpec))
+ .count();
+ // 1 is the only correct configuration, don't throw an exception
+ if (mutuallyExclusiveOptionsCount != 1) {
+ throw new AdminCommandFailedException(
+ "One and only one of the following options is required: " +
+ mutuallyExclusiveOptions.stream().map(opt ->
opt.options().get(0)).collect(Collectors.joining(", "))
+ );
+ }
+ // --partition if and only if --topic is used
+ if (commandOptions.options.has(commandOptions.topic) &&
!commandOptions.options.has(commandOptions.partition)) {
+ throw new AdminCommandFailedException(String.format("Missing
required option(s): %s",
+ commandOptions.partition.options().get(0)));
+ }
+
+ if (!commandOptions.options.has(commandOptions.topic) &&
commandOptions.options.has(commandOptions.partition)) {
+ throw new AdminCommandFailedException(String.format("Option %s is
only allowed if %s is used",
+ commandOptions.partition.options().get(0),
+ commandOptions.topic.options().get(0)
+ ));
+ }
+ }
+
+ static class LeaderElectionCommandOptions extends CommandDefaultOptions {
Review Comment:
```suggestion
static class LeaderElectionCommandOptions extends CommandDefaultOptions {
```
##########
tools/src/test/java/org/apache/kafka/tools/LeaderElectionCommandTest.java:
##########
@@ -0,0 +1,330 @@
+/*
+ * 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.tools;
+
+import kafka.common.AdminCommandFailedException;
+import kafka.server.KafkaConfig;
+import kafka.server.KafkaServer;
+import kafka.test.ClusterConfig;
+import kafka.test.ClusterInstance;
+import kafka.test.annotation.ClusterTest;
+import kafka.test.annotation.ClusterTestDefaults;
+import kafka.test.annotation.Type;
+import kafka.test.junit.ClusterTestExtensions;
+import kafka.utils.TestUtils;
+import org.apache.kafka.clients.admin.Admin;
+import org.apache.kafka.clients.admin.AdminClientConfig;
+import org.apache.kafka.clients.admin.CreateTopicsResult;
+import org.apache.kafka.clients.admin.NewTopic;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.errors.UnknownTopicOrPartitionException;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Tag;
+import org.junit.jupiter.api.extension.ExtendWith;
+import scala.collection.JavaConverters;
+
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.time.Duration;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ExecutionException;
+
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+
+@SuppressWarnings("deprecation")
+@ExtendWith(value = ClusterTestExtensions.class)
+@ClusterTestDefaults(clusterType = Type.ALL, brokers = 3)
+@Tag("integration")
+public class LeaderElectionCommandTest {
+ private final ClusterInstance cluster;
+ int broker1 = 0;
+ int broker2 = 1;
+ int broker3 = 2;
+
+ public LeaderElectionCommandTest(ClusterInstance cluster) {
+ this.cluster = cluster;
+ }
+
Review Comment:
```suggestion
```
##########
tools/src/main/java/org/apache/kafka/tools/LeaderElectionCommand.java:
##########
@@ -0,0 +1,335 @@
+/*
+ * 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.tools;
+
+import joptsimple.AbstractOptionSpec;
+import joptsimple.ArgumentAcceptingOptionSpec;
+import joptsimple.OptionSpecBuilder;
+import joptsimple.util.EnumConverter;
+import kafka.admin.AdminOperationException;
+import kafka.common.AdminCommandFailedException;
+import kafka.utils.CoreUtils;
+import kafka.utils.Json;
+import kafka.utils.json.DecodeJson;
+import kafka.utils.json.JsonObject;
+import kafka.utils.json.JsonValue;
+import org.apache.kafka.clients.admin.Admin;
+import org.apache.kafka.clients.admin.AdminClientConfig;
+import org.apache.kafka.common.ElectionType;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.errors.ClusterAuthorizationException;
+import org.apache.kafka.common.errors.ElectionNotNeededException;
+import org.apache.kafka.common.errors.TimeoutException;
+import org.apache.kafka.common.utils.Utils;
+import org.apache.kafka.server.util.CommandDefaultOptions;
+import org.apache.kafka.server.util.CommandLineUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import scala.Option;
+import scala.collection.Iterable;
+import scala.collection.Iterator;
+import scala.collection.JavaConverters;
+import scala.collection.mutable.Seq;
+
+import java.io.IOException;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Properties;
+import java.util.Set;
+import java.util.concurrent.ExecutionException;
+import java.util.stream.Collectors;
+
+public class LeaderElectionCommand {
+ private static final Logger log =
LoggerFactory.getLogger(LeaderElectionCommand.class);
Review Comment:
We can probably get rid of this. There is only one log statement that just
prints input parameters, not much useful.
##########
tools/src/main/java/org/apache/kafka/tools/LeaderElectionCommand.java:
##########
@@ -0,0 +1,335 @@
+/*
+ * 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.tools;
+
+import joptsimple.AbstractOptionSpec;
+import joptsimple.ArgumentAcceptingOptionSpec;
+import joptsimple.OptionSpecBuilder;
+import joptsimple.util.EnumConverter;
+import kafka.admin.AdminOperationException;
+import kafka.common.AdminCommandFailedException;
+import kafka.utils.CoreUtils;
+import kafka.utils.Json;
+import kafka.utils.json.DecodeJson;
+import kafka.utils.json.JsonObject;
+import kafka.utils.json.JsonValue;
+import org.apache.kafka.clients.admin.Admin;
+import org.apache.kafka.clients.admin.AdminClientConfig;
+import org.apache.kafka.common.ElectionType;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.errors.ClusterAuthorizationException;
+import org.apache.kafka.common.errors.ElectionNotNeededException;
+import org.apache.kafka.common.errors.TimeoutException;
+import org.apache.kafka.common.utils.Utils;
+import org.apache.kafka.server.util.CommandDefaultOptions;
+import org.apache.kafka.server.util.CommandLineUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import scala.Option;
+import scala.collection.Iterable;
+import scala.collection.Iterator;
+import scala.collection.JavaConverters;
+import scala.collection.mutable.Seq;
+
+import java.io.IOException;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Properties;
+import java.util.Set;
+import java.util.concurrent.ExecutionException;
+import java.util.stream.Collectors;
+
+public class LeaderElectionCommand {
+ private static final Logger log =
LoggerFactory.getLogger(LeaderElectionCommand.class);
+
+ public static void main(String... args) {
+ try {
+ run(Duration.ofSeconds(30), args);
+ } catch (TerseException e) {
+ System.err.println(e.getMessage());
+ } catch (Exception e) {
+ System.err.println(e.getMessage());
+ System.err.println(Utils.stackTrace(e));
+ }
+ }
+
+ static void run(Duration timeout, String... args) throws Exception {
+ LeaderElectionCommandOptions commandOptions = new
LeaderElectionCommandOptions(args);
+
+ CommandLineUtils.maybePrintHelpOrVersion(
+ commandOptions,
+ "This tool attempts to elect a new leader for a set of topic
partitions. The type of elections supported are preferred replicas and unclean
replicas."
+ );
+
+ validate(commandOptions);
+ ElectionType electionType =
commandOptions.options.valueOf(commandOptions.electionType);
+ Optional<Set<TopicPartition>> jsonFileTopicPartitions =
+
Optional.ofNullable(commandOptions.options.valueOf(commandOptions.pathToJsonFile))
+ .map(path -> parseReplicaElectionData(path));
+
+ Optional<String> topicOption =
Optional.ofNullable(commandOptions.options.valueOf(commandOptions.topic));
+ Optional<Integer> partitionOption =
Optional.ofNullable(commandOptions.options.valueOf(commandOptions.partition));
+ final Optional<Set<TopicPartition>> singleTopicPartition =
+ (topicOption.isPresent() && partitionOption.isPresent()) ?
+ Optional.of(Collections.singleton(new
TopicPartition(topicOption.get(), partitionOption.get()))) :
+ Optional.empty();
+
+ /* Note: No need to look at --all-topic-partitions as we want this to
be None if it is use.
+ * The validate function should be checking that this option is
required if the --topic and --path-to-json-file
+ * are not specified.
+ */
+ Optional<Set<TopicPartition>> topicPartitions =
jsonFileTopicPartitions.map(Optional::of).orElse(singleTopicPartition);
+
+ Properties props = new Properties();
+ if (commandOptions.options.has(commandOptions.adminClientConfig)) {
+
props.putAll(Utils.loadProps(commandOptions.options.valueOf(commandOptions.adminClientConfig)));
+ }
+ props.setProperty(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG,
commandOptions.options.valueOf(commandOptions.bootstrapServer));
+ props.setProperty(AdminClientConfig.DEFAULT_API_TIMEOUT_MS_CONFIG,
Long.toString(timeout.toMillis()));
+ props.setProperty(AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG,
Long.toString(timeout.toMillis() / 2));
+
+ Admin adminClient = Admin.create(props);
Review Comment:
The Admin interface extends AutoCloseable, so we can use try-with-resources
here.
##########
tools/src/main/java/org/apache/kafka/tools/LeaderElectionCommand.java:
##########
@@ -0,0 +1,335 @@
+/*
+ * 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.tools;
+
+import joptsimple.AbstractOptionSpec;
+import joptsimple.ArgumentAcceptingOptionSpec;
+import joptsimple.OptionSpecBuilder;
+import joptsimple.util.EnumConverter;
+import kafka.admin.AdminOperationException;
+import kafka.common.AdminCommandFailedException;
+import kafka.utils.CoreUtils;
+import kafka.utils.Json;
+import kafka.utils.json.DecodeJson;
+import kafka.utils.json.JsonObject;
+import kafka.utils.json.JsonValue;
+import org.apache.kafka.clients.admin.Admin;
+import org.apache.kafka.clients.admin.AdminClientConfig;
+import org.apache.kafka.common.ElectionType;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.errors.ClusterAuthorizationException;
+import org.apache.kafka.common.errors.ElectionNotNeededException;
+import org.apache.kafka.common.errors.TimeoutException;
+import org.apache.kafka.common.utils.Utils;
+import org.apache.kafka.server.util.CommandDefaultOptions;
+import org.apache.kafka.server.util.CommandLineUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import scala.Option;
+import scala.collection.Iterable;
+import scala.collection.Iterator;
+import scala.collection.JavaConverters;
+import scala.collection.mutable.Seq;
+
+import java.io.IOException;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Properties;
+import java.util.Set;
+import java.util.concurrent.ExecutionException;
+import java.util.stream.Collectors;
+
+public class LeaderElectionCommand {
+ private static final Logger log =
LoggerFactory.getLogger(LeaderElectionCommand.class);
+
+ public static void main(String... args) {
+ try {
+ run(Duration.ofSeconds(30), args);
+ } catch (TerseException e) {
+ System.err.println(e.getMessage());
+ } catch (Exception e) {
+ System.err.println(e.getMessage());
+ System.err.println(Utils.stackTrace(e));
+ }
+ }
+
+ static void run(Duration timeout, String... args) throws Exception {
+ LeaderElectionCommandOptions commandOptions = new
LeaderElectionCommandOptions(args);
+
+ CommandLineUtils.maybePrintHelpOrVersion(
+ commandOptions,
+ "This tool attempts to elect a new leader for a set of topic
partitions. The type of elections supported are preferred replicas and unclean
replicas."
+ );
+
+ validate(commandOptions);
+ ElectionType electionType =
commandOptions.options.valueOf(commandOptions.electionType);
+ Optional<Set<TopicPartition>> jsonFileTopicPartitions =
+
Optional.ofNullable(commandOptions.options.valueOf(commandOptions.pathToJsonFile))
+ .map(path -> parseReplicaElectionData(path));
+
+ Optional<String> topicOption =
Optional.ofNullable(commandOptions.options.valueOf(commandOptions.topic));
+ Optional<Integer> partitionOption =
Optional.ofNullable(commandOptions.options.valueOf(commandOptions.partition));
+ final Optional<Set<TopicPartition>> singleTopicPartition =
+ (topicOption.isPresent() && partitionOption.isPresent()) ?
+ Optional.of(Collections.singleton(new
TopicPartition(topicOption.get(), partitionOption.get()))) :
+ Optional.empty();
+
+ /* Note: No need to look at --all-topic-partitions as we want this to
be None if it is use.
Review Comment:
```suggestion
/* Note: No need to look at --all-topic-partitions as we want this
to be null if it is use.
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
