markap14 commented on code in PR #6131:
URL: https://github.com/apache/nifi/pull/6131#discussion_r976575896
##########
nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-2-6-processors/src/main/java/org/apache/nifi/processors/kafka/pubsub/ConsumerLease.java:
##########
@@ -591,51 +607,18 @@ private void writeRecordData(final ProcessSession
session, final List<ConsumerRe
}
try {
+ int recordCount = 0;
Record record;
while ((record = reader.nextRecord()) != null) {
- // Determine the bundle for this record.
- final RecordSchema recordSchema =
record.getSchema();
- final BundleInformation bundleInfo = new
BundleInformation(topicPartition, recordSchema, attributes, separateByKey ?
consumerRecord.key() : null);
-
- BundleTracker tracker = bundleMap.get(bundleInfo);
- if (tracker == null) {
- FlowFile flowFile = session.create();
- flowFile = session.putAllAttributes(flowFile,
attributes);
-
- final OutputStream rawOut =
session.write(flowFile);
-
- final RecordSchema writeSchema;
- try {
- writeSchema =
writerFactory.getSchema(flowFile.getAttributes(), recordSchema);
- } catch (final Exception e) {
- logger.error("Failed to obtain Schema for
FlowFile. Will roll back the Kafka message offsets.", e);
-
- rollback(topicPartition);
- yield();
-
- throw new ProcessException(e);
- }
-
- writer = writerFactory.createWriter(logger,
writeSchema, rawOut, flowFile);
- writer.beginRecordSet();
-
- tracker = new BundleTracker(consumerRecord,
topicPartition, keyEncoding, writer);
- tracker.updateFlowFile(flowFile);
- bundleMap.put(bundleInfo, tracker);
- } else {
- writer = tracker.recordWriter;
- }
-
- try {
- writer.write(record);
- } catch (final RuntimeException re) {
- handleParseFailure(consumerRecord, session,
re, "Failed to write message from Kafka using the configured Record Writer. "
- + "Will route message as its own FlowFile
to the 'parse.failure' relationship");
- continue;
+ if
((OutputStrategy.USE_WRAPPER.equals(outputStrategy)) && (consumerRecord.key()
!= null)) {
Review Comment:
We don't need the `&& (consumerRecord.key() != null)` check here. We should
be using the wrapper even if the key is null.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
