diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/CachingSessionStore.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/CachingSessionStore.java index 69506932e5350..c307f6d762865 100644 --- a/streams/src/main/java/org/apache/kafka/streams/state/internals/CachingSessionStore.java +++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/CachingSessionStore.java @@ -147,7 +147,7 @@ public void put(final Windowed key, byte[] value) { context.headers(), true, context.offset(), - key.window().end(), + context.timestamp(), context.partition(), context.topic()); cache.put(cacheName, cacheFunction.cacheKey(binaryKey), entry); diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/CachingWindowStore.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/CachingWindowStore.java index 1f08f5157b94b..07120dffbbffb 100644 --- a/streams/src/main/java/org/apache/kafka/streams/state/internals/CachingWindowStore.java +++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/CachingWindowStore.java @@ -156,7 +156,7 @@ public synchronized void put(final Bytes key, final byte[] value, final long tim context.headers(), true, context.offset(), - timestamp, + context.timestamp(), context.partition(), context.topic()); cache.put(name, cacheFunction.cacheKey(keyBytes), entry); diff --git a/streams/src/test/java/org/apache/kafka/streams/integration/KStreamAggregationIntegrationTest.java b/streams/src/test/java/org/apache/kafka/streams/integration/KStreamAggregationIntegrationTest.java index 10363f82ebe0f..a29332c5da746 100644 --- a/streams/src/test/java/org/apache/kafka/streams/integration/KStreamAggregationIntegrationTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/integration/KStreamAggregationIntegrationTest.java @@ -28,7 +28,6 @@ import org.apache.kafka.common.serialization.StringDeserializer; import org.apache.kafka.common.serialization.StringSerializer; import org.apache.kafka.common.utils.Bytes; -import org.apache.kafka.streams.kstream.Consumed; import org.apache.kafka.streams.KafkaStreams; import org.apache.kafka.streams.KeyValue; import org.apache.kafka.streams.StreamsBuilder; @@ -36,7 +35,7 @@ import org.apache.kafka.streams.integration.utils.EmbeddedKafkaCluster; import org.apache.kafka.streams.integration.utils.IntegrationTestUtils; import org.apache.kafka.streams.kstream.Aggregator; -import org.apache.kafka.streams.kstream.ForeachAction; +import org.apache.kafka.streams.kstream.Consumed; import org.apache.kafka.streams.kstream.Initializer; import org.apache.kafka.streams.kstream.KGroupedStream; import org.apache.kafka.streams.kstream.KStream; @@ -49,16 +48,16 @@ import org.apache.kafka.streams.kstream.SessionWindows; import org.apache.kafka.streams.kstream.TimeWindowedDeserializer; import org.apache.kafka.streams.kstream.TimeWindows; +import org.apache.kafka.streams.kstream.Transformer; import org.apache.kafka.streams.kstream.Windowed; import org.apache.kafka.streams.kstream.WindowedSerdes; import org.apache.kafka.streams.kstream.internals.SessionWindow; import org.apache.kafka.streams.kstream.internals.TimeWindow; +import org.apache.kafka.streams.processor.ProcessorContext; import org.apache.kafka.streams.state.KeyValueIterator; import org.apache.kafka.streams.state.KeyValueStore; import org.apache.kafka.streams.state.QueryableStoreTypes; import org.apache.kafka.streams.state.ReadOnlySessionStore; -import org.apache.kafka.streams.state.SessionStore; -import org.apache.kafka.streams.state.WindowStore; import org.apache.kafka.test.IntegrationTest; import org.apache.kafka.test.MockMapper; import org.apache.kafka.test.TestUtils; @@ -136,24 +135,9 @@ public void before() throws InterruptedException { mapper, Serialized.with(Serdes.String(), Serdes.String())); - reducer = new Reducer() { - @Override - public String apply(final String value1, final String value2) { - return value1 + ":" + value2; - } - }; - initializer = new Initializer() { - @Override - public Integer apply() { - return 0; - } - }; - aggregator = new Aggregator() { - @Override - public Integer apply(final String aggKey, final String value, final Integer aggregate) { - return aggregate + value.length(); - } - }; + reducer = (value1, value2) -> value1 + ":" + value2; + initializer = () -> 0; + aggregator = (aggKey, value, aggregate) -> aggregate + value.length(); } @After @@ -181,12 +165,7 @@ public void shouldReduce() throws Exception { new StringDeserializer(), 10); - Collections.sort(results, new Comparator>() { - @Override - public int compare(final KeyValue o1, final KeyValue o2) { - return KStreamAggregationIntegrationTest.compare(o1, o2); - } - }); + Collections.sort(results, KStreamAggregationIntegrationTest::compare); assertThat(results, is(Arrays.asList(KeyValue.pair("A", "A"), KeyValue.pair("A", "A:A"), @@ -218,7 +197,7 @@ public void shouldReduceWindowed() throws Exception { produceMessages(secondBatchTimestamp); produceMessages(secondBatchTimestamp); - Serde> windowedSerde = WindowedSerdes.timeWindowedSerdeFrom(String.class); + final Serde> windowedSerde = WindowedSerdes.timeWindowedSerdeFrom(String.class); groupedStream .windowedBy(TimeWindows.of(500L)) .reduce(reducer) @@ -228,34 +207,28 @@ public void shouldReduceWindowed() throws Exception { startStreams(); final List, String>> windowedOutput = receiveMessages( - new TimeWindowedDeserializer(), + new TimeWindowedDeserializer<>(), new StringDeserializer(), String.class, 15); // read from ConsoleConsumer - String resultFromConsoleConsumer = readWindowedKeyedMessagesViaConsoleConsumer( - new TimeWindowedDeserializer(), - new StringDeserializer(), - String.class, - 15); + final String resultFromConsoleConsumer = readWindowedKeyedMessagesViaConsoleConsumer( + new TimeWindowedDeserializer(), + new StringDeserializer(), + String.class, + 15, + false); final Comparator, String>> comparator = - new Comparator, String>>() { - @Override - public int compare(final KeyValue, String> o1, - final KeyValue, String> o2) { - final int keyComparison = o1.key.key().compareTo(o2.key.key()); - return keyComparison == 0 ? o1.value.compareTo(o2.value) : keyComparison; - } - }; + Comparator.comparing((KeyValue, String> o) -> o.key.key()).thenComparing(o -> o.value); Collections.sort(windowedOutput, comparator); final long firstBatchWindow = firstBatchTimestamp / 500 * 500; final long secondBatchWindow = secondBatchTimestamp / 500 * 500; - List, String>> expectResult = Arrays.asList( + final List, String>> expectResult = Arrays.asList( new KeyValue<>(new Windowed<>("A", new TimeWindow(firstBatchWindow, Long.MAX_VALUE)), "A"), new KeyValue<>(new Windowed<>("A", new TimeWindow(secondBatchWindow, Long.MAX_VALUE)), "A"), new KeyValue<>(new Windowed<>("A", new TimeWindow(secondBatchWindow, Long.MAX_VALUE)), "A:A"), @@ -274,13 +247,13 @@ public int compare(final KeyValue, String> o1, ); assertThat(windowedOutput, is(expectResult)); - Set expectResultString = new HashSet<>(expectResult.size()); - for (KeyValue, String> eachRecord: expectResult) { + final Set expectResultString = new HashSet<>(expectResult.size()); + for (final KeyValue, String> eachRecord: expectResult) { expectResultString.add(eachRecord.toString()); } // check every message is contained in the expect result - String[] allRecords = resultFromConsoleConsumer.split("\n"); + final String[] allRecords = resultFromConsoleConsumer.split("\n"); for (String record: allRecords) { record = "KeyValue(" + record + ")"; assertTrue(expectResultString.contains(record)); @@ -306,12 +279,7 @@ public void shouldAggregate() throws Exception { new IntegerDeserializer(), 10); - Collections.sort(results, new Comparator>() { - @Override - public int compare(final KeyValue o1, final KeyValue o2) { - return KStreamAggregationIntegrationTest.compare(o1, o2); - } - }); + Collections.sort(results, KStreamAggregationIntegrationTest::compare); assertThat(results, is(Arrays.asList( KeyValue.pair("A", 1), @@ -336,75 +304,68 @@ public void shouldAggregateWindowed() throws Exception { produceMessages(secondTimestamp); produceMessages(secondTimestamp); - Serde> windowedSerde = WindowedSerdes.timeWindowedSerdeFrom(String.class); + final Serde> windowedSerde = WindowedSerdes.timeWindowedSerdeFrom(String.class); groupedStream.windowedBy(TimeWindows.of(500L)) .aggregate( initializer, aggregator, - Materialized.>with(null, Serdes.Integer()) + Materialized.with(null, Serdes.Integer()) ) .toStream() .to(outputTopic, Produced.with(windowedSerde, Serdes.Integer())); startStreams(); - final List, Integer>> windowedMessages = receiveMessages( - new TimeWindowedDeserializer(), + final List, KeyValue>> windowedMessages = receiveMessagesWithTimestamp( + new TimeWindowedDeserializer<>(), new IntegerDeserializer(), String.class, 15); // read from ConsoleConsumer - String resultFromConsoleConsumer = readWindowedKeyedMessagesViaConsoleConsumer( - new TimeWindowedDeserializer(), - new IntegerDeserializer(), - String.class, - 15); + final String resultFromConsoleConsumer = readWindowedKeyedMessagesViaConsoleConsumer( + new TimeWindowedDeserializer(), + new IntegerDeserializer(), + String.class, + 15, + true); - final Comparator, Integer>> + final Comparator, KeyValue>> comparator = - new Comparator, Integer>>() { - @Override - public int compare(final KeyValue, Integer> o1, - final KeyValue, Integer> o2) { - final int keyComparison = o1.key.key().compareTo(o2.key.key()); - return keyComparison == 0 ? o1.value.compareTo(o2.value) : keyComparison; - } - }; + Comparator.comparing((KeyValue, KeyValue> o) -> o.key.key()).thenComparingInt(o -> o.value.key); Collections.sort(windowedMessages, comparator); final long firstWindow = firstTimestamp / 500 * 500; final long secondWindow = secondTimestamp / 500 * 500; - List, Integer>> expectResult = Arrays.asList( - new KeyValue<>(new Windowed<>("A", new TimeWindow(firstWindow, Long.MAX_VALUE)), 1), - new KeyValue<>(new Windowed<>("A", new TimeWindow(secondWindow, Long.MAX_VALUE)), 1), - new KeyValue<>(new Windowed<>("A", new TimeWindow(secondWindow, Long.MAX_VALUE)), 2), - new KeyValue<>(new Windowed<>("B", new TimeWindow(firstWindow, Long.MAX_VALUE)), 1), - new KeyValue<>(new Windowed<>("B", new TimeWindow(secondWindow, Long.MAX_VALUE)), 1), - new KeyValue<>(new Windowed<>("B", new TimeWindow(secondWindow, Long.MAX_VALUE)), 2), - new KeyValue<>(new Windowed<>("C", new TimeWindow(firstWindow, Long.MAX_VALUE)), 1), - new KeyValue<>(new Windowed<>("C", new TimeWindow(secondWindow, Long.MAX_VALUE)), 1), - new KeyValue<>(new Windowed<>("C", new TimeWindow(secondWindow, Long.MAX_VALUE)), 2), - new KeyValue<>(new Windowed<>("D", new TimeWindow(firstWindow, Long.MAX_VALUE)), 1), - new KeyValue<>(new Windowed<>("D", new TimeWindow(secondWindow, Long.MAX_VALUE)), 1), - new KeyValue<>(new Windowed<>("D", new TimeWindow(secondWindow, Long.MAX_VALUE)), 2), - new KeyValue<>(new Windowed<>("E", new TimeWindow(firstWindow, Long.MAX_VALUE)), 1), - new KeyValue<>(new Windowed<>("E", new TimeWindow(secondWindow, Long.MAX_VALUE)), 1), - new KeyValue<>(new Windowed<>("E", new TimeWindow(secondWindow, Long.MAX_VALUE)), 2)); + final List, KeyValue>> expectResult = Arrays.asList( + new KeyValue<>(new Windowed<>("A", new TimeWindow(firstWindow, Long.MAX_VALUE)), KeyValue.pair(1, firstTimestamp)), + new KeyValue<>(new Windowed<>("A", new TimeWindow(secondWindow, Long.MAX_VALUE)), KeyValue.pair(1, secondTimestamp)), + new KeyValue<>(new Windowed<>("A", new TimeWindow(secondWindow, Long.MAX_VALUE)), KeyValue.pair(2, secondTimestamp)), + new KeyValue<>(new Windowed<>("B", new TimeWindow(firstWindow, Long.MAX_VALUE)), KeyValue.pair(1, firstTimestamp)), + new KeyValue<>(new Windowed<>("B", new TimeWindow(secondWindow, Long.MAX_VALUE)), KeyValue.pair(1, secondTimestamp)), + new KeyValue<>(new Windowed<>("B", new TimeWindow(secondWindow, Long.MAX_VALUE)), KeyValue.pair(2, secondTimestamp)), + new KeyValue<>(new Windowed<>("C", new TimeWindow(firstWindow, Long.MAX_VALUE)), KeyValue.pair(1, firstTimestamp)), + new KeyValue<>(new Windowed<>("C", new TimeWindow(secondWindow, Long.MAX_VALUE)), KeyValue.pair(1, secondTimestamp)), + new KeyValue<>(new Windowed<>("C", new TimeWindow(secondWindow, Long.MAX_VALUE)), KeyValue.pair(2, secondTimestamp)), + new KeyValue<>(new Windowed<>("D", new TimeWindow(firstWindow, Long.MAX_VALUE)), KeyValue.pair(1, firstTimestamp)), + new KeyValue<>(new Windowed<>("D", new TimeWindow(secondWindow, Long.MAX_VALUE)), KeyValue.pair(1, secondTimestamp)), + new KeyValue<>(new Windowed<>("D", new TimeWindow(secondWindow, Long.MAX_VALUE)), KeyValue.pair(2, secondTimestamp)), + new KeyValue<>(new Windowed<>("E", new TimeWindow(firstWindow, Long.MAX_VALUE)), KeyValue.pair(1, firstTimestamp)), + new KeyValue<>(new Windowed<>("E", new TimeWindow(secondWindow, Long.MAX_VALUE)), KeyValue.pair(1, secondTimestamp)), + new KeyValue<>(new Windowed<>("E", new TimeWindow(secondWindow, Long.MAX_VALUE)), KeyValue.pair(2, secondTimestamp))); assertThat(windowedMessages, is(expectResult)); - Set expectResultString = new HashSet<>(expectResult.size()); - for (KeyValue, Integer> eachRecord: expectResult) { - expectResultString.add(eachRecord.toString()); + final Set expectResultString = new HashSet<>(expectResult.size()); + for (final KeyValue, KeyValue> eachRecord: expectResult) { + expectResultString.add("CreateTime:" + eachRecord.value.value + ", " + eachRecord.key.toString() + ", " + eachRecord.value.key); } // check every message is contained in the expect result - String[] allRecords = resultFromConsoleConsumer.split("\n"); - for (String record: allRecords) { - record = "KeyValue(" + record + ")"; + final String[] allRecords = resultFromConsoleConsumer.split("\n"); + for (final String record: allRecords) { assertTrue(expectResultString.contains(record)); } @@ -419,12 +380,7 @@ private void shouldCountHelper() throws Exception { new StringDeserializer(), new LongDeserializer(), 10); - Collections.sort(results, new Comparator>() { - @Override - public int compare(final KeyValue o1, final KeyValue o2) { - return KStreamAggregationIntegrationTest.compare(o1, o2); - } - }); + Collections.sort(results, KStreamAggregationIntegrationTest::compare); assertThat(results, is(Arrays.asList( KeyValue.pair("A", 1L), @@ -444,7 +400,7 @@ public int compare(final KeyValue o1, final KeyValue public void shouldCount() throws Exception { produceMessages(mockTime.milliseconds()); - groupedStream.count(Materialized.>as("count-by-key")) + groupedStream.count(Materialized.as("count-by-key")) .toStream() .to(outputTopic, Produced.with(Serdes.String(), Serdes.Long())); @@ -471,12 +427,7 @@ public void shouldGroupByKey() throws Exception { stream.groupByKey(Serialized.with(Serdes.Integer(), Serdes.String())) .windowedBy(TimeWindows.of(500L)) .count() - .toStream(new KeyValueMapper, Long, String>() { - @Override - public String apply(final Windowed windowedKey, final Long value) { - return windowedKey.key() + "@" + windowedKey.window().start(); - } - }).to(outputTopic, Produced.with(Serdes.String(), Serdes.Long())); + .toStream((windowedKey, value) -> windowedKey.key() + "@" + windowedKey.window().start()).to(outputTopic, Produced.with(Serdes.String(), Serdes.Long())); startStreams(); @@ -484,12 +435,7 @@ public String apply(final Windowed windowedKey, final Long value) { new StringDeserializer(), new LongDeserializer(), 10); - Collections.sort(results, new Comparator>() { - @Override - public int compare(final KeyValue o1, final KeyValue o2) { - return KStreamAggregationIntegrationTest.compare(o1, o2); - } - }); + Collections.sort(results, KStreamAggregationIntegrationTest::compare); final long window = timestamp / 500 * 500; assertThat(results, is(Arrays.asList( @@ -568,7 +514,7 @@ public void shouldCountSessionWindows() throws Exception { new Properties()), t4); - final Map, Long> results = new HashMap<>(); + final Map, KeyValue> results = new HashMap<>(); final CountDownLatch latch = new CountDownLatch(11); builder.stream(userSessionsStream, Consumed.with(Serdes.String(), Serdes.String())) @@ -576,23 +522,34 @@ public void shouldCountSessionWindows() throws Exception { .windowedBy(SessionWindows.with(sessionGap).until(maintainMillis)) .count() .toStream() - .foreach(new ForeachAction, Long>() { - @Override - public void apply(final Windowed key, final Long value) { - results.put(key, value); - latch.countDown(); - } - }); + .transform(() -> new Transformer, Long, KeyValue>() { + private ProcessorContext context; + + @Override + public void init(final ProcessorContext context) { + this.context = context; + } + + @Override + public KeyValue transform(final Windowed key, final Long value) { + results.put(key, KeyValue.pair(value, context.timestamp())); + latch.countDown(); + return null; + } + + @Override + public void close() {} + }); startStreams(); latch.await(30, TimeUnit.SECONDS); - assertThat(results.get(new Windowed<>("bob", new SessionWindow(t1, t1))), equalTo(1L)); - assertThat(results.get(new Windowed<>("penny", new SessionWindow(t1, t1))), equalTo(1L)); - assertThat(results.get(new Windowed<>("jo", new SessionWindow(t1, t1))), equalTo(1L)); - assertThat(results.get(new Windowed<>("jo", new SessionWindow(t4, t4))), equalTo(1L)); - assertThat(results.get(new Windowed<>("emily", new SessionWindow(t1, t2))), equalTo(2L)); - assertThat(results.get(new Windowed<>("bob", new SessionWindow(t3, t4))), equalTo(2L)); - assertThat(results.get(new Windowed<>("penny", new SessionWindow(t3, t3))), equalTo(1L)); + assertThat(results.get(new Windowed<>("bob", new SessionWindow(t1, t1))), equalTo(KeyValue.pair(1L, t1))); + assertThat(results.get(new Windowed<>("penny", new SessionWindow(t1, t1))), equalTo(KeyValue.pair(1L, t1))); + assertThat(results.get(new Windowed<>("jo", new SessionWindow(t1, t1))), equalTo(KeyValue.pair(1L, t1))); + assertThat(results.get(new Windowed<>("jo", new SessionWindow(t4, t4))), equalTo(KeyValue.pair(1L, t4))); + assertThat(results.get(new Windowed<>("emily", new SessionWindow(t1, t2))), equalTo(KeyValue.pair(2L, t2))); + assertThat(results.get(new Windowed<>("bob", new SessionWindow(t3, t4))), equalTo(KeyValue.pair(2L, t4))); + assertThat(results.get(new Windowed<>("penny", new SessionWindow(t3, t3))), equalTo(KeyValue.pair(1L, t3))); } @Test @@ -662,25 +619,17 @@ public void shouldReduceSessionWindows() throws Exception { builder.stream(userSessionsStream, Consumed.with(Serdes.String(), Serdes.String())) .groupByKey(Serialized.with(Serdes.String(), Serdes.String())) .windowedBy(SessionWindows.with(sessionGap).until(maintainMillis)) - .reduce(new Reducer() { - @Override - public String apply(final String value1, final String value2) { - return value1 + ":" + value2; - } - }, Materialized.>as(userSessionsStore)) + .reduce((value1, value2) -> value1 + ":" + value2, Materialized.as(userSessionsStore)) .toStream() - .foreach(new ForeachAction, String>() { - @Override - public void apply(final Windowed key, final String value) { - results.put(key, value); - latch.countDown(); - } + .foreach((key, value) -> { + results.put(key, value); + latch.countDown(); }); startStreams(); latch.await(30, TimeUnit.SECONDS); final ReadOnlySessionStore sessionStore - = kafkaStreams.store(userSessionsStore, QueryableStoreTypes.sessionStore()); + = kafkaStreams.store(userSessionsStore, QueryableStoreTypes.sessionStore()); // verify correct data received assertThat(results.get(new Windowed<>("bob", new SessionWindow(t1, t1))), equalTo("start")); @@ -732,16 +681,14 @@ private void startStreams() { } private List> receiveMessages(final Deserializer keyDeserializer, - final Deserializer valueDeserializer, - final int numMessages) + final Deserializer valueDeserializer, + final int numMessages) throws InterruptedException { return receiveMessages(keyDeserializer, valueDeserializer, null, numMessages); } - private List> receiveMessages(final Deserializer - keyDeserializer, - final Deserializer - valueDeserializer, + private List> receiveMessages(final Deserializer keyDeserializer, + final Deserializer valueDeserializer, final Class innerClass, final int numMessages) throws InterruptedException { final Properties consumerProperties = new Properties(); @@ -761,21 +708,44 @@ private List> receiveMessages(final Deserializer 60 * 1000); } + private List>> receiveMessagesWithTimestamp(final Deserializer keyDeserializer, + final Deserializer valueDeserializer, + final Class innerClass, + final int numMessages) throws InterruptedException { + final Properties consumerProperties = new Properties(); + consumerProperties.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, CLUSTER.bootstrapServers()); + consumerProperties.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "kgroupedstream-test-" + testNo); + consumerProperties.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); + consumerProperties.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, keyDeserializer.getClass().getName()); + consumerProperties.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, valueDeserializer.getClass().getName()); + if (keyDeserializer instanceof TimeWindowedDeserializer || keyDeserializer instanceof SessionWindowedDeserializer) { + consumerProperties.setProperty(StreamsConfig.DEFAULT_WINDOWED_KEY_SERDE_INNER_CLASS, + Serdes.serdeFrom(innerClass).getClass().getName()); + } + return IntegrationTestUtils.waitUntilMinKeyValueWithTimestampRecordsReceived( + consumerProperties, + outputTopic, + numMessages, + 60 * 1000); + } + private String readWindowedKeyedMessagesViaConsoleConsumer(final Deserializer keyDeserializer, final Deserializer valueDeserializer, final Class innerClass, - final int numMessages) { - ByteArrayOutputStream newConsole = new ByteArrayOutputStream(); - PrintStream originalStream = System.out; - try (PrintStream newStream = new PrintStream(newConsole)) { + final int numMessages, + final boolean printTimestamp) { + final ByteArrayOutputStream newConsole = new ByteArrayOutputStream(); + final PrintStream originalStream = System.out; + try (final PrintStream newStream = new PrintStream(newConsole)) { System.setOut(newStream); - String keySeparator = ", "; + final String keySeparator = ", "; // manually construct the console consumer argument array - String[] args = new String[] { + final String[] args = new String[] { "--bootstrap-server", CLUSTER.bootstrapServers(), "--from-beginning", "--property", "print.key=true", + "--property", "print.timestamp=" + printTimestamp, "--topic", outputTopic, "--max-messages", String.valueOf(numMessages), "--property", "key.deserializer=" + keyDeserializer.getClass().getName(), 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 a0b8f3d5aca30..749d74887e5b1 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 @@ -192,7 +192,7 @@ public static void produceKeyValuesSynchronouslyWithTimestamp(final Strin final Long timestamp, final boolean enabledTransactions) throws ExecutionException, InterruptedException { - try (Producer producer = new KafkaProducer<>(producerConfig)) { + try (final Producer producer = new KafkaProducer<>(producerConfig)) { if (enabledTransactions) { producer.initTransactions(); producer.beginTransaction(); @@ -310,14 +310,39 @@ public static List> waitUntilMinKeyValueRecordsReceived(fi final long waitTime) throws InterruptedException { final List> accumData = new ArrayList<>(); try (final Consumer consumer = createConsumer(consumerConfig)) { - final TestCondition valuesRead = new TestCondition() { - @Override - public boolean conditionMet() { - final List> readData = - readKeyValues(topic, consumer, waitTime, expectedNumRecords); - accumData.addAll(readData); - return accumData.size() >= expectedNumRecords; - } + final TestCondition valuesRead = () -> { + final List> readData = + readKeyValues(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 Topic to consume from + * @param expectedNumRecords Minimum number of expected records + * @param waitTime Upper bound in waiting time in milliseconds + * @return All the records consumed, or null if no records are consumed + * @throws AssertionError if the given wait time elapses + */ + public static List>> waitUntilMinKeyValueWithTimestampRecordsReceived(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 = + readKeyValuesWithTimestamp(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); @@ -331,14 +356,11 @@ public static List> waitUntilMinRecordsReceived(fina final long waitTime) throws InterruptedException { final List> accumData = new ArrayList<>(); try (final Consumer consumer = createConsumer(consumerConfig)) { - final TestCondition valuesRead = new TestCondition() { - @Override - public boolean conditionMet() { - final List> readData = - readRecords(topic, consumer, waitTime, expectedNumRecords); - accumData.addAll(readData); - return accumData.size() >= expectedNumRecords; - } + 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); @@ -369,14 +391,11 @@ public static List waitUntilMinValuesRecordsReceived(final Properties con final long waitTime) throws InterruptedException { final List accumData = new ArrayList<>(); try (final Consumer consumer = createConsumer(consumerConfig)) { - final TestCondition valuesRead = new TestCondition() { - @Override - public boolean conditionMet() { - final List readData = - readValues(topic, consumer, waitTime, expectedNumRecords); - accumData.addAll(readData); - return accumData.size() >= expectedNumRecords; - } + final TestCondition valuesRead = () -> { + final List readData = + readValues(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); @@ -401,23 +420,20 @@ public static void waitUntilMetadataIsPropagated(final List servers final String topic, final int partition, final long timeout) throws InterruptedException { - TestUtils.waitForCondition(new TestCondition() { - @Override - public boolean conditionMet() { - for (final KafkaServer server : servers) { - final MetadataCache metadataCache = server.apis().metadataCache(); - final Option partitionInfo = - metadataCache.getPartitionInfo(topic, partition); - if (partitionInfo.isEmpty()) { - return false; - } - final UpdateMetadataRequest.PartitionState metadataPartitionState = partitionInfo.get(); - if (!Request.isValidBrokerId(metadataPartitionState.basePartitionState.leader)) { - return false; - } + TestUtils.waitForCondition(() -> { + for (final KafkaServer server : servers) { + final MetadataCache metadataCache = server.apis().metadataCache(); + final Option partitionInfo = + metadataCache.getPartitionInfo(topic, partition); + if (partitionInfo.isEmpty()) { + return false; + } + final UpdateMetadataRequest.PartitionState metadataPartitionState = partitionInfo.get(); + if (!Request.isValidBrokerId(metadataPartitionState.basePartitionState.leader)) { + return false; } - return true; } + return true; }, timeout, "metadata for topic=" + topic + " partition=" + partition + " not propagated to all brokers"); } @@ -502,6 +518,28 @@ private static List> readKeyValues(final String topic, return consumedValues; } + /** + * Returns up to `maxMessages` by reading via the provided consumer (the topic(s) to read from + * are already configured in the consumer). + * + * @param topic Kafka topic to read messages from + * @param consumer Kafka consumer + * @param waitTime Maximum wait time in milliseconds + * @param maxMessages Maximum number of messages to read via the consumer + * @return The KeyValue elements retrieved via the consumer + */ + private static List>> readKeyValuesWithTimestamp(final String topic, + final Consumer consumer, + final long waitTime, + final int maxMessages) { + final List>> consumedValues = new ArrayList<>(); + final List> records = readRecords(topic, consumer, waitTime, maxMessages); + for (final ConsumerRecord record : records) { + consumedValues.add(new KeyValue<>(record.key(), KeyValue.pair(record.value(), record.timestamp()))); + } + return consumedValues; + } + private static List> readRecords(final String topic, final Consumer consumer, final long waitTime,