Repository: nifi Updated Branches: refs/heads/master 50ea1083e -> ba2bdf858
NIFI-3730: Allow exclusion of BEGIN/COMMIT event output in CaptureChangeMySQL This closes #1689. Signed-off-by: Koji Kawamura <[email protected]> Project: http://git-wip-us.apache.org/repos/asf/nifi/repo Commit: http://git-wip-us.apache.org/repos/asf/nifi/commit/ba2bdf85 Tree: http://git-wip-us.apache.org/repos/asf/nifi/tree/ba2bdf85 Diff: http://git-wip-us.apache.org/repos/asf/nifi/diff/ba2bdf85 Branch: refs/heads/master Commit: ba2bdf858672e935da60861cfc412e2b55f35652 Parents: 50ea108 Author: Matt Burgess <[email protected]> Authored: Mon Apr 24 09:23:02 2017 -0400 Committer: Koji Kawamura <[email protected]> Committed: Tue Apr 25 14:40:24 2017 +0900 ---------------------------------------------------------------------- .../mysql/processors/CaptureChangeMySQL.java | 37 ++++++++++--- .../processors/CaptureChangeMySQLTest.groovy | 56 ++++++++++++++++++++ 2 files changed, 85 insertions(+), 8 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/nifi/blob/ba2bdf85/nifi-nar-bundles/nifi-cdc/nifi-cdc-mysql-bundle/nifi-cdc-mysql-processors/src/main/java/org/apache/nifi/cdc/mysql/processors/CaptureChangeMySQL.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-cdc/nifi-cdc-mysql-bundle/nifi-cdc-mysql-processors/src/main/java/org/apache/nifi/cdc/mysql/processors/CaptureChangeMySQL.java b/nifi-nar-bundles/nifi-cdc/nifi-cdc-mysql-bundle/nifi-cdc-mysql-processors/src/main/java/org/apache/nifi/cdc/mysql/processors/CaptureChangeMySQL.java index 726d7c8..2e0cfea 100644 --- a/nifi-nar-bundles/nifi-cdc/nifi-cdc-mysql-bundle/nifi-cdc-mysql-processors/src/main/java/org/apache/nifi/cdc/mysql/processors/CaptureChangeMySQL.java +++ b/nifi-nar-bundles/nifi-cdc/nifi-cdc-mysql-bundle/nifi-cdc-mysql-processors/src/main/java/org/apache/nifi/cdc/mysql/processors/CaptureChangeMySQL.java @@ -125,9 +125,9 @@ import static com.github.shyiko.mysql.binlog.event.EventType.WRITE_ROWS; @Stateful(scopes = Scope.CLUSTER, description = "Information such as a 'pointer' to the current CDC event in the database is stored by this processor, such " + "that it can continue from the same location if restarted.") @WritesAttributes({ - @WritesAttribute(attribute = "cdc.sequence.id", description = "A sequence identifier (i.e. strictly increasing integer value) specifying the order " + @WritesAttribute(attribute = EventWriter.SEQUENCE_ID_KEY, description = "A sequence identifier (i.e. strictly increasing integer value) specifying the order " + "of the CDC event flow file relative to the other event flow file(s)."), - @WritesAttribute(attribute = "cdc.event.type", description = "A string indicating the type of CDC event that occurred, including (but not limited to) " + @WritesAttribute(attribute = EventWriter.CDC_EVENT_TYPE_ATTRIBUTE, description = "A string indicating the type of CDC event that occurred, including (but not limited to) " + "'begin', 'insert', 'update', 'delete', 'schema_change' and 'commit'."), @WritesAttribute(attribute = "mime.type", description = "The processor outputs flow file content in JSON format, and sets the mime.type attribute to " + "application/json") @@ -263,6 +263,17 @@ public class CaptureChangeMySQL extends AbstractSessionFactoryProcessor { .addValidator(StandardValidators.BOOLEAN_VALIDATOR) .build(); + public static final PropertyDescriptor INCLUDE_BEGIN_COMMIT = new PropertyDescriptor.Builder() + .name("capture-change-mysql-include-begin-commit") + .displayName("Include Begin/Commit Events") + .description("Specifies whether to emit events corresponding to a BEGIN or COMMIT event in the binary log. Set to true if the BEGIN/COMMIT events are necessary in the downstream flow, " + + "otherwise set to false, which suppresses generation of these events and can increase flow performance.") + .required(true) + .allowableValues("true", "false") + .defaultValue("false") + .addValidator(StandardValidators.BOOLEAN_VALIDATOR) + .build(); + public static final PropertyDescriptor STATE_UPDATE_INTERVAL = new PropertyDescriptor.Builder() .name("capture-change-mysql-state-update-interval") .displayName("State Update Interval") @@ -331,6 +342,7 @@ public class CaptureChangeMySQL extends AbstractSessionFactoryProcessor { private volatile TableInfo currentTable = null; private volatile Pattern databaseNamePattern; private volatile Pattern tableNamePattern; + private volatile boolean includeBeginCommit = false; private volatile boolean inTransaction = false; private volatile boolean skipTable = false; @@ -376,6 +388,7 @@ public class CaptureChangeMySQL extends AbstractSessionFactoryProcessor { pds.add(CONNECT_TIMEOUT); pds.add(DIST_CACHE_CLIENT); pds.add(RETRIEVE_ALL_RECORDS); + pds.add(INCLUDE_BEGIN_COMMIT); pds.add(STATE_UPDATE_INTERVAL); pds.add(INIT_SEQUENCE_ID); pds.add(INIT_BINLOG_FILENAME); @@ -420,6 +433,8 @@ public class CaptureChangeMySQL extends AbstractSessionFactoryProcessor { boolean getAllRecords = context.getProperty(RETRIEVE_ALL_RECORDS).asBoolean(); + includeBeginCommit = context.getProperty(INCLUDE_BEGIN_COMMIT).asBoolean(); + // Set current binlog filename to whatever is in State, falling back to the Retrieve All Records then Initial Binlog Filename if no State variable is present currentBinlogFile = stateMap.get(BinlogEventInfo.BINLOG_FILENAME_KEY); if (currentBinlogFile == null) { @@ -741,8 +756,10 @@ public class CaptureChangeMySQL extends AbstractSessionFactoryProcessor { xactBinlogPosition = currentBinlogPosition; xactSequenceId = currentSequenceId.get(); - BeginTransactionEventInfo beginEvent = new BeginTransactionEventInfo(timestamp, currentBinlogFile, currentBinlogPosition); - currentSequenceId.set(beginEventWriter.writeEvent(currentSession, transitUri, beginEvent, currentSequenceId.get(), REL_SUCCESS)); + if (includeBeginCommit) { + BeginTransactionEventInfo beginEvent = new BeginTransactionEventInfo(timestamp, currentBinlogFile, currentBinlogPosition); + currentSequenceId.set(beginEventWriter.writeEvent(currentSession, transitUri, beginEvent, currentSequenceId.get(), REL_SUCCESS)); + } inTransaction = true; } else if ("COMMIT".equals(sql)) { if (!inTransaction) { @@ -750,8 +767,10 @@ public class CaptureChangeMySQL extends AbstractSessionFactoryProcessor { + "This could indicate that your binlog position is invalid."); } // InnoDB generates XID events for "commit", but MyISAM generates Query events with "COMMIT", so handle that here - CommitTransactionEventInfo commitTransactionEvent = new CommitTransactionEventInfo(timestamp, currentBinlogFile, currentBinlogPosition); - currentSequenceId.set(commitEventWriter.writeEvent(currentSession, transitUri, commitTransactionEvent, currentSequenceId.get(), REL_SUCCESS)); + if (includeBeginCommit) { + CommitTransactionEventInfo commitTransactionEvent = new CommitTransactionEventInfo(timestamp, currentBinlogFile, currentBinlogPosition); + currentSequenceId.set(commitEventWriter.writeEvent(currentSession, transitUri, commitTransactionEvent, currentSequenceId.get(), REL_SUCCESS)); + } // Commit the NiFi session session.commit(); inTransaction = false; @@ -780,8 +799,10 @@ public class CaptureChangeMySQL extends AbstractSessionFactoryProcessor { throw new IOException("COMMIT event received while not processing a transaction (i.e. no corresponding BEGIN event). " + "This could indicate that your binlog position is invalid."); } - CommitTransactionEventInfo commitTransactionEvent = new CommitTransactionEventInfo(timestamp, currentBinlogFile, currentBinlogPosition); - currentSequenceId.set(commitEventWriter.writeEvent(currentSession, transitUri, commitTransactionEvent, currentSequenceId.get(), REL_SUCCESS)); + if (includeBeginCommit) { + CommitTransactionEventInfo commitTransactionEvent = new CommitTransactionEventInfo(timestamp, currentBinlogFile, currentBinlogPosition); + currentSequenceId.set(commitEventWriter.writeEvent(currentSession, transitUri, commitTransactionEvent, currentSequenceId.get(), REL_SUCCESS)); + } // Commit the NiFi session session.commit(); inTransaction = false; http://git-wip-us.apache.org/repos/asf/nifi/blob/ba2bdf85/nifi-nar-bundles/nifi-cdc/nifi-cdc-mysql-bundle/nifi-cdc-mysql-processors/src/test/groovy/org/apache/nifi/cdc/mysql/processors/CaptureChangeMySQLTest.groovy ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-cdc/nifi-cdc-mysql-bundle/nifi-cdc-mysql-processors/src/test/groovy/org/apache/nifi/cdc/mysql/processors/CaptureChangeMySQLTest.groovy b/nifi-nar-bundles/nifi-cdc/nifi-cdc-mysql-bundle/nifi-cdc-mysql-processors/src/test/groovy/org/apache/nifi/cdc/mysql/processors/CaptureChangeMySQLTest.groovy index 7273199..22959e6 100644 --- a/nifi-nar-bundles/nifi-cdc/nifi-cdc-mysql-bundle/nifi-cdc-mysql-processors/src/test/groovy/org/apache/nifi/cdc/mysql/processors/CaptureChangeMySQLTest.groovy +++ b/nifi-nar-bundles/nifi-cdc/nifi-cdc-mysql-bundle/nifi-cdc-mysql-processors/src/test/groovy/org/apache/nifi/cdc/mysql/processors/CaptureChangeMySQLTest.groovy @@ -94,6 +94,7 @@ class CaptureChangeMySQLTest { testRunner.setProperty(CaptureChangeMySQL.HOSTS, 'localhost:3306') testRunner.setProperty(CaptureChangeMySQL.USERNAME, 'root') testRunner.setProperty(CaptureChangeMySQL.CONNECT_TIMEOUT, '2 seconds') + testRunner.setProperty(CaptureChangeMySQL.INCLUDE_BEGIN_COMMIT, 'true') testRunner.run(1, false, true) @@ -116,6 +117,58 @@ class CaptureChangeMySQLTest { )) testRunner.run(1, true, false) + + def resultFiles = testRunner.getFlowFilesForRelationship(CaptureChangeMySQL.REL_SUCCESS) + assertEquals(2, resultFiles.size()) + } + + @Test + void testBeginCommitTransactionFiltered() throws Exception { + testRunner.setProperty(CaptureChangeMySQL.DRIVER_LOCATION, 'file:///path/to/mysql-connector-java-5.1.38-bin.jar') + testRunner.setProperty(CaptureChangeMySQL.HOSTS, 'localhost:3306') + testRunner.setProperty(CaptureChangeMySQL.USERNAME, 'root') + testRunner.setProperty(CaptureChangeMySQL.CONNECT_TIMEOUT, '2 seconds') + testRunner.setProperty(CaptureChangeMySQL.INCLUDE_BEGIN_COMMIT, 'false') + testRunner.setProperty(CaptureChangeMySQL.INIT_SEQUENCE_ID, '10') + + testRunner.run(1, false, true) + + // ROTATE + client.sendEvent(new Event( + [timestamp: new Date().time, eventType: EventType.ROTATE, nextPosition: 2] as EventHeaderV4, + [binlogFilename: 'master.000001', binlogPosition: 4L] as RotateEventData + )) + + // BEGIN + client.sendEvent(new Event( + [timestamp: new Date().time, eventType: EventType.QUERY, nextPosition: 4] as EventHeaderV4, + [database: 'myDB', sql: 'BEGIN'] as QueryEventData + )) + + client.sendEvent(new Event( + [timestamp: new Date().time, eventType: EventType.TABLE_MAP, nextPosition: 6] as EventHeaderV4, + [tableId: 1, database: 'myDB', table: 'myTable', columnTypes: [4, -4] as byte[]] as TableMapEventData + )) + + def cols = new BitSet() + cols.set(1) + client.sendEvent(new Event( + [timestamp: new Date().time, eventType: EventType.EXT_WRITE_ROWS, nextPosition: 8] as EventHeaderV4, + [tableId: 1, includedColumns: cols, + rows : [[2, 'Smith'] as Serializable[]] as List<Serializable[]>] as WriteRowsEventData + )) + + // COMMIT + client.sendEvent(new Event( + [timestamp: new Date().time, eventType: EventType.XID, nextPosition: 12] as EventHeaderV4, + {} as EventData + )) + + testRunner.run(1, true, false) + + def resultFiles = testRunner.getFlowFilesForRelationship(CaptureChangeMySQL.REL_SUCCESS) + assertEquals(1, resultFiles.size()) + assertEquals('10', resultFiles[0].getAttribute(EventWriter.SEQUENCE_ID_KEY)) } @Test @@ -226,6 +279,7 @@ class CaptureChangeMySQLTest { testRunner.setProperty(CaptureChangeMySQL.CONNECT_TIMEOUT, '2 seconds') testRunner.setProperty(CaptureChangeMySQL.INIT_BINLOG_FILENAME, 'master.000001') testRunner.setProperty(CaptureChangeMySQL.INIT_BINLOG_POSITION, '4') + testRunner.setProperty(CaptureChangeMySQL.INCLUDE_BEGIN_COMMIT, 'true') final DistributedMapCacheClientImpl cacheClient = createCacheClient() def clientProperties = [:] clientProperties.put(DistributedMapCacheClientService.HOSTNAME.getName(), 'localhost') @@ -415,6 +469,7 @@ class CaptureChangeMySQLTest { testRunner.setProperty(CaptureChangeMySQL.CONNECT_TIMEOUT, '2 seconds') testRunner.setProperty(CaptureChangeMySQL.DATABASE_NAME_PATTERN, "myDB") testRunner.setProperty(CaptureChangeMySQL.TABLE_NAME_PATTERN, "user") + testRunner.setProperty(CaptureChangeMySQL.INCLUDE_BEGIN_COMMIT, 'true') testRunner.run(1, false, true) @@ -511,6 +566,7 @@ class CaptureChangeMySQLTest { testRunner.setProperty(CaptureChangeMySQL.USERNAME, 'root') testRunner.setProperty(CaptureChangeMySQL.PASSWORD, 'password') testRunner.setProperty(CaptureChangeMySQL.CONNECT_TIMEOUT, '2 seconds') + testRunner.setProperty(CaptureChangeMySQL.INCLUDE_BEGIN_COMMIT, 'true') testRunner.run(1, false, true)
