static Schema removeFields(Schema schema, String... fields) {
    List<String> exclude = Arrays.stream(fields).collect(Collectors.toList());
    Schema.Builder builder = Schema.builder();
    for (Field field : schema.getFields()) {
      if (exclude.contains(field.getName())) {
        continue;
      }
      builder.addField(field);
    }
    return builder.build();
  }
@Test
  public void testPubsubRowToMessageParDo_attributesWithoutTimestamp() {
    Map<String, String> attributes = new HashMap<>();
    attributes.put("a", "1");
    attributes.put("b", "2");
    attributes.put("c", "3");

    Row withAttributes =
        Row.withSchema(NON_USER_WITH_BYTES_PAYLOAD)
            .addValues(attributes, Instant.now(), new byte[] {})
            .build();

    assertEquals(
        attributes,
        doFn(NON_USER_WITH_BYTES_PAYLOAD, null).attributesWithoutTimestamp(withAttributes));

    Schema withoutAttributesSchema =
        removeFields(NON_USER_WITH_BYTES_PAYLOAD, DEFAULT_ATTRIBUTES_KEY_NAME);
    Row withoutAttributes =
        Row.withSchema(withoutAttributesSchema).addValues(Instant.now(), new byte[] {}).build();

    assertEquals(
        ImmutableMap.of(),
        doFn(withoutAttributesSchema, null).attributesWithoutTimestamp(withoutAttributes));
  }