Make changes according to comments.

Author: Bingqin Zhou

kcbq-connector/src/main/java/com/wepay/kafka/connect/bigquery/BigQuerySinkConnector.java

@@ -111,7 +111,7 @@ private void ensureExistingTables() {
     BigQuery bigQuery = getBigQuery();
     Map<String, TableId> topicsToTableIds = TopicToTableResolver.getTopicsToTables(config);
     for (TableId tableId : topicsToTableIds.values()) {
-      if (!config.getBoolean(config.TABLE_CREATE_CONFIG) && bigQuery.getTable(tableId) == null) {
+      if (bigQuery.getTable(tableId) == null) {
         logger.warn(
                 "You may want to enable auto table creation by setting {}=true in the properties file",
                 config.TABLE_CREATE_CONFIG);
@@ -133,7 +133,9 @@ public void start(Map<String, String> properties) {
       );
     }
 
-    ensureExistingTables();
+    if (!config.getBoolean(config.TABLE_CREATE_CONFIG)) {
+      ensureExistingTables();
+    }
   }
 
   @Override

kcbq-connector/src/main/java/com/wepay/kafka/connect/bigquery/BigQuerySinkTask.java

@@ -131,7 +131,7 @@ public void flush(Map<TopicPartition, OffsetAndMetadata> offsets) {
     topicPartitionManager.resumeAll();
   }
 
-  private PartitionedTableId getRecordTable(boolean autoCreateTables, SinkRecord record) {
+  private PartitionedTableId getRecordTable(SinkRecord record, boolean autoCreateTables) {
     // Dynamically update topicToBaseTableIds mapping. topicToBaseTableIds was used to be
     // constructed when connector starts hence new topic configuration needed connector to restart.
     // Dynamic update shall not require connector restart and shall compute table id in runtime.
@@ -196,7 +196,7 @@ public void put(Collection<SinkRecord> records) {
 
     for (SinkRecord record : records) {
       if (record.value() != null) {
-        PartitionedTableId table = getRecordTable(autoCreateTables, record);
+        PartitionedTableId table = getRecordTable(record, autoCreateTables);
         if (schemaRetriever != null) {
           schemaRetriever.setLastSeenSchema(table.getBaseTableId(),
                                             record.topic(),