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 reorderRowToMessage() {
    Schema schema =
        Schema.builder()
            .addByteArrayField(RowHandler.MESSAGE_KEY_FIELD)
            .addByteArrayField(RowHandler.PAYLOAD_FIELD)
            .build();
    Schema rowSchema =
        Schema.builder()
            .addByteArrayField(RowHandler.PAYLOAD_FIELD)
            .addByteArrayField(RowHandler.MESSAGE_KEY_FIELD)
            .build();
    RowHandler rowHandler = new RowHandler(schema);
    Row row = Row.withSchema(rowSchema).attachValues("abc".getBytes(UTF_8), "def".getBytes(UTF_8));
    PubSubMessage expected =
        PubSubMessage.newBuilder()
            .setData(ByteString.copyFromUtf8("abc"))
            .setKey(ByteString.copyFromUtf8("def"))
            .build();
    assertEquals(expected, rowHandler.rowToMessage(row));
  }