You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by pv...@apache.org on 2017/11/02 08:17:26 UTC
nifi git commit: NIFI-4490: Ensure driver settings are correct before
connecting binlog
Repository: nifi
Updated Branches:
refs/heads/master 9a850c7ed -> 641333791
NIFI-4490: Ensure driver settings are correct before connecting binlog
Signed-off-by: Pierre Villard <pi...@gmail.com>
This closes #2244.
Project: http://git-wip-us.apache.org/repos/asf/nifi/repo
Commit: http://git-wip-us.apache.org/repos/asf/nifi/commit/64133379
Tree: http://git-wip-us.apache.org/repos/asf/nifi/tree/64133379
Diff: http://git-wip-us.apache.org/repos/asf/nifi/diff/64133379
Branch: refs/heads/master
Commit: 64133379188eabbb672f695b2ab9c168f49c10b4
Parents: 9a850c7
Author: Koji Kawamura <ij...@apache.org>
Authored: Wed Nov 1 21:28:04 2017 +0900
Committer: Pierre Villard <pi...@gmail.com>
Committed: Thu Nov 2 09:16:56 2017 +0100
----------------------------------------------------------------------
.../mysql/processors/CaptureChangeMySQL.java | 34 +++++++++++++++-----
.../processors/CaptureChangeMySQLTest.groovy | 4 +++
2 files changed, 30 insertions(+), 8 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/nifi/blob/64133379/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 f7f075e..5ee51fd 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
@@ -646,6 +646,19 @@ public class CaptureChangeMySQL extends AbstractSessionFactoryProcessor {
InetSocketAddress connectedHost = null;
Exception lastConnectException = new Exception("Unknown connection error");
+ if (createEnrichmentConnection) {
+ try {
+ // Ensure driverLocation and driverName are correct before establishing binlog connection
+ // to avoid failing after binlog messages are received.
+ // Actual JDBC connection is created after binlog client gets started, because we need
+ // the connect-able host same as the binlog client.
+ registerDriver(driverLocation, driverName);
+ } catch (InitializationException e) {
+ throw new RuntimeException("Failed to register JDBC driver. Ensure MySQL Driver Location(s)" +
+ " and MySQL Driver Class Name are configured correctly. " + e, e);
+ }
+ }
+
while (connectedHost == null && connectionAttempts < numHosts) {
if (binlogClient == null) {
@@ -1009,6 +1022,17 @@ public class CaptureChangeMySQL extends AbstractSessionFactoryProcessor {
*/
protected Connection getJdbcConnection(String locationString, String drvName, InetSocketAddress host, String username, String password, Map<String, String> customProperties)
throws InitializationException, SQLException {
+ Properties connectionProps = new Properties();
+ if (customProperties != null) {
+ connectionProps.putAll(customProperties);
+ }
+ connectionProps.put("user", username);
+ connectionProps.put("password", password);
+
+ return DriverManager.getConnection("jdbc:mysql://" + host.getHostString() + ":" + host.getPort(), connectionProps);
+ }
+
+ protected void registerDriver(String locationString, String drvName) throws InitializationException {
if (locationString != null && locationString.length() > 0) {
try {
// Split and trim the entries
@@ -1027,20 +1051,14 @@ public class CaptureChangeMySQL extends AbstractSessionFactoryProcessor {
final Driver driver = (Driver) clazz.newInstance();
DriverManager.registerDriver(new DriverShim(driver));
+ } catch (final InitializationException e) {
+ throw e;
} catch (final MalformedURLException e) {
throw new InitializationException("Invalid Database Driver Jar Url", e);
} catch (final Exception e) {
throw new InitializationException("Can't load Database Driver", e);
}
}
- Properties connectionProps = new Properties();
- if (customProperties != null) {
- connectionProps.putAll(customProperties);
- }
- connectionProps.put("user", username);
- connectionProps.put("password", password);
-
- return DriverManager.getConnection("jdbc:mysql://" + host.getHostString() + ":" + host.getPort(), connectionProps);
}
private static class DriverShim implements Driver {
http://git-wip-us.apache.org/repos/asf/nifi/blob/64133379/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 f876b6a..7e8607d 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
@@ -963,6 +963,10 @@ class CaptureChangeMySQLTest {
}
@Override
+ protected void registerDriver(String locationString, String drvName) throws InitializationException {
+ }
+
+ @Override
protected Connection getJdbcConnection(String locationString, String drvName, InetSocketAddress host, String username, String password, Map<String, String> customProperties)
throws InitializationException, SQLException {
Connection mockConnection = mock(Connection)