This is an automated email from the ASF dual-hosted git repository.
chia7712 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 123356bad50 MINOR: Scala to Java migration. Migrated some tests from
PlaintextAdminIntegrationTest to a new java class. (#22290)
123356bad50 is described below
commit 123356bad50da1418f42f2f675c963156d4bb78c
Author: Nikita Shupletsov <[email protected]>
AuthorDate: Sun Jul 26 05:46:16 2026 -0700
MINOR: Scala to Java migration. Migrated some tests from
PlaintextAdminIntegrationTest to a new java class. (#22290)
Contributing to the bigger effort of migrating scala to java
Reviewers: Sushant Mahajan <[email protected]>, Ken Huang
<[email protected]>, Chia-Ping Tsai <[email protected]>
---
.../kafka/clients/admin/AdminMetadataTest.java | 247 +++++++++++++++++++++
.../kafka/api/PlaintextAdminIntegrationTest.scala | 155 +------------
2 files changed, 248 insertions(+), 154 deletions(-)
diff --git
a/clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/admin/AdminMetadataTest.java
b/clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/admin/AdminMetadataTest.java
new file mode 100644
index 00000000000..239c8e31d50
--- /dev/null
+++
b/clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/admin/AdminMetadataTest.java
@@ -0,0 +1,247 @@
+/*
+ * 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.clients.consumer.Consumer;
+import org.apache.kafka.common.KafkaFuture;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.TopicCollection;
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.common.errors.TimeoutException;
+import org.apache.kafka.common.errors.UnknownTopicIdException;
+import org.apache.kafka.common.errors.UnknownTopicOrPartitionException;
+import org.apache.kafka.common.internals.Topic;
+import org.apache.kafka.common.test.ClusterInstance;
+import org.apache.kafka.common.test.api.ClusterTest;
+import org.apache.kafka.common.test.api.ClusterTestDefaults;
+import org.apache.kafka.common.test.api.Type;
+import org.apache.kafka.test.TestUtils;
+
+import java.time.Duration;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ExecutionException;
+
+import static org.apache.kafka.test.TestUtils.assertFutureThrows;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+@ClusterTestDefaults(brokers = 3, types = {Type.KRAFT})
+public class AdminMetadataTest {
+
+ // Incorrect broker port which can used by kafka clients in tests. This
port should not be used
+ // by any other service and hence we use a reserved port.
+ private static final int INCORRECT_BROKER_PORT = 225;
+
+ private final ClusterInstance clusterInstance;
+
+ AdminMetadataTest(ClusterInstance clusterInstance) {
+ this.clusterInstance = clusterInstance;
+ }
+
+ @ClusterTest
+ public void testListNodes() throws Exception {
+ try (Admin admin = clusterInstance.admin()) {
+ List<String> brokerStrs =
Arrays.stream(clusterInstance.bootstrapServers().split(","))
+ .sorted()
+ .toList();
+ TestUtils.waitForCondition(
+ () -> admin.describeCluster().nodes().get().size() >=
brokerStrs.size(),
+ "Timed out waiting for all brokers to be discovered");
+ List<String> nodeStrs =
admin.describeCluster().nodes().get().stream()
+ .map(node -> node.host() + ":" + node.port())
+ .sorted()
+ .toList();
+ assertEquals(String.join(",", brokerStrs), String.join(",",
nodeStrs));
+ }
+ }
+
+ @ClusterTest
+ public void testListNodesWithFencedBroker() throws Exception {
+ try (Admin admin = clusterInstance.admin()) {
+ int fencedBrokerId = Collections.max(clusterInstance.brokerIds());
+ int totalBrokers = clusterInstance.brokerIds().size();
+ clusterInstance.shutdownBroker(fencedBrokerId);
+
+ // It takes a few seconds for a broker to get fenced after being
killed,
+ // so we retry until only the non-fenced brokers are returned.
+ TestUtils.waitForCondition(
+ () -> admin.describeCluster().nodes().get().size() ==
totalBrokers - 1,
+ 20_000,
+ "Timed out waiting for broker " + fencedBrokerId + " to be
fenced");
+
+ // List nodes again but this time include the fenced broker.
+ Collection<Node> nodes = admin.describeCluster(
+ new
DescribeClusterOptions().includeFencedBrokers(true)).nodes().get();
+ assertEquals(totalBrokers, nodes.size());
+ for (Node node : nodes) {
+ if (node.id() == fencedBrokerId) {
+ assertTrue(node.isFenced());
+ } else {
+ assertFalse(node.isFenced());
+ }
+ }
+ }
+ }
+
+ @ClusterTest
+ public void testDescribeCluster() throws Exception {
+ try (Admin admin = clusterInstance.admin()) {
+ DescribeClusterResult result = admin.describeCluster();
+ Collection<Node> nodes = result.nodes().get();
+ assertEquals(clusterInstance.clusterId(),
result.clusterId().get());
+
+ // In KRaft, we return a random brokerId as the current controller.
+ Node controller = result.controller().get();
+ assertTrue(clusterInstance.brokerIds().contains(controller.id()));
+
+ Set<String> brokerEndpoints =
Set.of(clusterInstance.bootstrapServers().split(","));
+ assertEquals(brokerEndpoints.size(), nodes.size());
+ for (Node node : nodes) {
+ String hostStr = node.host() + ":" + node.port();
+ assertTrue(brokerEndpoints.contains(hostStr),
+ "Unknown host:port pair " + hostStr + " in
brokerVersionInfos");
+ }
+ }
+ }
+
+ @ClusterTest
+ public void testListTopicsWithOptionTimeoutMs() {
+ Admin admin = createInvalidAdminClient();
+ try {
+ ListTopicsOptions timeoutOption = new
ListTopicsOptions().timeoutMs(0);
+ ExecutionException exception =
assertThrows(ExecutionException.class,
+ () -> admin.listTopics(timeoutOption).names().get());
+ assertInstanceOf(TimeoutException.class, exception.getCause());
+ } finally {
+ admin.close(Duration.ZERO);
+ }
+ }
+
+ @ClusterTest
+ public void testListTopicsWithOptionListInternal() throws Exception {
+ try (Admin admin = clusterInstance.admin()) {
+ String topic = "test-topic";
+ admin.createTopics(List.of(new NewTopic(topic, 1, (short)
1))).all().get();
+ clusterInstance.waitTopicCreation(topic, 1);
+
+ try (Consumer<byte[], byte[]> consumer =
clusterInstance.consumer()) {
+ consumer.subscribe(List.of(topic));
+ consumer.poll(Duration.ofMillis(100));
+ }
+
+ TestUtils.waitForCondition(() -> {
+ Set<String> topicNames = admin.listTopics(new
ListTopicsOptions().listInternal(true)).names().get();
+ return topicNames.contains(Topic.GROUP_METADATA_TOPIC_NAME);
+ }, "Expected to see internal topic " +
Topic.GROUP_METADATA_TOPIC_NAME);
+ }
+ }
+
+ @ClusterTest
+ public void testDescribeTopicsWithOptionPartitionSizeLimitPerResponse()
throws Exception {
+ try (Admin admin = clusterInstance.admin()) {
+ String testTopic = "test-topic";
+ admin.createTopics(List.of(new NewTopic(testTopic, 3, (short)
1))).all().get();
+ clusterInstance.waitTopicCreation(testTopic, 3);
+
+ Map<String, TopicDescription> topics =
admin.describeTopics(List.of(testTopic),
+ new
DescribeTopicsOptions().partitionSizeLimitPerResponse(1)).allTopicNames().get();
+ assertEquals(1, topics.size());
+ assertEquals(3, topics.get(testTopic).partitions().size());
+ }
+ }
+
+ @ClusterTest
+ public void testDescribeTopicsWithOptionTimeoutMs() {
+ Admin admin = createInvalidAdminClient();
+ try {
+ DescribeTopicsOptions timeoutOption = new
DescribeTopicsOptions().timeoutMs(0);
+ ExecutionException exception =
assertThrows(ExecutionException.class,
+ () -> admin.describeTopics(List.of("test-topic"),
timeoutOption).allTopicNames().get());
+ assertInstanceOf(TimeoutException.class, exception.getCause());
+ } finally {
+ admin.close(Duration.ZERO);
+ }
+ }
+
+ /**
+ * describe should not auto create topics.
+ */
+ @ClusterTest
+ public void testDescribeNonExistingTopic() throws Exception {
+ try (Admin admin = clusterInstance.admin()) {
+ String existingTopic = "existing-topic";
+ admin.createTopics(List.of(new NewTopic(existingTopic, 1, (short)
1))).all().get();
+ clusterInstance.waitTopicCreation(existingTopic, 1);
+
+ String nonExistingTopic = "non-existing";
+ Map<String, KafkaFuture<TopicDescription>> results =
+ admin.describeTopics(List.of(nonExistingTopic,
existingTopic)).topicNameValues();
+ assertEquals(existingTopic,
results.get(existingTopic).get().name());
+ assertFutureThrows(UnknownTopicOrPartitionException.class,
results.get(nonExistingTopic));
+ }
+ }
+
+ @ClusterTest
+ public void testDescribeTopicsWithIds() throws Exception {
+ try (Admin admin = clusterInstance.admin()) {
+ String existingTopic = "existing-topic";
+ CreateTopicsResult createResult = admin.createTopics(List.of(new
NewTopic(existingTopic, 1, (short) 1)));
+ createResult.all().get();
+ clusterInstance.waitTopicCreation(existingTopic, 1);
+ clusterInstance.ensureConsistentMetadata();
+
+ Uuid existingTopicId = createResult.topicId(existingTopic).get();
+ Uuid nonExistingTopicId = Uuid.randomUuid();
+
+ Map<Uuid, KafkaFuture<TopicDescription>> results =
admin.describeTopics(
+ TopicCollection.ofTopicIds(List.of(existingTopicId,
nonExistingTopicId))).topicIdValues();
+ assertEquals(existingTopicId,
results.get(existingTopicId).get().topicId());
+ assertFutureThrows(UnknownTopicIdException.class,
results.get(nonExistingTopicId));
+ }
+ }
+
+ @ClusterTest
+ public void testDescribeTopicsWithNames() throws Exception {
+ try (Admin admin = clusterInstance.admin()) {
+ String existingTopic = "existing-topic";
+ CreateTopicsResult createResult = admin.createTopics(List.of(new
NewTopic(existingTopic, 1, (short) 1)));
+ createResult.all().get();
+ clusterInstance.waitTopicCreation(existingTopic, 1);
+ clusterInstance.ensureConsistentMetadata();
+
+ Uuid existingTopicId = createResult.topicId(existingTopic).get();
+ Map<String, KafkaFuture<TopicDescription>> results =
admin.describeTopics(
+
TopicCollection.ofTopicNames(List.of(existingTopic))).topicNameValues();
+ assertEquals(existingTopicId,
results.get(existingTopic).get().topicId());
+ }
+ }
+
+ private Admin createInvalidAdminClient() {
+ Map<String, Object> config = Map.of(
+ AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:" +
INCORRECT_BROKER_PORT
+ );
+ return Admin.create(config);
+ }
+}
diff --git
a/core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala
b/core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala
index 5b92f13c74c..4e8fcf7698e 100644
---
a/core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala
+++
b/core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala
@@ -48,7 +48,7 @@ import org.apache.kafka.common.requests.DeleteRecordsRequest
import org.apache.kafka.common.resource.{PatternType, ResourcePattern,
ResourceType}
import org.apache.kafka.common.serialization.{ByteArrayDeserializer,
ByteArraySerializer}
import org.apache.kafka.common.utils.{Time, Utils}
-import org.apache.kafka.common.{ConsumerGroupState, ElectionType, GroupState,
GroupType, IsolationLevel, TopicCollection, TopicPartition, TopicPartitionInfo,
TopicPartitionReplica, Uuid}
+import org.apache.kafka.common.{ConsumerGroupState, ElectionType, GroupState,
GroupType, IsolationLevel, TopicCollection, TopicPartition, TopicPartitionInfo,
TopicPartitionReplica}
import
org.apache.kafka.controller.ControllerRequestContextUtil.ANONYMOUS_CONTEXT
import org.apache.kafka.coordinator.group.{GroupConfig, GroupCoordinatorConfig}
import org.apache.kafka.network.SocketServerConfigs
@@ -630,41 +630,6 @@ class PlaintextAdminIntegrationTest extends
BaseAdminIntegrationTest {
client.close() // double close has no effect
}
- @Test
- def testListNodes(): Unit = {
- client = createAdminClient
- val brokerStrs = bootstrapServers().split(",").toList.sorted
- var nodeStrs: List[String] = null
- do {
- val nodes = client.describeCluster().nodes().get().asScala
- nodeStrs = nodes.map(node => s"${node.host}:${node.port}").toList.sorted
- } while (nodeStrs.size < brokerStrs.size)
- assertEquals(brokerStrs.mkString(","), nodeStrs.mkString(","))
- }
-
- @Test
- def testListNodesWithFencedBroker(): Unit = {
- client = createAdminClient
- val fencedBrokerId = brokers.last.config.brokerId
- killBroker(fencedBrokerId, JDuration.ofMillis(0))
- // It takes a few seconds for a broker to get fenced after being killed
- // So we retry until only 2 of 3 brokers returned in the result or the max
wait is reached
- TestUtils.retry(20000) {
-
assertTrue(client.describeCluster().nodes().get().asScala.size.equals(brokers.size
- 1))
- }
-
- // List nodes again but this time include the fenced broker
- val nodes = client.describeCluster(new
DescribeClusterOptions().includeFencedBrokers(true)).nodes().get().asScala
- assertTrue(nodes.size.equals(brokers.size))
- nodes.foreach(node => {
- if (node.id().equals(fencedBrokerId)) {
- assertTrue(node.isFenced)
- } else {
- assertFalse(node.isFenced)
- }
- })
- }
-
@Test
def testAdminClientHandlingBadIPWithoutTimeout(): Unit = {
val config = createConfig
@@ -725,124 +690,6 @@ class PlaintextAdminIntegrationTest extends
BaseAdminIntegrationTest {
} finally client.close(time.Duration.ZERO)
}
- @Test
- def testListTopicsWithOptionTimeoutMs(): Unit = {
- client = createInvalidAdminClient()
-
- try {
- val timeoutOption = new ListTopicsOptions().timeoutMs(0)
- val exception = assertThrows(classOf[ExecutionException], () =>
- client.listTopics(timeoutOption).names().get())
- assertInstanceOf(classOf[TimeoutException], exception.getCause)
- } finally client.close(time.Duration.ZERO)
- }
-
- @Test
- def testListTopicsWithOptionListInternal(): Unit = {
- client = createAdminClient
-
- val topicNames = client.listTopics(new
ListTopicsOptions().listInternal(true)).names().get()
- assertFalse(topicNames.isEmpty, "Expected to see internal topics")
- }
-
- @Test
- def testDescribeTopicsWithOptionPartitionSizeLimitPerResponse(): Unit = {
- client = createAdminClient
-
- val testTopics = Seq("test-topic")
- client.createTopics(testTopics.map(new NewTopic(_, 3,
1.toShort)).asJava).all.get()
- waitForTopics(client, testTopics, List())
-
- val topics = client.describeTopics(testTopics.asJava, new
DescribeTopicsOptions().partitionSizeLimitPerResponse(1)).allTopicNames().get()
- assertEquals(1, topics.size())
- assertEquals(3, topics.get("test-topic").partitions().size())
-
- client.deleteTopics(testTopics.asJava).all().get()
- waitForTopics(client, List(), testTopics)
- }
-
- @Test
- def testDescribeTopicsWithOptionTimeoutMs(): Unit = {
- client = createInvalidAdminClient()
-
- try {
- val timeoutOption = new DescribeTopicsOptions().timeoutMs(0)
- val exception = assertThrows(classOf[ExecutionException], () =>
- client.describeTopics(util.List.of("test-topic"),
timeoutOption).allTopicNames().get())
- assertInstanceOf(classOf[TimeoutException], exception.getCause)
- } finally client.close(time.Duration.ZERO)
- }
-
- /**
- * describe should not auto create topics
- */
- @Test
- def testDescribeNonExistingTopic(): Unit = {
- client = createAdminClient
-
- val existingTopic = "existing-topic"
- client.createTopics(Seq(existingTopic).map(new NewTopic(_, 1,
1.toShort)).asJava).all.get()
- waitForTopics(client, Seq(existingTopic), List())
-
- val nonExistingTopic = "non-existing"
- val results = client.describeTopics(util.List.of(nonExistingTopic,
existingTopic)).topicNameValues()
- assertEquals(existingTopic, results.get(existingTopic).get.name)
- assertFutureThrows(classOf[UnknownTopicOrPartitionException],
results.get(nonExistingTopic))
- }
-
- @Test
- def testDescribeTopicsWithIds(): Unit = {
- client = createAdminClient
-
- val existingTopic = "existing-topic"
- client.createTopics(Seq(existingTopic).map(new NewTopic(_, 1,
1.toShort)).asJava).all.get()
- waitForTopics(client, Seq(existingTopic), List())
- ensureConsistentKRaftMetadata()
-
- val existingTopicId = brokers.head.metadataCache.getTopicId(existingTopic)
-
- val nonExistingTopicId = Uuid.randomUuid()
-
- val results =
client.describeTopics(TopicCollection.ofTopicIds(util.List.of(existingTopicId,
nonExistingTopicId))).topicIdValues()
- assertEquals(existingTopicId, results.get(existingTopicId).get.topicId())
- assertFutureThrows(classOf[UnknownTopicIdException],
results.get(nonExistingTopicId))
- }
-
- @Test
- def testDescribeTopicsWithNames(): Unit = {
- client = createAdminClient
-
- val existingTopic = "existing-topic"
- client.createTopics(Seq(existingTopic).map(new NewTopic(_, 1,
1.toShort)).asJava).all.get()
- waitForTopics(client, Seq(existingTopic), List())
- ensureConsistentKRaftMetadata()
-
- val existingTopicId = brokers.head.metadataCache.getTopicId(existingTopic)
- val results =
client.describeTopics(TopicCollection.ofTopicNames(util.List.of(existingTopic))).topicNameValues()
- assertEquals(existingTopicId, results.get(existingTopic).get.topicId())
- }
-
- @Test
- def testDescribeCluster(): Unit = {
- client = createAdminClient
- val result = client.describeCluster
- val nodes = result.nodes.get()
- val clusterId = result.clusterId().get()
- assertEquals(brokers.head.dataPlaneRequestProcessor.clusterId, clusterId)
- val controller = result.controller().get()
-
- // In KRaft, we return a random brokerId as the current controller.
- val brokerIds = brokers.map(_.config.brokerId).toSet
- assertTrue(brokerIds.contains(controller.id))
-
- val brokerEndpoints = bootstrapServers().split(",")
- assertEquals(brokerEndpoints.size, nodes.size)
- for (node <- nodes.asScala) {
- val hostStr = s"${node.host}:${node.port}"
- assertTrue(brokerEndpoints.contains(hostStr), s"Unknown host:port pair
$hostStr in brokerVersionInfos")
- }
- }
-
@Test
def testDescribeLogDirs(): Unit = {
client = createAdminClient
