@Override
  public Serde<List<?>> getSerde(
      final PersistenceSchema schema,
      final Map<String, String> formatProps,
      final KsqlConfig config,
      final Supplier<SchemaRegistryClient> srFactory,
      final boolean isKey
  ) {
    SerdeUtils.throwOnUnsupportedFeatures(schema.features(), supportedFeatures());

    final ConnectSchema outerSchema = ConnectSchemas.columnsToConnectSchema(schema.columns());
    final ConnectSchema innerSchema = SerdeUtils
        .applySinglesUnwrapping(outerSchema, schema.features());

    final Class<?> targetType = SchemaConverters.connectToJavaTypeConverter()
        .toJavaType(innerSchema);

    return schema.features().enabled(SerdeFeature.UNWRAP_SINGLES)
        ? handleUnwrapped(innerSchema, formatProps, config, srFactory, targetType, isKey)
        : handleWrapped(innerSchema, formatProps, config, srFactory, targetType, isKey);
  }
@Test
  public void shouldThrowOnSerializationIfStructColumnValueDoesNotMatchSchema() {
    // Given:
    final SimpleColumn singleColumn = createColumn(
        "bob",
        SqlTypes.struct()
            .field("vic", SqlTypes.STRING)
            .build()
    );
    when(persistenceSchema.columns()).thenReturn(ImmutableList.of(singleColumn));

    final ConnectSchema connectSchema = (ConnectSchema) SchemaBuilder.struct()
        .field("vic", Schema.STRING_SCHEMA)
        .build();

    final Serializer<List<?>> serializer = format
        .getSerde(persistenceSchema, formatProps, config, srFactory, false)
        .serializer();

    final List<?> values = ImmutableList.of(new Struct(connectSchema));

    // When:
    final Exception e = assertThrows(
        SerializationException.class,
        () -> serializer.serialize("topicName", values)
    );

    // Then:
    assertThat(e.getMessage(), is(
        "Failed to prepare Struct value field 'bob' for serialization. "
            + "This could happen if the value was produced by a user-defined function "
            + "where the schema has non-optional return types. ksqlDB requires all "
            + "schemas to be optional at all levels of the Struct: the Struct itself, "
            + "schemas for all fields within the Struct, and so on."
    ));
  }