### Question:
DecompressPayload extends PTransform<PCollection<PubsubMessage>, PCollection<PubsubMessage>> { public static DecompressPayload enabled(ValueProvider<Boolean> enabled) { return new DecompressPayload(enabled); } private  DecompressPayload(ValueProvider<Boolean> enabled); static DecompressPayload enabled(ValueProvider<Boolean> enabled); @Override PCollection<PubsubMessage> expand(PCollection<PubsubMessage> input);  }### Answer:
@Test public void testOutput() { final List<String> input = Arrays.asList( "{\"attributeMap\":{\"host\":\"test\"},\"payload\":\"dGVzdA==\"}", "{\"payload\":\"H4sIAM1ekFsAAytJLS4BAAx+f9gEAAAA\"}"); final List<String> expected = Arrays.asList( "{\"attributeMap\":{\"host\":\"test\"},\"payload\":\"dGVzdA==\"}", "{\"attributeMap\":null,\"payload\":\"dGVzdA==\"}"); final PCollection<String> output = pipeline .apply(Create.of(input)) .apply(InputFileFormat.json.decode()) .apply(DecompressPayload.enabled(pipeline.newProvider(true))) .apply(OutputFileFormat.json.encode()); PAssert.that(output).containsInAnyOrder(expected); pipeline.run(); }