@Override
  public <T> Serde<T> createSerde(
      final Schema schema,
      final KsqlConfig ksqlConfig,
      final Supplier<SchemaRegistryClient> srFactory,
      final Class<T> targetType,
      final boolean isKey
  ) {
    validateSchema(schema);

    final Optional<Schema> physicalSchema;
    if (useSchemaRegistryFormat) {
      physicalSchema = properties.getSchemaId().isPresent() ? Optional.of(
          SerdeUtils.getAndTranslateSchemaById(srFactory, properties.getSchemaId()
              .get(), new JsonSchemaTranslator())) : Optional.empty();
    } else {
      physicalSchema = Optional.empty();
    }

    final Converter converter = useSchemaRegistryFormat
        ? getSchemaRegistryConverter(srFactory.get(), ksqlConfig, properties.getSchemaId(), isKey)
        : getConverter();

    // The translators are used in the serializer & deserializzer only for JSON_SR formats
    final ConnectDataTranslator dataTranslator = physicalSchema.isPresent()
            ? new ConnectSRSchemaDataTranslator(physicalSchema.get())
            : new ConnectDataTranslator(schema);

    final Supplier<Serializer<T>> serializer = () -> createSerializer(
        targetType,
        dataTranslator,
        converter
    );

    final Deserializer<T> deserializer = createDeserializer(
        ksqlConfig,
        schema,
        targetType,
        dataTranslator,
        converter
    );

    // Sanity check:
    serializer.get();

    return Serdes.serdeFrom(
        new ThreadLocalSerializer<>(serializer),
        deserializer
    );
  }
@Test
  public void shouldThrowOnNestedMapWithNoneStringKeys() {
    // Given
    final ConnectSchema schemaWithNestedInvalidMap = (ConnectSchema) SchemaBuilder
        .struct()
        .field("f0", SchemaBuilder
            .map(Schema.OPTIONAL_BOOLEAN_SCHEMA, Schema.OPTIONAL_STRING_SCHEMA)
            .optional()
            .build())
        .build();

    // When:
    final Exception e = assertThrows(
        KsqlException.class,
        () -> jsonFactory.createSerde(schemaWithNestedInvalidMap, config, srFactory, String.class, false)
    );

    // Then:
    assertThat(e.getMessage(), containsString(
        "JSON only supports MAP types with STRING keys"));
  }