Github user joewitt commented on a diff in the pull request:
https://github.com/apache/nifi/pull/1695#discussion_r114151491
--- Diff:
nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-10-processors/src/main/java/org/apache/nifi/processors/kafka/pubsub/ConsumerLease.java
---
@@ -392,13 +394,27 @@ private void writeDemarcatedData(final ProcessSession
session, final List<Consum
bundleMap.put(topicPartition, tracker);
}
+ private void rollback(final TopicPartition topicPartition) {
+ final OffsetAndMetadata offsetAndMetadata =
kafkaConsumer.committed(topicPartition);
--- End diff --
instead of asking the kafka client where we were as far as committed
offsets we should first check the uncommittedOffsets map and if nothing then
ask the kafkaclient otherwise use the last uncommittedOffset for this
topic/partition pair.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [email protected] or file a JIRA ticket
with INFRA.
---
