### Question:
DecodePubsubMessages { public static PTransform<PCollection<? extends String>, PCollection<PubsubMessage>> json() { return PTransform.compose("DecodePubsubMessages.Json", input -> input .apply(MapElements.into(TypeDescriptor.of(PubsubMessage.class)).via((String s) -> { try { return com.mozilla.telemetry.util.Json.readPubsubMessage(s); } catch (IOException e) { throw new UncheckedIOException(e); } }))); }  static PTransform<PCollection<? extends String>, PCollection<PubsubMessage>> text(); static PTransform<PCollection<? extends String>, PCollection<PubsubMessage>> json();  }### Answer:
@Test public void testJson() { List<String> inputLines = Lines.resources("testdata/decode-pubsub-messages/input-valid*"); List<String> validLines = Lines .resources("testdata/decode-pubsub-messages/output-normalized-json.ndjson"); PCollection<String> result = pipeline .apply(Create.of(inputLines)) .apply(InputFileFormat.json.decode()) .apply("EncodeJsonOutput", OutputFileFormat.json.encode()); PAssert.that(result).containsInAnyOrder(validLines); pipeline.run(); }