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_timestamp() {
    Instant timestamp = Instant.now();
    Row withTimestamp =
        Row.withSchema(NON_USER_WITH_BYTES_PAYLOAD)
            .addValues(ImmutableMap.of(), timestamp, new byte[] {})
            .build();
    assertEquals(
        timestamp.toString(),
        doFn(NON_USER_WITH_BYTES_PAYLOAD, null).timestamp(withTimestamp).toString());

    Schema withoutTimestampSchema =
        removeFields(NON_USER_WITH_BYTES_PAYLOAD, DEFAULT_EVENT_TIMESTAMP_KEY_NAME);
    Row withoutTimestamp =
        Row.withSchema(withoutTimestampSchema).addValues(ImmutableMap.of(), new byte[] {}).build();
    ReadableDateTime actual = doFn(withoutTimestampSchema, null).timestamp(withoutTimestamp);
    assertNotNull(actual);
  }