mattyb149 commented on code in PR #6907:
URL: https://github.com/apache/nifi/pull/6907#discussion_r1106123621
##########
nifi-nar-bundles/nifi-cdc/nifi-cdc-mysql-bundle/nifi-cdc-mysql-processors/src/main/java/org/apache/nifi/cdc/mysql/processors/CaptureChangeMySQL.java:
##########
@@ -967,13 +1026,19 @@ public void outputEvents(ProcessSession session,
ComponentLog log) throws IOExce
CommitTransactionEventInfo commitTransactionEvent
= useGtid
? new
CommitTransactionEventInfo(currentDatabase, timestamp, currentGtidSet)
: new
CommitTransactionEventInfo(currentDatabase, timestamp, currentBinlogFile,
currentBinlogPosition);
-
currentSequenceId.set(commitEventWriter.writeEvent(currentSession, transitUri,
commitTransactionEvent, currentSequenceId.get(), REL_SUCCESS));
+ currentEventInfo = commitTransactionEvent;
+ currentEventWriter = commitEventWriter;
+
currentSequenceId.set(commitEventWriter.writeEvent(currentSession, transitUri,
commitTransactionEvent, currentSequenceId.get(), REL_SUCCESS,
eventWriterConfiguration));
}
+
//update inTransaction value to state
inTransaction = false;
updateState(session);
- // Commit the NiFi session
- session.commitAsync();
+ // If there is no FlowFile open, commit the session
+ if (eventWriterConfiguration.getCurrentFlowFile() ==
null) {
+ // Commit the NiFi session
+ session.commitAsync();
Review Comment:
In that case we should call the property `Max Events Per FlowFile` and
document that when the event queue is empty (about to finish a call to
onTrigger), we will finish any open FlowFile and commit the session. @tpalfy
@exceptionfactory @mark-bathori thoughts?
--
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]
