Github user joewitt commented on a diff in the pull request:
https://github.com/apache/nifi/pull/1695#discussion_r114150192
--- 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);
+ final long offset = offsetAndMetadata.offset();
+ kafkaConsumer.seek(topicPartition, offset);
+ }
+
private void writeRecordData(final ProcessSession session, final
List<ConsumerRecord<byte[], byte[]>> records, final TopicPartition
topicPartition) {
FlowFile flowFile = session.create();
try {
final RecordSetWriter writer;
try {
writer = writerFactory.createWriter(logger, flowFile, new
ByteArrayInputStream(new byte[0]));
} catch (final Exception e) {
+ logger.error(
+ "Failed to obtain a Record Writer for serializing
Kafka messages. This generally happens because the "
+ + "Record Writer cannot obtain the appropriate
Schema, due to failure to connect to a remote Schema Registry "
+ + "or due to the Schema Access Strategy being
dependent upon FlowFile Attributes that are not available. "
+ + "Will roll back the Kafka session.", e);
+
+ rollback(topicPartition);
--- End diff --
i think we probably need to have error handling on this as well in case we
cannot talk back to kafka during this.
---
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.
---
