[SPARK-24404] Increase currentEpoch when meet a EpochMarker in ContinuousQueuedDataReader.next() in CP mode based on PR #21353 #21332 #21293 and the latest master

• Type: Bug
• Status: Closed
• Priority: Major
• Resolution: Won't Fix
• Affects Version/s: 2.3.0
• Component/s: Structured Streaming
• Assignee: Unassigned
• Reporter: Liangchang Zhu
• Created: 28/May/18 15:35
• Updated: 01/Feb/19 06:04
• Resolved: 01/Feb/19 06:03

Description

In CP mode, based on PR #21353 #21332 #21293 and the latest master ContinuousQueuedDataReader.next() will be invoked by ContinuousDataSourceRDD.compute to return UnsafeRow. When currentEntry polled from ArrayBlockingQueue is a EpochMarker, ContinuousQueuedDataReader will send `ReportPartitionOffset` message to epochCoordinator with currentEpoch of EpochTracker. The currentEpoch is a ThreadLocal variable, but now no place invoke `incrementCurrentEpoch` to increase currentEpoch in its thread, so `getCurrentEpoch` will return `None` all the time(because currentEpoch is -1). This will cause exception when invoke `None.get`. At the same time, in order to make the `ReportPartitionOffset` have correct semantics, we need increase currentEpoch before send this message

Issue Links

• links to: [Github] Pull Request #21445 (LiangchangZ)
• links to: GitHub Pull Request #21445