feat: Add sample about processing permanent writer failure (#2057)

yirutang · gcf-owl-bot[bot] · web-flow · commit 8eda9347a90f · 2023-04-03T13:52:32.000-07:00
* feat: Add sample about writer permanently failed * . * . * . * . * . * . * . * . * . * . * 🦉 Updates from OwlBot post-processor See https://github.com/googleapis/repo-automation-bots/blob/main/packages/owl-bot/README.md --------- Co-authored-by: Owl Bot <gcf-owl-bot[bot]@users.noreply.github.com>
diff --git a/samples/snippets/src/main/java/com/example/bigquerystorage/WriteToDefaultStream.java b/samples/snippets/src/main/java/com/example/bigquerystorage/WriteToDefaultStream.java
@@ -40,6 +40,7 @@
 import java.io.IOException;
 import java.util.Map;
 import java.util.concurrent.Phaser;
+import java.util.concurrent.atomic.AtomicInteger;
 import javax.annotation.concurrent.GuardedBy;
 import org.json.JSONArray;
 import org.json.JSONObject;
@@ -123,6 +124,7 @@ private static class AppendContext {
   private static class DataWriter {
 
     private static final int MAX_RETRY_COUNT = 3;
+    private static final int MAX_RECREATE_COUNT = 3;
     private static final ImmutableList<Code> RETRIABLE_ERROR_CODES =
         ImmutableList.of(
             Code.INTERNAL,
@@ -140,6 +142,8 @@ private static class DataWriter {
     @GuardedBy("lock")
     private RuntimeException error = null;
 
+    private AtomicInteger recreateCount = new AtomicInteger(0);
+
     public void initialize(TableName parentTable)
         throws DescriptorValidationException, IOException, InterruptedException {
       // Use the JSON stream writer to send records in JSON format. Specify the table name to write
@@ -151,8 +155,17 @@ public void initialize(TableName parentTable)
     }
 
     public void append(AppendContext appendContext)
-        throws DescriptorValidationException, IOException {
+        throws DescriptorValidationException, IOException, InterruptedException {
       synchronized (this.lock) {
+        if (!streamWriter.isUserClosed()
+            && streamWriter.isClosed()
+            && recreateCount.getAndIncrement() < MAX_RECREATE_COUNT) {
+          streamWriter =
+              JsonStreamWriter.newBuilder(
+                      streamWriter.getStreamName(), BigQueryWriteClient.create())
+                  .build();
+          this.error = null;
+        }
         // If earlier appends have failed, we need to reset before continuing.
         if (this.error != null) {
           throw this.error;
@@ -194,6 +207,7 @@ public AppendCompleteCallback(DataWriter parent, AppendContext appendContext) {
 
       public void onSuccess(AppendRowsResponse response) {
         System.out.format("Append success\n");
+        this.parent.recreateCount.set(0);
         done();
       }
 
@@ -241,6 +255,8 @@ public void onFailure(Throwable throwable) {
                 throw new RuntimeException(e);
               } catch (IOException e) {
                 throw new RuntimeException(e);
+              } catch (InterruptedException e) {
+                throw new RuntimeException(e);
               }
             }
             // Mark the existing attempt as done since we got a response for it
