Ups. That is definitely a bug in the test. Thanks for pointing it out! Do you wanna open a PR it? If not, I can take care, too.
-Matthias On 7/12/17 8:48 AM, Joel Dice . wrote: > After some debugging, I figured it out. The name of my custom store > was "mapStore", so the store tried to log changes to a topic with that > name, but my test case never created such a topic. Unsurprisingly, > the producer couldn't get metadata for a nonexistent topic, so it > failed with a timeout. > > With this trivial fix, the test passes: > https://github.com/dicej/kafka-eos-test/commit/0fdf400350e41092f1973a80e9e4196003859ddc > > BTW, in EosIntegrationTest, shouldn't this line read "outputTopic," > instead of "inputTopic,": > > https://github.com/apache/kafka/blob/44167399157b06d566401a3f6c17e9ca901a8c20/streams/src/test/java/org/apache/kafka/streams/integration/EosIntegrationTest.java#L195 > > As written, the test seems to only verify that data produced to > inputTopic can be consumed from inputTopic, whereas the intention of > the test seems to be that data is copied from inputTopic to > outputTopic, so I would expect it to verify that it actually made it > to outputTopic. > > > 2017-07-11 16:25 GMT-06:00 Matthias J. Sax <matth...@confluent.io>: >> Seems Streams cannot connect (or looses connection) to the brokers. Not >> sure why. >> >> You can also have a look here for our own EOS Streams integration test: >> https://github.com/apache/kafka/blob/trunk/streams/src/test/java/org/apache/kafka/streams/integration/EosIntegrationTest.java >> >> >> -Matthias >> >> >> On 7/11/17 2:13 PM, Joel Dice . wrote: >>> Hi all, >>> >>> I wrote an integration test to see how Kafka Streams handles exactly >>> once processing with a custom state store, and it's failing with this >>> error: >>> >>> org.apache.kafka.common.errors.TimeoutException: Failed to update >>> metadata after 60000 ms >>> >>> I'm not sure whether this is a bug in my code, a bug in Kafka, or a >>> bug in the EmbeddedKafkaCluster class I'm using for testing. >>> >>> The test is available here: https://github.com/dicej/kafka-eos-test, >>> and it can be run using "gradle test". >>> >>> The test class itself is >>> https://github.com/dicej/kafka-eos-test/blob/f892a856d25d6e17e16a6681066a2033b06c5b3f/src/test/java/KafkaIntegrationTest.java. >>> It's based loosely on the EosIntegrationTest class in the Kafka source >>> tree. >>> >>> This is the full trace: >>> >>> Exception in thread >>> "appId-1-9f749063-1a4d-4498-8df4-8619269e36f3-StreamThread-1" >>> org.apache.kafka.streams.errors.StreamsException: Exception caught in >>> process. taskId=0_0, processor=inputTopic, topic=inputTopic, >>> partition=0, offset=1 >>> >>> at >>> org.apache.kafka.streams.processor.internals.StreamTask.process(StreamTask.java:203) >>> >>> at >>> org.apache.kafka.streams.processor.internals.StreamThread.processAndPunctuate(StreamThread.java:679) >>> >>> at >>> org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:557) >>> >>> at >>> org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:527) >>> >>> Caused by: org.apache.kafka.streams.errors.StreamsException: task >>> [0_0] exception caught when producing >>> >>> at >>> org.apache.kafka.streams.processor.internals.RecordCollectorImpl.checkForException(RecordCollectorImpl.java:136) >>> >>> at >>> org.apache.kafka.streams.processor.internals.RecordCollectorImpl.send(RecordCollectorImpl.java:87) >>> >>> at KafkaIntegrationTest$MapStore.put(KafkaIntegrationTest.java:259) >>> >>> at KafkaIntegrationTest$MapStore.put(KafkaIntegrationTest.java:236) >>> >>> at >>> KafkaIntegrationTest$StatisticMachine.process(KafkaIntegrationTest.java:227) >>> >>> at >>> KafkaIntegrationTest$StatisticMachine.process(KafkaIntegrationTest.java:203) >>> >>> at >>> org.apache.kafka.streams.processor.internals.ProcessorNode$1.run(ProcessorNode.java:47) >>> >>> at >>> org.apache.kafka.streams.processor.internals.StreamsMetricsImpl.measureLatencyNs(StreamsMetricsImpl.java:187) >>> >>> at >>> org.apache.kafka.streams.processor.internals.ProcessorNode.process(ProcessorNode.java:133) >>> >>> at >>> org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:82) >>> >>> at >>> org.apache.kafka.streams.processor.internals.SourceNode.process(SourceNode.java:80) >>> >>> at >>> org.apache.kafka.streams.processor.internals.StreamTask.process(StreamTask.java:189) >>> >>> ... 3 more >>> >>> Caused by: org.apache.kafka.common.errors.TimeoutException: Failed >>> to update metadata after 60000 ms. >>> >>> >>> Any ideas what might be going wrong? >>> >>> Cheers, >>> Joel >>> >>
signature.asc
Description: OpenPGP digital signature
