[FLINK-2735] KafkaProducerITCase.testCustomPartitioning sporadically fails

• Type: Bug
• Status: Resolved
• Priority: Major
• Resolution: Cannot Reproduce
• Affects Version/s: 0.10.0
• Component/s: Connectors / Kafka
• Labels: test-stability
• Assignee: Unassigned
• Reporter: Robert Metzger
• Created: 22/Sep/15 15:31
• Updated: 24/Aug/16 09:12
• Resolved: 24/Aug/16 09:12

Description

In the following test run: https://s3.amazonaws.com/archive.travis-ci.org/jobs/81584444/log.txt
there was the following failure

Caused by: java.lang.Exception: Unable to get last offset for topic customPartitioningTestTopic and partitions [FetchPartition {partition=2, offset=-915623761776}]. 
Exception for partition 2: kafka.common.UnknownException
	at sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)
	at sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:62)
	at sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
	at java.lang.reflect.Constructor.newInstance(Constructor.java:408)
	at java.lang.Class.newInstance(Class.java:438)
	at kafka.common.ErrorMapping$.exceptionFor(ErrorMapping.scala:86)
	at kafka.common.ErrorMapping.exceptionFor(ErrorMapping.scala)
	at org.apache.flink.streaming.connectors.kafka.internals.LegacyFetcher$SimpleConsumerThread.getLastOffset(LegacyFetcher.java:521)
	at org.apache.flink.streaming.connectors.kafka.internals.LegacyFetcher$SimpleConsumerThread.run(LegacyFetcher.java:370)

	at org.apache.flink.streaming.connectors.kafka.internals.LegacyFetcher.run(LegacyFetcher.java:242)
	at org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer.run(FlinkKafkaConsumer.java:382)
	at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:58)
	at org.apache.flink.streaming.runtime.tasks.SourceStreamTask.run(SourceStreamTask.java:58)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:168)
	at org.apache.flink.runtime.taskmanager.Task.run(Task.java:579)
	at java.lang.Thread.run(Thread.java:745)
Caused by: java.lang.RuntimeException: Unable to get last offset for topic customPartitioningTestTopic and partitions [FetchPartition {partition=2, offset=-915623761776}]. 
Exception for partition 2: kafka.common.UnknownException
	at sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)
	at sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:62)
	at sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
	at java.lang.reflect.Constructor.newInstance(Constructor.java:408)
	at java.lang.Class.newInstance(Class.java:438)
	at kafka.common.ErrorMapping$.exceptionFor(ErrorMapping.scala:86)
	at kafka.common.ErrorMapping.exceptionFor(ErrorMapping.scala)
	at org.apache.flink.streaming.connectors.kafka.internals.LegacyFetcher$SimpleConsumerThread.getLastOffset(LegacyFetcher.java:521)
	at org.apache.flink.streaming.connectors.kafka.internals.LegacyFetcher$SimpleConsumerThread.run(LegacyFetcher.java:370)

	at org.apache.flink.streaming.connectors.kafka.internals.LegacyFetcher$SimpleConsumerThread.getLastOffset(LegacyFetcher.java:524)
	at org.apache.flink.streaming.connectors.kafka.internals.LegacyFetcher$SimpleConsumerThread.run(LegacyFetcher.java:370)
Tests run: 1, Failures: 1, Errors: 0, Skipped: 0, Time elapsed: 17.455 sec <<< FAILURE! - in org.apache.flink.streaming.connectors.kafka.KafkaProducerITCase
testCustomPartitioning(org.apache.flink.streaming.connectors.kafka.KafkaProducerITCase)  Time elapsed: 7.809 sec  <<< FAILURE!
java.lang.AssertionError: Test failed: The program execution failed: Job execution failed.
	at org.junit.Assert.fail(Assert.java:88)
	at org.apache.flink.streaming.connectors.kafka.KafkaTestBase.tryExecute(KafkaTestBase.java:313)
	at org.apache.flink.streaming.connectors.kafka.KafkaProducerITCase.testCustomPartitioning(KafkaProducerITCase.java:155)

From the broker logs it seems to be an issue in the Kafka broker

14:43:03,328 INFO  kafka.network.Processor                                       - Closing socket connection to /127.0.0.1.
14:43:03,334 WARN  kafka.server.KafkaApis                                        - [KafkaApi-0] Error while responding to offset request
java.lang.ArrayIndexOutOfBoundsException: 1
	at kafka.server.KafkaApis.fetchOffsetsBefore(KafkaApis.scala:469)
	at kafka.server.KafkaApis.fetchOffsets(KafkaApis.scala:449)
	at kafka.server.KafkaApis$$anonfun$17.apply(KafkaApis.scala:411)
	at kafka.server.KafkaApis$$anonfun$17.apply(KafkaApis.scala:402)
	at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:244)
	at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:244)
	at scala.collection.immutable.Map$Map1.foreach(Map.scala:109)
	at scala.collection.TraversableLike$class.map(TraversableLike.scala:244)
	at scala.collection.AbstractTraversable.map(Traversable.scala:105)
	at kafka.server.KafkaApis.handleOffsetRequest(KafkaApis.scala:402)
	at kafka.server.KafkaApis.handle(KafkaApis.scala:61)
	at kafka.server.KafkaRequestHandler.run(KafkaRequestHandler.scala:59)
	at java.lang.Thread.run(Thread.java:745)

Issue Links

• is blocked by: KAFKA-2236 offset request reply racing with segment rolling