class ConnectFormatSchemaTranslator implements SchemaTranslator {

  @Override
  public List<SimpleColumn> toColumns(final ParsedSchema schema) {
    Schema connectSchema = connectSrTranslator.toConnectSchema(schema);

    if (connectSchema.type() != Type.STRUCT) {
      if (!format.supportsFeature(SerdeFeature.UNWRAP_SINGLES)) {
        throw new KsqlException("Schema returned from schema registry is anonymous type, "
            + "but format " + format.name() + " does not support anonymous types. "
            + "schema: " + schema);
      }

      connectSchema = SerdeUtils.wrapSingle(connectSchema);
    }

    final Schema rowSchema = connectKsqlTranslator.apply(connectSchema);

    return rowSchema.fields().stream()
        .map(ConnectFormatSchemaTranslator::toColumn)
        .collect(Collectors.toList());
  }

    ConnectFormatSchemaTranslator(
      final ConnectFormat format,
      final Map<String, String> formatProps,
      final Function<Schema, Schema> connectKsqlTranslator
  );

}