@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 fullRowToRecord() {
    NestedPayloadKafkaTable table = newTable(FULL_WRITE_SCHEMA, Optional.empty());
    Instant now = Instant.now();
    Row row =
        Row.withSchema(FULL_WRITE_SCHEMA)
            .withFieldValue(Schemas.MESSAGE_KEY_FIELD, "val1".getBytes(UTF_8))
            .withFieldValue(Schemas.PAYLOAD_FIELD, "val2".getBytes(UTF_8))
            .withFieldValue(Schemas.EVENT_TIMESTAMP_FIELD, now)
            .withFieldValue(
                Schemas.HEADERS_FIELD,
                ImmutableList.of(
                    Row.withSchema(Schemas.HEADERS_ENTRY_SCHEMA)
                        .attachValues(
                            "key1",
                            ImmutableList.of("attr1".getBytes(UTF_8), "attr2".getBytes(UTF_8))),
                    Row.withSchema(Schemas.HEADERS_ENTRY_SCHEMA)
                        .attachValues("key2", ImmutableList.of("attr3".getBytes(UTF_8)))))
            .build();
    ProducerRecord<byte[], byte[]> result = table.transformOutput(row);
    assertEquals("val1", new String(result.key(), UTF_8));
    assertEquals("val2", new String(result.value(), UTF_8));
    assertEquals(now.getMillis(), result.timestamp().longValue());
    List<Header> key1Headers = ImmutableList.copyOf(result.headers().headers("key1"));
    List<Header> key2Headers = ImmutableList.copyOf(result.headers().headers("key2"));
    assertEquals(2, key1Headers.size());
    assertEquals(1, key2Headers.size());
    assertEquals("attr3", new String(key2Headers.get(0).value(), UTF_8));
  }