@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 fullRecordToRow() {
    NestedPayloadKafkaTable table = newTable(FULL_READ_SCHEMA, Optional.empty());
    Instant event = Instant.now();
    KafkaRecord<byte[], byte[]> record =
        readRecord(
            "key".getBytes(UTF_8),
            "value".getBytes(UTF_8),
            event.getMillis(),
            ImmutableListMultimap.of(
                "key1", "attr1".getBytes(UTF_8),
                "key1", "attr2".getBytes(UTF_8),
                "key2", "attr3".getBytes(UTF_8)));
    Row expected =
        Row.withSchema(FULL_READ_SCHEMA)
            .withFieldValue(Schemas.MESSAGE_KEY_FIELD, "key".getBytes(UTF_8))
            .withFieldValue(Schemas.PAYLOAD_FIELD, "value".getBytes(UTF_8))
            .withFieldValue(Schemas.EVENT_TIMESTAMP_FIELD, event)
            .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();
    assertEquals(expected, table.transformInput(record));
  }