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

    try {
      return format
          .getSerde(schema,
              formatInfo.getProperties(),
              ksqlConfig,
              schemaRegistryClientFactory,
              isKey
          );
    } 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
      );
    }
  }
@Test
  public void shouldInvokeFormatWithCorrectParams() {
    // When:
    serdeFactory.createFormatSerde("target", formatInfo, schema, config, srClientFactory, false);

    // Then:
    verify(format).getSerde(schema, formatProperties, config, srClientFactory, false);
  }