[jira] [Updated] (STORM-96) _msgIdToBatchId in RichSpoutBatchTriggerer.java did not give the right value

["Rick Kellogg (JIRA)" <ji...@apache.org>]

     [ https://issues.apache.org/jira/browse/STORM-96?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]

Rick Kellogg updated STORM-96:
------------------------------
    Component/s: storm-core

> _msgIdToBatchId in RichSpoutBatchTriggerer.java did not give the right value
> ----------------------------------------------------------------------------
>
>                 Key: STORM-96
>                 URL: https://issues.apache.org/jira/browse/STORM-96
>             Project: Apache Storm
>          Issue Type: Bug
>          Components: storm-core
>            Reporter: James Xu
>
> https://github.com/nathanmarz/storm/issues/734
> _msgIdToBatchId did not give the right value, and the fail/ack in the underline Spout will be never called, and the _finishConditions will never released.
> The fix will be like follows in RichSpoutBatchTriggerer.java
>     public List<Integer> emit(String ignore, List<Object> values, Object msgId) {
>         long batchIdVal = _rand.nextLong();
>         Object batchId = new RichSpoutBatchId(batchIdVal);
>         FinishCondition finish = new FinishCondition();
>         finish.msgId = msgId;
>         List<Integer> tasks = _collector.emit(_stream, new ConsList(batchId, values));
>         Set<Integer> outTasksSet = new HashSet<Integer>(tasks);
>         for(Integer t: _outputTasks) {
>             int count = 0;
>             if(outTasksSet.contains(t)) {
>                 count = 1;
>             }
>             long r = _rand.nextLong();
>             _collector.emitDirect(t, _coordStream, new Values(batchId, count), r);
>             finish.vals.add(r);
>             //Ming Li: Adding the follow line to init the _msgIdToBatchId with correct value
>             _msgIdToBatchId.put(r, batchIdVal);
>         }
>         _finishConditions.put(batchIdVal, finish);
>         return tasks;
>     }



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)