class GenericSerdeFactory {

  Serde<List<?>> createFormatSerde(
      final String target,
      final FormatInfo formatInfo,
      final PersistenceSchema schema,
      final KsqlConfig ksqlConfig,
      final Supplier<SchemaRegistryClient> schemaRegistryClientFactory
  ) {
    final Format format = formatFactory.apply(formatInfo);

    try {
      return format
          .getSerde(schema, formatInfo.getProperties(), ksqlConfig, schemaRegistryClientFactory);
    } catch (final Exception e) {
      throw new SchemaNotSupportedException(target + " format does not support schema."
          + System.lineSeparator()
          + "format: " + format.name()
          + System.lineSeparator()
          + "schema: " + schema
          + System.lineSeparator()
          + "reason: " + e.getMessage(),
          e
      );
    }
  }

    GenericSerdeFactory();
  @VisibleForTesting  GenericSerdeFactory(final Function<FormatInfo, Format> formatFactory);

}