[BEAM-4798] IndexOutOfBoundsException when Flink parallelism > 1

• Type: Bug
• Status: Resolved
• Priority: P2
• Resolution: Fixed
• Affects Version/s: 2.4.0, 2.5.0
• Fix Version/s: 2.7.0
• Component/s: runner-flink
• Assignee: Alexey Romanenko
• Reporter: Alexey Romanenko
• Created: 16/Jul/18 15:54
• Updated: 24/Jul/20 20:11
• Resolved: 10/Aug/18 14:40

Description

Running job on Flink in streaming mode and get data from a Kafka topic with parallelism > 1 causes an exception:

Caused by: java.lang.IndexOutOfBoundsException: Index: 0, Size: 0
	at java.util.ArrayList.rangeCheck(ArrayList.java:657)
	at java.util.ArrayList.get(ArrayList.java:433)
	at org.apache.beam.runners.flink.translation.wrappers.streaming.io.UnboundedSourceWrapper.run(UnboundedSourceWrapper.java:277)
	at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:87)
	at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:56)
	at org.apache.flink.streaming.runtime.tasks.SourceStreamTask.run(SourceStreamTask.java:99)
	at org.apache.flink.streaming.runtime.tasks.StoppableSourceStreamTask.run(StoppableSourceStreamTask.java:45)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:306)
	at org.apache.flink.runtime.taskmanager.Task.run(Task.java:703)
	at java.lang.Thread.run(Thread.java:748)

It happens when number of Kafka topic partitions is less than value of parallelism (number of task slots).
So, workaround for now can be to set parallelism <= number of topic partitions, thus if parallelism=2 then number_partitions >= 2

Issue Links

• links to: GitHub Pull Request #5996
• links to: GitHub Pull Request #6177