@Override
  public ParsedSchema toParsedSchema(final PersistenceSchema schema) {
    SerdeUtils.throwOnUnsupportedFeatures(schema.features(), format.supportedFeatures());

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

    return connectSrTranslator.fromConnectSchema(innerSchema);
  }
@Test
  public void shouldThrowOnUnsupportedFeatureWhenBuildingParsedSchema() {
    // When:
    final Exception e = assertThrows(IllegalArgumentException.class,
        () -> translator.toParsedSchema(
            PersistenceSchema.from(
                ImmutableList.of(createColumn("vic", SqlTypes.STRING)),
                SerdeFeatures.of(SerdeFeature.UNWRAP_SINGLES)
            )
        )
    );

    // Then:
    assertThat(e.getMessage(), containsString("Unsupported feature"));
  }