@VisibleForTesting
  Row transformInput(KafkaRecord<byte[], byte[]> record) {
    Row.FieldValueBuilder builder = Row.withSchema(getSchema()).withFieldValues(ImmutableMap.of());
    if (schema.hasField(Schemas.MESSAGE_KEY_FIELD)) {
      builder.withFieldValue(Schemas.MESSAGE_KEY_FIELD, record.getKV().getKey());
    }
    if (schema.hasField(Schemas.EVENT_TIMESTAMP_FIELD)) {
      builder.withFieldValue(
          Schemas.EVENT_TIMESTAMP_FIELD, Instant.ofEpochMilli(record.getTimestamp()));
    }
    if (schema.hasField(Schemas.HEADERS_FIELD)) {
      @Nullable Headers recordHeaders = record.getHeaders();
      if (recordHeaders != null) {
        ImmutableListMultimap.Builder<String, byte[]> headersBuilder =
            ImmutableListMultimap.builder();
        recordHeaders.forEach(header -> headersBuilder.put(header.key(), header.value()));
        ImmutableList.Builder<Row> listBuilder = ImmutableList.builder();
        headersBuilder
            .build()
            .asMap()
            .forEach(
                (key, values) -> {
                  Row entry =
                      Row.withSchema(Schemas.HEADERS_ENTRY_SCHEMA)
                          .withFieldValue(Schemas.HEADERS_KEY_FIELD, key)
                          .withFieldValue(Schemas.HEADERS_VALUES_FIELD, values)
                          .build();
                  listBuilder.add(entry);
                });
        builder.withFieldValue(Schemas.HEADERS_FIELD, listBuilder.build());
      }
    }
    if (payloadSerializer == null) {
      builder.withFieldValue(Schemas.PAYLOAD_FIELD, record.getKV().getValue());
    } else {
      byte[] payload = record.getKV().getValue();
      if (payload != null) {
        builder.withFieldValue(
            Schemas.PAYLOAD_FIELD, payloadSerializer.deserialize(record.getKV().getValue()));
      }
    }
    return builder.build();
  }
@Test
  public void recordToRowFailures() {
    {
      Schema payloadSchema = Schema.builder().addStringField("def").build();
      NestedPayloadKafkaTable table =
          newTable(
              Schema.builder()
                  .addRowField(Schemas.PAYLOAD_FIELD, payloadSchema)
                  .addField(Schemas.HEADERS_FIELD, Schemas.HEADERS_FIELD_TYPE)
                  .build(),
              Optional.of(serializer));
      doThrow(new IllegalArgumentException("")).when(serializer).deserialize(any());
      assertThrows(
          IllegalArgumentException.class,
          () ->
              table.transformInput(
                  readRecord(
                      new byte[] {}, "abc".getBytes(UTF_8), 123, ImmutableListMultimap.of())));
    }
    // Schema requires headers, missing in message
    {
      NestedPayloadKafkaTable table =
          newTable(
              Schema.builder()
                  .addByteArrayField(Schemas.PAYLOAD_FIELD)
                  .addField(Schemas.HEADERS_FIELD, Schemas.HEADERS_FIELD_TYPE)
                  .build(),
              Optional.empty());
      assertThrows(
          IllegalArgumentException.class,
          () ->
              table.transformInput(
                  new KafkaRecord<>(
                      TOPIC,
                      0,
                      0,
                      0,
                      KafkaTimestampType.LOG_APPEND_TIME,
                      null,
                      new byte[] {},
                      new byte[] {})));
    }
  }