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 2699fc40748 KAFKA-20553 Remove clients-integration-tests dependency on
core test output (#22884)
2699fc40748 is described below
commit 2699fc40748da5ce9dc4b6aac41687fbc603ab1d
Author: majialong <[email protected]>
AuthorDate: Fri Jul 24 18:03:41 2026 +0800
KAFKA-20553 Remove clients-integration-tests dependency on core test output
(#22884)
This PR removes the `clients:clients-integration-tests` dependency on
the core test output by replacing usages of `kafka.utils.TestUtils` in
`ProducerCompressionTest` and `ProducerIdExpirationTest`.
Reviewers: Chia-Ping Tsai <[email protected]>
---
build.gradle | 1 -
.../org/apache/kafka/clients/ClientsTestUtils.java | 11 +-
.../clients/producer/ProducerCompressionTest.java | 24 ++++-
.../clients/producer/ProducerIdExpirationTest.java | 114 +++++++++++++++------
.../test/scala/unit/kafka/utils/TestUtils.scala | 4 -
5 files changed, 113 insertions(+), 41 deletions(-)
diff --git a/build.gradle b/build.gradle
index 69e82b5c55b..6a4f33106b5 100644
--- a/build.gradle
+++ b/build.gradle
@@ -2175,7 +2175,6 @@ project(':clients:clients-integration-tests') {
testImplementation project(':server')
testImplementation project(':storage')
testImplementation testFixtures(project(':storage'))
- testImplementation project(':core').sourceSets.test.output
testImplementation testFixtures(project(':clients'))
implementation project(':server-common')
testImplementation testFixtures(project(':server-common'))
diff --git
a/clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/ClientsTestUtils.java
b/clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/ClientsTestUtils.java
index 5635f6e36b7..bef08022d19 100644
---
a/clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/ClientsTestUtils.java
+++
b/clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/ClientsTestUtils.java
@@ -75,6 +75,15 @@ public class ClientsTestUtils {
Consumer<K, V> consumer,
int numRecords,
int maxPollRecords
+ ) throws InterruptedException {
+ return consumeRecords(consumer, numRecords, maxPollRecords, 60000);
+ }
+
+ public static <K, V> List<ConsumerRecord<K, V>> consumeRecords(
+ Consumer<K, V> consumer,
+ int numRecords,
+ int maxPollRecords,
+ long maxWaitMs
) throws InterruptedException {
List<ConsumerRecord<K, V>> consumedRecords = new ArrayList<>();
TestUtils.waitForCondition(() -> {
@@ -82,7 +91,7 @@ public class ClientsTestUtils {
records.forEach(consumedRecords::add);
assertTrue(records.count() <= maxPollRecords);
return consumedRecords.size() >= numRecords;
- }, 60000, "Timed out before consuming expected " + numRecords + "
records.");
+ }, maxWaitMs, "Timed out before consuming expected " + numRecords + "
records.");
return consumedRecords;
}
diff --git
a/clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/producer/ProducerCompressionTest.java
b/clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/producer/ProducerCompressionTest.java
index d8d3ebe3d86..9a14d2b56cc 100644
---
a/clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/producer/ProducerCompressionTest.java
+++
b/clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/producer/ProducerCompressionTest.java
@@ -20,12 +20,14 @@ package org.apache.kafka.clients.producer;
import org.apache.kafka.clients.consumer.CloseOptions;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.header.Header;
import org.apache.kafka.common.header.internals.RecordHeader;
import org.apache.kafka.common.header.internals.RecordHeaders;
import org.apache.kafka.common.record.internal.CompressionType;
import org.apache.kafka.common.test.ClusterInstance;
+import org.apache.kafka.common.test.api.ClusterConfigProperty;
import org.apache.kafka.common.test.api.ClusterTest;
import org.apache.kafka.common.test.api.ClusterTestDefaults;
import org.apache.kafka.common.test.api.Type;
@@ -43,13 +45,19 @@ import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
-import static kafka.utils.TestUtils.consumeRecords;
+import static org.apache.kafka.clients.ClientsTestUtils.consumeRecords;
+import static
org.apache.kafka.coordinator.group.GroupCoordinatorConfig.OFFSETS_TOPIC_REPLICATION_FACTOR_CONFIG;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.fail;
-@ClusterTestDefaults(types = {Type.KRAFT})
+@ClusterTestDefaults(
+ types = {Type.KRAFT},
+ serverProperties = {
+ @ClusterConfigProperty(key = OFFSETS_TOPIC_REPLICATION_FACTOR_CONFIG,
value = "1")
+ }
+)
public class ProducerCompressionTest {
private final String topicName = "topic";
@@ -114,13 +122,20 @@ public class ProducerCompressionTest {
}
private void verifyConsumerRecords(Consumer<byte[], byte[]> consumer,
List<String> messages, long now,
- Header[] headerArr, int partition,
String topic, String compression) {
+ Header[] headerArr, int partition,
String topic, String compression) throws InterruptedException {
TopicPartition tp = new TopicPartition(topic, partition);
consumer.assign(List.of(tp));
consumer.seek(tp, 0);
AtomicInteger num = new AtomicInteger(0);
AtomicInteger flag = new AtomicInteger(0);
- consumeRecords(consumer, numRecords * 3,
TestUtils.DEFAULT_MAX_WAIT_MS).foreach(record -> {
+ List<ConsumerRecord<byte[], byte[]>> records = consumeRecords(
+ consumer,
+ numRecords * 3,
+ Integer.MAX_VALUE,
+ TestUtils.DEFAULT_MAX_WAIT_MS
+ );
+ assertEquals(numRecords * 3, records.size(), "Consumed more records
than expected");
+ records.forEach(record -> {
String messageValue = messages.get(num.get());
long offset = num.get() * 3L + flag.get();
if (flag.get() == 0) {
@@ -149,7 +164,6 @@ public class ProducerCompressionTest {
fail();
}
flagLoop(num, flag);
- return null;
});
}
diff --git
a/clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/producer/ProducerIdExpirationTest.java
b/clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/producer/ProducerIdExpirationTest.java
index 94bc50b4955..cdaa44cda4f 100644
---
a/clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/producer/ProducerIdExpirationTest.java
+++
b/clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/producer/ProducerIdExpirationTest.java
@@ -17,7 +17,6 @@
package org.apache.kafka.clients.producer;
import kafka.server.KafkaBroker;
-import kafka.utils.TestUtils;
import org.apache.kafka.clients.admin.Admin;
import org.apache.kafka.clients.admin.AlterConfigOp;
@@ -25,18 +24,22 @@ import org.apache.kafka.clients.admin.ConfigEntry;
import org.apache.kafka.clients.admin.ProducerState;
import org.apache.kafka.clients.consumer.CloseOptions;
import org.apache.kafka.clients.consumer.Consumer;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.KafkaException;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.config.ConfigResource;
import org.apache.kafka.common.errors.InvalidPidMappingException;
import org.apache.kafka.common.errors.TransactionalIdNotFoundException;
+import org.apache.kafka.common.header.internals.RecordHeader;
import org.apache.kafka.common.test.ClusterInstance;
import org.apache.kafka.common.test.api.ClusterConfigProperty;
import org.apache.kafka.common.test.api.ClusterTest;
import org.apache.kafka.common.test.api.ClusterTestDefaults;
+import org.apache.kafka.test.NoRetryException;
import org.opentest4j.AssertionFailedError;
+import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.Collection;
import java.util.Collections;
@@ -46,7 +49,7 @@ import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
-import static kafka.utils.TestUtils.consumeRecords;
+import static org.apache.kafka.clients.ClientsTestUtils.consumeRecords;
import static
org.apache.kafka.clients.consumer.ConsumerConfig.ISOLATION_LEVEL_CONFIG;
import static
org.apache.kafka.clients.producer.ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG;
import static
org.apache.kafka.coordinator.group.GroupCoordinatorConfig.GROUP_INITIAL_REBALANCE_DELAY_MS_CONFIG;
@@ -64,9 +67,12 @@ import static
org.apache.kafka.server.config.ServerConfigs.CONTROLLED_SHUTDOWN_E
import static
org.apache.kafka.server.config.ServerLogConfigs.AUTO_CREATE_TOPICS_ENABLE_CONFIG;
import static org.apache.kafka.test.TestUtils.DEFAULT_MAX_WAIT_MS;
import static org.apache.kafka.test.TestUtils.assertFutureThrows;
+import static org.apache.kafka.test.TestUtils.waitForCondition;
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
@ClusterTestDefaults(
brokers = 3,
@@ -94,6 +100,10 @@ import static org.junit.jupiter.api.Assertions.assertThrows;
}
)
public class ProducerIdExpirationTest {
+ private static final String HEADER_KEY = "transactionStatus";
+ private static final byte[] COMMITTED_VALUE =
"committed".getBytes(StandardCharsets.UTF_8);
+ private static final byte[] ABORTED_VALUE =
"aborted".getBytes(StandardCharsets.UTF_8);
+
private final String topic1 = "topic1";
private final int numPartitions = 1;
private final short replicationFactor = 3;
@@ -130,19 +140,22 @@ public class ProducerIdExpirationTest {
// Start and then abort a transaction to allow the producer ID to
expire.
producer.beginTransaction();
-
producer.send(TestUtils.producerRecordWithExpectedTransactionStatus(topic1, 0,
"2", "2", false));
+ producer.send(producerRecordWithExpectedTransactionStatus(topic1, 0,
"2", "2", false));
producer.flush();
Consumer<byte[], byte[]> consumer =
cluster.consumer(Map.of(ISOLATION_LEVEL_CONFIG, "read_committed"));
try (Admin admin = cluster.admin()) {
// Ensure producer IDs are added.
- TestUtils.waitUntilTrue(() -> {
+ waitForCondition(() -> {
try {
return producerStates(admin).size() == 1;
- } catch (ExecutionException | InterruptedException e) {
- throw new RuntimeException(e);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new NoRetryException(e);
+ } catch (ExecutionException e) {
+ throw new NoRetryException(e);
}
- }, () -> "Producer IDs were not added.", DEFAULT_MAX_WAIT_MS, 100);
+ }, DEFAULT_MAX_WAIT_MS, 100L, () -> "Producer IDs were not
added.");
producer.abortTransaction();
@@ -156,9 +169,8 @@ public class ProducerIdExpirationTest {
// due to the expired transactional ID, resulting in a fatal error.
producer.beginTransaction();
Future<RecordMetadata> failedFuture =
-
producer.send(TestUtils.producerRecordWithExpectedTransactionStatus(topic1, 0,
"1", "1", false));
- TestUtils.waitUntilTrue(failedFuture::isDone, () -> "Producer
future never completed.",
- DEFAULT_MAX_WAIT_MS, 100);
+
producer.send(producerRecordWithExpectedTransactionStatus(topic1, 0, "1", "1",
false));
+ waitForCondition(failedFuture::isDone, DEFAULT_MAX_WAIT_MS, 100L,
() -> "Producer future never completed.");
assertFutureThrows(InvalidPidMappingException.class, failedFuture);
// Assert that aborting the transaction throws a KafkaException
due to the fatal error.
@@ -170,8 +182,8 @@ public class ProducerIdExpirationTest {
producer.initTransactions();
producer.beginTransaction();
-
producer.send(TestUtils.producerRecordWithExpectedTransactionStatus(topic1, 0,
"4", "4", true));
-
producer.send(TestUtils.producerRecordWithExpectedTransactionStatus(topic1, 0,
"3", "3", true));
+ producer.send(producerRecordWithExpectedTransactionStatus(topic1,
0, "4", "4", true));
+ producer.send(producerRecordWithExpectedTransactionStatus(topic1,
0, "3", "3", true));
// Producer IDs should be retained.
assertFalse(producerStates(admin).isEmpty());
@@ -180,7 +192,16 @@ public class ProducerIdExpirationTest {
// Check we can still consume the transaction.
consumer.subscribe(List.of(topic1));
- consumeRecords(consumer, 2,
DEFAULT_MAX_WAIT_MS).foreach(TestUtils::assertCommittedAndGetValue);
+ List<ConsumerRecord<byte[], byte[]>> records = consumeRecords(
+ consumer,
+ 2,
+ Integer.MAX_VALUE,
+ DEFAULT_MAX_WAIT_MS
+ );
+ assertEquals(
+ List.of("4", "3"),
+
records.stream().map(ProducerIdExpirationTest::assertCommittedAndGetValue).toList()
+ );
} finally {
producer.close();
consumer.close(CloseOptions.timeout(Duration.ofSeconds(1)));
@@ -204,10 +225,14 @@ public class ProducerIdExpirationTest {
// Update the producer ID expiration ms to a very high value.
admin.incrementalAlterConfigs(producerIdExpirationConfig("100000"));
- cluster.brokers().values().forEach(broker ->
- TestUtils.waitUntilTrue(() ->
broker.logManager().producerStateManagerConfig().producerIdExpirationMs() ==
100000,
- () -> "Configuration was not updated.",
DEFAULT_MAX_WAIT_MS, 100)
- );
+ for (KafkaBroker broker : cluster.brokers().values()) {
+ waitForCondition(
+ () ->
broker.logManager().producerStateManagerConfig().producerIdExpirationMs() ==
100000,
+ DEFAULT_MAX_WAIT_MS,
+ 100L,
+ () -> "Configuration was not updated."
+ );
+ }
// Send more records to send producer ID back to brokers.
producer.send(new ProducerRecord<>(topic1, 0, null,
"key".getBytes(), "value".getBytes()));
producer.flush();
@@ -226,10 +251,14 @@ public class ProducerIdExpirationTest {
kafkaBroker.awaitShutdown();
kafkaBroker.startup();
cluster.waitForReadyBrokers();
- cluster.brokers().values().forEach(broker ->
- TestUtils.waitUntilTrue(() ->
broker.logManager().producerStateManagerConfig().producerIdExpirationMs() ==
100,
- () -> "Configuration was not updated.",
DEFAULT_MAX_WAIT_MS, 100)
- );
+ for (KafkaBroker broker : cluster.brokers().values()) {
+ waitForCondition(
+ () ->
broker.logManager().producerStateManagerConfig().producerIdExpirationMs() ==
100,
+ DEFAULT_MAX_WAIT_MS,
+ 100L,
+ () -> "Configuration was not updated."
+ );
+ }
// Ensure producer ID expires quickly again.
waitProducerIdExpire(admin);
@@ -237,18 +266,21 @@ public class ProducerIdExpirationTest {
}
- private void waitProducerIdExpire(Admin admin) {
+ private void waitProducerIdExpire(Admin admin) throws InterruptedException
{
waitProducerIdExpire(admin, DEFAULT_MAX_WAIT_MS);
}
- private void waitProducerIdExpire(Admin admin, long timeout) {
- TestUtils.waitUntilTrue(() -> {
+ private void waitProducerIdExpire(Admin admin, long timeout) throws
InterruptedException {
+ waitForCondition(() -> {
try {
return producerStates(admin).isEmpty();
- } catch (ExecutionException | InterruptedException e) {
- throw new RuntimeException(e);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new NoRetryException(e);
+ } catch (ExecutionException e) {
+ throw new NoRetryException(e);
}
- }, () -> "Producer ID expired.", timeout, 100);
+ }, timeout, 100L, () -> "Producer ID expired.");
}
private Map<ConfigResource, Collection<AlterConfigOp>>
producerIdExpirationConfig(String configValue) {
@@ -257,8 +289,8 @@ public class ProducerIdExpirationTest {
}
- private void waitUntilTransactionalStateExpires(Admin admin) {
- TestUtils.waitUntilTrue(() -> {
+ private void waitUntilTransactionalStateExpires(Admin admin) throws
InterruptedException {
+ waitForCondition(() -> {
boolean removedTransactionState = false;
try {
admin.describeTransactions(List.of(transactionalId))
@@ -268,7 +300,29 @@ public class ProducerIdExpirationTest {
removedTransactionState = e.getCause() instanceof
TransactionalIdNotFoundException;
}
return removedTransactionState;
- }, () -> "Transaction state never expired.", DEFAULT_MAX_WAIT_MS, 100);
+ }, DEFAULT_MAX_WAIT_MS, 100L, () -> "Transaction state never
expired.");
+ }
+
+ private static ProducerRecord<byte[], byte[]>
producerRecordWithExpectedTransactionStatus(
+ String topic,
+ Integer partition,
+ String key,
+ String value,
+ boolean willBeCommitted
+ ) {
+ var header = new RecordHeader(HEADER_KEY, willBeCommitted ?
COMMITTED_VALUE : ABORTED_VALUE);
+ return new ProducerRecord<>(topic, partition,
+ key.getBytes(StandardCharsets.UTF_8),
+ value.getBytes(StandardCharsets.UTF_8),
+ Collections.singleton(header));
+ }
+
+ private static String assertCommittedAndGetValue(ConsumerRecord<byte[],
byte[]> record) {
+ var headers = record.headers().headers(HEADER_KEY).iterator();
+ assertTrue(headers.hasNext(), "Expected transaction status header");
+ var header = headers.next();
+ assertArrayEquals(COMMITTED_VALUE, header.value(), "Expected committed
status");
+ return new String(record.value(), StandardCharsets.UTF_8);
}
private Map<String, Object> transactionalProducerConfig() {
diff --git a/core/src/test/scala/unit/kafka/utils/TestUtils.scala
b/core/src/test/scala/unit/kafka/utils/TestUtils.scala
index bc41c2ed2ae..c4e35f74705 100755
--- a/core/src/test/scala/unit/kafka/utils/TestUtils.scala
+++ b/core/src/test/scala/unit/kafka/utils/TestUtils.scala
@@ -1116,10 +1116,6 @@ object TestUtils extends Logging {
new ProducerRecord[Array[Byte], Array[Byte]](topic, partition, key, value,
util.Set.of(header))
}
- def producerRecordWithExpectedTransactionStatus(topic: String, partition:
Integer, key: String, value: String, willBeCommitted: Boolean):
ProducerRecord[Array[Byte], Array[Byte]] = {
- producerRecordWithExpectedTransactionStatus(topic, partition,
asBytes(key), asBytes(value), willBeCommitted)
- }
-
// Collect the current positions for all partition in the consumers current
assignment.
def consumerPositions(consumer: Consumer[Array[Byte], Array[Byte]]) :
Map[TopicPartition, OffsetAndMetadata] = {
val offsetsToCommit = new mutable.HashMap[TopicPartition,
OffsetAndMetadata]()
