[jira] [Created] (FLINK-2008) PersistentKafkaSource is sometimes emitting tuples multiple times

["Robert Metzger (JIRA)" <ji...@apache.org>]

Robert Metzger created FLINK-2008:
-------------------------------------

             Summary: PersistentKafkaSource is sometimes emitting tuples multiple times
                 Key: FLINK-2008
                 URL: https://issues.apache.org/jira/browse/FLINK-2008
             Project: Flink
          Issue Type: Bug
          Components: Kafka Connector, Streaming
    Affects Versions: 0.9
            Reporter: Robert Metzger
            Assignee: Robert Metzger


The PersistentKafkaSource is expected to emit records exactly once.

Two test cases of the KafkaITCase are sporadically failing because records are emitted multiple times.
Affected tests:
{{testPersistentSourceWithOffsetUpdates()}}, after the offsets have been changed manually in ZK:
{code}
java.lang.RuntimeException: Expected v to be 3, but was 4 on element 0 array=[4, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 2]
{code}

{{brokerFailureTest()}} also fails:
{code}
05/13/2015 08:13:16	Custom source -> Stream Sink(1/1) switched to FAILED 
java.lang.AssertionError: Received tuple with value 21 twice
	at org.junit.Assert.fail(Assert.java:88)
	at org.junit.Assert.assertTrue(Assert.java:41)
	at org.junit.Assert.assertFalse(Assert.java:64)
	at org.apache.flink.streaming.connectors.kafka.KafkaITCase$15.invoke(KafkaITCase.java:877)
	at org.apache.flink.streaming.connectors.kafka.KafkaITCase$15.invoke(KafkaITCase.java:859)
	at org.apache.flink.streaming.api.operators.StreamSink.callUserFunction(StreamSink.java:39)
	at org.apache.flink.streaming.api.operators.StreamOperator.callUserFunctionAndLogException(StreamOperator.java:137)
	at org.apache.flink.streaming.api.operators.ChainableStreamOperator.collect(ChainableStreamOperator.java:54)
	at org.apache.flink.streaming.api.collector.CollectorWrapper.collect(CollectorWrapper.java:39)
	at org.apache.flink.streaming.connectors.kafka.api.persistent.PersistentKafkaSource.run(PersistentKafkaSource.java:173)
	at org.apache.flink.streaming.api.operators.StreamSource.callUserFunction(StreamSource.java:40)
	at org.apache.flink.streaming.api.operators.StreamOperator.callUserFunctionAndLogException(StreamOperator.java:137)
	at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:34)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:139)
	at org.apache.flink.runtime.taskmanager.Task.run(Task.java:562)
	at java.lang.Thread.run(Thread.java:745)
{code}



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)