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 9f692648b3c KAFKA-20553 Remove tools module dependency on core test
output (#22899)
9f692648b3c is described below
commit 9f692648b3cfe0df9c5a9c61220dcc885900d449
Author: majialong <[email protected]>
AuthorDate: Thu Jul 23 12:33:41 2026 +0800
KAFKA-20553 Remove tools module dependency on core test output (#22899)
This PR removes the `tools` module dependency on the core test output by
replacing usages of `kafka.utils.TestUtils` in
`ConfigCommandIntegrationTest`, `ConsoleProducerTest`,
`DumpLogSegmentsTest`, and `ReplicationQuotasTestRig`.
Reviewers: Ken Huang <[email protected]>, Chia-Ping Tsai
<[email protected]>
---
build.gradle | 1 -
.../kafka/tools/ConfigCommandIntegrationTest.java | 6 +--
.../apache/kafka/tools/ConsoleProducerTest.java | 3 +-
.../apache/kafka/tools/DumpLogSegmentsTest.java | 29 +++++++----
.../tools/other/ReplicationQuotasTestRig.java | 56 +++++++++++-----------
5 files changed, 50 insertions(+), 45 deletions(-)
diff --git a/build.gradle b/build.gradle
index d974b67578f..1cbde4ed2d0 100644
--- a/build.gradle
+++ b/build.gradle
@@ -2657,7 +2657,6 @@ project(':tools') {
testImplementation project(':server')
testImplementation project(':core')
testImplementation testFixtures(project(':server'))
- testImplementation project(':core').sourceSets.test.output
testImplementation project(':test-common:test-common-internal-api')
testImplementation project(':test-common:test-common-runtime')
testImplementation project(':test-common:test-common-util')
diff --git
a/tools/src/test/java/org/apache/kafka/tools/ConfigCommandIntegrationTest.java
b/tools/src/test/java/org/apache/kafka/tools/ConfigCommandIntegrationTest.java
index 85a718c139f..7dd92a38b10 100644
---
a/tools/src/test/java/org/apache/kafka/tools/ConfigCommandIntegrationTest.java
+++
b/tools/src/test/java/org/apache/kafka/tools/ConfigCommandIntegrationTest.java
@@ -746,13 +746,11 @@ public class ConfigCommandIntegrationTest {
"--add-config", "log.cleaner.threads=2",
"--entity-type", "brokers",
"--entity-default"))));
- kafka.utils.TestUtils.waitUntilTrue(
+ TestUtils.waitForCondition(
() -> cluster.brokers().values().stream()
.map(KafkaBroker::config)
.allMatch(config ->
config.getInt("log.cleaner.threads") == 2),
- () -> "Timeout waiting for topic config propagating to
broker",
- org.apache.kafka.test.TestUtils.DEFAULT_MAX_WAIT_MS,
- 100L);
+ () -> "Timeout waiting for topic config propagating to
broker");
}
}
diff --git
a/tools/src/test/java/org/apache/kafka/tools/ConsoleProducerTest.java
b/tools/src/test/java/org/apache/kafka/tools/ConsoleProducerTest.java
index 4baa8079e29..d503ec53786 100644
--- a/tools/src/test/java/org/apache/kafka/tools/ConsoleProducerTest.java
+++ b/tools/src/test/java/org/apache/kafka/tools/ConsoleProducerTest.java
@@ -16,12 +16,11 @@
*/
package org.apache.kafka.tools;
-import kafka.utils.TestUtils;
-
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.utils.internals.Exit;
+import org.apache.kafka.test.TestUtils;
import org.apache.kafka.tools.ConsoleProducer.ConsoleProducerOptions;
import org.apache.kafka.tools.api.RecordReader;
diff --git
a/tools/src/test/java/org/apache/kafka/tools/DumpLogSegmentsTest.java
b/tools/src/test/java/org/apache/kafka/tools/DumpLogSegmentsTest.java
index 629bf9b62d6..2840ea8904a 100644
--- a/tools/src/test/java/org/apache/kafka/tools/DumpLogSegmentsTest.java
+++ b/tools/src/test/java/org/apache/kafka/tools/DumpLogSegmentsTest.java
@@ -16,8 +16,6 @@
*/
package org.apache.kafka.tools;
-import kafka.utils.TestUtils;
-
import org.apache.kafka.clients.consumer.ConsumerPartitionAssignor.Assignment;
import
org.apache.kafka.clients.consumer.ConsumerPartitionAssignor.Subscription;
import org.apache.kafka.clients.consumer.internals.ConsumerProtocol;
@@ -98,6 +96,7 @@ import
org.apache.kafka.storage.internals.log.TransactionIndex;
import org.apache.kafka.storage.internals.log.UnifiedLog;
import org.apache.kafka.storage.internals.log.VerificationGuard;
import org.apache.kafka.storage.log.metrics.BrokerTopicStats;
+import org.apache.kafka.test.TestUtils;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Test;
@@ -144,7 +143,7 @@ public class DumpLogSegmentsTest {
boolean hasValues
) { }
- private final File tmpDir = TestUtils.tempDir();
+ private final File tmpDir = TestUtils.tempDirectory();
private final File logDir = TestUtils.randomPartitionLogDir(tmpDir);
private final String segmentName = "00000000000000000000";
private final String logFilePath = new File(logDir, segmentName +
".log").getAbsolutePath();
@@ -818,7 +817,7 @@ public class DumpLogSegmentsTest {
private Record serializedRecord(ApiMessage key, ApiMessageAndVersion
value) {
byte[] valueBytes = value == null ? null :
MessageUtil.toVersionPrefixedBytes(value.version(), value.message());
- return TestUtils.singletonRecords(
+ return singletonRecords(
valueBytes,
MessageUtil.toCoordinatorTypePrefixedBytes(key),
Compression.NONE,
@@ -835,7 +834,7 @@ public class DumpLogSegmentsTest {
assertEquals(
"Failed to decode message at offset 0 using the specified decoder
(message had a missing key)",
assertThrows(RuntimeException.class, () ->
- parser.parse(TestUtils.singletonRecords(null, null,
Compression.NONE, RecordBatch.NO_TIMESTAMP,
+ parser.parse(singletonRecords(null, null, Compression.NONE,
RecordBatch.NO_TIMESTAMP,
RecordBatch.CURRENT_MAGIC_VALUE).records().iterator().next())
).getMessage()
);
@@ -939,7 +938,7 @@ public class DumpLogSegmentsTest {
// An unknown record type should be handled and reported as such.
assertParseResult(
parser.parse(
- TestUtils.singletonRecords(
+ singletonRecords(
new byte[0],
ByteBuffer.allocate(2).putShort(Short.MAX_VALUE).array(),
Compression.NONE,
@@ -974,7 +973,7 @@ public class DumpLogSegmentsTest {
assertEquals(
"Failed to decode message at offset 0 using the specified decoder
(message had a missing key)",
assertThrows(RuntimeException.class, () ->
- parser.parse(TestUtils.singletonRecords(null, null,
Compression.NONE, RecordBatch.NO_TIMESTAMP,
+ parser.parse(singletonRecords(null, null, Compression.NONE,
RecordBatch.NO_TIMESTAMP,
RecordBatch.CURRENT_MAGIC_VALUE).records().iterator().next())
).getMessage()
);
@@ -1007,7 +1006,7 @@ public class DumpLogSegmentsTest {
// An unknown record type should be handled and reported as such.
assertParseResult(
parser.parse(
- TestUtils.singletonRecords(
+ singletonRecords(
new byte[0],
ByteBuffer.allocate(2).putShort(Short.MAX_VALUE).array(),
Compression.NONE,
@@ -1196,7 +1195,7 @@ public class DumpLogSegmentsTest {
assertEquals(
"Failed to decode message at offset 0 using the specified decoder
(message had a missing key)",
assertThrows(RuntimeException.class, () ->
- parser.parse(TestUtils.singletonRecords(null, null,
Compression.NONE, RecordBatch.NO_TIMESTAMP,
+ parser.parse(singletonRecords(null, null, Compression.NONE,
RecordBatch.NO_TIMESTAMP,
RecordBatch.CURRENT_MAGIC_VALUE).records().iterator().next())
).getMessage()
);
@@ -1272,7 +1271,7 @@ public class DumpLogSegmentsTest {
// An unknown record type should be handled and reported as such.
assertParseResult(
parser.parse(
- TestUtils.singletonRecords(
+ singletonRecords(
new byte[0],
ByteBuffer.allocate(2).putShort(Short.MAX_VALUE).array(),
Compression.NONE,
@@ -1616,4 +1615,14 @@ public class DumpLogSegmentsTest {
() -> runDumpLogSegments(new String[]{"--files",
noDotFile.getAbsolutePath()}));
assertTrue(errOutput.contains("Ignoring unknown file"), errOutput);
}
+
+ private static MemoryRecords singletonRecords(byte[] value, byte[] key,
Compression compression, long timestamp, byte magicValue) {
+ return MemoryRecords.withRecords(
+ magicValue,
+ 0L,
+ compression,
+ TimestampType.CREATE_TIME,
+ new SimpleRecord(timestamp, key, value)
+ );
+ }
}
diff --git
a/tools/src/test/java/org/apache/kafka/tools/other/ReplicationQuotasTestRig.java
b/tools/src/test/java/org/apache/kafka/tools/other/ReplicationQuotasTestRig.java
index 8e646534263..5df994ec1aa 100644
---
a/tools/src/test/java/org/apache/kafka/tools/other/ReplicationQuotasTestRig.java
+++
b/tools/src/test/java/org/apache/kafka/tools/other/ReplicationQuotasTestRig.java
@@ -18,17 +18,16 @@ package org.apache.kafka.tools.other;
import kafka.server.BrokerServer;
import kafka.server.KafkaBroker;
-import kafka.utils.TestUtils;
import org.apache.kafka.clients.admin.Admin;
import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.MetricName;
import org.apache.kafka.common.Node;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.TopicPartitionInfo;
-import org.apache.kafka.common.security.auth.SecurityProtocol;
import org.apache.kafka.common.serialization.ByteArraySerializer;
import org.apache.kafka.common.test.KafkaClusterTestKit;
import org.apache.kafka.common.test.TestKitNodes;
@@ -39,6 +38,8 @@ import org.apache.kafka.image.TopicImage;
import org.apache.kafka.metadata.PartitionRegistration;
import org.apache.kafka.server.quota.QuotaType;
import org.apache.kafka.storage.internals.log.UnifiedLog;
+import org.apache.kafka.test.NoRetryException;
+import org.apache.kafka.test.TestUtils;
import org.apache.kafka.tools.reassign.ReassignPartitionsCommand;
import org.apache.logging.log4j.core.config.Configurator;
@@ -73,8 +74,6 @@ import java.util.stream.IntStream;
import javax.imageio.ImageIO;
-import scala.Option;
-
import static java.nio.file.StandardOpenOption.APPEND;
import static java.nio.file.StandardOpenOption.CREATE;
import static org.apache.kafka.test.TestUtils.DEFAULT_MAX_WAIT_MS;
@@ -215,14 +214,14 @@ public class ReplicationQuotasTestRig {
startBrokers(config.brokers);
adminClient.createTopics(Set.of(new NewTopic(TOPIC_NAME,
replicas))).all().get();
- TestUtils.waitUntilTrue(
+ TestUtils.waitForCondition(
() -> cluster.brokers().values().stream().allMatch(server
-> {
TopicImage image =
server.metadataCache().currentImage().topics().getTopic(TOPIC_NAME);
return image != null &&
image.partitions().values().stream().allMatch(PartitionRegistration::hasLeader);
}),
- () -> "Timed out waiting for topic listing",
DEFAULT_MAX_WAIT_MS,
- 500L
+ 500L,
+ () -> "Timed out waiting for topic listing"
);
System.out.println("Writing Data");
@@ -306,15 +305,18 @@ public class ReplicationQuotasTestRig {
System.out.println("Worst case duration is " +
config.targetBytesPerBrokerMB * 1000 * 1000 / config.throttle);
}
- void waitForReassignmentToComplete() {
- TestUtils.waitUntilTrue(() -> {
+ void waitForReassignmentToComplete() throws InterruptedException {
+ TestUtils.waitForCondition(() -> {
printRateMetrics();
try {
return
adminClient.listPartitionReassignments().reassignments().get().isEmpty();
- } catch (InterruptedException | ExecutionException e) {
- throw new RuntimeException(e);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new NoRetryException(e);
+ } catch (ExecutionException e) {
+ throw new NoRetryException(e);
}
- }, () -> "Partition reassignments didn't complete.", 60 * 60 *
1000, 1000L);
+ }, 60 * 60 * 1000, 1000L, () -> "Partition reassignments didn't
complete.");
}
void renderChart(Map<Integer, List<Double>> data, String name, Journal
journal, boolean displayChartsOnScreen) throws Exception {
@@ -398,23 +400,21 @@ public class ReplicationQuotasTestRig {
}
KafkaProducer<byte[], byte[]> createProducer() {
- return TestUtils.createProducer(
- cluster.bootstrapServers(),
- 1,
- 60 * 1000L,
- 1024L * 1024L,
- Integer.MAX_VALUE,
- 30 * 1000,
- 0,
- 16384,
- "none",
- 20 * 1000,
- SecurityProtocol.PLAINTEXT,
- Option.empty(),
- Option.empty(),
- new ByteArraySerializer(),
+ return new KafkaProducer<>(
+ Map.of(
+ ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
cluster.bootstrapServers(),
+ ProducerConfig.ACKS_CONFIG, "1",
+ ProducerConfig.MAX_BLOCK_MS_CONFIG, 60 * 1000L,
+ ProducerConfig.BUFFER_MEMORY_CONFIG, 1024L * 1024L,
+ ProducerConfig.RETRIES_CONFIG, Integer.MAX_VALUE,
+ ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, 30 * 1000,
+ ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, 20 * 1000,
+ ProducerConfig.LINGER_MS_CONFIG, 0,
+ ProducerConfig.BATCH_SIZE_CONFIG, 16384,
+ ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, false
+ ),
new ByteArraySerializer(),
- false
+ new ByteArraySerializer()
);
}
}
