initial plumbing

Author: ron-gal

v2/googlecloud-to-googlecloud/src/main/java/com/google/cloud/teleport/v2/templates/bigtablechangestreamstopubsub/BigtableChangeStreamsToPubSub.java

@@ -560,7 +560,7 @@ private static PubsubMessage createTestChangeMessage(PubSubUtils pubSub) throws
       case PROTOCOL_BUFFERS:
         return pubSub.mapChangeJsonStringToPubSubMessageAsProto(mod.getChangeJson());
       case JSON:
-        return pubSub.mapChangeJsonStringToPubSubMessageAsJson(mod.getChangeJson());
+        return pubSub.mapChangeJsonStringToPubSubMessageAsJson(mod.getChangeJson(), null);
       default:
         throw new IllegalArgumentException(
             "Unexpected message format: " + pubSub.getDestination().getMessageFormat());

v2/googlecloud-to-googlecloud/src/main/java/com/google/cloud/teleport/v2/templates/bigtablechangestreamstopubsub/FailsafePublisher.java

@@ -112,11 +112,15 @@ public void tearDown() {
       }
 
       @ProcessElement
-      public void processElement(ProcessContext context) {
+      public void processElement(ProcessContext context, OutputReceiver<String> dlq) {
         FailsafeElement<String, String> failsafeModJsonString = context.element();
 
         try {
-          PubsubMessage pubSubMessage = newPubsubMessage(failsafeModJsonString.getPayload());
+          PubsubMessage pubSubMessage = newPubsubMessage(failsafeModJsonString.getPayload(), dlq);
+          if (pubSubMessage == null) {
+            return;
+          }
+
           throttled.success(LOG, publisher.publish(pubSubMessage).get());
         } catch (Exception e) {
           throttled.failure(LOG, e);
@@ -128,7 +132,8 @@ public void processElement(ProcessContext context) {
       }
 
       /* Schema Details:  */
-      private PubsubMessage newPubsubMessage(String modJsonString) throws Exception {
+      private PubsubMessage newPubsubMessage(String modJsonString, OutputReceiver<String> dlq)
+          throws Exception {
         String changeJsonString = Mod.fromJson(modJsonString).getChangeJson();
         MessageFormat messageFormat = pubSubUtils.getDestination().getMessageFormat();
 
@@ -138,7 +143,7 @@ private PubsubMessage newPubsubMessage(String modJsonString) throws Exception {
           case PROTOCOL_BUFFERS:
             return pubSubUtils.mapChangeJsonStringToPubSubMessageAsProto(changeJsonString);
           case JSON:
-            return pubSubUtils.mapChangeJsonStringToPubSubMessageAsJson(changeJsonString);
+            return pubSubUtils.mapChangeJsonStringToPubSubMessageAsJson(changeJsonString, dlq);
           default:
             final String errorMessage =
                 "Invalid message format:"

v2/googlecloud-to-googlecloud/src/main/java/com/google/cloud/teleport/v2/templates/bigtablechangestreamstopubsub/schemautils/PubSubUtils.java

@@ -37,6 +37,7 @@
 import org.apache.avro.io.BinaryEncoder;
 import org.apache.avro.io.Encoder;
 import org.apache.avro.io.EncoderFactory;
+import org.apache.beam.sdk.transforms.DoFn.OutputReceiver;
 import org.apache.commons.lang3.Validate;
 import org.json.JSONObject;
 
@@ -256,8 +257,8 @@ public PubsubMessage mapChangeJsonStringToPubSubMessageAsAvro(String changeJsonS
     return PubsubMessage.newBuilder().setData(data).build();
   }
 
-  public PubsubMessage mapChangeJsonStringToPubSubMessageAsJson(String changeJsonSting)
-      throws Exception {
+  public PubsubMessage mapChangeJsonStringToPubSubMessageAsJson(
+      String changeJsonSting, OutputReceiver<String> dlq) throws Exception {
     JSONObject changeJsonParsed = new JSONObject(changeJsonSting);
 
     var changelogEntryTextBuilder =
@@ -310,7 +311,10 @@ public PubsubMessage mapChangeJsonStringToPubSubMessageAsJson(String changeJsonS
         value = (String) FORMATTERS.get(PubSubFields.VALUE_STRING).format(this, changeJsonParsed);
       }
       if (value == null) {
-        // somehow go to DLQ
+        if (dlq != null) {
+          dlq.output(changeJsonSting);
+          return null;
+        }
       }
       changelogEntryTextBuilder.setValue(value);
     }