awelless commented on code in PR #10904:
URL: https://github.com/apache/nifi/pull/10904#discussion_r2812504325
##########
nifi-extension-bundles/nifi-aws-bundle/nifi-aws-kinesis/src/main/java/org/apache/nifi/processors/aws/kinesis/ConsumeKinesis.java:
##########
@@ -680,11 +706,10 @@ private void processRecordsFromBuffer(final
ProcessSession session, final Lease
}
final String shardId = lease.shardId();
- final ReaderRecordProcessor processor = readerRecordProcessor;
- if (processor != null) {
- processRecordsWithReader(processor, session, shardId, records);
- } else {
- processRecordsAsRaw(session, shardId, records);
+ switch (processingStrategy) {
+ case FLOW_FILE -> processRecordsAsRaw(session, shardId,
records);
+ case RECORD -> processRecordsWithReader(session, shardId,
records);
+ case DEMARCATOR -> processRecordsAsDemarcated(session,
shardId, records);
}
Review Comment:
There are a few switches based on the processing strategy in the flow.
An alternative would be to introduce an functional interface with `(session,
shardId, record) -> void` interface. And then configure 3 implementations, one
per processing strategy. Each implementation will transfer flow files to the
relationship.
Therefore, it will be guaranteed all objects are properly initialized for
each processing strategy.
I'm hesitant to introduce such change in this PR. But can do this as a
follow up, if it makes sense
--
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]
