PubSubMessage rowToMessage(Row row) {
    row = castRow(row, row.getSchema(), schema);
    PubSubMessage.Builder builder = PubSubMessage.newBuilder();
    if (schema.hasField(MESSAGE_KEY_FIELD)) {
      byte[] bytes = row.getBytes(MESSAGE_KEY_FIELD);
      if (bytes != null) {
        builder.setKey(ByteString.copyFrom(bytes));
      }
    }
    if (schema.hasField(EVENT_TIMESTAMP_FIELD)) {
      ReadableDateTime time = row.getDateTime(EVENT_TIMESTAMP_FIELD);
      if (time != null) {
        builder.setEventTime(Timestamps.fromMillis(time.getMillis()));
      }
    }
    if (schema.hasField(ATTRIBUTES_FIELD)) {
      Collection<Row> attributes = row.getArray(ATTRIBUTES_FIELD);
      if (attributes != null) {
        attributes.forEach(
            entry -> {
              AttributeValues.Builder valuesBuilder = AttributeValues.newBuilder();
              Collection<byte[]> values =
                  checkArgumentNotNull(entry.getArray(ATTRIBUTES_VALUES_FIELD));
              values.forEach(bytes -> valuesBuilder.addValues(ByteString.copyFrom(bytes)));
              builder.putAttributes(
                  checkArgumentNotNull(entry.getString(ATTRIBUTES_KEY_FIELD)),
                  valuesBuilder.build());
            });
      }
    }
    if (payloadSerializer == null) {
      byte[] payload = row.getBytes(PAYLOAD_FIELD);
      if (payload != null) {
        builder.setData(ByteString.copyFrom(payload));
      }
    } else {
      Row payload = row.getRow(PAYLOAD_FIELD);
      if (payload != null) {
        builder.setData(ByteString.copyFrom(payloadSerializer.serialize(payload)));
      }
    }
    return builder.build();
  }
@Test
  public void fullRowToMessage() {
    RowHandler rowHandler = new RowHandler(FULL_WRITE_SCHEMA);
    Instant now = Instant.now();
    Row row =
        Row.withSchema(FULL_WRITE_SCHEMA)
            .withFieldValue(RowHandler.MESSAGE_KEY_FIELD, "val1".getBytes(UTF_8))
            .withFieldValue(RowHandler.PAYLOAD_FIELD, "val2".getBytes(UTF_8))
            .withFieldValue(RowHandler.EVENT_TIMESTAMP_FIELD, now)
            .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();
    PubSubMessage expected =
        PubSubMessage.newBuilder()
            .setKey(ByteString.copyFromUtf8("val1"))
            .setData(ByteString.copyFromUtf8("val2"))
            .setEventTime(Timestamps.fromMillis(now.getMillis()))
            .putAttributes(
                "key1",
                AttributeValues.newBuilder()
                    .addValues(ByteString.copyFromUtf8("attr1"))
                    .addValues(ByteString.copyFromUtf8("attr2"))
                    .build())
            .putAttributes(
                "key2",
                AttributeValues.newBuilder().addValues(ByteString.copyFromUtf8("attr3")).build())
            .build();
    assertEquals(expected, rowHandler.rowToMessage(row));
  }