@Override
  public PubsubMessage apply(Row row) {
    return new PubsubMessage(payloadExtractor.apply(row), attributesExtractor.apply(row));
  }
@Test
  public void rowPayloadTransformed() {
    Schema payloadSchema = Schema.builder().addStringField("fieldName").build();
    Row payload = Row.withSchema(payloadSchema).attachValues("abc");
    Schema schema =
        Schema.builder()
            .addRowField(PAYLOAD_FIELD, payloadSchema)
            .addField(ATTRIBUTES_FIELD, ATTRIBUTE_MAP_FIELD_TYPE)
            .build();
    Row row = Row.withSchema(schema).attachValues(payload, ATTRIBUTES);
    when(SERIALIZER.serialize(payload)).thenReturn("abc".getBytes(UTF_8));
    PubsubMessage message = new PubsubMessage("abc".getBytes(UTF_8), ATTRIBUTES);
    assertEquals(message, new NestedRowToMessage(SERIALIZER, schema).apply(row));
  }