[FLINK-9349] KafkaConnector Exception while fetching from multiple kafka topics

• Type: Bug
• Status: Resolved
• Priority: Critical
• Resolution: Fixed
• Affects Version/s: 1.4.0
• Fix Version/s: 1.4.3, 1.5.0
• Component/s: Connectors / Kafka
• Assignee: Sergey Nuyanzin
• Reporter: Vishal Santoshi
• Created: 13/May/18 14:49
• Updated: 11/Jul/19 07:45
• Resolved: 23/May/18 07:15

Description

./flink-connectors/flink-connector-kafka-0.9/src/main/java/org/apache/flink/streaming/connectors/kafka/internal/Kafka09Fetcher.java
 
It seems the List subscribedPartitionStates was being modified when runFetchLoop iterated the List.
This can happen if, e.g., FlinkKafkaConsumer runs the following code concurrently:
                kafkaFetcher.addDiscoveredPartitions(discoveredPartitions);
 

 java.util.ConcurrentModificationException
	at java.util.LinkedList$ListItr.checkForComodification(LinkedList.java:966)
	at java.util.LinkedList$ListItr.next(LinkedList.java:888)
	at org.apache.flink.streaming.connectors.kafka.internal.Kafka09Fetcher.runFetchLoop(Kafka09Fetcher.java:134)

Issue Links

• duplicates: FLINK-13204 The subscribedPartitionStates should be thread safe.
• links to: GitHub Pull Request #6040