Make changes according to comments.

Author: Bingqin Zhou

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

@@ -113,8 +113,8 @@ private void ensureExistingTables() {
     for (TableId tableId : topicsToTableIds.values()) {
       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);
+          "You may want to enable auto table creation by setting {}=true in the properties file",
+          config.TABLE_CREATE_CONFIG);
         throw new BigQueryConnectException("Table '" + tableId + "' does not exist");
       }
     }

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

@@ -141,11 +141,7 @@ private PartitionedTableId getRecordTable(SinkRecord record, boolean autoCreateT
 
     TableId baseTableId = topicsToBaseTableIds.get(record.topic());
 
-    BigQuery bigQuery = getBigQuery();
-    if (autoCreateTables && bigQuery.getTable(baseTableId) == null) {
-      getSchemaManager(bigQuery).createTable(baseTableId, record.topic());
-      logger.info("Table {} does not exist, auto-created table for topic {}", baseTableId, record.topic());
-    }
+    maybeCreateTable(record, baseTableId, autoCreateTables);
 
     PartitionedTableId.Builder builder = new PartitionedTableId.Builder(baseTableId);
     if (useMessageTimeDatePartitioning) {
@@ -162,6 +158,20 @@ private PartitionedTableId getRecordTable(SinkRecord record, boolean autoCreateT
     return builder.build();
   }
 
+  /**
+   * Create the table which doesn't exist in BigQuery for a (record's) topic when autoCreateTables config is set to true.
+   * @param record Kafka Sink Record to be streamed into BigQuery.
+   * @param baseTableId BaseTableId in BigQuery.
+   * @param autoCreateTables If this config is set to true, auto-creating the table that doesn't not exist.
+   */
+  private void maybeCreateTable(SinkRecord record, TableId baseTableId, boolean autoCreateTables) {
+    BigQuery bigQuery = getBigQuery();
+    if (autoCreateTables && bigQuery.getTable(baseTableId) == null) {
+      getSchemaManager(bigQuery).createTable(baseTableId, record.topic());
+      logger.info("Table {} does not exist, auto-created table for topic {}", baseTableId, record.topic());
+    }
+  }
+
   private RowToInsert getRecordRow(SinkRecord record) {
     Map<String, Object> convertedRecord = recordConverter.convertRecord(record, KafkaSchemaRecordType.VALUE);
     Optional<String> kafkaKeyFieldName = config.getKafkaKeyFieldName();

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

@@ -625,18 +625,13 @@ private void verifyBucketSpecified() throws ConfigException {
   }
 
   private void checkAutoCreateTables() {
-    Class<?> schemaRetriever = getClass(BigQuerySinkConfig.SCHEMA_RETRIEVER_CONFIG);
 
+    Class<?> schemaRetriever = getClass(BigQuerySinkConfig.SCHEMA_RETRIEVER_CONFIG);
     boolean autoCreateTables = getBoolean(TABLE_CREATE_CONFIG);
+
     if (autoCreateTables && schemaRetriever == null) {
       throw new ConfigException(
-              "Cannot specify automatic table creation without a schema retriever"
-      );
-    }
-
-    if (schemaRetriever == null) {
-      logger.warn(
-              "No schema retriever class provided; auto table creation is impossible"
+        "Cannot specify automatic table creation without a schema retriever"
       );
     }
   }