Add logging around which DbDialect gets used

Author: shikhar

src/main/java/io/confluent/connect/jdbc/sink/JdbcDbWriter.java

@@ -41,10 +41,10 @@ public class JdbcDbWriter {
 
   Connection connection;
 
-  JdbcDbWriter(final JdbcSinkConfig config) {
+  JdbcDbWriter(final JdbcSinkConfig config, DbDialect dbDialect, DbStructure dbStructure) {
     this.config = config;
-    dbDialect = DbDialect.fromConnectionString(config.connectionUrl);
-    dbStructure = new DbStructure(dbDialect);
+    this.dbDialect = dbDialect;
+    this.dbStructure = dbStructure;
   }
 
   void write(final Collection<SinkRecord> records) throws SQLException {

src/main/java/io/confluent/connect/jdbc/sink/JdbcSinkTask.java

@@ -29,6 +29,8 @@
 import java.util.Collection;
 import java.util.Map;
 
+import io.confluent.connect.jdbc.sink.dialect.DbDialect;
+
 public class JdbcSinkTask extends SinkTask {
   private static final Logger log = LoggerFactory.getLogger(JdbcSinkTask.class);
 
@@ -40,10 +42,17 @@ public class JdbcSinkTask extends SinkTask {
   public void start(final Map<String, String> props) {
     log.info("Starting task");
     config = new JdbcSinkConfig(props);
-    writer = new JdbcDbWriter(config);
+    initWriter();
     remainingRetries = config.maxRetries;
   }
 
+  private void initWriter() {
+    final DbDialect dbDialect = DbDialect.fromConnectionString(config.connectionUrl);
+    final DbStructure dbStructure = new DbStructure(dbDialect);
+    log.info("Initializing writer using SQL dialect: {}", dbDialect.getClass().getSimpleName());
+    writer = new JdbcDbWriter(config, dbDialect, dbStructure);
+  }
+
   @Override
   public void put(Collection<SinkRecord> records) {
     if (records.isEmpty()) {
@@ -61,7 +70,7 @@ public void put(Collection<SinkRecord> records) {
         throw new ConnectException(sqle);
       } else {
         writer.closeQuietly();
-        writer = new JdbcDbWriter(config);
+        initWriter();
         remainingRetries--;
         context.timeout(config.retryBackoffMs);
         throw new RetriableException(sqle);

src/test/java/io/confluent/connect/jdbc/sink/JdbcDbWriterTest.java

@@ -34,6 +34,8 @@
 import java.util.Map;
 import java.util.concurrent.ThreadLocalRandom;
 
+import io.confluent.connect.jdbc.sink.dialect.DbDialect;
+import io.confluent.connect.jdbc.sink.dialect.SqliteDialect;
 import io.confluent.connect.jdbc.sink.metadata.DbTable;
 
 import static junit.framework.TestCase.assertTrue;
@@ -53,6 +55,13 @@ public void tearDown() throws IOException, SQLException {
     sqliteHelper.tearDown();
   }
 
+  private JdbcDbWriter newWriter(Map<String, String> props) {
+    final JdbcSinkConfig config = new JdbcSinkConfig(props);
+    final DbDialect dbDialect = new SqliteDialect();
+    final DbStructure dbStructure = new DbStructure(dbDialect);
+    return new JdbcDbWriter(config, dbDialect, dbStructure);
+  }
+
   @Test
   public void autoCreateWithAutoEvolve() throws SQLException {
     String topic = "books";
@@ -64,7 +73,7 @@ public void autoCreateWithAutoEvolve() throws SQLException {
     props.put("pk.mode", "record_key");
     props.put("pk.fields", "id"); // assigned name for the primitive key
 
-    JdbcDbWriter writer = new JdbcDbWriter(new JdbcSinkConfig(props));
+    JdbcDbWriter writer = newWriter(props);
 
     Schema keySchema = Schema.INT64_SCHEMA;
 
@@ -151,7 +160,7 @@ private void writeSameRecordTwiceExpectingSingleUpdate(
     props.put("pk.fields", pkFields);
     props.put("insert.mode", insertMode.toString());
 
-    JdbcDbWriter writer = new JdbcDbWriter(new JdbcSinkConfig(props));
+    JdbcDbWriter writer = newWriter(props);
 
     Schema keySchema = SchemaBuilder.struct()
         .field("id", SchemaBuilder.INT64_SCHEMA);
@@ -231,7 +240,8 @@ public void sameRecordNTimes() throws SQLException {
     props.put("table.name.format", tableName);
     props.put("batch.size", String.valueOf(ThreadLocalRandom.current().nextInt(20, 100)));
 
-    JdbcDbWriter writer = new JdbcDbWriter(new JdbcSinkConfig(props));
+    JdbcDbWriter writer = newWriter(props);
+
     writer.write(Collections.nCopies(
         numRecords,
         new SinkRecord("topic", 0, null, null, schema, struct, 0)