You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@nifi.apache.org by GitBox <gi...@apache.org> on 2022/09/21 14:21:58 UTC
[GitHub] [nifi] markap14 commented on a diff in pull request #6131: NIFI-9822 - ConsumeKafkaRecord allows writing out Kafka record key
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org