Merge pull request confluentinc#22 from confluentinc/nullable-column-check

Add check for NOT NULL columns on target tables when specified as increasing or timestamp column for offsets.

Author: gwenshap

src/main/java/io/confluent/connect/jdbc/JdbcSourceConnectorConfig.java

@@ -64,12 +64,14 @@ public class JdbcSourceConnectorConfig extends AbstractConfig {
   public static final String INCREASING_COLUMN_NAME_CONFIG = "increasing.column.name";
   private static final String INCREASING_COLUMN_NAME_DOC =
       "The name of the strictly increasing column to use to detect new rows. Any empty value "
-      + "indicates the column should be autodetected by looking for an auto-incrementing column.";
+      + "indicates the column should be autodetected by looking for an auto-incrementing column. "
+      + "This column may not be nullable.";
   public static final String INCREASING_COLUMN_NAME_DEFAULT = "";
 
   public static final String TIMESTAMP_COLUMN_NAME_CONFIG = "timestamp.column.name";
   private static final String TIMESTAMP_COLUMN_NAME_DOC =
-      "The name of the timestamp column to use to detect new or modified rows.";
+      "The name of the timestamp column to use to detect new or modified rows. This column may "
+      + "not be nullable.";
   public static final String TIMESTAMP_COLUMN_NAME_DEFAULT = "";
 
   public static final String TABLE_POLL_INTERVAL_MS_CONFIG = "table.poll.interval.ms";

src/main/java/io/confluent/connect/jdbc/JdbcSourceTask.java

@@ -108,14 +108,27 @@ public void start(Map<String, String> properties) {
       offsets = context.offsetStorageReader().offsets(partitions);
     }
 
+    // Must setup the connection now to validate NOT NULL columns. At this point we've already
+    // caught any easy-to-find errors so deferring the connection creation won't save any effort
+    String dbUrl = config.getString(JdbcSourceTaskConfig.CONNECTION_URL_CONFIG);
+    log.debug("Trying to connect to {}", dbUrl);
+    try {
+      db = DriverManager.getConnection(dbUrl);
+    } catch (SQLException e) {
+      log.error("Couldn't open connection to {}: {}", dbUrl, e);
+      throw new ConnectException(e);
+    }
+
     String increasingColumn
         = config.getString(JdbcSourceTaskConfig.INCREASING_COLUMN_NAME_CONFIG);
     String timestampColumn
         = config.getString(JdbcSourceTaskConfig.TIMESTAMP_COLUMN_NAME_CONFIG);
+
     for (String tableOrQuery : tablesOrQuery) {
       final Map<String, String> partition;
       switch (queryMode) {
         case TABLE:
+          validateNonNullable(mode, tableOrQuery, increasingColumn, timestampColumn);
           partition = Collections.singletonMap(
               JdbcSourceConnectorConstants.TABLE_NAME_KEY, tableOrQuery);
           break;
@@ -149,15 +162,6 @@ public void start(Map<String, String> properties) {
       }
     }
 
-    String dbUrl = config.getString(JdbcSourceTaskConfig.CONNECTION_URL_CONFIG);
-    log.debug("Trying to connect to {}", dbUrl);
-    try {
-      db = DriverManager.getConnection(dbUrl);
-    } catch (SQLException e) {
-      log.error("Couldn't open connection to {}: {}", dbUrl, e);
-      throw new ConnectException(e);
-    }
-
     stop = new AtomicBoolean(false);
   }
 
@@ -234,4 +238,29 @@ public List<SourceRecord> poll() throws InterruptedException {
     return null;
   }
 
+  private void validateNonNullable(String incrementalMode, String table, String increasingColumn,
+                                   String timestampColumn) {
+    try {
+      // Validate that requested columns for offsets are NOT NULL. Currently this is only performed
+      // for table-based copying because custom query mode doesn't allow this to be looked up
+      // without a query or parsing the query since we don't have a table name.
+      if ((incrementalMode.equals(JdbcSourceConnectorConfig.MODE_INCREASING) ||
+           incrementalMode.equals(JdbcSourceConnectorConfig.MODE_TIMESTAMP_INCREASING)) &&
+          JdbcUtils.isColumnNullable(db, table, increasingColumn)) {
+        throw new ConnectException("Cannot make incremental queries using incrementing column " +
+                                   increasingColumn + " on " + table + " because this column is "
+                                   + "nullable.");
+      }
+      if ((incrementalMode.equals(JdbcSourceConnectorConfig.MODE_TIMESTAMP) ||
+           incrementalMode.equals(JdbcSourceConnectorConfig.MODE_TIMESTAMP_INCREASING)) &&
+          JdbcUtils.isColumnNullable(db, table, timestampColumn)) {
+        throw new ConnectException("Cannot make incremental queries using timestamp column " +
+                                   timestampColumn + " on " + table + " because this column is "
+                                   + "nullable.");
+      }
+    } catch (SQLException e) {
+      throw new ConnectException("Failed trying to validate that columns used for offsets are NOT"
+                                 + " NULL", e);
+    }
+  }
 }

