diff --git a/connect/mirror/src/test/java/org/apache/kafka/connect/mirror/integration/MirrorConnectorsIntegrationBaseTest.java b/connect/mirror/src/test/java/org/apache/kafka/connect/mirror/integration/MirrorConnectorsIntegrationBaseTest.java index 56d2bf4974093..67e8d477b0d55 100644 --- a/connect/mirror/src/test/java/org/apache/kafka/connect/mirror/integration/MirrorConnectorsIntegrationBaseTest.java +++ b/connect/mirror/src/test/java/org/apache/kafka/connect/mirror/integration/MirrorConnectorsIntegrationBaseTest.java @@ -33,7 +33,6 @@ import org.apache.kafka.clients.consumer.ConsumerRecords; import org.apache.kafka.clients.consumer.KafkaConsumer; import org.apache.kafka.clients.consumer.OffsetAndMetadata; -import org.apache.kafka.clients.producer.KafkaProducer; import org.apache.kafka.clients.producer.Producer; import org.apache.kafka.clients.producer.ProducerRecord; import org.apache.kafka.common.MetricName; @@ -1016,26 +1015,26 @@ public void testReplicateFromLatest() throws Exception { } @Test - public void testConnectorMetricsNew() throws InterruptedException, ExecutionException { + public void testConnectorMetricsNew() throws InterruptedException { testConnectorMetrics(METRIC_NAMES_NEW, () -> assertMetrics(false)); } @Test - public void testConnectorMetricsLegacy() throws InterruptedException, ExecutionException { + public void testConnectorMetricsLegacy() throws InterruptedException { testConnectorMetrics(METRIC_NAMES_LEGACY, () -> assertMetrics(true)); } @Test - public void testConnectorMetricsDefault() throws InterruptedException, ExecutionException { + public void testConnectorMetricsDefault() throws InterruptedException { testConnectorMetrics(null, () -> assertMetrics(true)); } @Test - public void testConnectorMetricsNewAndLegacy() throws InterruptedException, ExecutionException { + public void testConnectorMetricsNewAndLegacy() throws InterruptedException { testConnectorMetrics(METRIC_NAMES_NEW + "," + METRIC_NAMES_LEGACY, () -> assertMetrics(true) && assertMetrics(false)); } - private void testConnectorMetrics(String format, Supplier assertions) throws InterruptedException, ExecutionException { + private void testConnectorMetrics(String format, Supplier assertions) throws InterruptedException { // one way replication from primary to backup mm2Props.put(BACKUP_CLUSTER_ALIAS + "->" + PRIMARY_CLUSTER_ALIAS + ".enabled", "false"); if (format != null) { @@ -1049,11 +1048,9 @@ private void testConnectorMetrics(String format, Supplier assertions) t String topic = "test-topic-metrics"; primary.kafka().createTopic(topic); - try (KafkaProducer producer = primary.kafka().createProducer(Map.of())) { - for (int i = 0; i < NUM_RECORDS_PRODUCED; i++) { - producer.send(new ProducerRecord<>(topic, ("value" + i).getBytes())).get(); - } - } + primary.kafka().produce(IntStream.range(0, NUM_RECORDS_PRODUCED) + .mapToObj(i -> new ProducerRecord(topic, ("value" + i).getBytes())) + .toList()); waitUntilMirrorMakerIsRunning(backup, List.of(MirrorSourceConnector.class, MirrorCheckpointConnector.class), @@ -1326,8 +1323,9 @@ protected void produceMessages(Producer producer, List record : records) { - producer.send(record).get(timer.remainingMs(), TimeUnit.MILLISECONDS); + var futures = records.stream().map(producer::send).toList(); + for (var future : futures) { + future.get(timer.remainingMs(), TimeUnit.MILLISECONDS); timer.update(); } } catch (ExecutionException | InterruptedException | TimeoutException e) { diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/integration/ErrorHandlingIntegrationTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/integration/ErrorHandlingIntegrationTest.java index fd7f14d122d3f..35b53caea224d 100644 --- a/connect/runtime/src/test/java/org/apache/kafka/connect/integration/ErrorHandlingIntegrationTest.java +++ b/connect/runtime/src/test/java/org/apache/kafka/connect/integration/ErrorHandlingIntegrationTest.java @@ -19,6 +19,7 @@ import org.apache.kafka.clients.consumer.ConsumerConfig; import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.clients.consumer.ConsumerRecords; +import org.apache.kafka.clients.producer.ProducerRecord; import org.apache.kafka.common.config.ConfigDef; import org.apache.kafka.common.header.Headers; import org.apache.kafka.connect.connector.ConnectRecord; @@ -41,6 +42,7 @@ import java.util.Map; import java.util.Set; import java.util.concurrent.TimeUnit; +import java.util.stream.IntStream; import static org.apache.kafka.connect.runtime.ConnectorConfig.CONNECTOR_CLASS_CONFIG; import static org.apache.kafka.connect.runtime.ConnectorConfig.CONNECTOR_CLIENT_CONSUMER_OVERRIDES_PREFIX; @@ -144,9 +146,9 @@ public void testSkipRetryAndDLQWithHeaders() throws Exception { "Connector task was not assigned a partition."); // produce some strings into test topic - for (int i = 0; i < NUM_RECORDS_PRODUCED; i++) { - connect.kafka().produce("test-topic", "key-" + i, "value-" + i); - } + connect.kafka().produce(IntStream.range(0, NUM_RECORDS_PRODUCED) + .mapToObj(i -> new ProducerRecord<>("test-topic", ("key-" + i).getBytes(), ("value-" + i).getBytes())) + .toList()); // consume all records from test topic log.info("Consuming records from test topic"); @@ -223,9 +225,9 @@ public void testErrantRecordReporter() throws Exception { "Connector task was not assigned a partition."); // produce some strings into test topic - for (int i = 0; i < NUM_RECORDS_PRODUCED; i++) { - connect.kafka().produce("test-topic", "key-" + i, "value-" + i); - } + connect.kafka().produce(IntStream.range(0, NUM_RECORDS_PRODUCED) + .mapToObj(i -> new ProducerRecord<>("test-topic", ("key-" + i).getBytes(), ("value-" + i).getBytes())) + .toList()); // consume all records from test topic log.info("Consuming records from test topic"); diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/integration/ExampleConnectIntegrationTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/integration/ExampleConnectIntegrationTest.java index 6263c8ab96cc1..c8c4944c60ab7 100644 --- a/connect/runtime/src/test/java/org/apache/kafka/connect/integration/ExampleConnectIntegrationTest.java +++ b/connect/runtime/src/test/java/org/apache/kafka/connect/integration/ExampleConnectIntegrationTest.java @@ -16,6 +16,7 @@ */ package org.apache.kafka.connect.integration; +import org.apache.kafka.clients.producer.ProducerRecord; import org.apache.kafka.connect.runtime.rest.entities.ConnectorStateInfo; import org.apache.kafka.connect.storage.StringConverter; import org.apache.kafka.connect.util.clusters.EmbeddedConnectCluster; @@ -31,6 +32,7 @@ import java.util.Map; import java.util.Properties; import java.util.concurrent.TimeUnit; +import java.util.stream.IntStream; import static org.apache.kafka.connect.runtime.ConnectorConfig.CONNECTOR_CLASS_CONFIG; import static org.apache.kafka.connect.runtime.ConnectorConfig.KEY_CONVERTER_CLASS_CONFIG; @@ -146,9 +148,9 @@ public void testSinkConnector() throws Exception { "Connector tasks were not assigned a partition each."); // produce some messages into source topic partitions - for (int i = 0; i < NUM_RECORDS_PRODUCED; i++) { - connect.kafka().produce("test-topic", i % NUM_TOPIC_PARTITIONS, "key", "simple-message-value-" + i); - } + connect.kafka().produce(IntStream.range(0, NUM_RECORDS_PRODUCED) + .mapToObj(i -> new ProducerRecord<>("test-topic", i % NUM_TOPIC_PARTITIONS, "key".getBytes(), ("simple-message-value-" + i).getBytes())) + .toList()); // consume all records from the source topic or fail, to ensure that they were correctly produced. assertEquals(NUM_RECORDS_PRODUCED, diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/integration/MonitorableSinkIntegrationTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/integration/MonitorableSinkIntegrationTest.java index e9d74836e0101..d59cf871e02d8 100644 --- a/connect/runtime/src/test/java/org/apache/kafka/connect/integration/MonitorableSinkIntegrationTest.java +++ b/connect/runtime/src/test/java/org/apache/kafka/connect/integration/MonitorableSinkIntegrationTest.java @@ -16,6 +16,7 @@ */ package org.apache.kafka.connect.integration; +import org.apache.kafka.clients.producer.ProducerRecord; import org.apache.kafka.common.MetricName; import org.apache.kafka.common.metrics.KafkaMetric; import org.apache.kafka.connect.runtime.rest.entities.ConnectorStateInfo; @@ -32,6 +33,7 @@ import java.util.Map; import java.util.concurrent.TimeUnit; +import java.util.stream.IntStream; import static org.apache.kafka.connect.runtime.ConnectorConfig.CONNECTOR_CLASS_CONFIG; import static org.apache.kafka.connect.runtime.ConnectorConfig.KEY_CONVERTER_CLASS_CONFIG; @@ -111,9 +113,9 @@ public void testMonitorableSinkConnectorAndTask() throws Exception { assertEquals(MonitorableSinkConnector.VALUE, kafkaMetric.metricValue()); // produce some records - for (int i = 0; i < NUM_RECORDS_PRODUCED; i++) { - connect.kafka().produce("test-topic", "key-" + i, "value-" + i); - } + connect.kafka().produce(IntStream.range(0, NUM_RECORDS_PRODUCED) + .mapToObj(i -> new ProducerRecord<>("test-topic", ("key-" + i).getBytes(), ("value-" + i).getBytes())) + .toList()); // wait for records to reach the task connectorHandle.taskHandle(TASK_ID).awaitRecords(CONSUME_MAX_DURATION_MS); diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/integration/TransformationIntegrationTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/integration/TransformationIntegrationTest.java index 5eca2c24e8401..3905a150aee82 100644 --- a/connect/runtime/src/test/java/org/apache/kafka/connect/integration/TransformationIntegrationTest.java +++ b/connect/runtime/src/test/java/org/apache/kafka/connect/integration/TransformationIntegrationTest.java @@ -17,6 +17,7 @@ package org.apache.kafka.connect.integration; import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.clients.producer.ProducerRecord; import org.apache.kafka.connect.storage.StringConverter; import org.apache.kafka.connect.transforms.Filter; import org.apache.kafka.connect.transforms.predicates.HasHeaderKey; @@ -33,6 +34,7 @@ import java.util.Map; import java.util.Properties; import java.util.concurrent.TimeUnit; +import java.util.stream.IntStream; import static org.apache.kafka.connect.runtime.ConnectorConfig.CONNECTOR_CLASS_CONFIG; import static org.apache.kafka.connect.runtime.ConnectorConfig.KEY_CONVERTER_CLASS_CONFIG; @@ -146,12 +148,12 @@ public void testFilterOnTopicNameWithSinkConnector() throws Exception { assertConnectorRunning(); // produce some messages into source topic partitions - for (int i = 0; i < numBarRecords; i++) { - connect.kafka().produce(barTopic, i % NUM_TOPIC_PARTITIONS, "key", "simple-message-value-" + i); - } - for (int i = 0; i < numFooRecords; i++) { - connect.kafka().produce(fooTopic, i % NUM_TOPIC_PARTITIONS, "key", "simple-message-value-" + i); - } + connect.kafka().produce(IntStream.range(0, numBarRecords) + .mapToObj(i -> new ProducerRecord<>(barTopic, i % NUM_TOPIC_PARTITIONS, "key".getBytes(), ("simple-message-value-" + i).getBytes())) + .toList()); + connect.kafka().produce(IntStream.range(0, numFooRecords) + .mapToObj(i -> new ProducerRecord<>(fooTopic, i % NUM_TOPIC_PARTITIONS, "key".getBytes(), ("simple-message-value-" + i).getBytes())) + .toList()); // consume all records from the source topic or fail, to ensure that they were correctly produced. assertEquals( @@ -235,9 +237,9 @@ public void testFilterOnTombstonesWithSinkConnector() throws Exception { assertConnectorRunning(); // produce some messages into source topic partitions - for (int i = 0; i < numRecords; i++) { - connect.kafka().produce(topic, i % NUM_TOPIC_PARTITIONS, "key", i % 2 == 0 ? "simple-message-value-" + i : null); - } + connect.kafka().produce(IntStream.range(0, numRecords) + .mapToObj(i -> new ProducerRecord<>(topic, i % NUM_TOPIC_PARTITIONS, "key".getBytes(), i % 2 == 0 ? ("simple-message-value-" + i).getBytes() : null)) + .toList()); // consume all records from the source topic or fail, to ensure that they were correctly produced. assertEquals( diff --git a/connect/runtime/src/testFixtures/java/org/apache/kafka/connect/util/clusters/EmbeddedKafkaCluster.java b/connect/runtime/src/testFixtures/java/org/apache/kafka/connect/util/clusters/EmbeddedKafkaCluster.java index 7913d60fc2837..562a83528448a 100644 --- a/connect/runtime/src/testFixtures/java/org/apache/kafka/connect/util/clusters/EmbeddedKafkaCluster.java +++ b/connect/runtime/src/testFixtures/java/org/apache/kafka/connect/util/clusters/EmbeddedKafkaCluster.java @@ -414,11 +414,17 @@ public void produce(String topic, String key, String value) { } public void produce(String topic, Integer partition, String key, String value) { - ProducerRecord msg = new ProducerRecord<>(topic, partition, key == null ? null : key.getBytes(), value == null ? null : value.getBytes()); - try { - producer.send(msg).get(DEFAULT_PRODUCE_SEND_DURATION_MS, TimeUnit.MILLISECONDS); - } catch (Exception e) { - throw new KafkaException("Could not produce message: " + msg, e); + produce(List.of(new ProducerRecord<>(topic, partition, key == null ? null : key.getBytes(), value == null ? null : value.getBytes()))); + } + + public void produce(List> records) { + var futures = records.stream().map(producer::send).toList(); + for (int i = 0; i < futures.size(); i++) { + try { + futures.get(i).get(DEFAULT_PRODUCE_SEND_DURATION_MS, TimeUnit.MILLISECONDS); + } catch (Exception e) { + throw new KafkaException("Could not produce message: " + records.get(i), e); + } } }