class ConnectFormat implements Format {

  @Override
  public Serde<List<?>> getSerde(
      final PersistenceSchema schema,
      final Map<String, String> formatProps,
      final KsqlConfig config,
      final Supplier<SchemaRegistryClient> srFactory
  ) {
    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)
        : handleWrapped(innerSchema, formatProps, config, srFactory, targetType);
  }

}