From 6c13b3749c1654f95df702ec8ccb92cf3d370186 Mon Sep 17 00:00:00 2001 From: Lee Dongjin Date: Wed, 19 Sep 2018 22:37:42 +0900 Subject: [PATCH 1/2] 1. Add Javadoc to undocumented methods. 2. Add documentation on parameters in methods. 3. Fix typo and method ordering. --- .../utils/IntegrationTestUtils.java | 218 +++++++++++++++--- 1 file changed, 187 insertions(+), 31 deletions(-) diff --git a/streams/src/test/java/org/apache/kafka/streams/integration/utils/IntegrationTestUtils.java b/streams/src/test/java/org/apache/kafka/streams/integration/utils/IntegrationTestUtils.java index a99fd5afc3435..efa4582af3374 100644 --- a/streams/src/test/java/org/apache/kafka/streams/integration/utils/IntegrationTestUtils.java +++ b/streams/src/test/java/org/apache/kafka/streams/integration/utils/IntegrationTestUtils.java @@ -98,7 +98,7 @@ public boolean createdToRevokedSeen() { } /** - * Removes local state stores. Useful to reset state in-between integration test runs. + * Removes local state stores. Useful to reset state in-between integration test runs. * * @param streamsConfiguration Streams configuration settings */ @@ -140,6 +140,7 @@ public static void cleanStateAfterTest(final EmbeddedKafkaCluster cluster, final * @param topic Kafka topic to write the data records to * @param records Data records to write to Kafka * @param producerConfig Kafka producer configuration + * @param time Timestamp provider * @param Key type of the data records * @param Value type of the data records */ @@ -149,6 +150,15 @@ public static void produceKeyValuesSynchronously( IntegrationTestUtils.produceKeyValuesSynchronously(topic, records, producerConfig, time, false); } + /** + * @param topic Kafka topic to write the data records to + * @param records Data records to write to Kafka + * @param producerConfig Kafka producer configuration + * @param headers {@link Headers} of the data records + * @param time Timestamp provider + * @param Key type of the data records + * @param Value type of the data records + */ public static void produceKeyValuesSynchronously( final String topic, final Collection> records, final Properties producerConfig, final Headers headers, final Time time) throws ExecutionException, InterruptedException { @@ -159,6 +169,7 @@ public static void produceKeyValuesSynchronously( * @param topic Kafka topic to write the data records to * @param records Data records to write to Kafka * @param producerConfig Kafka producer configuration + * @param time Timestamp provider * @param enableTransactions Send messages in a transaction * @param Key type of the data records * @param Value type of the data records @@ -169,6 +180,16 @@ public static void produceKeyValuesSynchronously( IntegrationTestUtils.produceKeyValuesSynchronously(topic, records, producerConfig, null, time, enableTransactions); } + /** + * @param topic Kafka topic to write the data records to + * @param records Data records to write to Kafka + * @param producerConfig Kafka producer configuration + * @param headers {@link Headers} of the data records + * @param time Timestamp provider + * @param enableTransactions Send messages in a transaction + * @param Key type of the data records + * @param Value type of the data records + */ public static void produceKeyValuesSynchronously(final String topic, final Collection> records, final Properties producerConfig, @@ -187,6 +208,14 @@ public static void produceKeyValuesSynchronously(final String topic, } } + /** + * @param topic Kafka topic to write the data records to + * @param records Data records to write to Kafka + * @param producerConfig Kafka producer configuration + * @param timestamp Timestamp of the record + * @param Key type of the data records + * @param Value type of the data records + */ public static void produceKeyValuesSynchronouslyWithTimestamp(final String topic, final Collection> records, final Properties producerConfig, @@ -195,6 +224,33 @@ public static void produceKeyValuesSynchronouslyWithTimestamp(final Strin IntegrationTestUtils.produceKeyValuesSynchronouslyWithTimestamp(topic, records, producerConfig, timestamp, false); } + /** + * @param topic Kafka topic to write the data records to + * @param records Data records to write to Kafka + * @param producerConfig Kafka producer configuration + * @param headers {@link Headers} of the data records + * @param timestamp Timestamp of the record + * @param Key type of the data records + * @param Value type of the data records + */ + public static void produceKeyValuesSynchronouslyWithTimestamp(final String topic, + final Collection> records, + final Properties producerConfig, + final Headers headers, + final Long timestamp) + throws ExecutionException, InterruptedException { + IntegrationTestUtils.produceKeyValuesSynchronouslyWithTimestamp(topic, records, producerConfig, headers, timestamp, false); + } + + /** + * @param topic Kafka topic to write the data records to + * @param records Data records to write to Kafka + * @param producerConfig Kafka producer configuration + * @param timestamp Timestamp of the record + * @param enableTransactions Send messages in a transaction + * @param Key type of the data records + * @param Value type of the data records + */ public static void produceKeyValuesSynchronouslyWithTimestamp(final String topic, final Collection> records, final Properties producerConfig, @@ -204,15 +260,25 @@ public static void produceKeyValuesSynchronouslyWithTimestamp(final Strin IntegrationTestUtils.produceKeyValuesSynchronouslyWithTimestamp(topic, records, producerConfig, null, timestamp, enableTransactions); } + /** + * @param topic Kafka topic to write the data records to + * @param records Data records to write to Kafka + * @param producerConfig Kafka producer configuration + * @param headers {@link Headers} of the data records + * @param timestamp Timestamp of the record + * @param enableTransactions Send messages in a transaction + * @param Key type of the data records + * @param Value type of the data records + */ public static void produceKeyValuesSynchronouslyWithTimestamp(final String topic, final Collection> records, final Properties producerConfig, final Headers headers, final Long timestamp, - final boolean enabledTransactions) + final boolean enableTransactions) throws ExecutionException, InterruptedException { try (final Producer producer = new KafkaProducer<>(producerConfig)) { - if (enabledTransactions) { + if (enableTransactions) { producer.initTransactions(); producer.beginTransaction(); } @@ -221,7 +287,7 @@ public static void produceKeyValuesSynchronouslyWithTimestamp(final Strin new ProducerRecord<>(topic, null, timestamp, record.key, record.value, headers)); f.get(); } - if (enabledTransactions) { + if (enableTransactions) { producer.commitTransaction(); } producer.flush(); @@ -263,6 +329,17 @@ public static void produceSynchronously(final Properties producerConfig, } } + /** + * Produce data records and send them synchronously in an aborted transaction; that is, a transaction is started for + * each data record but not committed. + * + * @param topic Kafka topic to write the data records to + * @param records Data records to write to Kafka + * @param producerConfig Kafka producer configuration + * @param timestamp Timestamp of the record + * @param Key type of the data records + * @param Value type of the data records + */ public static void produceAbortedKeyValuesSynchronouslyWithTimestamp(final String topic, final Collection> records, final Properties producerConfig, @@ -280,6 +357,13 @@ public static void produceAbortedKeyValuesSynchronouslyWithTimestamp(fina } } + /** + * @param topic Kafka topic to write the data records to + * @param records Data records to write to Kafka + * @param producerConfig Kafka producer configuration + * @param time Timestamp provider + * @param Value type of the data records + */ public static void produceValuesSynchronously(final String topic, final Collection records, final Properties producerConfig, @@ -288,6 +372,14 @@ public static void produceValuesSynchronously(final String topic, IntegrationTestUtils.produceValuesSynchronously(topic, records, producerConfig, time, false); } + /** + * @param topic Kafka topic to write the data records to + * @param records Data records to write to Kafka + * @param producerConfig Kafka producer configuration + * @param time Timestamp provider + * @param enableTransactions Send messages in a transaction + * @param Value type of the data records + */ public static void produceValuesSynchronously(final String topic, final Collection records, final Properties producerConfig, @@ -336,12 +428,61 @@ public static void waitForCompletion(final KafkaStreams streams, } } + /** + * Wait until enough data (consumer records) has been consumed. + * + * @param consumerConfig Kafka Consumer configuration + * @param topic Kafka topic to consume from + * @param expectedNumRecords Minimum number of expected records + * @param Key type of the data records + * @param Value type of the data records + * @return All the records consumed, or null if no records are consumed + */ public static List> waitUntilMinRecordsReceived(final Properties consumerConfig, final String topic, final int expectedNumRecords) throws InterruptedException { return waitUntilMinRecordsReceived(consumerConfig, topic, expectedNumRecords, DEFAULT_TIMEOUT); } + /** + * Wait until enough data (consumer records) has been consumed. + * + * @param consumerConfig Kafka Consumer configuration + * @param topic Kafka topic to consume from + * @param expectedNumRecords Minimum number of expected records + * @param waitTime Upper bound of waiting time in milliseconds + * @param Key type of the data records + * @param Value type of the data records + * @return All the records consumed, or null if no records are consumed + */ + public static List> waitUntilMinRecordsReceived(final Properties consumerConfig, + final String topic, + final int expectedNumRecords, + final long waitTime) throws InterruptedException { + final List> accumData = new ArrayList<>(); + try (final Consumer consumer = createConsumer(consumerConfig)) { + final TestCondition valuesRead = () -> { + final List> readData = + readRecords(topic, consumer, waitTime, expectedNumRecords); + accumData.addAll(readData); + return accumData.size() >= expectedNumRecords; + }; + final String conditionDetails = "Did not receive all " + expectedNumRecords + " records from topic " + topic; + TestUtils.waitForCondition(valuesRead, waitTime, conditionDetails); + } + return accumData; + } + + /** + * Wait until enough data (key-value records) has been consumed. + * + * @param consumerConfig Kafka Consumer configuration + * @param topic Kafka topic to consume from + * @param expectedNumRecords Minimum number of expected records + * @param Key type of the data records + * @param Value type of the data records + * @return All the records consumed, or null if no records are consumed + */ public static List> waitUntilMinKeyValueRecordsReceived(final Properties consumerConfig, final String topic, final int expectedNumRecords) throws InterruptedException { @@ -352,11 +493,13 @@ public static List> waitUntilMinKeyValueRecordsReceived(fi * Wait until enough data (key-value records) has been consumed. * * @param consumerConfig Kafka Consumer configuration - * @param topic Topic to consume from + * @param topic Kafka topic to consume from * @param expectedNumRecords Minimum number of expected records - * @param waitTime Upper bound in waiting time in milliseconds + * @param waitTime Upper bound of waiting time in milliseconds + * @param Key type of the data records + * @param Value type of the data records * @return All the records consumed, or null if no records are consumed - * @throws AssertionError if the given wait time elapses + * @throws AssertionError if the given wait time elapses */ public static List> waitUntilMinKeyValueRecordsReceived(final Properties consumerConfig, final String topic, @@ -377,14 +520,15 @@ public static List> waitUntilMinKeyValueRecordsReceived(fi } /** - * Wait until enough data (key-value records) has been consumed. + * Wait until enough data (timestamped key-value records) has been consumed. * * @param consumerConfig Kafka Consumer configuration - * @param topic Topic to consume from + * @param topic Kafka topic to consume from * @param expectedNumRecords Minimum number of expected records - * @param waitTime Upper bound in waiting time in milliseconds + * @param waitTime Upper bound of waiting time in milliseconds * @return All the records consumed, or null if no records are consumed - * @throws AssertionError if the given wait time elapses + * @param Key type of the data records + * @param Value type of the data records */ public static List>> waitUntilMinKeyValueWithTimestampRecordsReceived(final Properties consumerConfig, final String topic, @@ -404,12 +548,33 @@ public static List>> waitUntilMinKeyValueWi return accumData; } + /** + * Wait until final key-value mappings have been consumed. + * + * @param consumerConfig Kafka Consumer configuration + * @param topic Kafka topic to consume from + * @param expectedRecords Expected key-value mappings + * @param Key type of the data records + * @param Value type of the data records + * @return All the mappings consumed, or null if no records are consumed + */ public static List> waitUntilFinalKeyValueRecordsReceived(final Properties consumerConfig, final String topic, final List> expectedRecords) throws InterruptedException { return waitUntilFinalKeyValueRecordsReceived(consumerConfig, topic, expectedRecords, DEFAULT_TIMEOUT); } + /** + * Wait until final key-value mappings have been consumed. + * + * @param consumerConfig Kafka Consumer configuration + * @param topic Kafka topic to consume from + * @param expectedRecords Expected key-value mappings + * @param waitTime Upper bound of waiting time in milliseconds + * @param Key type of the data records + * @param Value type of the data records + * @return All the mappings consumed, or null if no records are consumed + */ public static List> waitUntilFinalKeyValueRecordsReceived(final Properties consumerConfig, final String topic, final List> expectedRecords, @@ -445,24 +610,15 @@ public static List> waitUntilFinalKeyValueRecordsReceived( return accumData; } - public static List> waitUntilMinRecordsReceived(final Properties consumerConfig, - final String topic, - final int expectedNumRecords, - final long waitTime) throws InterruptedException { - final List> accumData = new ArrayList<>(); - try (final Consumer consumer = createConsumer(consumerConfig)) { - final TestCondition valuesRead = () -> { - final List> readData = - readRecords(topic, consumer, waitTime, expectedNumRecords); - accumData.addAll(readData); - return accumData.size() >= expectedNumRecords; - }; - final String conditionDetails = "Did not receive all " + expectedNumRecords + " records from topic " + topic; - TestUtils.waitForCondition(valuesRead, waitTime, conditionDetails); - } - return accumData; - } - + /** + * Wait until enough data (value records) has been consumed. + * + * @param consumerConfig Kafka Consumer configuration + * @param topic Topic to consume from + * @param expectedNumRecords Minimum number of expected records + * @return All the records consumed, or null if no records are consumed + * @throws AssertionError if the given wait time elapses + */ public static List waitUntilMinValuesRecordsReceived(final Properties consumerConfig, final String topic, final int expectedNumRecords) throws InterruptedException { @@ -476,9 +632,9 @@ public static List waitUntilMinValuesRecordsReceived(final Properties con * @param consumerConfig Kafka Consumer configuration * @param topic Topic to consume from * @param expectedNumRecords Minimum number of expected records - * @param waitTime Upper bound in waiting time in milliseconds + * @param waitTime Upper bound of waiting time in milliseconds * @return All the records consumed, or null if no records are consumed - * @throws AssertionError if the given wait time elapses + * @throws AssertionError if the given wait time elapses */ public static List waitUntilMinValuesRecordsReceived(final Properties consumerConfig, final String topic, From 7e458fdb90f6f13d2701a6294e1564c416fe8a26 Mon Sep 17 00:00:00 2001 From: Lee Dongjin Date: Thu, 24 Jan 2019 21:33:58 +0900 Subject: [PATCH 2/2] Remove IntegrationTestUtils#produceKeyValuesSynchronouslyWithTimestamp(String, Collection>, Properties, Headers, Long) --- .../utils/IntegrationTestUtils.java | 18 ------------------ 1 file changed, 18 deletions(-) diff --git a/streams/src/test/java/org/apache/kafka/streams/integration/utils/IntegrationTestUtils.java b/streams/src/test/java/org/apache/kafka/streams/integration/utils/IntegrationTestUtils.java index efa4582af3374..e61b312146e16 100644 --- a/streams/src/test/java/org/apache/kafka/streams/integration/utils/IntegrationTestUtils.java +++ b/streams/src/test/java/org/apache/kafka/streams/integration/utils/IntegrationTestUtils.java @@ -224,24 +224,6 @@ public static void produceKeyValuesSynchronouslyWithTimestamp(final Strin IntegrationTestUtils.produceKeyValuesSynchronouslyWithTimestamp(topic, records, producerConfig, timestamp, false); } - /** - * @param topic Kafka topic to write the data records to - * @param records Data records to write to Kafka - * @param producerConfig Kafka producer configuration - * @param headers {@link Headers} of the data records - * @param timestamp Timestamp of the record - * @param Key type of the data records - * @param Value type of the data records - */ - public static void produceKeyValuesSynchronouslyWithTimestamp(final String topic, - final Collection> records, - final Properties producerConfig, - final Headers headers, - final Long timestamp) - throws ExecutionException, InterruptedException { - IntegrationTestUtils.produceKeyValuesSynchronouslyWithTimestamp(topic, records, producerConfig, headers, timestamp, false); - } - /** * @param topic Kafka topic to write the data records to * @param records Data records to write to Kafka