### Question:
AddMetadata { public static MapWithFailures<PubsubMessage, PubsubMessage, PubsubMessage> of() { return MapElements.into(TypeDescriptor.of(PubsubMessage.class)).via((PubsubMessage msg) -> { msg = PubsubConstraints.ensureNonNull(msg); byte[] metadata = Json .asBytes(NestedMetadata.attributesToMetadataPayload(msg.getAttributeMap())); byte[] mergedPayload = NestedMetadata.mergedPayload(msg.getPayload(), metadata); return new PubsubMessage(mergedPayload, msg.getAttributeMap()); }).exceptionsInto(TypeDescriptor.of(PubsubMessage.class)) .exceptionsVia((WithFailures.ExceptionElement<PubsubMessage> ee) -> { try { throw ee.exception(); } catch (UncheckedIOException e) { return FailureMessage.of(AddMetadata.class.getSimpleName(), ee.element(), ee.exception()); } }); } private  AddMetadata(); static MapWithFailures<PubsubMessage, PubsubMessage, PubsubMessage> of();  }### Answer:
@Test public void testOutput() { final List<String> input = Arrays.asList("{}", "{\"id\":null}", "[]", "{"); Map<String, String> attributes = ImmutableMap.<String, String>builder().put("sample_id", "18") .put("geo_country", "CA").put("isp_name", "service provider").put("x_debug_id", "mysession") .put("normalized_channel", "release").build(); WithFailures.Result<PCollection<PubsubMessage>, PubsubMessage> output = pipeline .apply(Create.of(input)) .apply("DecodeTextInput", InputFileFormat.text.decode()) .apply("AddAttributes", MapElements.into(TypeDescriptor.of(PubsubMessage.class)) .via(element -> new PubsubMessage(element.getPayload(), attributes))) .apply(AddMetadata.of()); final List<String> expectedMain = ImmutableList.of( "{\"metadata\":{\"geo\":{\"country\":\"CA\"}" + ",\"isp\":{\"name\":\"service provider\"}" + ",\"user_agent\":{}" + ",\"header\":{\"x_debug_id\":\"mysession\"}}" + ",\"normalized_channel\":\"release\"" + ",\"sample_id\":18}", "{\"metadata\":{\"geo\":{\"country\":\"CA\"}" + ",\"isp\":{\"name\":\"service provider\"}" + ",\"user_agent\":{}" + ",\"header\":{\"x_debug_id\":\"mysession\"}}" + ",\"normalized_channel\":\"release\"" + ",\"sample_id\":18" + ",\"id\":null}"); final List<String> expectedError = Arrays.asList("{", "[]"); final PCollection<String> error = output.failures() .apply("EncodeTextError", OutputFileFormat.text.encode()); PAssert.that(error).containsInAnyOrder(expectedError); final PCollection<String> main = output.output() .apply("EncodeTextMain", OutputFileFormat.text.encode()); PAssert.that(main).containsInAnyOrder(expectedMain); pipeline.run(); }