Feature/bigtable error handling

sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/errorhandling/BadRecordRouter.java

@@ -19,7 +19,9 @@
 
 import java.io.Serializable;
 import org.apache.beam.sdk.coders.Coder;
+import org.apache.beam.sdk.transforms.DoFn;
 import org.apache.beam.sdk.transforms.DoFn.MultiOutputReceiver;
+import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
 import org.apache.beam.sdk.util.Preconditions;
 import org.apache.beam.sdk.values.TupleTag;
 import org.checkerframework.checker.nullness.qual.Nullable;
@@ -40,6 +42,15 @@ <RecordT> void route(
       String description)
       throws Exception;
 
+  <RecordT> void route(
+      DoFn<?, ?>.FinishBundleContext c,
+      RecordT record,
+      @Nullable Coder<RecordT> coder,
+      @Nullable Exception exception,
+      String description,
+      BoundedWindow window)
+      throws Exception;
+
   class ThrowingBadRecordRouter implements BadRecordRouter {
 
     @Override
@@ -65,6 +76,31 @@ public <RecordT> void route(
         throw new RuntimeException("Encountered Bad Record: " + encodedRecord);
       }
     }
+
+    @Override
+    public <RecordT> void route(
+        DoFn<?, ?>.FinishBundleContext c,
+        RecordT record,
+        @Nullable Coder<RecordT> coder,
+        @Nullable Exception exception,
+        String description,
+        BoundedWindow window)
+        throws Exception {
+      if (exception != null) {
+        throw exception;
+      } else {
+        Preconditions.checkArgumentNotNull(record);
+        String encodedRecord =
+            BadRecord.Record.builder()
+                .addHumanReadableJson(record)
+                .build()
+                .getHumanReadableJsonRecord();
+        if (encodedRecord == null) {
+          encodedRecord = "Unable to serialize bad record";
+        }
+        throw new RuntimeException("Encountered Bad Record: " + encodedRecord);
+      }
+    }
   }
 
   class RecordingBadRecordRouter implements BadRecordRouter {
@@ -81,5 +117,21 @@ public <RecordT> void route(
           .get(BAD_RECORD_TAG)
           .output(BadRecord.fromExceptionInformation(record, coder, exception, description));
     }
+
+    @Override
+    public <RecordT> void route(
+        DoFn<?, ?>.FinishBundleContext c,
+        RecordT record,
+        @Nullable Coder<RecordT> coder,
+        @Nullable Exception exception,
+        String description,
+        BoundedWindow window)
+        throws Exception {
+      c.output(
+          BAD_RECORD_TAG,
+          BadRecord.fromExceptionInformation(record, coder, exception, description),
+          window.maxTimestamp(),
+          window);
+    }
   }
 }