@Override
  public List<SimpleColumn> toColumns(
      final ParsedSchema schema,
      final SerdeFeatures serdeFeatures,
      final boolean isKey) {
    SerdeUtils.throwOnUnsupportedFeatures(serdeFeatures, format.supportedFeatures());

    Schema connectSchema = connectSrTranslator.toConnectSchema(schema);

    if (serdeFeatures.enabled(SerdeFeature.UNWRAP_SINGLES)) {
      connectSchema = SerdeUtils.wrapSingle(connectSchema, isKey);
    }

    if (connectSchema.type() != Type.STRUCT) {
      if (isKey) {
        throw new IllegalStateException("Key schemas are always unwrapped.");
      }

      throw new KsqlException("Schema returned from schema registry is anonymous type. "
          + "To use this schema with ksqlDB, set '" + CommonCreateConfigs.WRAP_SINGLE_VALUE
          + "=false' in the WITH clause properties.");
    }

    final Schema rowSchema = connectKsqlTranslator.toKsqlSchema(connectSchema);

    return rowSchema.fields().stream()
        .map(ConnectFormatSchemaTranslator::toColumn)
        .collect(Collectors.toList());
  }
@Test
  public void shouldThrowOnUnwrappedSchemaIfUnwrapSingleIsFalse() {
    // Given:
    when(connectSchema.type()).thenReturn(Type.INT32);
    when(format.supportedFeatures()).thenReturn(Collections.singleton(SerdeFeature.UNWRAP_SINGLES));

    // When:
    final Exception e = assertThrows(KsqlException.class,
        () -> translator.toColumns(parsedSchema, SerdeFeatures.of(), false)
    );

    // Then:
    assertThat(e.getMessage(), is("Schema returned from schema registry is anonymous type. "
        + "To use this schema with ksqlDB, set 'WRAP_SINGLE_VALUE=false' in the "
        + "WITH clause properties."));
  }