STORM-2296 Kafka spout no dup on leader changes, 1.x-branch #1888
Conversation
ernisv
changed the title
| waitForRefresh(); | ||
| when(reader.getBrokerInfo()).thenReturn(TestUtils.buildPartitionInfoList(TestUtils.buildPartitionInfo(totalTasks, 9093))); | ||
| List<List<PartitionManager>> partitionManagersAfterRefresh = getPartitionManagers(coordinatorList); | ||
| assertEquals(partitionManagersAfterRefresh.size(), partitionManagersAfterRefresh.size()); |
There was a problem hiding this comment.
It compares exactly same thing. Did you miss here, or it's redundant line?
There was a problem hiding this comment.
Right, there's the same error in another test that got copy-pasted here :)
Fixed in both places.
| Iterator<List<PartitionManager>> iterator = partitionManagersAfterRefresh.iterator(); | ||
| for (List<PartitionManager> partitionManagersBefore : partitionManagersBeforeRefresh) { | ||
| List<PartitionManager> partitionManagersAfter = iterator.next(); | ||
| for (PartitionManager before : partitionManagersBefore) |
There was a problem hiding this comment.
I'd rather create a new hash map for before or after which key is partition, and compare based on the map and other list. The depth of if statement is already 4.
| private void assertStateIsTheSame(PartitionManager managerBefore, PartitionManager managerAfter) { | ||
| // check if state was actually moved from old PartitionManager | ||
| assertNotSame(managerBefore, managerAfter); | ||
| assertSame(managerBefore._waitingToEmit, managerAfter._waitingToEmit); |
There was a problem hiding this comment.
Let's check all the copying fields, not only _waitingToEmit.
There was a problem hiding this comment.
I've added checks for _emittedToOffset and _committedTo fields.
However _failedMsgRetryManager and _pending are private, so we either do not check them or increase the fields visibility to "package". Do you think it warrants visibility change ?
There was a problem hiding this comment.
No I think that's enough.
|
@ernisv |
|
+1 |
|
@ernisv I can handle it if you mind, but then you need to close this PR manually after merging, since Github auto-close doesn't work against non-master branch. |
ernisv
force-pushed
the
kafka_spout_no_dup_on_leader_changes_1_x
branch
from
2bdcbdd
to
b076399
Compare
|
Ok, I've squashed all the changes to single commit. |
Current behavior of Kafka spout emits duplicate tuples whenever Kafka topic leader's change.
In case of exception caused by leader changes, PartitionManagers are simply recreated losing the state about which tuples were already emitted and new PartitionManager re-emits them again.
This is fine as at-least-once is fulfilled, but still it would be better to not emit duplicate data if possible.
Moreover this could be easily avoided by moving the state related to emitted tuples from old PartitionManager to new one.