diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/RecordCollectorImpl.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/RecordCollectorImpl.java index 358af6b1b3aed..38445fde9bf0e 100644 --- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/RecordCollectorImpl.java +++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/RecordCollectorImpl.java @@ -219,12 +219,13 @@ public void send(final String topic, } if (!topic.endsWith("-changelog")) { - // we may not have created a sensor yet if the node uses dynamic topic routing final Map producedSensorByTopic = sinkNodeToProducedSensorByTopic.get(processorNodeId); if (producedSensorByTopic == null) { log.error("Unable to records bytes produced to topic {} by sink node {} as the node is not recognized.\n" + "Known sink nodes are {}.", topic, processorNodeId, sinkNodeToProducedSensorByTopic.keySet()); } else { + // we may not have created a sensor during initialization if the node uses dynamic topic routing, + // as all topics are not known up front, so create the sensor for that topic if absent final Sensor topicProducedSensor = producedSensorByTopic.computeIfAbsent( topic, t -> TopicMetrics.producedSensor( diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/metrics/StreamsMetricsImpl.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/metrics/StreamsMetricsImpl.java index cb77fc9bdd14a..3260bfc1b8276 100644 --- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/metrics/StreamsMetricsImpl.java +++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/metrics/StreamsMetricsImpl.java @@ -117,7 +117,7 @@ public int hashCode() { public static final String THREAD_ID_TAG = "thread-id"; public static final String TASK_ID_TAG = "task-id"; public static final String PROCESSOR_NODE_ID_TAG = "processor-node-id"; - public static final String TOPIC_NAME_TAG = "topic-name"; + public static final String TOPIC_NAME_TAG = "topic"; public static final String STORE_ID_TAG = "state-id"; public static final String RECORD_CACHE_ID_TAG = "record-cache-id"; diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/metrics/TopicMetricsTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/metrics/TopicMetricsTest.java index c359affe5dae1..b698b26192a08 100644 --- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/metrics/TopicMetricsTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/metrics/TopicMetricsTest.java @@ -39,7 +39,7 @@ public class TopicMetricsTest { private static final String THREAD_ID = "test-thread"; private static final String TASK_ID = "test-task"; private static final String PROCESSOR_NODE_ID = "test-processor"; - private static final String TOPIC_NAME = "topic"; + private static final String TOPIC = "topic"; private final Map tagMap = Collections.singletonMap("hello", "world"); @@ -59,14 +59,14 @@ public void shouldGetRecordsAndBytesConsumedSensor() { final String descriptionOfRecordsTotal = "The total number of records consumed from this topic"; final String descriptionOfBytesTotal = "The total number of bytes consumed from this topic"; - when(streamsMetrics.topicLevelSensor(THREAD_ID, TASK_ID, PROCESSOR_NODE_ID, TOPIC_NAME, "consumed", RecordingLevel.INFO)) + when(streamsMetrics.topicLevelSensor(THREAD_ID, TASK_ID, PROCESSOR_NODE_ID, TOPIC, "consumed", RecordingLevel.INFO)) .thenReturn(expectedSensor); - when(streamsMetrics.topicLevelSensor(THREAD_ID, TASK_ID, PROCESSOR_NODE_ID, TOPIC_NAME, "consumed", RecordingLevel.INFO)) + when(streamsMetrics.topicLevelSensor(THREAD_ID, TASK_ID, PROCESSOR_NODE_ID, TOPIC, "consumed", RecordingLevel.INFO)) .thenReturn(expectedSensor); - when(streamsMetrics.topicLevelTagMap(THREAD_ID, TASK_ID, PROCESSOR_NODE_ID, TOPIC_NAME)).thenReturn(tagMap); + when(streamsMetrics.topicLevelTagMap(THREAD_ID, TASK_ID, PROCESSOR_NODE_ID, TOPIC)).thenReturn(tagMap); verifySensor( - () -> TopicMetrics.consumedSensor(THREAD_ID, TASK_ID, PROCESSOR_NODE_ID, TOPIC_NAME, streamsMetrics) + () -> TopicMetrics.consumedSensor(THREAD_ID, TASK_ID, PROCESSOR_NODE_ID, TOPIC, streamsMetrics) ); STREAMS_METRICS_STATIC_MOCK.verify( @@ -89,13 +89,13 @@ public void shouldGetRecordsAndBytesProducedSensor() { final String descriptionOfRecordsTotal = "The total number of records produced to this topic"; final String descriptionOfBytesTotal = "The total number of bytes produced to this topic"; - when(streamsMetrics.topicLevelSensor(THREAD_ID, TASK_ID, PROCESSOR_NODE_ID, TOPIC_NAME, "produced", RecordingLevel.INFO)) + when(streamsMetrics.topicLevelSensor(THREAD_ID, TASK_ID, PROCESSOR_NODE_ID, TOPIC, "produced", RecordingLevel.INFO)) .thenReturn(expectedSensor); - when(streamsMetrics.topicLevelSensor(THREAD_ID, TASK_ID, PROCESSOR_NODE_ID, TOPIC_NAME, "produced", RecordingLevel.INFO)) + when(streamsMetrics.topicLevelSensor(THREAD_ID, TASK_ID, PROCESSOR_NODE_ID, TOPIC, "produced", RecordingLevel.INFO)) .thenReturn(expectedSensor); - when(streamsMetrics.topicLevelTagMap(THREAD_ID, TASK_ID, PROCESSOR_NODE_ID, TOPIC_NAME)).thenReturn(tagMap); + when(streamsMetrics.topicLevelTagMap(THREAD_ID, TASK_ID, PROCESSOR_NODE_ID, TOPIC)).thenReturn(tagMap); - verifySensor(() -> TopicMetrics.producedSensor(THREAD_ID, TASK_ID, PROCESSOR_NODE_ID, TOPIC_NAME, streamsMetrics)); + verifySensor(() -> TopicMetrics.producedSensor(THREAD_ID, TASK_ID, PROCESSOR_NODE_ID, TOPIC, streamsMetrics)); STREAMS_METRICS_STATIC_MOCK.verify( () -> StreamsMetricsImpl.addTotalCountAndSumMetricsToSensor( diff --git a/streams/src/test/java/org/apache/kafka/test/MockRecordCollector.java b/streams/src/test/java/org/apache/kafka/test/MockRecordCollector.java index c99a32c980f35..8a7f543496356 100644 --- a/streams/src/test/java/org/apache/kafka/test/MockRecordCollector.java +++ b/streams/src/test/java/org/apache/kafka/test/MockRecordCollector.java @@ -50,12 +50,14 @@ public void send(final String topic, final Serializer valueSerializer, final String processorNodeId, final InternalProcessorContext context) { - collected.add(new ProducerRecord<>(topic, - partition, - timestamp, - key, - value, - headers)); + collected.add(new ProducerRecord<>( + topic, + partition, + timestamp, + key, + value, + headers) + ); } @Override @@ -69,12 +71,14 @@ public void send(final String topic, final String processorNodeId, final InternalProcessorContext context, final StreamPartitioner partitioner) { - collected.add(new ProducerRecord<>(topic, + collected.add(new ProducerRecord<>( + topic, 0, // partition id timestamp, key, value, - headers)); + headers) + ); } @Override diff --git a/streams/streams-scala/src/test/scala/org/apache/kafka/streams/scala/kstream/KStreamSplitTest.scala b/streams/streams-scala/src/test/scala/org/apache/kafka/streams/scala/kstream/KStreamSplitTest.scala index 89adbd6b1ef29..bbcc1b503f4fa 100644 --- a/streams/streams-scala/src/test/scala/org/apache/kafka/streams/scala/kstream/KStreamSplitTest.scala +++ b/streams/streams-scala/src/test/scala/org/apache/kafka/streams/scala/kstream/KStreamSplitTest.scala @@ -17,7 +17,6 @@ package org.apache.kafka.streams.scala.kstream import org.apache.kafka.streams.kstream.Named -import org.apache.kafka.streams.KeyValue import org.apache.kafka.streams.scala.ImplicitConversions._ import org.apache.kafka.streams.scala.StreamsBuilder import org.apache.kafka.streams.scala.serialization.Serdes._ @@ -36,7 +35,7 @@ class KStreamSplitTest extends TestDriver { val sinkTopic = Array("default", "even", "three"); val m = builder - .stream[Int, Int](sourceTopic) + .stream[Integer, Integer](sourceTopic) .split(Named.as("_")) .branch((_, v) => v % 2 == 0) .branch((_, v) => v % 3 == 0) @@ -47,17 +46,14 @@ class KStreamSplitTest extends TestDriver { m("_2").to(sinkTopic(2)) val testDriver = createTestDriver(builder) - val testInput = testDriver.createInput[Int, Int](sourceTopic) - val testOutput = sinkTopic.map(name => testDriver.createOutput[Int, Int](name)) - - testInput pipeKeyValueList List( - new KeyValue(1, 1), - new KeyValue(1, 2), - new KeyValue(1, 3), - new KeyValue(1, 4), - new KeyValue(1, 5) - ).asJava - + val testInput = testDriver.createInput[Integer, Integer](sourceTopic) + val testOutput = sinkTopic.map(name => testDriver.createOutput[Integer, Integer](name)) + + testInput.pipeValueList( + List(1, 2, 3, 4, 5) + .map(Integer.valueOf) + .asJava + ) assertEquals(List(1, 5), testOutput(0).readValuesToList().asScala) assertEquals(List(2, 4), testOutput(1).readValuesToList().asScala) assertEquals(List(3), testOutput(2).readValuesToList().asScala) @@ -71,7 +67,7 @@ class KStreamSplitTest extends TestDriver { val sourceTopic = "source" val m = builder - .stream[Int, Int](sourceTopic) + .stream[Integer, Integer](sourceTopic) .split(Named.as("_")) .branch((_, v) => v % 2 == 0, Branched.withConsumer(ks => ks.to("even"), "consumedEvens")) .branch((_, v) => v % 3 == 0, Branched.withFunction(ks => ks.mapValues(x => x * x), "mapped")) @@ -80,18 +76,15 @@ class KStreamSplitTest extends TestDriver { m("_mapped").to("mapped") val testDriver = createTestDriver(builder) - val testInput = testDriver.createInput[Int, Int](sourceTopic) - testInput pipeKeyValueList List( - new KeyValue(1, 1), - new KeyValue(1, 2), - new KeyValue(1, 3), - new KeyValue(1, 4), - new KeyValue(1, 5), - new KeyValue(1, 9) - ).asJava - - val even = testDriver.createOutput[Int, Int]("even") - val mapped = testDriver.createOutput[Int, Int]("mapped") + val testInput = testDriver.createInput[Integer, Integer](sourceTopic) + testInput.pipeValueList( + List(1, 2, 3, 4, 5, 9) + .map(Integer.valueOf) + .asJava + ) + + val even = testDriver.createOutput[Integer, Integer]("even") + val mapped = testDriver.createOutput[Integer, Integer]("mapped") assertEquals(List(2, 4), even.readValuesToList().asScala) assertEquals(List(9, 81), mapped.readValuesToList().asScala) @@ -105,7 +98,7 @@ class KStreamSplitTest extends TestDriver { val sourceTopic = "source" val m = builder - .stream[Int, Int](sourceTopic) + .stream[Integer, Integer](sourceTopic) .split(Named.as("_")) .branch((_, v) => v % 2 == 0, Branched.withConsumer(ks => ks.to("even"))) .branch((_, v) => v % 3 == 0, Branched.withFunction(ks => ks.mapValues(x => x * x))) @@ -114,23 +107,19 @@ class KStreamSplitTest extends TestDriver { m("_2").to("mapped") val testDriver = createTestDriver(builder) - val testInput = testDriver.createInput[Int, Int](sourceTopic) - testInput pipeKeyValueList List( - new KeyValue(1, 1), - new KeyValue(1, 2), - new KeyValue(1, 3), - new KeyValue(1, 4), - new KeyValue(1, 5), - new KeyValue(1, 9) - ).asJava - - val even = testDriver.createOutput[Int, Int]("even") - val mapped = testDriver.createOutput[Int, Int]("mapped") + val testInput = testDriver.createInput[Integer, Integer](sourceTopic) + testInput.pipeValueList( + List(1, 2, 3, 4, 5, 9) + .map(Integer.valueOf) + .asJava + ) + + val even = testDriver.createOutput[Integer, Integer]("even") + val mapped = testDriver.createOutput[Integer, Integer]("mapped") assertEquals(List(2, 4), even.readValuesToList().asScala) assertEquals(List(9, 81), mapped.readValuesToList().asScala) testDriver.close() } - } diff --git a/streams/test-utils/src/test/java/org/apache/kafka/streams/TestTopicsTest.java b/streams/test-utils/src/test/java/org/apache/kafka/streams/TestTopicsTest.java index 84b8c5ebd6024..ac82c938d7d3d 100644 --- a/streams/test-utils/src/test/java/org/apache/kafka/streams/TestTopicsTest.java +++ b/streams/test-utils/src/test/java/org/apache/kafka/streams/TestTopicsTest.java @@ -41,7 +41,6 @@ import java.util.Map; import java.util.NoSuchElementException; import java.util.Properties; -import java.util.stream.Collectors; import static org.hamcrest.CoreMatchers.containsString; import static org.hamcrest.CoreMatchers.equalTo; @@ -86,12 +85,12 @@ public void tearDown() { @Test public void testValue() { - final TestInputTopic inputTopic = - testDriver.createInputTopic(INPUT_TOPIC, longSerde.serializer(), stringSerde.serializer()); + final TestInputTopic inputTopic = + testDriver.createInputTopic(INPUT_TOPIC, stringSerde.serializer(), stringSerde.serializer()); final TestOutputTopic outputTopic = testDriver.createOutputTopic(OUTPUT_TOPIC, stringSerde.deserializer(), stringSerde.deserializer()); - //Feed word "Hello" to inputTopic, timestamp and key irrelevant in this case - inputTopic.pipeInput(1L, "Hello"); + //Feed word "Hello" to inputTopic and no kafka key, timestamp is irrelevant in this case + inputTopic.pipeInput("Hello"); assertThat(outputTopic.readValue(), equalTo("Hello")); //No more output in topic assertThat(outputTopic.isEmpty(), is(true)); @@ -99,20 +98,16 @@ public void testValue() { @Test public void testValueList() { - final TestInputTopic inputTopic = - testDriver.createInputTopic(INPUT_TOPIC, longSerde.serializer(), stringSerde.serializer()); + final TestInputTopic inputTopic = + testDriver.createInputTopic(INPUT_TOPIC, stringSerde.serializer(), stringSerde.serializer()); final TestOutputTopic outputTopic = testDriver.createOutputTopic(OUTPUT_TOPIC, stringSerde.deserializer(), stringSerde.deserializer()); - final List> inputList = Arrays.asList( - new KeyValue<>(1L, "This"), - new KeyValue<>(2L, "is"), - new KeyValue<>(3L, "an"), - new KeyValue<>(4L, "example")); - //Feed list of words to inputTopic, key and timestamp are irrelevant in this case - inputTopic.pipeKeyValueList(inputList); + final List inputList = Arrays.asList("This", "is", "an", "example"); + //Feed list of words to inputTopic and no kafka key, timestamp is irrelevant in this case + inputTopic.pipeValueList(inputList); final List output = outputTopic.readValuesToList(); assertThat(output, hasItems("This", "is", "an", "example")); - assertThat(output, is(equalTo(inputList.stream().map(kv -> kv.value).collect(Collectors.toList())))); + assertThat(output, is(equalTo(inputList))); } @Test @@ -229,8 +224,8 @@ public void testTimestamp() { testDriver.createInputTopic(INPUT_TOPIC, longSerde.serializer(), stringSerde.serializer()); final TestOutputTopic outputTopic = testDriver.createOutputTopic(OUTPUT_TOPIC, longSerde.deserializer(), stringSerde.deserializer()); - inputTopic.pipeInput(1L, "Hello", baseTime); - assertThat(outputTopic.readRecord(), is(equalTo(new TestRecord<>(1L, "Hello", null, baseTime)))); + inputTopic.pipeInput(null, "Hello", baseTime); + assertThat(outputTopic.readRecord(), is(equalTo(new TestRecord<>(null, "Hello", null, baseTime)))); inputTopic.pipeInput(2L, "Kafka", ++baseTime); assertThat(outputTopic.readRecord(), is(equalTo(new TestRecord<>(2L, "Kafka", null, baseTime)))); @@ -238,15 +233,13 @@ public void testTimestamp() { inputTopic.pipeInput(2L, "Kafka", testBaseTime); assertThat(outputTopic.readRecord(), is(equalTo(new TestRecord<>(2L, "Kafka", testBaseTime)))); - final List> inputList = Arrays.asList( - new KeyValue<>(1L, "Advancing"), - new KeyValue<>(2L, "time")); + final List inputList = Arrays.asList("Advancing", "time"); //Feed list of words to inputTopic and no kafka key, timestamp advancing from testInstant final Duration advance = Duration.ofSeconds(15); final Instant recordInstant = testBaseTime.plus(Duration.ofDays(1)); - inputTopic.pipeKeyValueList(inputList, recordInstant, advance); - assertThat(outputTopic.readRecord(), is(equalTo(new TestRecord<>(1L, "Advancing", recordInstant)))); - assertThat(outputTopic.readRecord(), is(equalTo(new TestRecord<>(2L, "time", null, recordInstant.plus(advance))))); + inputTopic.pipeValueList(inputList, recordInstant, advance); + assertThat(outputTopic.readRecord(), is(equalTo(new TestRecord<>(null, "Advancing", recordInstant)))); + assertThat(outputTopic.readRecord(), is(equalTo(new TestRecord<>(null, "time", null, recordInstant.plus(advance))))); } @Test @@ -294,8 +287,8 @@ public void testTimestampAutoAdvance() { testDriver.createInputTopic(INPUT_TOPIC, longSerde.serializer(), stringSerde.serializer(), testBaseTime, advance); final TestOutputTopic outputTopic = testDriver.createOutputTopic(OUTPUT_TOPIC, longSerde.deserializer(), stringSerde.deserializer()); - inputTopic.pipeInput(1L, "Hello"); - assertThat(outputTopic.readRecord(), is(equalTo(new TestRecord<>(1L, "Hello", testBaseTime)))); + inputTopic.pipeInput("Hello"); + assertThat(outputTopic.readRecord(), is(equalTo(new TestRecord<>(null, "Hello", testBaseTime)))); inputTopic.pipeInput(2L, "Kafka"); assertThat(outputTopic.readRecord(), is(equalTo(new TestRecord<>(2L, "Kafka", testBaseTime.plus(advance))))); } @@ -339,12 +332,12 @@ public void testNonUsedOutputTopic() { @Test public void testEmptyTopic() { - final TestInputTopic inputTopic = - testDriver.createInputTopic(INPUT_TOPIC, longSerde.serializer(), stringSerde.serializer()); + final TestInputTopic inputTopic = + testDriver.createInputTopic(INPUT_TOPIC, stringSerde.serializer(), stringSerde.serializer()); final TestOutputTopic outputTopic = testDriver.createOutputTopic(OUTPUT_TOPIC, stringSerde.deserializer(), stringSerde.deserializer()); //Feed word "Hello" to inputTopic and no kafka key, timestamp is irrelevant in this case - inputTopic.pipeInput(1L, "Hello"); + inputTopic.pipeInput("Hello"); assertThat(outputTopic.readValue(), equalTo("Hello")); //No more output in topic assertThrows(NoSuchElementException.class, outputTopic::readRecord, "Empty topic");