docs(tutorials): Add IT to JsonWriteDefaultStream tutorial (#1522)

Includes two small code fixes:

- Change a field name in the schema.
- Detect schema updates while streaming.

pom.xml

@@ -253,6 +253,7 @@
       <id>include-samples</id>
       <modules>
         <module>samples</module>
+        <module>tutorials</module>
       </modules>
     </profile>
   </profiles>

tutorials/JsonWriterDefaultStream/pom.xml

@@ -41,6 +41,19 @@
       <artifactId>arrow-memory-netty</artifactId>
       <version>6.0.1</version>
     </dependency>
+    <!-- Test Dependencies -->
+    <dependency>
+      <groupId>junit</groupId>
+      <artifactId>junit</artifactId>
+      <version>4.13.2</version>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>com.google.truth</groupId>
+      <artifactId>truth</artifactId>
+      <version>1.1.3</version>
+      <scope>test</scope>
+    </dependency>
   </dependencies>
 
   <build>
@@ -57,7 +70,6 @@
             </goals>
             <configuration>
               <sources>
-                <source>.</source>
                 <source>../../samples/snippets/src/main/java/com/example/bigquerystorage</source>
               </sources>
             </configuration>

tutorials/JsonWriterDefaultStream/src/main/java/com/example/JsonWriterDefaultStream.java

@@ -67,7 +67,7 @@ public static void createDestinationTable(
                 .build(),
             Field.of("author", StandardSQLTypeName.STRING),
             Field.of("committer", StandardSQLTypeName.STRING),
-            Field.of("ts", StandardSQLTypeName.DATETIME),
+            Field.of("commit_date", StandardSQLTypeName.DATETIME),
             Field.of("subject", StandardSQLTypeName.STRING),
             Field.of("message", StandardSQLTypeName.STRING),
             Field.of("repo_name", StandardSQLTypeName.STRING));
@@ -114,6 +114,12 @@ public static void writeToDefaultStream(
         } // batch
         ApiFuture<AppendRowsResponse> future = writer.append(jsonArr);
         AppendRowsResponse response = future.get();
+        if (response.hasUpdatedSchema()) {
+          // The destination table schema has changed. The client library automatically
+          // reestablishes a connection to the backend using the new schema, so we can continue
+          // to send data without interruption.
+          System.out.println("Table schema changed.");
+        }
       }
       System.out.println("Appended records successfully.");
     } catch (ExecutionException e) {

tutorials/JsonWriterDefaultStream/src/test/java/com/example/JsonWriterDefaultStreamIT.java

@@ -16,7 +16,65 @@
 
 package com.example.bigquerystorage;
 
+import static com.google.common.truth.Truth.assertThat;
+
+import com.google.cloud.bigquery.BigQuery;
+import com.google.cloud.bigquery.BigQuery.DatasetDeleteOption;
+import com.google.cloud.bigquery.BigQueryOptions;
+import com.google.cloud.bigquery.DatasetId;
+import com.google.cloud.bigquery.DatasetInfo;
+import java.io.ByteArrayOutputStream;
+import java.io.PrintStream;
+import java.nio.file.FileSystems;
+import java.nio.file.Path;
+import java.util.UUID;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
 @RunWith(JUnit4.class)
 public class JsonWriterDefaultStreamIT {
-  // TODO(mwasson): ADD Integration Test
+
+  private static final String GOOGLE_CLOUD_PROJECT = System.getenv("GOOGLE_CLOUD_PROJECT");
+
+  private ByteArrayOutputStream bout;
+  private PrintStream out;
+  private BigQuery bigquery;
+  private String datasetName;
+
+  @BeforeClass
+  public static void beforeClass() {}
+
+  @Before
+  public void setUp() {
+    bout = new ByteArrayOutputStream();
+    out = new PrintStream(bout);
+    System.setOut(out);
+
+    bigquery = BigQueryOptions.getDefaultInstance().getService();
+
+    // Create a new dataset for each test.
+    datasetName = "JAVA_WRITER_DEFAULT_STREAM_TEST" + UUID.randomUUID().toString().substring(0, 8);
+    bigquery.create(DatasetInfo.newBuilder(datasetName).build());
+  }
+
+  @Test
+  public void testJsonWriterDefaultStream() throws Exception {
+    Path dataFilePath = FileSystems.getDefault().getPath("src/test/resources", "TestData.json");
+
+    System.out.println(dataFilePath.toString());
+    String[] args = {GOOGLE_CLOUD_PROJECT, datasetName, "github", dataFilePath.toString()};
+    JsonWriterDefaultStream.main(args);
+    assertThat(bout.toString()).contains("Appended records successfully.");
+  }
+
+  @After
+  public void tearDown() {
+    bigquery.delete(
+        DatasetId.of(GOOGLE_CLOUD_PROJECT, datasetName), DatasetDeleteOption.deleteContents());
+    System.setOut(null);
+  }
 }

tutorials/JsonWriterDefaultStream/src/test/resources/TestData.json

@@ -0,0 +1,5 @@
+{"commit":"0001","parent":["00001"],"author":"user1","committer":"GitHub","subject":"Commit 1","message":"This is a commit.\n\n","repo_name":"googleapis/java-bigquerystorage","commit_date":"2019-07-23T20:28:01"}
+{"commit":"0002","parent":["00002"],"author":"user1","committer":"GitHub","subject":"Commit 2","message":"This is a commit.\n\n","repo_name":"googleapis/java-bigquerystorage","commit_date":"2019-12-05T16:05:16"}
+{"commit":"0003","parent":["00003"],"author":"user1","committer":"GitHub","subject":"Commit 3","message":"This is a commit.\n\n","repo_name":"googleapis/java-bigquerystorage","commit_date":"2019-03-21T16:59:23"}
+{"commit":"0004","parent":["00004"],"author":"user1","committer":"GitHub","subject":"Commit 4","message":"This is a commit.\n\n","repo_name":"googleapis/java-bigquerystorage","commit_date":"2019-01-11T01:31:39"}
+{"commit":"0005","parent":["00005"],"author":"user1","committer":"GitHub","subject":"Commit 5","message":"This is a commit.\n\n","repo_name":"googleapis/java-bigquerystorage","commit_date":"2019-07-31T19:09:09"}