fix another gemini review

Author: derrickaw

sdks/java/io/datadog/src/main/java/org/apache/beam/sdk/io/datadog/DatadogWriteSchemaTransformProvider.java

@@ -22,11 +22,13 @@
 import com.google.auto.service.AutoService;
 import java.util.Collections;
 import java.util.List;
+import org.apache.beam.sdk.coders.RowCoder;
 import org.apache.beam.sdk.schemas.Schema;
 import org.apache.beam.sdk.schemas.transforms.SchemaTransform;
 import org.apache.beam.sdk.schemas.transforms.SchemaTransformProvider;
 import org.apache.beam.sdk.schemas.transforms.TypedSchemaTransformProvider;
 import org.apache.beam.sdk.schemas.transforms.providers.ErrorHandling;
+import org.apache.beam.sdk.transforms.Create;
 import org.apache.beam.sdk.transforms.DoFn;
 import org.apache.beam.sdk.transforms.ParDo;
 import org.apache.beam.sdk.values.PCollection;
@@ -172,7 +174,11 @@ public PCollectionRowTuple expand(PCollectionRowTuple input) {
         return PCollectionRowTuple.of(errorHandling.getOutput(), allErrors);
       } else {
         writeErrors.apply("Fail on Write Error", ParDo.of(new FailOnWriteErrorFn()));
-        return PCollectionRowTuple.empty(input.getPipeline());
+        PCollection<Row> emptyErrors =
+            input
+                .getPipeline()
+                .apply("Empty Errors Placeholder", Create.empty(RowCoder.of(dynamicErrorSchema)));
+        return PCollectionRowTuple.of(ERROR, emptyErrors);
       }
     }
   }

sdks/java/io/datadog/src/test/java/org/apache/beam/sdk/io/datadog/DatadogWriteSchemaTransformProviderTest.java

@@ -158,7 +158,8 @@ public void testWriteBuildTransformAndRun() {
     PCollection<Row> input = p.apply("Create", Create.of(ROWS).withRowSchema(SCHEMA));
     PCollectionRowTuple inputTuple = PCollectionRowTuple.of(INPUT, input);
     PCollectionRowTuple output = transform.expand(inputTuple);
-    assertTrue(output.getAll().isEmpty());
+    assertEquals(1, output.getAll().size());
+    assertTrue(output.has(ERROR));
 
     assertThrows(PipelineExecutionException.class, () -> p.run().waitUntilFinish());
   }
@@ -298,7 +299,8 @@ public void testBuildTransformWithAllParameters() {
     PCollection<Row> input = p.apply("Create", Create.of(ROWS).withRowSchema(SCHEMA));
     PCollectionRowTuple inputTuple = PCollectionRowTuple.of(INPUT, input);
     PCollectionRowTuple output = transform.expand(inputTuple);
-    assertTrue(output.getAll().isEmpty());
+    assertEquals(1, output.getAll().size());
+    assertTrue(output.has(ERROR));
 
     assertThrows(PipelineExecutionException.class, () -> p.run().waitUntilFinish());
   }