[STORM-3047] Ensure Trident emitter refreshPartitions is only called with partitions assigned to the emitter

• Type: Bug
• Status: Resolved
• Priority: Major
• Resolution: Fixed
• Affects Version/s: 1.1.2, 1.2.1
• Fix Version/s: 1.2.3, 1.1.4
• Component/s: storm-core
• Labels: pull-request-available
• Assignee: Stig Rohde Døssing
• Reporter: Stig Rohde Døssing
• Created: 28/Apr/18 12:30
• Updated: 10/Nov/18 14:47
• Resolved: 09/Jul/18 08:07

Description

This is a backport of the changes made to OpaquePartitionedTridentSpoutExecutor in https://github.com/apache/storm/pull/2300/files.

The description of the issue is copied here for convenience:

The changes in https://github.com/apache/storm/pull/2009 released in 1.1.0 made some changes to the OpaquePartitionedTridentSpoutExecutor that likely broke IOpaquePartitionedTridentSpout implementations other than storm-kafka-client. The changed code used to request sorted partitions from the spout via getOrderedPartitions, do a round-robin partitioning, and assign partitions via refreshPartitions https://github.com/apache/storm/blob/v1.0.4/storm-core/src/jvm/org/apache/storm/trident/spout/OpaquePartitionedTridentSpoutExecutor.java#L100. The new code just passes the output of getOrderedPartitions into refreshPartitions https://github.com/apache/storm/blob/v1.1.0/storm-core/src/jvm/org/apache/storm/trident/spout/OpaquePartitionedTridentSpoutExecutor.java#L120. It looks to me like refreshPartitions is passed the list of all partitions assigned to any spout task, rather than just the partitions assigned to the current task.

The proposed fix will use getOrderedPartitions to get the sorted partitions list, pass the list into getPartitionsForTask, and pass the resulting list of assigned partitions back into refreshPartitions.

Issue Links

• links to: GitHub Pull Request #2650
• links to: GitHub Pull Request #2653