You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by mi...@apache.org on 2024/04/15 15:25:25 UTC
(kafka) branch trunk updated: MINOR: Various cleanups in tools (#15709)
This is an automated email from the ASF dual-hosted git repository.
mimaison pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new 395fdae5f23 MINOR: Various cleanups in tools (#15709)
395fdae5f23 is described below
commit 395fdae5f23bb8202cfc5b0a0e4630e9c32e0edd
Author: Mickael Maison <mi...@users.noreply.github.com>
AuthorDate: Mon Apr 15 17:25:17 2024 +0200
MINOR: Various cleanups in tools (#15709)
Reviewers: Chia-Ping Tsai <ch...@gmail.com>
---
.../kafka/tools/ClientCompatibilityTest.java | 2 +-
.../apache/kafka/tools/ClientMetricsCommand.java | 25 ++++++++--------------
.../org/apache/kafka/tools/GetOffsetShell.java | 6 ++----
.../tools/consumer/group/ConsumerGroupCommand.java | 3 +--
.../kafka/tools/ClientMetricsCommandTest.java | 15 +++++--------
.../kafka/tools/DelegationTokenCommandTest.java | 2 +-
.../kafka/tools/DeleteRecordsCommandTest.java | 10 ++++-----
.../java/org/apache/kafka/tools/JmxToolTest.java | 3 ++-
.../kafka/tools/ProducerPerformanceTest.java | 2 +-
.../kafka/tools/PushHttpMetricsReporterTest.java | 3 ++-
.../apache/kafka/tools/StreamsResetterTest.java | 10 ++++-----
.../org/apache/kafka/tools/ToolsTestUtils.java | 8 +++----
.../kafka/tools/TopicCommandIntegrationTest.java | 14 ++++++------
.../org/apache/kafka/tools/TopicCommandTest.java | 4 ++--
.../kafka/tools/TransactionsCommandTest.java | 8 +------
.../org/apache/kafka/tools/api/RecordReader.java | 2 +-
16 files changed, 46 insertions(+), 71 deletions(-)
diff --git a/tools/src/main/java/org/apache/kafka/tools/ClientCompatibilityTest.java b/tools/src/main/java/org/apache/kafka/tools/ClientCompatibilityTest.java
index 9a04f068a66..6dbbc52c417 100644
--- a/tools/src/main/java/org/apache/kafka/tools/ClientCompatibilityTest.java
+++ b/tools/src/main/java/org/apache/kafka/tools/ClientCompatibilityTest.java
@@ -415,7 +415,7 @@ public class ClientCompatibilityTest {
new ClientCompatibilityTestDeserializer(testConfig.expectClusterId);
try (final KafkaConsumer<byte[], byte[]> consumer = new KafkaConsumer<>(consumerProps, deserializer, deserializer)) {
final List<PartitionInfo> partitionInfos = consumer.partitionsFor(testConfig.topic);
- if (partitionInfos.size() < 1)
+ if (partitionInfos.isEmpty())
throw new RuntimeException("Expected at least one partition for topic " + testConfig.topic);
final Map<TopicPartition, Long> timestampsToSearch = new HashMap<>();
final LinkedList<TopicPartition> topicPartitions = new LinkedList<>();
diff --git a/tools/src/main/java/org/apache/kafka/tools/ClientMetricsCommand.java b/tools/src/main/java/org/apache/kafka/tools/ClientMetricsCommand.java
index 395206d383b..2b29bb8c2b6 100644
--- a/tools/src/main/java/org/apache/kafka/tools/ClientMetricsCommand.java
+++ b/tools/src/main/java/org/apache/kafka/tools/ClientMetricsCommand.java
@@ -49,6 +49,7 @@ import java.util.Properties;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
+import java.util.stream.Stream;
public class ClientMetricsCommand {
private static final Logger LOG = LoggerFactory.getLogger(ClientMetricsCommand.class);
@@ -74,9 +75,8 @@ public class ClientMetricsCommand {
Properties config = opts.commandConfig();
config.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, opts.bootstrapServer());
- ClientMetricsService service = new ClientMetricsService(config);
int exitCode = 0;
- try {
+ try (ClientMetricsService service = new ClientMetricsService(config)) {
if (opts.hasAlterOption()) {
service.alterClientMetrics(opts);
} else if (opts.hasDescribeOption()) {
@@ -84,7 +84,7 @@ public class ClientMetricsCommand {
} else if (opts.hasDeleteOption()) {
service.deleteClientMetrics(opts);
} else if (opts.hasListOption()) {
- service.listClientMetrics(opts);
+ service.listClientMetrics();
}
} catch (ExecutionException e) {
Throwable cause = e.getCause();
@@ -98,7 +98,6 @@ public class ClientMetricsCommand {
printException(t);
exitCode = 1;
} finally {
- service.close();
Exit.exit(exitCode);
}
}
@@ -120,8 +119,8 @@ public class ClientMetricsCommand {
Map<String, String> configsToBeSet = new HashMap<>();
opts.interval().map(intervalVal -> configsToBeSet.put("interval.ms", intervalVal.toString()));
- opts.metrics().map(metricslist -> configsToBeSet.put("metrics", metricslist.stream().collect(Collectors.joining(","))));
- opts.match().map(matchlist -> configsToBeSet.put("match", matchlist.stream().collect(Collectors.joining(","))));
+ opts.metrics().map(metricslist -> configsToBeSet.put("metrics", String.join(",", metricslist)));
+ opts.match().map(matchlist -> configsToBeSet.put("match", String.join(",", matchlist)));
ConfigResource configResource = new ConfigResource(ConfigResource.Type.CLIENT_METRICS, entityName);
AlterConfigsOptions alterOptions = new AlterConfigsOptions().timeoutMs(30000).validateOnly(false);
@@ -168,7 +167,7 @@ public class ClientMetricsCommand {
}
}
- public void listClientMetrics(ClientMetricsCommandOptions opts) throws Exception {
+ public void listClientMetrics() throws Exception {
Collection<ClientMetricsResourceListing> resources = adminClient.listClientMetricsResources()
.all().get(30, TimeUnit.SECONDS);
String results = resources.stream().map(ClientMetricsResourceListing::name).collect(Collectors.joining("\n"));
@@ -212,8 +211,6 @@ public class ClientMetricsCommand {
private final ArgumentAcceptingOptionSpec<Integer> intervalOpt;
- private final String nl;
-
private final ArgumentAcceptingOptionSpec<String> matchOpt;
private final ArgumentAcceptingOptionSpec<String> metricsOpt;
@@ -244,15 +241,14 @@ public class ClientMetricsCommand {
.describedAs("push interval")
.ofType(java.lang.Integer.class);
- nl = System.getProperty("line.separator");
+ String nl = System.lineSeparator();
String[] matchSelectors = new String[] {
"client_id", "client_instance_id", "client_software_name",
"client_software_version", "client_source_address", "client_source_port"
};
String matchSelectorNames = Arrays.stream(matchSelectors).map(config -> "\t" + config).collect(Collectors.joining(nl));
- matchOpt = parser.accepts("match", "Matching selector 'k1=v1,k2=v2'. The following is a list of valid selector names: " +
- nl + matchSelectorNames)
+ matchOpt = parser.accepts("match", "Matching selector 'k1=v1,k2=v2'. The following is a list of valid selector names: " + nl + matchSelectorNames)
.withRequiredArg()
.describedAs("k1=v1,k2=v2")
.ofType(String.class)
@@ -347,10 +343,7 @@ public class ClientMetricsCommand {
CommandLineUtils.maybePrintHelpOrVersion(this, "This tool helps to manipulate and describe client metrics configurations.");
// should have exactly one action
- long actions =
- Arrays.asList(alterOpt, deleteOpt, describeOpt, listOpt)
- .stream().filter(options::has)
- .count();
+ long actions = Stream.of(alterOpt, deleteOpt, describeOpt, listOpt).filter(options::has).count();
if (actions != 1)
CommandLineUtils.printUsageAndExit(parser, "Command must include exactly one action: --alter, --delete, --describe or --list.");
diff --git a/tools/src/main/java/org/apache/kafka/tools/GetOffsetShell.java b/tools/src/main/java/org/apache/kafka/tools/GetOffsetShell.java
index 0d90327bbe7..4ed3d3a5ca5 100644
--- a/tools/src/main/java/org/apache/kafka/tools/GetOffsetShell.java
+++ b/tools/src/main/java/org/apache/kafka/tools/GetOffsetShell.java
@@ -95,8 +95,6 @@ public class GetOffsetShell {
}
private static class GetOffsetShellOptions extends CommandDefaultOptions {
- private final OptionSpec<String> brokerListOpt;
- private final OptionSpec<String> bootstrapServerOpt;
private final OptionSpec<String> topicPartitionsOpt;
private final OptionSpec<String> topicOpt;
private final OptionSpec<String> partitionsOpt;
@@ -108,11 +106,11 @@ public class GetOffsetShell {
public GetOffsetShellOptions(String[] args) throws TerseException {
super(args);
- brokerListOpt = parser.accepts("broker-list", "DEPRECATED, use --bootstrap-server instead; ignored if --bootstrap-server is specified. The server(s) to connect to in the form HOST1:PORT1,HOST2:PORT2.")
+ OptionSpec<String> brokerListOpt = parser.accepts("broker-list", "DEPRECATED, use --bootstrap-server instead; ignored if --bootstrap-server is specified. The server(s) to connect to in the form HOST1:PORT1,HOST2:PORT2.")
.withRequiredArg()
.describedAs("HOST1:PORT1,...,HOST3:PORT3")
.ofType(String.class);
- bootstrapServerOpt = parser.accepts("bootstrap-server", "REQUIRED. The server(s) to connect to in the form HOST1:PORT1,HOST2:PORT2.")
+ OptionSpec<String> bootstrapServerOpt = parser.accepts("bootstrap-server", "REQUIRED. The server(s) to connect to in the form HOST1:PORT1,HOST2:PORT2.")
.requiredUnless("broker-list")
.withRequiredArg()
.describedAs("HOST1:PORT1,...,HOST3:PORT3")
diff --git a/tools/src/main/java/org/apache/kafka/tools/consumer/group/ConsumerGroupCommand.java b/tools/src/main/java/org/apache/kafka/tools/consumer/group/ConsumerGroupCommand.java
index 5aafde1c0fe..49dd4efc69d 100644
--- a/tools/src/main/java/org/apache/kafka/tools/consumer/group/ConsumerGroupCommand.java
+++ b/tools/src/main/java/org/apache/kafka/tools/consumer/group/ConsumerGroupCommand.java
@@ -354,8 +354,7 @@ public class ConsumerGroupCommand {
}
}
- String format = "\n%" + (-maxGroupLen) + "s %" + (-maxTopicLen) + "s %-10s %-15s %-15s %-15s %" + (-maxConsumerIdLen) + "s %" + (-maxHostLen) + "s %s";
- return format;
+ return "\n%" + (-maxGroupLen) + "s %" + (-maxTopicLen) + "s %-10s %-15s %-15s %-15s %" + (-maxConsumerIdLen) + "s %" + (-maxHostLen) + "s %s";
}
private void printMembers(Map<String, Entry<Optional<String>, Optional<Collection<MemberAssignmentState>>>> members, boolean verbose) {
diff --git a/tools/src/test/java/org/apache/kafka/tools/ClientMetricsCommandTest.java b/tools/src/test/java/org/apache/kafka/tools/ClientMetricsCommandTest.java
index ce71674819b..4fc81156fba 100644
--- a/tools/src/test/java/org/apache/kafka/tools/ClientMetricsCommandTest.java
+++ b/tools/src/test/java/org/apache/kafka/tools/ClientMetricsCommandTest.java
@@ -28,10 +28,8 @@ import org.apache.kafka.common.config.ConfigResource;
import org.apache.kafka.common.protocol.Errors;
import org.junit.jupiter.api.Test;
-import java.util.Arrays;
import java.util.Collections;
import java.util.concurrent.ExecutionException;
-import java.util.stream.Collectors;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
@@ -42,8 +40,8 @@ import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
public class ClientMetricsCommandTest {
- private String bootstrapServer = "localhost:9092";
- private String clientMetricsName = "cm";
+ private final String bootstrapServer = "localhost:9092";
+ private final String clientMetricsName = "cm";
@Test
public void testOptionsNoActionFails() {
@@ -237,13 +235,12 @@ public class ClientMetricsCommandTest {
String capturedOutput = ToolsTestUtils.captureStandardOut(() -> {
try {
- service.listClientMetrics(new ClientMetricsCommand.ClientMetricsCommandOptions(
- new String[]{"--bootstrap-server", bootstrapServer, "--list"}));
+ service.listClientMetrics();
} catch (Throwable t) {
fail(t);
}
});
- assertEquals("one,two", Arrays.stream(capturedOutput.split("\n")).collect(Collectors.joining(",")));
+ assertEquals("one,two", String.join(",", capturedOutput.split("\n")));
}
@Test
@@ -254,9 +251,7 @@ public class ClientMetricsCommandTest {
ListClientMetricsResourcesResult result = AdminClientTestUtils.listClientMetricsResourcesResult(Errors.UNSUPPORTED_VERSION.exception());
when(adminClient.listClientMetricsResources()).thenReturn(result);
- assertThrows(ExecutionException.class,
- () -> service.listClientMetrics(new ClientMetricsCommand.ClientMetricsCommandOptions(
- new String[] {"--bootstrap-server", bootstrapServer, "--list"})));
+ assertThrows(ExecutionException.class, () -> service.listClientMetrics());
}
private void assertInitializeInvalidOptionsExitCode(int expected, String[] options) {
diff --git a/tools/src/test/java/org/apache/kafka/tools/DelegationTokenCommandTest.java b/tools/src/test/java/org/apache/kafka/tools/DelegationTokenCommandTest.java
index fb923629dca..4361c5f54f8 100644
--- a/tools/src/test/java/org/apache/kafka/tools/DelegationTokenCommandTest.java
+++ b/tools/src/test/java/org/apache/kafka/tools/DelegationTokenCommandTest.java
@@ -91,7 +91,7 @@ public class DelegationTokenCommandTest {
args.add("--command-config");
args.add("testfile");
args.add("--describe");
- if (!owner.equals("")) {
+ if (!owner.isEmpty()) {
args.add("--owner-principal");
args.add(owner);
}
diff --git a/tools/src/test/java/org/apache/kafka/tools/DeleteRecordsCommandTest.java b/tools/src/test/java/org/apache/kafka/tools/DeleteRecordsCommandTest.java
index 0f4e551a48a..104f21e327d 100644
--- a/tools/src/test/java/org/apache/kafka/tools/DeleteRecordsCommandTest.java
+++ b/tools/src/test/java/org/apache/kafka/tools/DeleteRecordsCommandTest.java
@@ -124,10 +124,9 @@ class DeleteRecordsCommandUnitTest {
"--bootstrap-server", "localhost:9092",
"--offset-json-file", "/not/existing/file"
}, System.out));
- assertEquals(1, DeleteRecordsCommand.mainNoExit(new String[]{
+ assertEquals(1, DeleteRecordsCommand.mainNoExit(
"--bootstrap-server", "localhost:9092",
- "--offset-json-file", "/not/existing/file"
- }));
+ "--offset-json-file", "/not/existing/file"));
}
@Test
@@ -137,11 +136,10 @@ class DeleteRecordsCommandUnitTest {
"--offset-json-file", "/not/existing/file",
"--command-config", "/another/not/existing/file"
}, System.out));
- assertEquals(1, DeleteRecordsCommand.mainNoExit(new String[] {
+ assertEquals(1, DeleteRecordsCommand.mainNoExit(
"--bootstrap-server", "localhost:9092",
"--offset-json-file", "/not/existing/file",
- "--command-config", "/another/not/existing/file"
- }));
+ "--command-config", "/another/not/existing/file"));
}
@Test
diff --git a/tools/src/test/java/org/apache/kafka/tools/JmxToolTest.java b/tools/src/test/java/org/apache/kafka/tools/JmxToolTest.java
index b789d75fae5..59c34df33d5 100644
--- a/tools/src/test/java/org/apache/kafka/tools/JmxToolTest.java
+++ b/tools/src/test/java/org/apache/kafka/tools/JmxToolTest.java
@@ -42,6 +42,7 @@ import java.util.Map;
import static java.lang.String.format;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;
public class JmxToolTest {
@@ -167,7 +168,7 @@ public class JmxToolTest {
assertNormalExit();
Map<String, String> csv = parseCsv(out);
- assertTrue(csv.size() > 0);
+ assertFalse(csv.isEmpty());
}
@Test
diff --git a/tools/src/test/java/org/apache/kafka/tools/ProducerPerformanceTest.java b/tools/src/test/java/org/apache/kafka/tools/ProducerPerformanceTest.java
index f97e34dda9c..39893e33bb1 100644
--- a/tools/src/test/java/org/apache/kafka/tools/ProducerPerformanceTest.java
+++ b/tools/src/test/java/org/apache/kafka/tools/ProducerPerformanceTest.java
@@ -228,7 +228,7 @@ public class ProducerPerformanceTest {
}
@Test
- public void testStatsInitializationWithLargeNumRecords() throws Exception {
+ public void testStatsInitializationWithLargeNumRecords() {
long numRecords = Long.MAX_VALUE;
assertDoesNotThrow(() -> new ProducerPerformance.Stats(numRecords, 5000));
}
diff --git a/tools/src/test/java/org/apache/kafka/tools/PushHttpMetricsReporterTest.java b/tools/src/test/java/org/apache/kafka/tools/PushHttpMetricsReporterTest.java
index e4ed958d93b..80765eaaedf 100644
--- a/tools/src/test/java/org/apache/kafka/tools/PushHttpMetricsReporterTest.java
+++ b/tools/src/test/java/org/apache/kafka/tools/PushHttpMetricsReporterTest.java
@@ -31,6 +31,7 @@ import static org.mockito.Mockito.when;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import java.io.IOException;
+import java.util.Comparator;
import java.util.List;
import org.apache.kafka.common.MetricName;
import org.apache.kafka.common.config.ConfigException;
@@ -235,7 +236,7 @@ public class PushHttpMetricsReporterTest {
assertEquals(3, metrics.size());
List<JsonNode> metricsList = Arrays.asList(metrics.get(0), metrics.get(1), metrics.get(2));
// Sort metrics based on name so that we can verify the value for each metric below
- metricsList.sort((m1, m2) -> m1.get("name").textValue().compareTo(m2.get("name").textValue()));
+ metricsList.sort(Comparator.comparing(m -> m.get("name").textValue()));
JsonNode m1 = metricsList.get(0);
assertEquals("name1", m1.get("name").textValue());
diff --git a/tools/src/test/java/org/apache/kafka/tools/StreamsResetterTest.java b/tools/src/test/java/org/apache/kafka/tools/StreamsResetterTest.java
index 84a7690bd48..b8bc84308f9 100644
--- a/tools/src/test/java/org/apache/kafka/tools/StreamsResetterTest.java
+++ b/tools/src/test/java/org/apache/kafka/tools/StreamsResetterTest.java
@@ -24,7 +24,6 @@ import org.apache.kafka.clients.consumer.OffsetAndTimestamp;
import org.apache.kafka.clients.consumer.OffsetResetStrategy;
import org.apache.kafka.common.Cluster;
import org.apache.kafka.common.Node;
-import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.TopicPartitionInfo;
import org.junit.jupiter.api.BeforeEach;
@@ -39,12 +38,11 @@ import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ExecutionException;
-import java.util.concurrent.TimeUnit;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertTrue;
-@Timeout(value = 600, unit = TimeUnit.SECONDS)
+@Timeout(value = 600)
public class StreamsResetterTest {
private static final String TOPIC = "topic1";
private final StreamsResetter streamsResetter = new StreamsResetter();
@@ -256,7 +254,7 @@ public class StreamsResetterTest {
public void shouldDeleteTopic() throws InterruptedException, ExecutionException {
final Cluster cluster = createCluster(1);
try (final MockAdminClient adminClient = new MockAdminClient(cluster.nodes(), cluster.nodeById(0))) {
- final TopicPartitionInfo topicPartitionInfo = new TopicPartitionInfo(0, cluster.nodeById(0), cluster.nodes(), Collections.<Node>emptyList());
+ final TopicPartitionInfo topicPartitionInfo = new TopicPartitionInfo(0, cluster.nodeById(0), cluster.nodes(), Collections.emptyList());
adminClient.addTopic(false, TOPIC, Collections.singletonList(topicPartitionInfo), null);
streamsResetter.doDelete(Collections.singletonList(TOPIC), adminClient);
assertEquals(Collections.emptySet(), adminClient.listTopics().names().get());
@@ -300,8 +298,8 @@ public class StreamsResetterTest {
nodes.put(i, new Node(i, "localhost", 8121 + i));
}
return new Cluster("mockClusterId", nodes.values(),
- Collections.<PartitionInfo>emptySet(), Collections.<String>emptySet(),
- Collections.<String>emptySet(), nodes.get(0));
+ Collections.emptySet(), Collections.emptySet(),
+ Collections.emptySet(), nodes.get(0));
}
private static class EmptyPartitionConsumer<K, V> extends MockConsumer<K, V> {
diff --git a/tools/src/test/java/org/apache/kafka/tools/ToolsTestUtils.java b/tools/src/test/java/org/apache/kafka/tools/ToolsTestUtils.java
index bbb70b76ad6..2e90fe23366 100644
--- a/tools/src/test/java/org/apache/kafka/tools/ToolsTestUtils.java
+++ b/tools/src/test/java/org/apache/kafka/tools/ToolsTestUtils.java
@@ -48,7 +48,7 @@ public class ToolsTestUtils {
/** @see TestInfoUtils#TestWithParameterizedQuorumAndGroupProtocolNames() */
public static final String TEST_WITH_PARAMETERIZED_QUORUM_AND_GROUP_PROTOCOL_NAMES = "{displayName}.quorum={0}.groupProtocol={1}";
- private static int randomPort = 0;
+ private static final int RANDOM_PORT = 0;
public static String captureStandardOut(Runnable runnable) {
return captureStandardStream(false, runnable);
@@ -68,7 +68,7 @@ public class ToolsTestUtils {
System.setOut(tempStream);
try {
runnable.run();
- return new String(outputStream.toByteArray()).trim();
+ return outputStream.toString().trim();
} finally {
if (isErr)
System.setErr(currentStream);
@@ -100,11 +100,11 @@ public class ToolsTestUtils {
List<Properties> result = new ArrayList<>();
int endingIdNumber = startingIdNumber + numConfigs - 1;
for (int node = startingIdNumber; node <= endingIdNumber; node++) {
- result.add(TestUtils.createBrokerConfig(node, zkConnect, true, true, randomPort,
+ result.add(TestUtils.createBrokerConfig(node, zkConnect, true, true, RANDOM_PORT,
scala.Option.empty(),
scala.Option.empty(),
scala.Option.empty(),
- true, false, randomPort, false, randomPort, false, randomPort,
+ true, false, RANDOM_PORT, false, RANDOM_PORT, false, RANDOM_PORT,
scala.Option.apply(rackInfo.get(node)),
logDirCount, enableToken, numPartitions, defaultReplicationFactor, false));
}
diff --git a/tools/src/test/java/org/apache/kafka/tools/TopicCommandIntegrationTest.java b/tools/src/test/java/org/apache/kafka/tools/TopicCommandIntegrationTest.java
index 5a57046705b..9da09a54fc4 100644
--- a/tools/src/test/java/org/apache/kafka/tools/TopicCommandIntegrationTest.java
+++ b/tools/src/test/java/org/apache/kafka/tools/TopicCommandIntegrationTest.java
@@ -83,14 +83,12 @@ import static org.mockito.Mockito.spy;
@Tag("integration")
@SuppressWarnings("deprecation") // Added for Scala 2.12 compatibility for usages of JavaConverters
public class TopicCommandIntegrationTest extends kafka.integration.KafkaServerTestHarness implements Logging, RackAwareTest {
- private short defaultReplicationFactor = 1;
- private int defaultNumPartitions = 1;
+ private final short defaultReplicationFactor = 1;
+ private final int defaultNumPartitions = 1;
private TopicCommand.TopicService topicService;
private Admin adminClient;
private String bootstrapServer;
private String testTopicName;
- private long defaultTimeout = 10000;
-
private Buffer<KafkaBroker> scalaBrokers;
private Seq<ControllerServer> scalaControllers;
@@ -123,8 +121,8 @@ public class TopicCommandIntegrationTest extends kafka.integration.KafkaServerTe
}
private TopicCommand.TopicCommandOptions buildTopicCommandOptionsWithBootstrap(String... opts) {
- String[] finalOptions = Stream.concat(Arrays.asList(opts).stream(),
- Arrays.asList("--bootstrap-server", bootstrapServer).stream()
+ String[] finalOptions = Stream.concat(Arrays.stream(opts),
+ Stream.of("--bootstrap-server", bootstrapServer)
).toArray(String[]::new);
return new TopicCommand.TopicCommandOptions(finalOptions);
}
@@ -411,7 +409,7 @@ public class TopicCommandIntegrationTest extends kafka.integration.KafkaServerTe
org.apache.kafka.test.TestUtils.DEFAULT_MAX_WAIT_MS, 100L);
TopicDescription topicDescription = adminClient.describeTopics(Collections.singletonList(testTopicName)).topicNameValues().get(testTopicName).get();
- assertTrue(topicDescription.partitions().size() == 3, "Expected partition count to be 3. Got: " + topicDescription.partitions().size());
+ assertEquals(3, topicDescription.partitions().size(), "Expected partition count to be 3. Got: " + topicDescription.partitions().size());
List<Integer> partitionReplicas = getPartitionReplicas(topicDescription.partitions(), 2);
assertEquals(Arrays.asList(4, 2), partitionReplicas, "Expected to have replicas 4,2. Got: " + partitionReplicas);
}
@@ -1111,7 +1109,7 @@ public class TopicCommandIntegrationTest extends kafka.integration.KafkaServerTe
.collect(Collectors.toList());
partitionRackMap.put(partitionId, partitionRackValues);
} else {
- System.err.println(String.format("No mapping found for %s in `brokerRackMapping`", brokerId));
+ System.err.printf("No mapping found for %s in `brokerRackMapping`%n", brokerId);
}
});
});
diff --git a/tools/src/test/java/org/apache/kafka/tools/TopicCommandTest.java b/tools/src/test/java/org/apache/kafka/tools/TopicCommandTest.java
index 7541d0a1121..151d22ebb4f 100644
--- a/tools/src/test/java/org/apache/kafka/tools/TopicCommandTest.java
+++ b/tools/src/test/java/org/apache/kafka/tools/TopicCommandTest.java
@@ -64,8 +64,8 @@ import static org.mockito.Mockito.verify;
@Timeout(value = 60)
public class TopicCommandTest {
- private String bootstrapServer = "localhost:9092";
- private String topicName = "topicName";
+ private final String bootstrapServer = "localhost:9092";
+ private final String topicName = "topicName";
@Test
public void testIsNotUnderReplicatedWhenAdding() {
diff --git a/tools/src/test/java/org/apache/kafka/tools/TransactionsCommandTest.java b/tools/src/test/java/org/apache/kafka/tools/TransactionsCommandTest.java
index fde6434cf02..130f498ef85 100644
--- a/tools/src/test/java/org/apache/kafka/tools/TransactionsCommandTest.java
+++ b/tools/src/test/java/org/apache/kafka/tools/TransactionsCommandTest.java
@@ -443,13 +443,7 @@ public class TransactionsCommandTest {
AbortTransactionResult abortTransactionResult = Mockito.mock(AbortTransactionResult.class);
KafkaFuture<Void> abortFuture = completedFuture(null);
- final int expectedCoordinatorEpoch;
- if (coordinatorEpoch < 0) {
- expectedCoordinatorEpoch = 0;
- } else {
- expectedCoordinatorEpoch = coordinatorEpoch;
- }
-
+ int expectedCoordinatorEpoch = Math.max(coordinatorEpoch, 0);
AbortTransactionSpec expectedAbortSpec = new AbortTransactionSpec(
topicPartition, producerId, producerEpoch, expectedCoordinatorEpoch);
diff --git a/tools/tools-api/src/main/java/org/apache/kafka/tools/api/RecordReader.java b/tools/tools-api/src/main/java/org/apache/kafka/tools/api/RecordReader.java
index cda150064ad..305e68af7ad 100644
--- a/tools/tools-api/src/main/java/org/apache/kafka/tools/api/RecordReader.java
+++ b/tools/tools-api/src/main/java/org/apache/kafka/tools/api/RecordReader.java
@@ -36,7 +36,7 @@ public interface RecordReader extends Closeable, Configurable {
/**
* read byte array from input stream and then generate an iterator of producer record
- * @param {@link InputStream} of messages. the implementation does not need to close the input stream.
+ * @param inputStream {@link InputStream} of messages. the implementation does not need to close the input stream.
* @return an iterator of producer record. It should implement following rules. 1) the hasNext() method must be idempotent.
* 2) the convert error should be thrown by next() method.
*/