static Serde<List<?>> createSerde(final PersistenceSchema schema) {
    final List<SimpleColumn> columns = schema.columns();
    if (columns.isEmpty()) {
      // No columns:
      return new KsqlVoidSerde<>();
    }

    if (columns.size() != 1) {
      throw new KsqlException("The '" + FormatFactory.KAFKA.name()
          + "' format only supports a single field. Got: " + columns);
    }

    final SimpleColumn singleColumn = columns.get(0);

    final Class<?> javaType = SchemaConverters.sqlToJavaConverter()
        .toJavaType(singleColumn.type());

    return createSerde(singleColumn, javaType);
  }
@Test
  public void shouldThroIfMultipleFields() {
    // Given:
    final PersistenceSchema schema = getPersistenceSchema(LogicalSchema.builder()
        .keyColumn(SystemColumns.ROWKEY_NAME, SqlTypes.STRING)
        .valueColumn(ColumnName.of("f0"), SqlTypes.INTEGER)
        .valueColumn(ColumnName.of("f1"), SqlTypes.BIGINT)
        .build());

    // When:
    final Exception e = assertThrows(
        KsqlException.class,
        () -> KafkaSerdeFactory.createSerde(schema)
    );

    // Then:
    assertThat(e.getMessage(), containsString(
        "The 'KAFKA' format only supports a single field. Got: [`f0` INTEGER, `f1` BIGINT]"));
  }