@Test public void testVersionInPayload() { ValueProvider<String> schemasLocation = pipeline.newProvider("schemas.tar.gz"); String input = "{\"attributeMap\":" + "{\"document_namespace\":\"telemetry\"" + ",\"app_name\":\"Firefox\"" + ",\"document_id\":\"2c3a0767-d84a-4d02-8a92-fa54a3376049\"" + ",\"document_type\":\"main\"" + "},\"payload\":\"eyJ2ZXJzaW9uIjo0fQ==\"}"; 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("org.everit.json.schema.ValidationException"); 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);  }