@Override
  public PubsubMessage apply(Row row) {
    return new PubsubMessage(payloadExtractor.apply(row), attributesExtractor.apply(row));
  }
@Test
  public void entriesAttributesTransformed() {
    Schema schema =
        Schema.builder()
            .addByteArrayField(PAYLOAD_FIELD)
            .addField(ATTRIBUTES_FIELD, ATTRIBUTE_ARRAY_FIELD_TYPE)
            .build();
    Row row =
        Row.withSchema(schema)
            .attachValues(
                "abc".getBytes(UTF_8),
                ImmutableList.of(
                    Row.withSchema(ATTRIBUTE_ARRAY_ENTRY_SCHEMA).attachValues("k1", "v1"),
                    Row.withSchema(ATTRIBUTE_ARRAY_ENTRY_SCHEMA).attachValues("k2", "v2")));
    PubsubMessage message = new PubsubMessage("abc".getBytes(UTF_8), ATTRIBUTES);
    assertEquals(message, new NestedRowToMessage(SERIALIZER, schema).apply(row));
  }