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 shouldThrowIfDecimal() {
    // Given:
    final PersistenceSchema schema = schemaWithFieldOfType(SqlTypes.decimal(1, 1));

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

    // Then:
    assertThat(e.getMessage(), containsString("The 'KAFKA' format does not support type 'DECIMAL'"));
  }