From 996d9f6de5d8b35fb5b504513a098667523e0ea4 Mon Sep 17 00:00:00 2001
From: Scott Wegner <swegner@google.com>
Date: Mon, 13 Jun 2016 11:05:00 -0700
Subject: [PATCH] Fix AutoComplete example streaming configuration

---
 .../org/apache/beam/examples/complete/AutoComplete.java     | 6 +++++-
 1 file changed, 5 insertions(+), 1 deletion(-)

diff --git a/examples/java/src/main/java/org/apache/beam/examples/complete/AutoComplete.java b/examples/java/src/main/java/org/apache/beam/examples/complete/AutoComplete.java
index ac8384b53b14..d725e0a54bb3 100644
--- a/examples/java/src/main/java/org/apache/beam/examples/complete/AutoComplete.java
+++ b/examples/java/src/main/java/org/apache/beam/examples/complete/AutoComplete.java
@@ -31,6 +31,7 @@
 import org.apache.beam.sdk.options.Default;
 import org.apache.beam.sdk.options.Description;
 import org.apache.beam.sdk.options.PipelineOptionsFactory;
+import org.apache.beam.sdk.options.Validation;
 import org.apache.beam.sdk.transforms.Count;
 import org.apache.beam.sdk.transforms.DoFn;
 import org.apache.beam.sdk.transforms.Filter;
@@ -416,6 +417,7 @@ public void processElement(ProcessContext c) {
    */
   private static interface Options extends ExamplePubsubTopicOptions, ExampleBigQueryTableOptions {
     @Description("Input text file")
+    @Validation.Required
     String getInputFile();
     void setInputFile(String value);
 
@@ -494,7 +496,9 @@ public static void main(String[] args) throws IOException {
                .to(tableRef)
                .withSchema(FormatForBigquery.getSchema())
                .withCreateDisposition(BigQueryIO.Write.CreateDisposition.CREATE_IF_NEEDED)
-               .withWriteDisposition(BigQueryIO.Write.WriteDisposition.WRITE_TRUNCATE));
+               .withWriteDisposition(options.isStreaming()
+                   ? BigQueryIO.Write.WriteDisposition.WRITE_APPEND
+                   : BigQueryIO.Write.WriteDisposition.WRITE_TRUNCATE));
     }
 
     // Run the pipeline.