Sophie Blee-Goldman created KAFKA-10395:
-------------------------------------------
Summary: TopologyTestDriver does not work with dynamic topic
routing
Key: KAFKA-10395
URL: https://issues.apache.org/jira/browse/KAFKA-10395
Project: Kafka
Issue Type: Bug
Components: streams
Reporter: Sophie Blee-Goldman
The TopologyTestDriver#read(topic) methods all call #getRecordsQueue which
checks
{code:java}
final Queue<ProducerRecord<byte[], byte[]>> outputRecords =
outputRecordsByTopic.get(topicName);
if (outputRecords == null) {
if (!processorTopology.sinkTopics().contains(topicName)) {
throw new IllegalArgumentException("Unknown topic: " + topicName);
}
}
{code}
The outputRecordsByTopic map keeps track of all topics that are actually
produced to, but obviously doesn't capture any topics that haven't yet received
output. The `processorTopology#sinkTopics` is supposed to account for that by
checking to make sure the topic is actually registered in the topology, and
throw an exception if not in case the user supplied the wrong topic name to
read from.
Unfortunately the TopicNameExtractor allows for dynamic routing of records to
any topic, so the topology isn't aware of all the possible output topics. If
trying to read from one of these topics that happens to not have received any
output yet, the test will throw the above misleading IllegalArgumentException.
We could just relax this check, but warning users who may actually have
accidentally passed in the wrong topic to read from seems quite useful. A
better solution would be to require registering all possible output topics to
the TTD up front. This would obviously require a KIP, but it would be a very
small one and shouldn't be too much trouble
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
