@Test public void testErrors() { ValueProvider<String> schemasLocation = pipeline.newProvider("schemas.tar.gz"); final List<String> input = Arrays.asList( "{\"attributeMap\":" + "{\"document_namespace\":\"eng-workflow\"" + ",\"document_version\":\"1\"" + ",\"document_id\":\"2c3a0767-d84a-4d02-8a92-fa54a3376049\"" + ",\"document_type\":\"hgpush\"" + "},\"payload\":\"\"}", "{\"attributeMap\":{\"app_name\":\"Firefox\"" + ",\"app_version\":\"61.0a1\"" + ",\"document_type\":\"main\"},\"payload\":\"e30K\"}", "{\"attributeMap\":{},\"payload\":\"e30K\"}", "{\"attributeMap\":null,\"payload\":\"e30K\"}"); Result<PCollection<PubsubMessage>, PubsubMessage> result = pipeline .apply(Create.of(input)) .apply(InputFileFormat.json.decode()) .apply(ParsePayload.of(schemasLocation)); PCollection<String> exceptions = result.failures().apply(MapElements .into(TypeDescriptors.strings()).via(message -> message.getAttribute("exception_class"))); PAssert.that(result.output()).empty(); PAssert.that(exceptions).containsInAnyOrder("java.io.IOException", "com.mozilla.telemetry.ingestion.core.schema.SchemaNotFoundException", "com.mozilla.telemetry.ingestion.core.schema.SchemaNotFoundException", "com.mozilla.telemetry.ingestion.core.schema.SchemaNotFoundException"); pipeline.run(); }
public static ParsePayload of(ValueProvider<String> schemasLocation) { return new ParsePayload(schemasLocation); }
ParsePayload extends
    PTransform<PCollection<PubsubMessage>, Result<PCollection<PubsubMessage>, PubsubMessage>> { public static ParsePayload of(ValueProvider<String> schemasLocation) { return new ParsePayload(schemasLocation); } }
ParsePayload extends
    PTransform<PCollection<PubsubMessage>, Result<PCollection<PubsubMessage>, PubsubMessage>> { public static ParsePayload of(ValueProvider<String> schemasLocation) { return new ParsePayload(schemasLocation); } private  ParsePayload(ValueProvider<String> schemasLocation); }
ParsePayload extends
    PTransform<PCollection<PubsubMessage>, Result<PCollection<PubsubMessage>, PubsubMessage>> { public static ParsePayload of(ValueProvider<String> schemasLocation) { return new ParsePayload(schemasLocation); } private  ParsePayload(ValueProvider<String> schemasLocation); static ParsePayload of(ValueProvider<String> schemasLocation); @Override Result<PCollection<PubsubMessage>, PubsubMessage> expand(
      PCollection<PubsubMessage> messages); static String getVersionFromTelemetryPayload(JsonNode json); static void addClientIdFromPayload(Map<String, String> attributes, ObjectNode json); }
ParsePayload extends
    PTransform<PCollection<PubsubMessage>, Result<PCollection<PubsubMessage>, PubsubMessage>> { public static ParsePayload of(ValueProvider<String> schemasLocation) { return new ParsePayload(schemasLocation); } private  ParsePayload(ValueProvider<String> schemasLocation); static ParsePayload of(ValueProvider<String> schemasLocation); @Override Result<PCollection<PubsubMessage>, PubsubMessage> expand(
      PCollection<PubsubMessage> messages); static String getVersionFromTelemetryPayload(JsonNode json); static void addClientIdFromPayload(Map<String, String> attributes, ObjectNode json);  }