Row messageToRow(SequencedMessage message) {
    // Transform this to a FieldValueBuilder, because otherwise individual withFieldValue calls will
    // not mutate the original object.
    Row.FieldValueBuilder builder = Row.withSchema(schema).withFieldValues(ImmutableMap.of());
    if (schema.hasField(PUBLISH_TIMESTAMP_FIELD)) {
      builder.withFieldValue(
          PUBLISH_TIMESTAMP_FIELD,
          Instant.ofEpochMilli(Timestamps.toMillis(message.getPublishTime())));
    }
    if (schema.hasField(MESSAGE_KEY_FIELD)) {
      builder.withFieldValue(MESSAGE_KEY_FIELD, message.getMessage().getKey().toByteArray());
    }
    if (schema.hasField(EVENT_TIMESTAMP_FIELD) && message.getMessage().hasEventTime()) {
      builder.withFieldValue(
          EVENT_TIMESTAMP_FIELD,
          Instant.ofEpochMilli(Timestamps.toMillis(message.getMessage().getEventTime())));
    }
    if (schema.hasField(ATTRIBUTES_FIELD)) {
      ImmutableList.Builder<Row> listBuilder = ImmutableList.builder();
      message
          .getMessage()
          .getAttributesMap()
          .forEach(
              (key, values) -> {
                Row entry =
                    Row.withSchema(ATTRIBUTES_ENTRY_SCHEMA)
                        .withFieldValue(ATTRIBUTES_KEY_FIELD, key)
                        .withFieldValue(
                            ATTRIBUTES_VALUES_FIELD,
                            values.getValuesList().stream()
                                .map(ByteString::toByteArray)
                                .collect(Collectors.toList()))
                        .build();
                listBuilder.add(entry);
              });
      builder.withFieldValue(ATTRIBUTES_FIELD, listBuilder.build());
    }
    if (payloadSerializer == null) {
      builder.withFieldValue(PAYLOAD_FIELD, message.getMessage().getData().toByteArray());
    } else {
      builder.withFieldValue(
          PAYLOAD_FIELD,
          payloadSerializer.deserialize(message.getMessage().getData().toByteArray()));
    }
    return builder.build();
  }
@Test
  public void fullMessageToRow() {
    RowHandler rowHandler = new RowHandler(FULL_READ_SCHEMA);
    Instant event = Instant.now();
    Instant publish = Instant.now();
    PubSubMessage userMessage =
        PubSubMessage.newBuilder()
            .setKey(ByteString.copyFromUtf8("val1"))
            .setData(ByteString.copyFromUtf8("val2"))
            .setEventTime(Timestamps.fromMillis(event.getMillis()))
            .putAttributes(
                "key1",
                AttributeValues.newBuilder()
                    .addValues(ByteString.copyFromUtf8("attr1"))
                    .addValues(ByteString.copyFromUtf8("attr2"))
                    .build())
            .putAttributes(
                "key2",
                AttributeValues.newBuilder().addValues(ByteString.copyFromUtf8("attr3")).build())
            .build();
    SequencedMessage sequencedMessage =
        SequencedMessage.newBuilder()
            .setMessage(userMessage)
            .setPublishTime(Timestamps.fromMillis(publish.getMillis()))
            .build();
    Row expected =
        Row.withSchema(FULL_READ_SCHEMA)
            .withFieldValue(RowHandler.MESSAGE_KEY_FIELD, "val1".getBytes(UTF_8))
            .withFieldValue(RowHandler.PAYLOAD_FIELD, "val2".getBytes(UTF_8))
            .withFieldValue(RowHandler.EVENT_TIMESTAMP_FIELD, event)
            .withFieldValue(RowHandler.PUBLISH_TIMESTAMP_FIELD, publish)
            .withFieldValue(
                RowHandler.ATTRIBUTES_FIELD,
                ImmutableList.of(
                    Row.withSchema(RowHandler.ATTRIBUTES_ENTRY_SCHEMA)
                        .attachValues(
                            "key1",
                            ImmutableList.of("attr1".getBytes(UTF_8), "attr2".getBytes(UTF_8))),
                    Row.withSchema(RowHandler.ATTRIBUTES_ENTRY_SCHEMA)
                        .attachValues("key2", ImmutableList.of("attr3".getBytes(UTF_8)))))
            .build();
    assertEquals(expected, rowHandler.messageToRow(sequencedMessage));
  }