[
https://issues.apache.org/jira/browse/APEXMALHAR-2156?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15402453#comment-15402453
]
ASF GitHub Bot commented on APEXMALHAR-2156:
--------------------------------------------
Github user sanjaypujare commented on a diff in the pull request:
https://github.com/apache/apex-malhar/pull/347#discussion_r73017194
--- Diff:
contrib/src/main/java/com/datatorrent/contrib/kafka/AbstractKafkaInputOperator.java
---
@@ -250,12 +250,7 @@ public void setup(OperatorContext context)
consumer.create();
// reset the offsets to checkpointed one
if (consumer instanceof SimpleKafkaConsumer && !offsetStats.isEmpty())
{
- Map<KafkaPartition, Long> currentOffsets = new HashMap<>();
- // Increment the offsets and set it to consumer
- for (Map.Entry<KafkaPartition, Long> e: offsetStats.entrySet()) {
- currentOffsets.put(e.getKey(), e.getValue() + 1);
- }
- ((SimpleKafkaConsumer)consumer).resetOffset(currentOffsets);
+ ((SimpleKafkaConsumer)consumer).resetOffset(offsetStats);
--- End diff --
These are not mine - I ended up merging changes from upstream/master
(because of conflicts) which show up in the pull request for this branch.
> JMS Input operator enhancements
> -------------------------------
>
> Key: APEXMALHAR-2156
> URL: https://issues.apache.org/jira/browse/APEXMALHAR-2156
> Project: Apache Apex Malhar
> Issue Type: Improvement
> Reporter: Sanjay M Pujare
> Assignee: Sanjay M Pujare
> Original Estimate: 672h
> Remaining Estimate: 672h
>
> This is described in
> https://drive.google.com/open?id=1gapzjxTiePP0VwaOrc-FWfw4xNK6bcIuxkNkGKoZAiA
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
