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

["James Xu (JIRA)" <ji...@apache.org>]

James Xu created STORM-96:
-----------------------------

             Summary: _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 (Incubating)
          Issue Type: Bug
            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.1.4#6159)