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 shouldHandleEmptyKey() {
    // Given:
    final LogicalSchema logical = LogicalSchema.builder()
        .valueColumn(ColumnName.of("f0"), SqlTypes.INTEGER)
        .build();

    final PersistenceSchema schema = PhysicalSchema
        .from(logical, SerdeFeatures.of(), SerdeFeatures.of())
        .keySchema();

    final Serde<List<?>> serde = KafkaSerdeFactory.createSerde(schema);

    // When:
    final byte[] bytes = serde.serializer().serialize("topic", ImmutableList.of());
    final Object result = serde.deserializer().deserialize("topic", null);

    // Then:
    assertThat(bytes, is(nullValue()));
    assertThat(result, is(nullValue()));
  }