Add DoFn to transform (#643)

feast-dev · Apr 27, 2020 · 45e6500 · 45e6500
1 parent 11151e8
commit 45e6500
Showing 1 changed file with 9 additions and 0 deletions.
diff --git a/ingestion/src/main/java/feast/ingestion/transform/metrics/WriteFailureMetricsTransform.java b/ingestion/src/main/java/feast/ingestion/transform/metrics/WriteFailureMetricsTransform.java
@@ -19,6 +19,7 @@
 import com.google.auto.value.AutoValue;
 import feast.ingestion.options.ImportOptions;
 import feast.storage.api.writer.FailedElement;
+import org.apache.beam.sdk.transforms.DoFn;
 import org.apache.beam.sdk.transforms.PTransform;
 import org.apache.beam.sdk.transforms.ParDo;
 import org.apache.beam.sdk.values.PCollection;
@@ -46,6 +47,14 @@ public PDone expand(PCollection<FailedElement> input) {
                   .setStatsdPort(options.getStatsdPort())
                   .setStoreName(getStoreName())
                   .build()));
+    } else {
+      input.apply(
+          "Noop",
+          ParDo.of(
+              new DoFn<FailedElement, Void>() {
+                @ProcessElement
+                public void processElement(ProcessContext c) {}
+              }));
     }
     return PDone.in(input.getPipeline());
   }