src/main/java/io/confluent/connect/jdbc/JdbcUtils.java

@@ -55,6 +55,7 @@ public class JdbcUtils {
   private static final int GET_TABLES_NAME_COLUMN = 3;
 
   private static final int GET_COLUMNS_COLUMN_NAME = 4;
+  private static final int GET_COLUMNS_IS_NULLABLE = 18;
   private static final int GET_COLUMNS_IS_AUTOINCREMENT = 23;
 
 
@@ -147,6 +148,21 @@ public static String getAutoincrementColumn(Connection conn, String table) throw
     return (matches == 1 ? result : null);
   }
 
+  public static boolean isColumnNullable(Connection conn, String table, String column)
+      throws SQLException {
+    ResultSet rs = conn.getMetaData().getColumns(null, null, table, column);
+    if (rs.getMetaData().getColumnCount() > GET_COLUMNS_IS_NULLABLE) {
+      // Should only be one match
+      if (!rs.next()) {
+        return false;
+      }
+      String val = rs.getString(GET_COLUMNS_IS_NULLABLE);
+      return rs.getString(GET_COLUMNS_IS_NULLABLE).equals("YES");
+    }
+
+    return false;
+  }
+
   /**
    * Format the given Date assuming UTC timezone in a format supported by SQL.
    * @param date the date to convert to a String

src/test/java/io/confluent/connect/jdbc/JdbcSourceTaskUpdateTest.java

@@ -17,6 +17,7 @@
 package io.confluent.connect.jdbc;
 
 import org.apache.kafka.connect.data.Struct;
+import org.apache.kafka.connect.errors.ConnectException;
 import org.apache.kafka.connect.source.SourceRecord;
 import org.junit.After;
 import org.junit.Test;
@@ -75,9 +76,36 @@ public void testBulkPeriodicLoad() throws Exception {
     assertRecordsTopic(records, TOPIC_PREFIX + SINGLE_TABLE_NAME);
   }
 
+  @Test(expected = ConnectException.class)
+  public void testIncreasingInvalidColumn() throws Exception {
+    expectInitializeNoOffsets(Arrays.asList(SINGLE_TABLE_PARTITION));
+
+    PowerMock.replayAll();
+
+    // Increasing column must be NOT NULL
+    db.createTable(SINGLE_TABLE_NAME, "id", "INT");
+
+    startTask(null, "id", null);
+
+    PowerMock.verifyAll();
+  }
+
+  @Test(expected = ConnectException.class)
+  public void testTimestampInvalidColumn() throws Exception {
+    expectInitializeNoOffsets(Arrays.asList(SINGLE_TABLE_PARTITION));
+
+    PowerMock.replayAll();
+
+    // Timestamp column must be NOT NULL
+    db.createTable(SINGLE_TABLE_NAME, "modified", "TIMESTAMP");
+
+    startTask("modified", null, null);
+
+    PowerMock.verifyAll();
+  }
+
   @Test
   public void testManualIncreasing() throws Exception {
-
     expectInitializeNoOffsets(Arrays.asList(SINGLE_TABLE_PARTITION));
 
     PowerMock.replayAll();

src/test/java/io/confluent/connect/jdbc/JdbcUtilsTest.java

@@ -25,7 +25,9 @@
 import java.util.HashSet;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
 
 public class JdbcUtilsTest {
 
@@ -84,4 +86,19 @@ public void testGetAutoincrement() throws Exception {
                    "bar", "INTEGER");
     assertEquals("id", JdbcUtils.getAutoincrementColumn(db.getConnection(), "mixed"));
   }
+
+  @Test
+  public void testIsColumnNullable() throws Exception {
+    db.createTable("test", "id", "INTEGER NOT NULL GENERATED ALWAYS AS IDENTITY", "bar", "INTEGER");
+    assertFalse(JdbcUtils.isColumnNullable(db.getConnection(), "test", "id"));
+    assertTrue(JdbcUtils.isColumnNullable(db.getConnection(), "test", "bar"));
+
+    // Derby does not seem to allow null
+    db.createTable("tstest", "ts", "TIMESTAMP NOT NULL", "tsdefault", "TIMESTAMP",
+                   "tsnull", "TIMESTAMP DEFAULT NULL");
+    assertFalse(JdbcUtils.isColumnNullable(db.getConnection(), "tstest", "ts"));
+    // The default for TIMESTAMP columns can vary between databases, but for Derby it is nullable
+    assertTrue(JdbcUtils.isColumnNullable(db.getConnection(), "tstest", "tsdefault"));
+    assertTrue(JdbcUtils.isColumnNullable(db.getConnection(), "tstest", "tsnull"));
+  }
 }