Github user srdo commented on a diff in the pull request: https://github.com/apache/storm/pull/2465#discussion_r157375639 --- Diff: external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/internal/OffsetManager.java --- @@ -140,10 +140,16 @@ public OffsetAndMetadata findNextCommitOffset() { OffsetAndMetadata nextCommitOffsetAndMetadata = null; if (found) { - nextCommitOffsetAndMetadata = new OffsetAndMetadata(nextCommitOffset, - nextCommitMsg.getMetadata(Thread.currentThread())); + try { + nextCommitOffsetAndMetadata = new OffsetAndMetadata(nextCommitOffset, + JSON_MAPPER.writeValueAsString(new KafkaSpout.Info(Thread.currentThread(), context))); --- End diff -- I think maybe I was being a little unclear. I'm saying that right now the information contained in the Info object is constant for the lifetime of the spout. We should be able to just create one Info object in KafkaSpout.open and reuse it. This means a single serialization for the lifetime of the spout, which is more efficient than doing it on every commit.
---