@SuppressWarnings("argument")
  @VisibleForTesting
  ProducerRecord<byte[], byte[]> transformOutput(Row row) {
    row = castRow(row, row.getSchema(), schema);
    String topic = Iterables.getOnlyElement(getTopics());
    byte[] key = null;
    byte[] payload;
    List<Header> headers = ImmutableList.of();
    Long timestampMillis = null;
    if (schema.hasField(Schemas.MESSAGE_KEY_FIELD)) {
      key = row.getBytes(Schemas.MESSAGE_KEY_FIELD);
    }
    if (schema.hasField(Schemas.EVENT_TIMESTAMP_FIELD)) {
      ReadableDateTime time = row.getDateTime(Schemas.EVENT_TIMESTAMP_FIELD);
      if (time != null) {
        timestampMillis = time.getMillis();
      }
    }
    if (schema.hasField(Schemas.HEADERS_FIELD)) {
      Collection<Row> headerRows = checkArgumentNotNull(row.getArray(Schemas.HEADERS_FIELD));
      ImmutableList.Builder<Header> headersBuilder = ImmutableList.builder();
      headerRows.forEach(
          entry -> {
            String headerKey = checkArgumentNotNull(entry.getString(Schemas.HEADERS_KEY_FIELD));
            Collection<byte[]> values =
                checkArgumentNotNull(entry.getArray(Schemas.HEADERS_VALUES_FIELD));
            values.forEach(value -> headersBuilder.add(new RecordHeader(headerKey, value)));
          });
      headers = headersBuilder.build();
    }
    if (payloadSerializer == null) {
      payload = row.getBytes(Schemas.PAYLOAD_FIELD);
    } else {
      payload =
          payloadSerializer.serialize(checkArgumentNotNull(row.getRow(Schemas.PAYLOAD_FIELD)));
    }
    return new ProducerRecord<>(topic, null, timestampMillis, key, payload, headers);
  }
@Test
  public void rowToRecordFailures() {
    Schema payloadSchema = Schema.builder().addStringField("def").build();
    Schema schema =
        Schema.builder()
            .addRowField(Schemas.PAYLOAD_FIELD, payloadSchema)
            .addField(Schemas.HEADERS_FIELD, Schemas.HEADERS_FIELD_TYPE.withNullable(true))
            .build();
    NestedPayloadKafkaTable table = newTable(schema, Optional.of(serializer));
    // badRow cannot be cast to schema
    Schema badRowSchema = Schema.builder().addStringField("xxx").build();
    Row badRow =
        Row.withSchema(badRowSchema).attachValues(Row.withSchema(badRowSchema).attachValues("abc"));
    assertThrows(IllegalArgumentException.class, () -> table.transformOutput(badRow));

    Row goodRow =
        Row.withSchema(schema)
            .withFieldValue(
                Schemas.PAYLOAD_FIELD,
                Row.withSchema(payloadSchema).withFieldValue("def", "abc").build())
            .build();
    doThrow(new IllegalArgumentException("")).when(serializer).serialize(any());
    assertThrows(IllegalArgumentException.class, () -> table.transformOutput(goodRow));
  }