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 messageToRowFailures() {
    {
      Schema payloadSchema = Schema.builder().addStringField("def").build();
      RowHandler rowHandler =
          new RowHandler(
              Schema.builder().addRowField(RowHandler.PAYLOAD_FIELD, payloadSchema).build(),
              serializer);
      doThrow(new IllegalArgumentException("")).when(serializer).deserialize(any());
      assertThrows(
          IllegalArgumentException.class,
          () -> rowHandler.messageToRow(SequencedMessage.getDefaultInstance()));
    }
    // Schema requires event time, missing in message
    {
      RowHandler rowHandler =
          new RowHandler(
              Schema.builder()
                  .addByteArrayField(RowHandler.PAYLOAD_FIELD)
                  .addDateTimeField(RowHandler.EVENT_TIMESTAMP_FIELD)
                  .build());
      assertThrows(
          IllegalArgumentException.class,
          () -> rowHandler.messageToRow(SequencedMessage.getDefaultInstance()));
    }
  }