dariuszseweryn commented on code in PR #10053:
URL: https://github.com/apache/nifi/pull/10053#discussion_r2206934994
##########
nifi-extension-bundles/nifi-aws-bundle/nifi-aws-processors/src/main/java/org/apache/nifi/processors/aws/kinesis/stream/record/KinesisRecordProcessorRecord.java:
##########
@@ -75,51 +77,79 @@ public KinesisRecordProcessorRecord(final
ProcessSessionFactory sessionFactory,
@Override
void startProcessingRecords() {
super.startProcessingRecords();
- outputStream = null;
- writer = null;
+ if (currentFlowFileState != null) {
+ // this may happen if the previous processing has not been
completed successfully, close the leftover state
+ closeSafe(currentFlowFileState, "FlowFile State");
+ currentFlowFileState = null;
+ }
+ }
+
+ @Override
+ void finishProcessingRecords(final ProcessSession session, final
List<FlowFile> flowFiles, final StopWatch stopWatch) {
+ super.finishProcessingRecords(session, flowFiles, stopWatch);
+ try {
+ if (currentFlowFileState == null) {
+ return;
+ }
+ if (!flowFiles.contains(currentFlowFileState.flowFile)) {
+ // this is unexpected, flowFiles have been altered not in this
class after the start of processing
+ throw new IllegalStateException("%s is not available in
provided FlowFiles [%d]".formatted(currentFlowFileState.flowFile,
flowFiles.size()));
+ }
+ completeFlowFile(flowFiles, session, stopWatch);
+ } catch (final FlowFileCompletionException e) {
+ if
(!currentFlowFileState.containsDataFromExactlyOneKinesisRecord()) {
+ throw new KinesisBatchUnrecoverableException(e.getMessage(),
e);
+ }
+ final boolean removeCurrentStateFlowFileIfAvailable = true;
Review Comment:
I'll refactor the code so it won't need this parameter
--
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]
