Add unit tests.

Author: Bingqin Zhou

build.gradle

@@ -217,6 +217,8 @@ project(':kcbq-connector') {
                 "junit:junit:$junitVersion",
                 "org.mockito:mockito-core:$mockitoVersion"
         )
+
+        testImplementation 'org.mockito:mockito-inline:2.13.0'
     }
 
     artifacts {

kcbq-connector/src/test/java/com/wepay/kafka/connect/bigquery/write/row/BigQueryWriterTest.java

@@ -27,11 +27,13 @@
 
 import com.google.cloud.bigquery.BigQuery;
 import com.google.cloud.bigquery.BigQueryError;
+import com.google.cloud.bigquery.BigQueryException;
 import com.google.cloud.bigquery.InsertAllRequest;
 import com.google.cloud.bigquery.InsertAllResponse;
 import com.google.cloud.storage.Storage;
 
 import com.wepay.kafka.connect.bigquery.BigQuerySinkTask;
+import com.wepay.kafka.connect.bigquery.SchemaManager;
 import com.wepay.kafka.connect.bigquery.SinkTaskPropertiesFactory;
 import com.wepay.kafka.connect.bigquery.config.BigQuerySinkConfig;
 import com.wepay.kafka.connect.bigquery.config.BigQuerySinkTaskConfig;
@@ -95,6 +97,75 @@ public void testBigQueryNoFailure() {
     verify(bigQuery, times(1)).insertAll(anyObject());
   }
 
+  @Test
+  public void testAutoCreateTables() {
+    final String topic = "test_topic";
+    final String dataset = "scratch";
+    final Map<String, String> properties = makeProperties("3", "2000", topic, dataset);
+    properties.put(BigQuerySinkTaskConfig.TABLE_CREATE_CONFIG, "true");
+
+    BigQuery bigQuery = mock(BigQuery.class);
+    Map<Long, List<BigQueryError>> emptyMap = mock(Map.class);
+    when(emptyMap.isEmpty()).thenReturn(true);
+
+    InsertAllResponse insertAllResponse = mock(InsertAllResponse.class);
+    when(insertAllResponse.hasErrors()).thenReturn(false);
+    when(insertAllResponse.getInsertErrors()).thenReturn(emptyMap);
+
+    BigQueryException missTableException = mock(BigQueryException.class);
+    when(missTableException.getReason()).thenReturn("notFound");
+
+    when(bigQuery.insertAll(anyObject())).thenThrow(missTableException).thenReturn(insertAllResponse);
+
+    SinkTaskContext sinkTaskContext = mock(SinkTaskContext.class);
+
+    Storage storage = mock(Storage.class);
+    SchemaManager schemaManager = mock(SchemaManager.class);
+    BigQuerySinkTask testTask = new BigQuerySinkTask(bigQuery, null, storage, schemaManager);
+    testTask.initialize(sinkTaskContext);
+    testTask.start(properties);
+    testTask.put(
+            Collections.singletonList(spoofSinkRecord(topic, 0, 0, "some_field", "some_value")));
+    testTask.flush(Collections.emptyMap());
+
+    verify(schemaManager, times(1)).createTable(anyObject(), anyObject());
+    verify(bigQuery, times(2)).insertAll(anyObject());
+  }
+
+  @Test
+  public void testNonAutoCreateTables() {
+    final String topic = "test_topic";
+    final String dataset = "scratch";
+    final Map<String, String> properties = makeProperties("3", "2000", topic, dataset);
+
+    BigQuery bigQuery = mock(BigQuery.class);
+
+    Map<Long, List<BigQueryError>> emptyMap = mock(Map.class);
+    when(emptyMap.isEmpty()).thenReturn(true);
+    InsertAllResponse insertAllResponse = mock(InsertAllResponse.class);
+    when(insertAllResponse.hasErrors()).thenReturn(false);
+    when(insertAllResponse.getInsertErrors()).thenReturn(emptyMap);
+
+    BigQueryException missTableException = mock(BigQueryException.class);
+    when(missTableException.getReason()).thenReturn("notFound");
+
+    when(bigQuery.insertAll(anyObject())).thenThrow(missTableException).thenReturn(insertAllResponse);
+
+    SinkTaskContext sinkTaskContext = mock(SinkTaskContext.class);
+
+    Storage storage = mock(Storage.class);
+    SchemaManager schemaManager = mock(SchemaManager.class);
+    BigQuerySinkTask testTask = new BigQuerySinkTask(bigQuery, null, storage, schemaManager);
+    testTask.initialize(sinkTaskContext);
+    testTask.start(properties);
+    testTask.put(
+            Collections.singletonList(spoofSinkRecord(topic, 0, 0, "some_field", "some_value")));
+    testTask.flush(Collections.emptyMap());
+
+    verify(schemaManager, times(0)).createTable(anyObject(), anyObject());
+    verify(bigQuery, times(2)).insertAll(anyObject());
+  }
+
   @Test
   public void testBigQueryPartialFailure() {
     final String topic = "test_topic";