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 shouldThrowIfGetSerdeThrows() {
    // Given:
    when(format.getSerde(any(), any(), any(), any(), eq(false))).thenThrow(new RuntimeException("boom"));

    // When:
    final Exception actual = assertThrows(
        SchemaNotSupportedException.class,
        () -> serdeFactory
            .createFormatSerde("Target-A", formatInfo, schema, config, srClientFactory, false)
    );

    // Then:
    assertThat(actual.getMessage(), is("Target-A format does not support schema."
        + System.lineSeparator()
        + "format: FormatName"
        + System.lineSeparator()
        + "schema: schemaText"
        + System.lineSeparator()
        + "reason: boom"
    ));
  }