### Question:
RemoveAttributes extends PTransform<PCollection<PubsubMessage>, PCollection<PubsubMessage>> { public static RemoveAttributes of() { return INSTANCE; }  static RemoveAttributes of(); @Override PCollection<PubsubMessage> expand(PCollection<PubsubMessage> input);  }### Answer:
@Test public void testRemoveAttributes() { final List<String> input = Collections.singletonList("{\"attributeMap\":" + "{\"submission_timestamp\":\"2000-01-01T00:00:00.000000Z\"" + ",\"remote_addr\":\"127.0.0.1\"" + ",\"x_forwarded_for\":\"1,2,3\"" + ",\"x_forwarded_for\":\"1,2,3\"" + ",\"geo_country\":\"US\"" + ",\"normalized_country_code\":\"US\"" + ",\"client_id\":\"abc\"" + ",\"client_ip\":\"2\"" + ",\"app_name\":\"app_name\"" + ",\"os\":\"os\"" + "},\"payload\":\"\"}"); PCollection<String> output = pipeline.apply(Create.of(input)) .apply(InputFileFormat.json.decode()).apply(RemoveAttributes.of()) .apply(OutputFileFormat.json.encode()); final List<String> expected = Collections.singletonList("{\"attributeMap\":" + "{\"client_id\":\"abc\"" + ",\"client_ip\":\"2\"" + ",\"geo_country\":\"US\"" + ",\"submission_timestamp\":\"2000-01-01T00:00:00.000000Z\"" + "},\"payload\":\"\"}"); PAssert.that(output).containsInAnyOrder(expected); pipeline.run(); }