You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@nifi.apache.org by "mattyb149 (via GitHub)" <gi...@apache.org> on 2023/01/30 22:03:24 UTC
[GitHub] [nifi] mattyb149 opened a new pull request, #6907: NIFI-11094: Allow CaptureChangeMySQL to send multiple events per FlowFile
mattyb149 opened a new pull request, #6907:
URL: https://github.com/apache/nifi/pull/6907
# Summary
[NIFI-11094](https://issues.apache.org/jira/browse/NIFI-11094) This PR refactors CaptureChangeMySQL to support two modes of operation, N events per FlowFile and one transaction per FlowFile. The default is N events per FlowFile with N set to 1 to retain existing behavior (1 event per FlowFile). However now this number is configurable, or the user can choose to output a FlowFile with all events between BEGIN and COMMIT events. The latter can help with downstream processing, as PutDatabaseRecord treats a FlowFile as a transaction so failure handling can be more consistent.
# Tracking
Please complete the following tracking steps prior to pull request creation.
### Issue Tracking
- [x] [Apache NiFi Jira](https://issues.apache.org/jira/browse/NIFI) issue created
### Pull Request Tracking
- [x] Pull Request title starts with Apache NiFi Jira issue number, such as `NIFI-00000`
- [x] Pull Request commit message starts with Apache NiFi Jira issue number, as such `NIFI-00000`
### Pull Request Formatting
- [x] Pull Request based on current revision of the `main` branch
- [x] Pull Request refers to a feature branch with one commit containing changes
# Verification
Please indicate the verification steps performed prior to pull request creation.
### Build
- [ ] Build completed using `mvn clean install -P contrib-check`
- [ ] JDK 8
- [x] JDK 11
- [ ] JDK 17
### Licensing
- [ ] New dependencies are compatible with the [Apache License 2.0](https://apache.org/licenses/LICENSE-2.0) according to the [License Policy](https://www.apache.org/legal/resolved.html)
- [ ] New dependencies are documented in applicable `LICENSE` and `NOTICE` files
### Documentation
- [x] Documentation formatting appears as expected in rendered files
--
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
[GitHub] [nifi] asfgit closed pull request #6907: NIFI-11094: Allow CaptureChangeMySQL to send multiple events per FlowFile
Posted by "asfgit (via GitHub)" <gi...@apache.org>.
asfgit closed pull request #6907: NIFI-11094: Allow CaptureChangeMySQL to send multiple events per FlowFile
URL: https://github.com/apache/nifi/pull/6907
--
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
[GitHub] [nifi] mattyb149 commented on a diff in pull request #6907: NIFI-11094: Allow CaptureChangeMySQL to send multiple events per FlowFile
Posted by "mattyb149 (via GitHub)" <gi...@apache.org>.
mattyb149 commented on code in PR #6907:
URL: https://github.com/apache/nifi/pull/6907#discussion_r1106124965
##########
nifi-nar-bundles/nifi-cdc/nifi-cdc-mysql-bundle/nifi-cdc-mysql-processors/src/main/java/org/apache/nifi/cdc/mysql/processors/CaptureChangeMySQL.java:
##########
@@ -1017,15 +1096,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));
Review Comment:
Good catch, will fix.
--
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
[GitHub] [nifi] tpalfy commented on a diff in pull request #6907: NIFI-11094: Allow CaptureChangeMySQL to send multiple events per FlowFile
Posted by "tpalfy (via GitHub)" <gi...@apache.org>.
tpalfy commented on code in PR #6907:
URL: https://github.com/apache/nifi/pull/6907#discussion_r1106198842
##########
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:
Yeah, I think that is more desirable anyway.
--
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
[GitHub] [nifi] tpalfy commented on a diff in pull request #6907: NIFI-11094: Allow CaptureChangeMySQL to send multiple events per FlowFile
Posted by "tpalfy (via GitHub)" <gi...@apache.org>.
tpalfy commented on code in PR #6907:
URL: https://github.com/apache/nifi/pull/6907#discussion_r1114556227
##########
nifi-nar-bundles/nifi-cdc/nifi-cdc-mysql-bundle/nifi-cdc-mysql-processors/src/main/java/org/apache/nifi/cdc/mysql/event/io/AbstractBinlogEventWriter.java:
##########
@@ -44,35 +51,114 @@ protected void writeJson(T event) throws IOException {
}
protected Map<String, String> getCommonAttributes(final long sequenceId, BinlogEventInfo eventInfo) {
- return new HashMap<String, String>() {
- {
- put(SEQUENCE_ID_KEY, Long.toString(sequenceId));
- put(CDC_EVENT_TYPE_ATTRIBUTE, eventInfo.getEventType());
- String gtidSet = eventInfo.getBinlogGtidSet();
- if (gtidSet == null) {
- put(BinlogEventInfo.BINLOG_FILENAME_KEY, eventInfo.getBinlogFilename());
- put(BinlogEventInfo.BINLOG_POSITION_KEY, Long.toString(eventInfo.getBinlogPosition()));
- } else {
- put(BinlogEventInfo.BINLOG_GTIDSET_KEY, gtidSet);
- }
- put(CoreAttributes.MIME_TYPE.key(), APPLICATION_JSON);
- }
- };
+ final Map<String, String> commonAttributeMap = new HashMap<>();
+
+ commonAttributeMap.put(SEQUENCE_ID_KEY, Long.toString(sequenceId));
+ commonAttributeMap.put(CDC_EVENT_TYPE_ATTRIBUTE, eventInfo.getEventType());
+ String gtidSet = eventInfo.getBinlogGtidSet();
+ if (gtidSet == null) {
+ commonAttributeMap.put(BinlogEventInfo.BINLOG_FILENAME_KEY, eventInfo.getBinlogFilename());
+ commonAttributeMap.put(BinlogEventInfo.BINLOG_POSITION_KEY, Long.toString(eventInfo.getBinlogPosition()));
+ } else {
+ commonAttributeMap.put(BinlogEventInfo.BINLOG_GTIDSET_KEY, gtidSet);
+ }
+ commonAttributeMap.put(CoreAttributes.MIME_TYPE.key(), APPLICATION_JSON);
+
+ return commonAttributeMap;
}
// Default implementation for binlog events
@Override
- public long writeEvent(ProcessSession session, String transitUri, T eventInfo, long currentSequenceId, Relationship relationship) {
- FlowFile flowFile = session.create();
- flowFile = session.write(flowFile, (outputStream) -> {
+ public long writeEvent(ProcessSession session, String transitUri, T eventInfo, long currentSequenceId, Relationship relationship,
+ final EventWriterConfiguration eventWriterConfiguration) {
+ configureEventWriter(eventWriterConfiguration, session, eventInfo);
+
+ OutputStream outputStream = eventWriterConfiguration.getFlowFileOutputStream();
+ try {
super.startJson(outputStream, eventInfo);
writeJson(eventInfo);
// Nothing in the body
super.endJson();
- });
- flowFile = session.putAllAttributes(flowFile, getCommonAttributes(currentSequenceId, eventInfo));
- session.transfer(flowFile, relationship);
- session.getProvenanceReporter().receive(flowFile, transitUri);
+ } catch (IOException ioe) {
+ throw new UncheckedIOException("Write JSON start array failed", ioe);
+ }
+
+ eventWriterConfiguration.incrementNumberOfEventsWritten();
+
+ // Check if it is time to finish the FlowFile
+ if (nEventsPerFlowFile(eventWriterConfiguration)
+ && eventWriterConfiguration.getNumberOfEventsWritten() == eventWriterConfiguration.getNumberOfEventsPerFlowFile()) {
+ finishAndTransferFlowFile(session, eventWriterConfiguration, transitUri, currentSequenceId, eventInfo, relationship);
+ }
return currentSequenceId + 1;
}
+
+ public void finishAndTransferFlowFile(final ProcessSession session, final EventWriterConfiguration eventWriterConfiguration, final String transitUri, final long seqId,
+ final BinlogEventInfo eventInfo, final Relationship relationship) {
+ if (writtenMultipleEvents(eventWriterConfiguration)) {
+ try {
+ jsonGenerator.writeEndArray();
+ } catch (IOException ioe) {
+ throw new UncheckedIOException("Write JSON end array failed", ioe);
+ }
+ }
+ try {
+ endFile();
+
+ FlowFile flowFile = eventWriterConfiguration.getCurrentFlowFile();
+ if (session == null || flowFile == null) {
+ throw new ProcessException("No open FlowFile or ProcessSession to write to");
+ }
+ flowFile = session.putAllAttributes(flowFile, getCommonAttributes(seqId, eventInfo));
+ session.transfer(flowFile, relationship);
+ session.getProvenanceReporter().receive(flowFile, transitUri);
+
+ eventWriterConfiguration.cleanUp();
+ } catch (IOException ioe) {
+ throw new FlowFileAccessException("Failed to close event writer", ioe);
+ }
+ }
+
+ protected void configureEventWriter(final EventWriterConfiguration eventWriterConfiguration, final ProcessSession session, final EventInfo eventInfo) {
+ FlowFile flowFile = eventWriterConfiguration.getCurrentFlowFile();
+ if (flowFile == null) {
+ flowFile = session.create();
+ OutputStream flowFileOutputStream = session.write(flowFile);
+ if (eventWriterConfiguration.getJsonGenerator() == null) {
+ try {
+ jsonGenerator = createJsonGenerator(flowFileOutputStream);
+ } catch (IOException ioe) {
+ throw new UncheckedIOException("JSON Generator creation failed", ioe);
+ }
+ }
+ if (multipleEventsPerFlowFile(eventWriterConfiguration)) {
+ try {
+ jsonGenerator.writeStartArray();
+ } catch (IOException ioe) {
+ throw new UncheckedIOException("Write JSON start array failed", ioe);
+ }
+ }
+ eventWriterConfiguration.startNewFlowFile(flowFile, flowFileOutputStream, jsonGenerator);
+ }
+ jsonGenerator = eventWriterConfiguration.getJsonGenerator();
+ }
+
+ private boolean multipleEventsPerFlowFile(EventWriterConfiguration eventWriterConfiguration) {
+ return (nEventsPerFlowFile(eventWriterConfiguration)
+ && eventWriterConfiguration.getNumberOfEventsPerFlowFile() > 1)
+ || oneTransactionPerFlowFile(eventWriterConfiguration);
+ }
+
+ private boolean writtenMultipleEvents(EventWriterConfiguration eventWriterConfiguration) {
+ return eventWriterConfiguration.getNumberOfEventsWritten() > 1
+ || oneTransactionPerFlowFile(eventWriterConfiguration);
+ }
+
+ protected boolean nEventsPerFlowFile(EventWriterConfiguration eventWriterConfiguration) {
Review Comment:
```suggestion
protected boolean maxEventsPerFlowFile(EventWriterConfiguration eventWriterConfiguration) {
```
--
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
[GitHub] [nifi] mattyb149 commented on a diff in pull request #6907: NIFI-11094: Allow CaptureChangeMySQL to send multiple events per FlowFile
Posted by "mattyb149 (via GitHub)" <gi...@apache.org>.
mattyb149 commented on code in PR #6907:
URL: https://github.com/apache/nifi/pull/6907#discussion_r1114665037
##########
nifi-nar-bundles/nifi-cdc/nifi-cdc-mysql-bundle/nifi-cdc-mysql-processors/src/main/java/org/apache/nifi/cdc/mysql/event/io/AbstractBinlogEventWriter.java:
##########
@@ -44,35 +51,114 @@ protected void writeJson(T event) throws IOException {
}
protected Map<String, String> getCommonAttributes(final long sequenceId, BinlogEventInfo eventInfo) {
- return new HashMap<String, String>() {
- {
- put(SEQUENCE_ID_KEY, Long.toString(sequenceId));
- put(CDC_EVENT_TYPE_ATTRIBUTE, eventInfo.getEventType());
- String gtidSet = eventInfo.getBinlogGtidSet();
- if (gtidSet == null) {
- put(BinlogEventInfo.BINLOG_FILENAME_KEY, eventInfo.getBinlogFilename());
- put(BinlogEventInfo.BINLOG_POSITION_KEY, Long.toString(eventInfo.getBinlogPosition()));
- } else {
- put(BinlogEventInfo.BINLOG_GTIDSET_KEY, gtidSet);
- }
- put(CoreAttributes.MIME_TYPE.key(), APPLICATION_JSON);
- }
- };
+ final Map<String, String> commonAttributeMap = new HashMap<>();
+
+ commonAttributeMap.put(SEQUENCE_ID_KEY, Long.toString(sequenceId));
+ commonAttributeMap.put(CDC_EVENT_TYPE_ATTRIBUTE, eventInfo.getEventType());
+ String gtidSet = eventInfo.getBinlogGtidSet();
+ if (gtidSet == null) {
+ commonAttributeMap.put(BinlogEventInfo.BINLOG_FILENAME_KEY, eventInfo.getBinlogFilename());
+ commonAttributeMap.put(BinlogEventInfo.BINLOG_POSITION_KEY, Long.toString(eventInfo.getBinlogPosition()));
+ } else {
+ commonAttributeMap.put(BinlogEventInfo.BINLOG_GTIDSET_KEY, gtidSet);
+ }
+ commonAttributeMap.put(CoreAttributes.MIME_TYPE.key(), APPLICATION_JSON);
+
+ return commonAttributeMap;
}
// Default implementation for binlog events
@Override
- public long writeEvent(ProcessSession session, String transitUri, T eventInfo, long currentSequenceId, Relationship relationship) {
- FlowFile flowFile = session.create();
- flowFile = session.write(flowFile, (outputStream) -> {
+ public long writeEvent(ProcessSession session, String transitUri, T eventInfo, long currentSequenceId, Relationship relationship,
+ final EventWriterConfiguration eventWriterConfiguration) {
+ configureEventWriter(eventWriterConfiguration, session, eventInfo);
+
+ OutputStream outputStream = eventWriterConfiguration.getFlowFileOutputStream();
+ try {
super.startJson(outputStream, eventInfo);
writeJson(eventInfo);
// Nothing in the body
super.endJson();
- });
- flowFile = session.putAllAttributes(flowFile, getCommonAttributes(currentSequenceId, eventInfo));
- session.transfer(flowFile, relationship);
- session.getProvenanceReporter().receive(flowFile, transitUri);
+ } catch (IOException ioe) {
+ throw new UncheckedIOException("Write JSON start array failed", ioe);
+ }
+
+ eventWriterConfiguration.incrementNumberOfEventsWritten();
+
+ // Check if it is time to finish the FlowFile
+ if (nEventsPerFlowFile(eventWriterConfiguration)
+ && eventWriterConfiguration.getNumberOfEventsWritten() == eventWriterConfiguration.getNumberOfEventsPerFlowFile()) {
+ finishAndTransferFlowFile(session, eventWriterConfiguration, transitUri, currentSequenceId, eventInfo, relationship);
+ }
return currentSequenceId + 1;
}
+
+ public void finishAndTransferFlowFile(final ProcessSession session, final EventWriterConfiguration eventWriterConfiguration, final String transitUri, final long seqId,
+ final BinlogEventInfo eventInfo, final Relationship relationship) {
+ if (writtenMultipleEvents(eventWriterConfiguration)) {
+ try {
+ jsonGenerator.writeEndArray();
+ } catch (IOException ioe) {
+ throw new UncheckedIOException("Write JSON end array failed", ioe);
+ }
+ }
+ try {
+ endFile();
+
+ FlowFile flowFile = eventWriterConfiguration.getCurrentFlowFile();
+ if (session == null || flowFile == null) {
+ throw new ProcessException("No open FlowFile or ProcessSession to write to");
+ }
+ flowFile = session.putAllAttributes(flowFile, getCommonAttributes(seqId, eventInfo));
+ session.transfer(flowFile, relationship);
+ session.getProvenanceReporter().receive(flowFile, transitUri);
+
+ eventWriterConfiguration.cleanUp();
+ } catch (IOException ioe) {
+ throw new FlowFileAccessException("Failed to close event writer", ioe);
+ }
+ }
+
+ protected void configureEventWriter(final EventWriterConfiguration eventWriterConfiguration, final ProcessSession session, final EventInfo eventInfo) {
+ FlowFile flowFile = eventWriterConfiguration.getCurrentFlowFile();
+ if (flowFile == null) {
+ flowFile = session.create();
+ OutputStream flowFileOutputStream = session.write(flowFile);
+ if (eventWriterConfiguration.getJsonGenerator() == null) {
+ try {
+ jsonGenerator = createJsonGenerator(flowFileOutputStream);
+ } catch (IOException ioe) {
+ throw new UncheckedIOException("JSON Generator creation failed", ioe);
+ }
+ }
+ if (multipleEventsPerFlowFile(eventWriterConfiguration)) {
+ try {
+ jsonGenerator.writeStartArray();
+ } catch (IOException ioe) {
+ throw new UncheckedIOException("Write JSON start array failed", ioe);
+ }
+ }
+ eventWriterConfiguration.startNewFlowFile(flowFile, flowFileOutputStream, jsonGenerator);
+ }
+ jsonGenerator = eventWriterConfiguration.getJsonGenerator();
+ }
+
+ private boolean multipleEventsPerFlowFile(EventWriterConfiguration eventWriterConfiguration) {
+ return (nEventsPerFlowFile(eventWriterConfiguration)
+ && eventWriterConfiguration.getNumberOfEventsPerFlowFile() > 1)
+ || oneTransactionPerFlowFile(eventWriterConfiguration);
+ }
+
+ private boolean writtenMultipleEvents(EventWriterConfiguration eventWriterConfiguration) {
+ return eventWriterConfiguration.getNumberOfEventsWritten() > 1
+ || oneTransactionPerFlowFile(eventWriterConfiguration);
+ }
+
+ protected boolean nEventsPerFlowFile(EventWriterConfiguration eventWriterConfiguration) {
Review Comment:
Good catch! Will fix
--
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
[GitHub] [nifi] mattyb149 commented on pull request #6907: NIFI-11094: Allow CaptureChangeMySQL to send multiple events per FlowFile
Posted by "mattyb149 (via GitHub)" <gi...@apache.org>.
mattyb149 commented on PR #6907:
URL: https://github.com/apache/nifi/pull/6907#issuecomment-1440439004
Hopefully good to go now, thanks all for the reviews! I added Tamas as a co-author to the last commit, please retain that (after the "This closes" line if done that way) as I didn't successfully incorporate his patch as a commit.
--
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
[GitHub] [nifi] mark-bathori commented on a diff in pull request #6907: NIFI-11094: Allow CaptureChangeMySQL to send multiple events per FlowFile
Posted by "mark-bathori (via GitHub)" <gi...@apache.org>.
mark-bathori commented on code in PR #6907:
URL: https://github.com/apache/nifi/pull/6907#discussion_r1101506026
##########
nifi-nar-bundles/nifi-cdc/nifi-cdc-mysql-bundle/nifi-cdc-mysql-processors/src/main/java/org/apache/nifi/cdc/mysql/event/io/AbstractBinlogEventWriter.java:
##########
@@ -62,17 +67,112 @@ protected Map<String, String> getCommonAttributes(final long sequenceId, BinlogE
// Default implementation for binlog events
@Override
- public long writeEvent(ProcessSession session, String transitUri, T eventInfo, long currentSequenceId, Relationship relationship) {
- FlowFile flowFile = session.create();
- flowFile = session.write(flowFile, (outputStream) -> {
+ public long writeEvent(ProcessSession session, String transitUri, T eventInfo, long currentSequenceId, Relationship relationship,
+ final EventWriterConfiguration eventWriterConfiguration) {
+ FlowFile flowFile = eventWriterConfiguration.getCurrentFlowFile();
+ if (flowFile == null) {
+ flowFile = session.create();
+ OutputStream flowFileOutputStream = session.write(flowFile);
+ eventWriterConfiguration.setFlowFileOutputStream(flowFileOutputStream);
+ eventWriterConfiguration.setCurrentFlowFile(flowFile);
+ if (eventWriterConfiguration.getJsonGenerator() == null) {
+ try {
+ jsonGenerator = createJsonGenerator(flowFileOutputStream);
+ eventWriterConfiguration.setJsonGenerator(jsonGenerator);
+ } catch (IOException ioe) {
+ throw new FlowFileAccessException("Couldn't create JSON generator", ioe);
+ }
+ }
+ if ((FlowFileEventWriteStrategy.N_EVENTS_PER_FLOWFILE.equals(eventWriterConfiguration.getFlowFileEventWriteStrategy())
+ && eventWriterConfiguration.getNumberOfEventsPerFlowFile() > 1)
+ || FlowFileEventWriteStrategy.ONE_TRANSACTION_PER_FLOWFILE.equals(eventWriterConfiguration.getFlowFileEventWriteStrategy())) {
+ try {
+ jsonGenerator.writeStartArray();
+ } catch (IOException ioe) {
+ throw new FlowFileAccessException("Couldn't write start of event array", ioe);
+ }
+ }
+ }
+ jsonGenerator = eventWriterConfiguration.getJsonGenerator();
+
+ OutputStream outputStream = eventWriterConfiguration.getFlowFileOutputStream();
+ try {
super.startJson(outputStream, eventInfo);
writeJson(eventInfo);
// Nothing in the body
super.endJson();
- });
- flowFile = session.putAllAttributes(flowFile, getCommonAttributes(currentSequenceId, eventInfo));
+ } catch (IOException ioe) {
+ throw new FlowFileAccessException("Couldn't write start of event array", ioe);
+ }
+
+ eventWriterConfiguration.incrementNumberOfEventsWritten();
+
+ // Check if it is time to finish the FlowFile
+ if (FlowFileEventWriteStrategy.N_EVENTS_PER_FLOWFILE.equals(eventWriterConfiguration.getFlowFileEventWriteStrategy())
+ && eventWriterConfiguration.getNumberOfEventsWritten() == eventWriterConfiguration.getNumberOfEventsPerFlowFile()) {
+ flowFile = finishAndTransferFlowFile(eventWriterConfiguration, transitUri, currentSequenceId, eventInfo, relationship);
+ }
+ eventWriterConfiguration.setCurrentFlowFile(flowFile);
+ return currentSequenceId + 1;
+ }
+
+ public FlowFile finishAndTransferFlowFile(final EventWriterConfiguration eventWriterConfiguration, final String transitUri, final long seqId,
+ final BinlogEventInfo eventInfo, final Relationship relationship) {
+ // If writing multiple events, end the array
+ if (eventWriterConfiguration.getNumberOfEventsWritten() > 1
+ || FlowFileEventWriteStrategy.ONE_TRANSACTION_PER_FLOWFILE.equals(eventWriterConfiguration.getFlowFileEventWriteStrategy())) {
+ try {
+ jsonGenerator.writeEndArray();
+ } catch (IOException ioe) {
+ throw new FlowFileAccessException("Couldn't write end of event array", ioe);
+ }
+ }
+ try {
+ endFile();
+ eventWriterConfiguration.setJsonGenerator(null);
+ eventWriterConfiguration.getFlowFileOutputStream().close();
+ } catch (IOException ioe) {
+ throw new FlowFileAccessException("Couldn't flush and close file", ioe);
+ }
+ FlowFile flowFile = eventWriterConfiguration.getCurrentFlowFile();
+ ProcessSession session = eventWriterConfiguration.getWorkingSession();
+ if (session == null && flowFile == null) {
Review Comment:
I think the `&&` operator should be `||` here since if either condition is false we will got NullPointerException in the next line. The exception's description also assumes 'OR' relation.
--
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
[GitHub] [nifi] exceptionfactory commented on a diff in pull request #6907: NIFI-11094: Allow CaptureChangeMySQL to send multiple events per FlowFile
Posted by "exceptionfactory (via GitHub)" <gi...@apache.org>.
exceptionfactory commented on code in PR #6907:
URL: https://github.com/apache/nifi/pull/6907#discussion_r1101893920
##########
nifi-nar-bundles/nifi-cdc/nifi-cdc-mysql-bundle/nifi-cdc-mysql-processors/src/main/java/org/apache/nifi/cdc/mysql/event/io/AbstractBinlogEventWriter.java:
##########
@@ -44,7 +49,7 @@ protected void writeJson(T event) throws IOException {
}
protected Map<String, String> getCommonAttributes(final long sequenceId, BinlogEventInfo eventInfo) {
- return new HashMap<String, String>() {
+ return new HashMap<>() {
{
put(SEQUENCE_ID_KEY, Long.toString(sequenceId));
put(CDC_EVENT_TYPE_ATTRIBUTE, eventInfo.getEventType());
Review Comment:
This change fails to compile on Java 8. This is a good opportunity to refactor and avoid the use of anonymous extensions of `HashMap`.
##########
nifi-nar-bundles/nifi-cdc/nifi-cdc-mysql-bundle/nifi-cdc-mysql-processors/src/main/java/org/apache/nifi/cdc/mysql/event/io/AbstractBinlogEventWriter.java:
##########
@@ -62,17 +67,112 @@ protected Map<String, String> getCommonAttributes(final long sequenceId, BinlogE
// Default implementation for binlog events
@Override
- public long writeEvent(ProcessSession session, String transitUri, T eventInfo, long currentSequenceId, Relationship relationship) {
- FlowFile flowFile = session.create();
- flowFile = session.write(flowFile, (outputStream) -> {
+ public long writeEvent(ProcessSession session, String transitUri, T eventInfo, long currentSequenceId, Relationship relationship,
+ final EventWriterConfiguration eventWriterConfiguration) {
+ FlowFile flowFile = eventWriterConfiguration.getCurrentFlowFile();
+ if (flowFile == null) {
+ flowFile = session.create();
+ OutputStream flowFileOutputStream = session.write(flowFile);
+ eventWriterConfiguration.setFlowFileOutputStream(flowFileOutputStream);
+ eventWriterConfiguration.setCurrentFlowFile(flowFile);
+ if (eventWriterConfiguration.getJsonGenerator() == null) {
+ try {
+ jsonGenerator = createJsonGenerator(flowFileOutputStream);
+ eventWriterConfiguration.setJsonGenerator(jsonGenerator);
+ } catch (IOException ioe) {
+ throw new FlowFileAccessException("Couldn't create JSON generator", ioe);
Review Comment:
Recommend avoiding conjunctions in error message wording. Is `FlowFileAccessException` the best choice in this scenario? Perhaps an `UncheckedIOException` would be better.
```suggestion
throw new UncheckedIOException("JSON Generator creation failed", ioe);
```
##########
nifi-nar-bundles/nifi-cdc/nifi-cdc-mysql-bundle/nifi-cdc-mysql-processors/src/main/java/org/apache/nifi/cdc/mysql/event/io/AbstractBinlogEventWriter.java:
##########
@@ -62,17 +67,112 @@ protected Map<String, String> getCommonAttributes(final long sequenceId, BinlogE
// Default implementation for binlog events
@Override
- public long writeEvent(ProcessSession session, String transitUri, T eventInfo, long currentSequenceId, Relationship relationship) {
- FlowFile flowFile = session.create();
- flowFile = session.write(flowFile, (outputStream) -> {
+ public long writeEvent(ProcessSession session, String transitUri, T eventInfo, long currentSequenceId, Relationship relationship,
+ final EventWriterConfiguration eventWriterConfiguration) {
+ FlowFile flowFile = eventWriterConfiguration.getCurrentFlowFile();
+ if (flowFile == null) {
+ flowFile = session.create();
+ OutputStream flowFileOutputStream = session.write(flowFile);
+ eventWriterConfiguration.setFlowFileOutputStream(flowFileOutputStream);
+ eventWriterConfiguration.setCurrentFlowFile(flowFile);
+ if (eventWriterConfiguration.getJsonGenerator() == null) {
+ try {
+ jsonGenerator = createJsonGenerator(flowFileOutputStream);
+ eventWriterConfiguration.setJsonGenerator(jsonGenerator);
+ } catch (IOException ioe) {
+ throw new FlowFileAccessException("Couldn't create JSON generator", ioe);
+ }
+ }
+ if ((FlowFileEventWriteStrategy.N_EVENTS_PER_FLOWFILE.equals(eventWriterConfiguration.getFlowFileEventWriteStrategy())
+ && eventWriterConfiguration.getNumberOfEventsPerFlowFile() > 1)
+ || FlowFileEventWriteStrategy.ONE_TRANSACTION_PER_FLOWFILE.equals(eventWriterConfiguration.getFlowFileEventWriteStrategy())) {
+ try {
+ jsonGenerator.writeStartArray();
+ } catch (IOException ioe) {
+ throw new FlowFileAccessException("Couldn't write start of event array", ioe);
+ }
+ }
+ }
+ jsonGenerator = eventWriterConfiguration.getJsonGenerator();
+
+ OutputStream outputStream = eventWriterConfiguration.getFlowFileOutputStream();
+ try {
super.startJson(outputStream, eventInfo);
writeJson(eventInfo);
// Nothing in the body
super.endJson();
- });
- flowFile = session.putAllAttributes(flowFile, getCommonAttributes(currentSequenceId, eventInfo));
+ } catch (IOException ioe) {
+ throw new FlowFileAccessException("Couldn't write start of event array", ioe);
+ }
+
+ eventWriterConfiguration.incrementNumberOfEventsWritten();
+
+ // Check if it is time to finish the FlowFile
+ if (FlowFileEventWriteStrategy.N_EVENTS_PER_FLOWFILE.equals(eventWriterConfiguration.getFlowFileEventWriteStrategy())
+ && eventWriterConfiguration.getNumberOfEventsWritten() == eventWriterConfiguration.getNumberOfEventsPerFlowFile()) {
+ flowFile = finishAndTransferFlowFile(eventWriterConfiguration, transitUri, currentSequenceId, eventInfo, relationship);
+ }
+ eventWriterConfiguration.setCurrentFlowFile(flowFile);
+ return currentSequenceId + 1;
+ }
+
+ public FlowFile finishAndTransferFlowFile(final EventWriterConfiguration eventWriterConfiguration, final String transitUri, final long seqId,
+ final BinlogEventInfo eventInfo, final Relationship relationship) {
+ // If writing multiple events, end the array
+ if (eventWriterConfiguration.getNumberOfEventsWritten() > 1
+ || FlowFileEventWriteStrategy.ONE_TRANSACTION_PER_FLOWFILE.equals(eventWriterConfiguration.getFlowFileEventWriteStrategy())) {
+ try {
+ jsonGenerator.writeEndArray();
+ } catch (IOException ioe) {
+ throw new FlowFileAccessException("Couldn't write end of event array", ioe);
+ }
+ }
+ try {
+ endFile();
+ eventWriterConfiguration.setJsonGenerator(null);
+ eventWriterConfiguration.getFlowFileOutputStream().close();
+ } catch (IOException ioe) {
+ throw new FlowFileAccessException("Couldn't flush and close file", ioe);
Review Comment:
```suggestion
throw new UncheckedIOException("Failed to close event writer", ioe);
```
##########
nifi-nar-bundles/nifi-cdc/nifi-cdc-mysql-bundle/nifi-cdc-mysql-processors/src/main/java/org/apache/nifi/cdc/mysql/event/io/AbstractBinlogEventWriter.java:
##########
@@ -62,17 +67,112 @@ protected Map<String, String> getCommonAttributes(final long sequenceId, BinlogE
// Default implementation for binlog events
@Override
- public long writeEvent(ProcessSession session, String transitUri, T eventInfo, long currentSequenceId, Relationship relationship) {
- FlowFile flowFile = session.create();
- flowFile = session.write(flowFile, (outputStream) -> {
+ public long writeEvent(ProcessSession session, String transitUri, T eventInfo, long currentSequenceId, Relationship relationship,
+ final EventWriterConfiguration eventWriterConfiguration) {
+ FlowFile flowFile = eventWriterConfiguration.getCurrentFlowFile();
+ if (flowFile == null) {
+ flowFile = session.create();
+ OutputStream flowFileOutputStream = session.write(flowFile);
+ eventWriterConfiguration.setFlowFileOutputStream(flowFileOutputStream);
+ eventWriterConfiguration.setCurrentFlowFile(flowFile);
+ if (eventWriterConfiguration.getJsonGenerator() == null) {
+ try {
+ jsonGenerator = createJsonGenerator(flowFileOutputStream);
+ eventWriterConfiguration.setJsonGenerator(jsonGenerator);
+ } catch (IOException ioe) {
+ throw new FlowFileAccessException("Couldn't create JSON generator", ioe);
+ }
+ }
+ if ((FlowFileEventWriteStrategy.N_EVENTS_PER_FLOWFILE.equals(eventWriterConfiguration.getFlowFileEventWriteStrategy())
+ && eventWriterConfiguration.getNumberOfEventsPerFlowFile() > 1)
+ || FlowFileEventWriteStrategy.ONE_TRANSACTION_PER_FLOWFILE.equals(eventWriterConfiguration.getFlowFileEventWriteStrategy())) {
+ try {
+ jsonGenerator.writeStartArray();
+ } catch (IOException ioe) {
+ throw new FlowFileAccessException("Couldn't write start of event array", ioe);
+ }
+ }
+ }
+ jsonGenerator = eventWriterConfiguration.getJsonGenerator();
+
+ OutputStream outputStream = eventWriterConfiguration.getFlowFileOutputStream();
+ try {
super.startJson(outputStream, eventInfo);
writeJson(eventInfo);
// Nothing in the body
super.endJson();
- });
- flowFile = session.putAllAttributes(flowFile, getCommonAttributes(currentSequenceId, eventInfo));
+ } catch (IOException ioe) {
+ throw new FlowFileAccessException("Couldn't write start of event array", ioe);
+ }
+
+ eventWriterConfiguration.incrementNumberOfEventsWritten();
+
+ // Check if it is time to finish the FlowFile
+ if (FlowFileEventWriteStrategy.N_EVENTS_PER_FLOWFILE.equals(eventWriterConfiguration.getFlowFileEventWriteStrategy())
+ && eventWriterConfiguration.getNumberOfEventsWritten() == eventWriterConfiguration.getNumberOfEventsPerFlowFile()) {
+ flowFile = finishAndTransferFlowFile(eventWriterConfiguration, transitUri, currentSequenceId, eventInfo, relationship);
+ }
+ eventWriterConfiguration.setCurrentFlowFile(flowFile);
+ return currentSequenceId + 1;
+ }
+
+ public FlowFile finishAndTransferFlowFile(final EventWriterConfiguration eventWriterConfiguration, final String transitUri, final long seqId,
+ final BinlogEventInfo eventInfo, final Relationship relationship) {
+ // If writing multiple events, end the array
+ if (eventWriterConfiguration.getNumberOfEventsWritten() > 1
+ || FlowFileEventWriteStrategy.ONE_TRANSACTION_PER_FLOWFILE.equals(eventWriterConfiguration.getFlowFileEventWriteStrategy())) {
+ try {
+ jsonGenerator.writeEndArray();
+ } catch (IOException ioe) {
+ throw new FlowFileAccessException("Couldn't write end of event array", ioe);
Review Comment:
```suggestion
throw new UncheckedIOException("Write JSON start array failed", ioe);
```
##########
nifi-nar-bundles/nifi-cdc/nifi-cdc-mysql-bundle/nifi-cdc-mysql-processors/src/main/java/org/apache/nifi/cdc/mysql/event/io/DDLEventWriter.java:
##########
@@ -16,28 +16,46 @@
*/
package org.apache.nifi.cdc.mysql.event.io;
+import org.apache.nifi.cdc.event.io.EventWriterConfiguration;
+import org.apache.nifi.cdc.event.io.FlowFileEventWriteStrategy;
import org.apache.nifi.flowfile.FlowFile;
import org.apache.nifi.processor.ProcessSession;
import org.apache.nifi.processor.Relationship;
import org.apache.nifi.cdc.mysql.event.DDLEventInfo;
+import org.apache.nifi.processor.exception.FlowFileAccessException;
+
+import java.io.IOException;
+import java.io.OutputStream;
+
/**
* A writer class to output MySQL binlog Data Definition Language (DDL) events to flow file(s).
*/
public class DDLEventWriter extends AbstractBinlogTableEventWriter<DDLEventInfo> {
@Override
- public long writeEvent(ProcessSession session, String transitUri, DDLEventInfo eventInfo, long currentSequenceId, Relationship relationship) {
- FlowFile flowFile = session.create();
- flowFile = session.write(flowFile, (outputStream) -> {
+ public long writeEvent(ProcessSession session, String transitUri, DDLEventInfo eventInfo, long currentSequenceId, Relationship relationship,
+ final EventWriterConfiguration eventWriterConfiguration) {
+ FlowFile flowFile = configureEventWriter(eventWriterConfiguration, session, eventInfo);
+ OutputStream outputStream = eventWriterConfiguration.getFlowFileOutputStream();
+
+ try {
super.startJson(outputStream, eventInfo);
super.writeJson(eventInfo);
jsonGenerator.writeStringField("query", eventInfo.getQuery());
super.endJson();
- });
- flowFile = session.putAllAttributes(flowFile, getCommonAttributes(currentSequenceId, eventInfo));
- session.transfer(flowFile, relationship);
- session.getProvenanceReporter().receive(flowFile, transitUri);
+ } catch (IOException ioe) {
+ throw new FlowFileAccessException("Couldn't write start of event array", ioe);
Review Comment:
```suggestion
throw new UncheckedIOException("Write JSON start array failed", ioe);
```
##########
nifi-nar-bundles/nifi-cdc/nifi-cdc-mysql-bundle/nifi-cdc-mysql-processors/src/main/java/org/apache/nifi/cdc/mysql/processors/CaptureChangeMySQL.java:
##########
@@ -269,6 +284,31 @@ public class CaptureChangeMySQL extends AbstractSessionFactoryProcessor {
.expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY)
.build();
+ public static final PropertyDescriptor EVENTS_PER_FLOWFILE_STRATEGY = new PropertyDescriptor.Builder()
+ .name("capture-change-mysql-events-per-flowfile")
+ .displayName("Events Per FlowFile Strategy")
+ .description("Specifies the strategy to use when writing events to FlowFile(s)")
+ .required(true)
+ .sensitive(false)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .allowableValues(N_EVENTS_PER_FLOWFILE_STRATEGY, ONE_TRANSACTION_PER_FLOWFILE_STRATEGY)
+ .defaultValue(N_EVENTS_PER_FLOWFILE_STRATEGY.getValue())
+ .expressionLanguageSupported(ExpressionLanguageScope.NONE)
+ .build();
+
+ public static final PropertyDescriptor NUMBER_OF_EVENTS_PER_FLOWFILE = new PropertyDescriptor.Builder()
+ .name("capture-change-mysql-number-of-events-per-flowfile")
+ .displayName("Number of Events Per FlowFile")
Review Comment:
This seems more verbose than necessary, what do you think about shorter names?
```suggestion
.name("events-per-flowfile")
.displayName("Events Per FlowFile")
```
##########
nifi-nar-bundles/nifi-cdc/nifi-cdc-mysql-bundle/nifi-cdc-mysql-processors/src/main/java/org/apache/nifi/cdc/mysql/processors/CaptureChangeMySQL.java:
##########
@@ -1241,6 +1335,23 @@ protected TableInfo loadTableInfo(TableInfoCacheKey key) throws SQLException {
return tableInfo;
}
+ protected Map<String, String> getCommonAttributes(final long sequenceId, BinlogEventInfo eventInfo) {
+ return new HashMap<>() {
+ {
+ put(SEQUENCE_ID_KEY, Long.toString(sequenceId));
+ put(CDC_EVENT_TYPE_ATTRIBUTE, eventInfo.getEventType());
+ String gtidSet = eventInfo.getBinlogGtidSet();
+ if (gtidSet == null) {
+ put(BinlogEventInfo.BINLOG_FILENAME_KEY, eventInfo.getBinlogFilename());
+ put(BinlogEventInfo.BINLOG_POSITION_KEY, Long.toString(eventInfo.getBinlogPosition()));
+ } else {
+ put(BinlogEventInfo.BINLOG_GTIDSET_KEY, gtidSet);
+ }
+ put(CoreAttributes.MIME_TYPE.key(), APPLICATION_JSON);
+ }
+ };
Review Comment:
This does not compile on Java 8. Although the goal is to move to Java 11 soon, using an anonymous extension of HashMap should be avoided.
##########
nifi-nar-bundles/nifi-cdc/nifi-cdc-mysql-bundle/nifi-cdc-mysql-processors/src/main/java/org/apache/nifi/cdc/mysql/processors/CaptureChangeMySQL.java:
##########
@@ -269,6 +284,31 @@ public class CaptureChangeMySQL extends AbstractSessionFactoryProcessor {
.expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY)
.build();
+ public static final PropertyDescriptor EVENTS_PER_FLOWFILE_STRATEGY = new PropertyDescriptor.Builder()
+ .name("capture-change-mysql-events-per-flowfile")
+ .displayName("Events Per FlowFile Strategy")
Review Comment:
What do you think about calling this `Event Processing Strategy`?
##########
nifi-nar-bundles/nifi-cdc/nifi-cdc-mysql-bundle/nifi-cdc-mysql-processors/src/main/java/org/apache/nifi/cdc/mysql/event/io/AbstractBinlogEventWriter.java:
##########
@@ -62,17 +67,112 @@ protected Map<String, String> getCommonAttributes(final long sequenceId, BinlogE
// Default implementation for binlog events
@Override
- public long writeEvent(ProcessSession session, String transitUri, T eventInfo, long currentSequenceId, Relationship relationship) {
- FlowFile flowFile = session.create();
- flowFile = session.write(flowFile, (outputStream) -> {
+ public long writeEvent(ProcessSession session, String transitUri, T eventInfo, long currentSequenceId, Relationship relationship,
+ final EventWriterConfiguration eventWriterConfiguration) {
+ FlowFile flowFile = eventWriterConfiguration.getCurrentFlowFile();
+ if (flowFile == null) {
+ flowFile = session.create();
+ OutputStream flowFileOutputStream = session.write(flowFile);
+ eventWriterConfiguration.setFlowFileOutputStream(flowFileOutputStream);
+ eventWriterConfiguration.setCurrentFlowFile(flowFile);
+ if (eventWriterConfiguration.getJsonGenerator() == null) {
+ try {
+ jsonGenerator = createJsonGenerator(flowFileOutputStream);
+ eventWriterConfiguration.setJsonGenerator(jsonGenerator);
+ } catch (IOException ioe) {
+ throw new FlowFileAccessException("Couldn't create JSON generator", ioe);
+ }
+ }
+ if ((FlowFileEventWriteStrategy.N_EVENTS_PER_FLOWFILE.equals(eventWriterConfiguration.getFlowFileEventWriteStrategy())
+ && eventWriterConfiguration.getNumberOfEventsPerFlowFile() > 1)
+ || FlowFileEventWriteStrategy.ONE_TRANSACTION_PER_FLOWFILE.equals(eventWriterConfiguration.getFlowFileEventWriteStrategy())) {
+ try {
+ jsonGenerator.writeStartArray();
+ } catch (IOException ioe) {
+ throw new FlowFileAccessException("Couldn't write start of event array", ioe);
Review Comment:
```suggestion
throw new UncheckedIOException("Write JSON start array failed", ioe);
```
##########
nifi-nar-bundles/nifi-cdc/nifi-cdc-mysql-bundle/nifi-cdc-mysql-processors/src/main/java/org/apache/nifi/cdc/mysql/event/io/AbstractBinlogEventWriter.java:
##########
@@ -62,17 +67,112 @@ protected Map<String, String> getCommonAttributes(final long sequenceId, BinlogE
// Default implementation for binlog events
@Override
- public long writeEvent(ProcessSession session, String transitUri, T eventInfo, long currentSequenceId, Relationship relationship) {
- FlowFile flowFile = session.create();
- flowFile = session.write(flowFile, (outputStream) -> {
+ public long writeEvent(ProcessSession session, String transitUri, T eventInfo, long currentSequenceId, Relationship relationship,
+ final EventWriterConfiguration eventWriterConfiguration) {
+ FlowFile flowFile = eventWriterConfiguration.getCurrentFlowFile();
+ if (flowFile == null) {
+ flowFile = session.create();
+ OutputStream flowFileOutputStream = session.write(flowFile);
+ eventWriterConfiguration.setFlowFileOutputStream(flowFileOutputStream);
+ eventWriterConfiguration.setCurrentFlowFile(flowFile);
+ if (eventWriterConfiguration.getJsonGenerator() == null) {
+ try {
+ jsonGenerator = createJsonGenerator(flowFileOutputStream);
+ eventWriterConfiguration.setJsonGenerator(jsonGenerator);
+ } catch (IOException ioe) {
+ throw new FlowFileAccessException("Couldn't create JSON generator", ioe);
+ }
+ }
+ if ((FlowFileEventWriteStrategy.N_EVENTS_PER_FLOWFILE.equals(eventWriterConfiguration.getFlowFileEventWriteStrategy())
+ && eventWriterConfiguration.getNumberOfEventsPerFlowFile() > 1)
+ || FlowFileEventWriteStrategy.ONE_TRANSACTION_PER_FLOWFILE.equals(eventWriterConfiguration.getFlowFileEventWriteStrategy())) {
+ try {
+ jsonGenerator.writeStartArray();
+ } catch (IOException ioe) {
+ throw new FlowFileAccessException("Couldn't write start of event array", ioe);
+ }
+ }
+ }
+ jsonGenerator = eventWriterConfiguration.getJsonGenerator();
+
+ OutputStream outputStream = eventWriterConfiguration.getFlowFileOutputStream();
+ try {
super.startJson(outputStream, eventInfo);
writeJson(eventInfo);
// Nothing in the body
super.endJson();
- });
- flowFile = session.putAllAttributes(flowFile, getCommonAttributes(currentSequenceId, eventInfo));
+ } catch (IOException ioe) {
+ throw new FlowFileAccessException("Couldn't write start of event array", ioe);
+ }
+
+ eventWriterConfiguration.incrementNumberOfEventsWritten();
+
+ // Check if it is time to finish the FlowFile
+ if (FlowFileEventWriteStrategy.N_EVENTS_PER_FLOWFILE.equals(eventWriterConfiguration.getFlowFileEventWriteStrategy())
+ && eventWriterConfiguration.getNumberOfEventsWritten() == eventWriterConfiguration.getNumberOfEventsPerFlowFile()) {
+ flowFile = finishAndTransferFlowFile(eventWriterConfiguration, transitUri, currentSequenceId, eventInfo, relationship);
+ }
+ eventWriterConfiguration.setCurrentFlowFile(flowFile);
+ return currentSequenceId + 1;
+ }
+
+ public FlowFile finishAndTransferFlowFile(final EventWriterConfiguration eventWriterConfiguration, final String transitUri, final long seqId,
+ final BinlogEventInfo eventInfo, final Relationship relationship) {
+ // If writing multiple events, end the array
+ if (eventWriterConfiguration.getNumberOfEventsWritten() > 1
+ || FlowFileEventWriteStrategy.ONE_TRANSACTION_PER_FLOWFILE.equals(eventWriterConfiguration.getFlowFileEventWriteStrategy())) {
+ try {
+ jsonGenerator.writeEndArray();
+ } catch (IOException ioe) {
+ throw new FlowFileAccessException("Couldn't write end of event array", ioe);
+ }
+ }
+ try {
+ endFile();
+ eventWriterConfiguration.setJsonGenerator(null);
+ eventWriterConfiguration.getFlowFileOutputStream().close();
+ } catch (IOException ioe) {
+ throw new FlowFileAccessException("Couldn't flush and close file", ioe);
+ }
+ FlowFile flowFile = eventWriterConfiguration.getCurrentFlowFile();
+ ProcessSession session = eventWriterConfiguration.getWorkingSession();
+ if (session == null && flowFile == null) {
+ throw new FlowFileAccessException("No open FlowFile or ProcessSession to write to");
Review Comment:
Is `FlowFileAccessException` the best option? Perhaps `ProcessException`?
```suggestion
throw new ProcessException("No open FlowFile or ProcessSession found for writing");
```
--
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
[GitHub] [nifi] mark-bathori commented on a diff in pull request #6907: NIFI-11094: Allow CaptureChangeMySQL to send multiple events per FlowFile
Posted by "mark-bathori (via GitHub)" <gi...@apache.org>.
mark-bathori commented on code in PR #6907:
URL: https://github.com/apache/nifi/pull/6907#discussion_r1101506026
##########
nifi-nar-bundles/nifi-cdc/nifi-cdc-mysql-bundle/nifi-cdc-mysql-processors/src/main/java/org/apache/nifi/cdc/mysql/event/io/AbstractBinlogEventWriter.java:
##########
@@ -62,17 +67,112 @@ protected Map<String, String> getCommonAttributes(final long sequenceId, BinlogE
// Default implementation for binlog events
@Override
- public long writeEvent(ProcessSession session, String transitUri, T eventInfo, long currentSequenceId, Relationship relationship) {
- FlowFile flowFile = session.create();
- flowFile = session.write(flowFile, (outputStream) -> {
+ public long writeEvent(ProcessSession session, String transitUri, T eventInfo, long currentSequenceId, Relationship relationship,
+ final EventWriterConfiguration eventWriterConfiguration) {
+ FlowFile flowFile = eventWriterConfiguration.getCurrentFlowFile();
+ if (flowFile == null) {
+ flowFile = session.create();
+ OutputStream flowFileOutputStream = session.write(flowFile);
+ eventWriterConfiguration.setFlowFileOutputStream(flowFileOutputStream);
+ eventWriterConfiguration.setCurrentFlowFile(flowFile);
+ if (eventWriterConfiguration.getJsonGenerator() == null) {
+ try {
+ jsonGenerator = createJsonGenerator(flowFileOutputStream);
+ eventWriterConfiguration.setJsonGenerator(jsonGenerator);
+ } catch (IOException ioe) {
+ throw new FlowFileAccessException("Couldn't create JSON generator", ioe);
+ }
+ }
+ if ((FlowFileEventWriteStrategy.N_EVENTS_PER_FLOWFILE.equals(eventWriterConfiguration.getFlowFileEventWriteStrategy())
+ && eventWriterConfiguration.getNumberOfEventsPerFlowFile() > 1)
+ || FlowFileEventWriteStrategy.ONE_TRANSACTION_PER_FLOWFILE.equals(eventWriterConfiguration.getFlowFileEventWriteStrategy())) {
+ try {
+ jsonGenerator.writeStartArray();
+ } catch (IOException ioe) {
+ throw new FlowFileAccessException("Couldn't write start of event array", ioe);
+ }
+ }
+ }
+ jsonGenerator = eventWriterConfiguration.getJsonGenerator();
+
+ OutputStream outputStream = eventWriterConfiguration.getFlowFileOutputStream();
+ try {
super.startJson(outputStream, eventInfo);
writeJson(eventInfo);
// Nothing in the body
super.endJson();
- });
- flowFile = session.putAllAttributes(flowFile, getCommonAttributes(currentSequenceId, eventInfo));
+ } catch (IOException ioe) {
+ throw new FlowFileAccessException("Couldn't write start of event array", ioe);
+ }
+
+ eventWriterConfiguration.incrementNumberOfEventsWritten();
+
+ // Check if it is time to finish the FlowFile
+ if (FlowFileEventWriteStrategy.N_EVENTS_PER_FLOWFILE.equals(eventWriterConfiguration.getFlowFileEventWriteStrategy())
+ && eventWriterConfiguration.getNumberOfEventsWritten() == eventWriterConfiguration.getNumberOfEventsPerFlowFile()) {
+ flowFile = finishAndTransferFlowFile(eventWriterConfiguration, transitUri, currentSequenceId, eventInfo, relationship);
+ }
+ eventWriterConfiguration.setCurrentFlowFile(flowFile);
+ return currentSequenceId + 1;
+ }
+
+ public FlowFile finishAndTransferFlowFile(final EventWriterConfiguration eventWriterConfiguration, final String transitUri, final long seqId,
+ final BinlogEventInfo eventInfo, final Relationship relationship) {
+ // If writing multiple events, end the array
+ if (eventWriterConfiguration.getNumberOfEventsWritten() > 1
+ || FlowFileEventWriteStrategy.ONE_TRANSACTION_PER_FLOWFILE.equals(eventWriterConfiguration.getFlowFileEventWriteStrategy())) {
+ try {
+ jsonGenerator.writeEndArray();
+ } catch (IOException ioe) {
+ throw new FlowFileAccessException("Couldn't write end of event array", ioe);
+ }
+ }
+ try {
+ endFile();
+ eventWriterConfiguration.setJsonGenerator(null);
+ eventWriterConfiguration.getFlowFileOutputStream().close();
+ } catch (IOException ioe) {
+ throw new FlowFileAccessException("Couldn't flush and close file", ioe);
+ }
+ FlowFile flowFile = eventWriterConfiguration.getCurrentFlowFile();
+ ProcessSession session = eventWriterConfiguration.getWorkingSession();
+ if (session == null && flowFile == null) {
Review Comment:
I think the `&&` operator should be `||` here since either condition is false we will got NullPointerException in the next line. The exception's description also assumes 'OR' relation.
##########
nifi-nar-bundles/nifi-cdc/nifi-cdc-api/src/main/java/org/apache/nifi/cdc/event/io/EventWriterConfiguration.java:
##########
@@ -0,0 +1,102 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.cdc.event.io;
+
+import com.fasterxml.jackson.core.JsonGenerator;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.ProcessSession;
+
+import java.io.OutputStream;
+
+public class EventWriterConfiguration {
+
+ private FlowFileEventWriteStrategy flowFileEventWriteStrategy;
+ private int numberOfEventsWritten = 0;
+ private int numberOfEventsPerFlowFile = 1000;
+ private FlowFile currentFlowFile;
+ private OutputStream flowFileOutputStream;
+ private ProcessSession workingSession;
+ private JsonGenerator jsonGenerator;
+
+ public EventWriterConfiguration(FlowFileEventWriteStrategy flowFileEventWriteStrategy, int numberOfEventsWritten, int numberOfEventsPerFlowFile, FlowFile currentFlowFile) {
+ this.flowFileEventWriteStrategy = flowFileEventWriteStrategy;
+ this.numberOfEventsWritten = numberOfEventsWritten;
+ this.numberOfEventsPerFlowFile = numberOfEventsPerFlowFile;
+ this.currentFlowFile = currentFlowFile;
+ }
+
+ public FlowFileEventWriteStrategy getFlowFileEventWriteStrategy() {
+ return flowFileEventWriteStrategy;
+ }
+
+ public void setFlowFileEventWriteStrategy(FlowFileEventWriteStrategy flowFileEventWriteStrategy) {
+ this.flowFileEventWriteStrategy = flowFileEventWriteStrategy;
+ }
+
+ public int getNumberOfEventsWritten() {
+ return numberOfEventsWritten;
+ }
+
+ public void setNumberOfEventsWritten(int numberOfEventsWritten) {
Review Comment:
This method is always called with the same value. Maybe remove the method param and set the `numberOfEventsWritten` to 0 within the method's body and rename the function according to it's modified functionality.
##########
nifi-nar-bundles/nifi-cdc/nifi-cdc-api/src/main/java/org/apache/nifi/cdc/event/io/EventWriterConfiguration.java:
##########
@@ -0,0 +1,102 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.cdc.event.io;
+
+import com.fasterxml.jackson.core.JsonGenerator;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.ProcessSession;
+
+import java.io.OutputStream;
+
+public class EventWriterConfiguration {
+
+ private FlowFileEventWriteStrategy flowFileEventWriteStrategy;
+ private int numberOfEventsWritten = 0;
+ private int numberOfEventsPerFlowFile = 1000;
+ private FlowFile currentFlowFile;
+ private OutputStream flowFileOutputStream;
+ private ProcessSession workingSession;
+ private JsonGenerator jsonGenerator;
+
+ public EventWriterConfiguration(FlowFileEventWriteStrategy flowFileEventWriteStrategy, int numberOfEventsWritten, int numberOfEventsPerFlowFile, FlowFile currentFlowFile) {
+ this.flowFileEventWriteStrategy = flowFileEventWriteStrategy;
+ this.numberOfEventsWritten = numberOfEventsWritten;
+ this.numberOfEventsPerFlowFile = numberOfEventsPerFlowFile;
+ this.currentFlowFile = currentFlowFile;
+ }
+
+ public FlowFileEventWriteStrategy getFlowFileEventWriteStrategy() {
+ return flowFileEventWriteStrategy;
+ }
+
+ public void setFlowFileEventWriteStrategy(FlowFileEventWriteStrategy flowFileEventWriteStrategy) {
Review Comment:
This method is unused and since the write strategy is coming from a processor property I don't think it should be modifiable after setting it's initial value.
##########
nifi-nar-bundles/nifi-cdc/nifi-cdc-api/src/main/java/org/apache/nifi/cdc/event/io/EventWriterConfiguration.java:
##########
@@ -0,0 +1,102 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.cdc.event.io;
+
+import com.fasterxml.jackson.core.JsonGenerator;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.ProcessSession;
+
+import java.io.OutputStream;
+
+public class EventWriterConfiguration {
+
+ private FlowFileEventWriteStrategy flowFileEventWriteStrategy;
+ private int numberOfEventsWritten = 0;
+ private int numberOfEventsPerFlowFile = 1000;
+ private FlowFile currentFlowFile;
+ private OutputStream flowFileOutputStream;
+ private ProcessSession workingSession;
+ private JsonGenerator jsonGenerator;
+
+ public EventWriterConfiguration(FlowFileEventWriteStrategy flowFileEventWriteStrategy, int numberOfEventsWritten, int numberOfEventsPerFlowFile, FlowFile currentFlowFile) {
Review Comment:
Currently this constructor method is only called with `numberOfEventsWritten=0` and `currentFlowFile=null`, these parameters can be removed from the constructor and have their values initialized at class level.
##########
nifi-nar-bundles/nifi-cdc/nifi-cdc-mysql-bundle/nifi-cdc-mysql-processors/src/main/java/org/apache/nifi/cdc/mysql/processors/CaptureChangeMySQL.java:
##########
@@ -1241,6 +1335,23 @@ protected TableInfo loadTableInfo(TableInfoCacheKey key) throws SQLException {
return tableInfo;
}
+ protected Map<String, String> getCommonAttributes(final long sequenceId, BinlogEventInfo eventInfo) {
Review Comment:
This method is unused.
##########
nifi-nar-bundles/nifi-cdc/nifi-cdc-api/src/main/java/org/apache/nifi/cdc/event/io/EventWriterConfiguration.java:
##########
@@ -0,0 +1,102 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.cdc.event.io;
+
+import com.fasterxml.jackson.core.JsonGenerator;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.ProcessSession;
+
+import java.io.OutputStream;
+
+public class EventWriterConfiguration {
+
+ private FlowFileEventWriteStrategy flowFileEventWriteStrategy;
+ private int numberOfEventsWritten = 0;
+ private int numberOfEventsPerFlowFile = 1000;
+ private FlowFile currentFlowFile;
+ private OutputStream flowFileOutputStream;
+ private ProcessSession workingSession;
+ private JsonGenerator jsonGenerator;
+
+ public EventWriterConfiguration(FlowFileEventWriteStrategy flowFileEventWriteStrategy, int numberOfEventsWritten, int numberOfEventsPerFlowFile, FlowFile currentFlowFile) {
+ this.flowFileEventWriteStrategy = flowFileEventWriteStrategy;
+ this.numberOfEventsWritten = numberOfEventsWritten;
+ this.numberOfEventsPerFlowFile = numberOfEventsPerFlowFile;
+ this.currentFlowFile = currentFlowFile;
+ }
+
+ public FlowFileEventWriteStrategy getFlowFileEventWriteStrategy() {
+ return flowFileEventWriteStrategy;
+ }
+
+ public void setFlowFileEventWriteStrategy(FlowFileEventWriteStrategy flowFileEventWriteStrategy) {
+ this.flowFileEventWriteStrategy = flowFileEventWriteStrategy;
+ }
+
+ public int getNumberOfEventsWritten() {
+ return numberOfEventsWritten;
+ }
+
+ public void setNumberOfEventsWritten(int numberOfEventsWritten) {
+ this.numberOfEventsWritten = numberOfEventsWritten;
+ }
+
+ public void incrementNumberOfEventsWritten() {
+ this.numberOfEventsWritten++;
+ }
+
+ public int getNumberOfEventsPerFlowFile() {
+ return numberOfEventsPerFlowFile;
+ }
+
+ public void setNumberOfEventsPerFlowFile(int numberOfEventsPerFlowFile) {
Review Comment:
Same as the write strategy field. This method is unused and since this is coming from a processor property I don't think it should be modifiable after setting it's initial value.
##########
nifi-nar-bundles/nifi-cdc/nifi-cdc-mysql-bundle/nifi-cdc-mysql-processors/src/main/java/org/apache/nifi/cdc/mysql/event/io/AbstractBinlogEventWriter.java:
##########
@@ -44,7 +49,7 @@ protected void writeJson(T event) throws IOException {
}
protected Map<String, String> getCommonAttributes(final long sequenceId, BinlogEventInfo eventInfo) {
- return new HashMap<String, String>() {
Review Comment:
This is causing compilation problem with java8.
--
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
[GitHub] [nifi] mattyb149 commented on a diff in pull request #6907: NIFI-11094: Allow CaptureChangeMySQL to send multiple events per FlowFile
Posted by "mattyb149 (via GitHub)" <gi...@apache.org>.
mattyb149 commented on code in PR #6907:
URL: https://github.com/apache/nifi/pull/6907#discussion_r1106117706
##########
nifi-nar-bundles/nifi-cdc/nifi-cdc-mysql-bundle/nifi-cdc-mysql-processors/src/main/java/org/apache/nifi/cdc/mysql/processors/CaptureChangeMySQL.java:
##########
@@ -185,6 +194,12 @@ public class CaptureChangeMySQL extends AbstractSessionFactoryProcessor {
SSLMode.VERIFY_IDENTITY.toString(),
"Connect with TLS or fail when server support not enabled. Verify server hostname matches presented X.509 certificate names or fail when not matched");
+ private static final AllowableValue N_EVENTS_PER_FLOWFILE_STRATEGY = new AllowableValue(FlowFileEventWriteStrategy.N_EVENTS_PER_FLOWFILE.name(), "N Events Per FlowFile",
+ "This strategy causes the number of events specified in the Events per FlowFile each binlog event to be written to its own FlowFile");
Review Comment:
Good point, will add that to the doc
--
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
[GitHub] [nifi] tpalfy commented on pull request #6907: NIFI-11094: Allow CaptureChangeMySQL to send multiple events per FlowFile
Posted by "tpalfy (via GitHub)" <gi...@apache.org>.
tpalfy commented on PR #6907:
URL: https://github.com/apache/nifi/pull/6907#issuecomment-1440573119
LGTM
Thank you for your work @mattyb149 and your reviews @exceptionfactory and @mark-bathori !
Merged into main.
--
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
[GitHub] [nifi] tpalfy commented on a diff in pull request #6907: NIFI-11094: Allow CaptureChangeMySQL to send multiple events per FlowFile
Posted by "tpalfy (via GitHub)" <gi...@apache.org>.
tpalfy commented on code in PR #6907:
URL: https://github.com/apache/nifi/pull/6907#discussion_r1100221116
##########
nifi-nar-bundles/nifi-cdc/nifi-cdc-mysql-bundle/nifi-cdc-mysql-processors/src/main/java/org/apache/nifi/cdc/mysql/processors/CaptureChangeMySQL.java:
##########
@@ -185,6 +194,12 @@ public class CaptureChangeMySQL extends AbstractSessionFactoryProcessor {
SSLMode.VERIFY_IDENTITY.toString(),
"Connect with TLS or fail when server support not enabled. Verify server hostname matches presented X.509 certificate names or fail when not matched");
+ private static final AllowableValue N_EVENTS_PER_FLOWFILE_STRATEGY = new AllowableValue(FlowFileEventWriteStrategy.N_EVENTS_PER_FLOWFILE.name(), "N Events Per FlowFile",
+ "This strategy causes the number of events specified in the Events per FlowFile each binlog event to be written to its own FlowFile");
Review Comment:
```suggestion
"This strategy causes the number of events specified in the 'Number of Events Per FlowFile' property to be written per FlowFile");
```
Btw, don't we want to let the processor to write _fewer_ events than specified (if only that much is available during a single `onTrigger` run)?
##########
nifi-nar-bundles/nifi-cdc/nifi-cdc-mysql-bundle/nifi-cdc-mysql-processors/src/main/java/org/apache/nifi/cdc/mysql/event/io/AbstractBinlogEventWriter.java:
##########
@@ -44,7 +49,7 @@ protected void writeJson(T event) throws IOException {
}
protected Map<String, String> getCommonAttributes(final long sequenceId, BinlogEventInfo eventInfo) {
- return new HashMap<String, String>() {
+ return new HashMap<>() {
Review Comment:
This change doesn't work on Java 8.
##########
nifi-nar-bundles/nifi-cdc/nifi-cdc-mysql-bundle/nifi-cdc-mysql-processors/src/main/java/org/apache/nifi/cdc/mysql/processors/CaptureChangeMySQL.java:
##########
@@ -1017,15 +1096,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));
Review Comment:
We are in a block that only runs when the emission of BEGIN and COMMIT is requested by the user (on the processor).
That means that this does NOT run when it is not requested, and in that case the `commitEventWriter.writeEvent` wont run and thus the flowfile will not be sent sout.
In short, it doesn't work when the user wants ONE_TRANSACTION_PER_FLOWFILE but doesn't want BEGIN and COMMIT events.
##########
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:
The process session is committed only after a transaction which is somewhat desirable. But because of this the N events per flowfile can hold back a flowfile with N events already in it until a commit comes. But the commit may never actually come.
Consider the following case:
- N=5
- 1st transaction inserts 4 files
- 2nd transaction inserts 2 files.
- When the event for the first insert of the 2nd transaction is received we finish up the current flowfile (has 4 events from the 1st transaction and 1 event from the 2nd transaction). However we don't commit the process session - we are not handling a commit event! - so the flowfile is held back.
- When the commit for the 2nd transaction is received we have only 1 event in the flowfile and it is still worked on (not null). So we don't commit the process session.
- A bunch of flowfiles can queue up this way until we reach a point where the commit comes when the flowfile has exactly N events or when we stop the processor.
--
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
[GitHub] [nifi] mattyb149 commented on a diff in pull request #6907: NIFI-11094: Allow CaptureChangeMySQL to send multiple events per FlowFile
Posted by "mattyb149 (via GitHub)" <gi...@apache.org>.
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org