@Override
  public Serde<GenericKey> create(
      final FormatInfo format,
      final PersistenceSchema schema,
      final KsqlConfig ksqlConfig,
      final Supplier<SchemaRegistryClient> schemaRegistryClientFactory,
      final String loggerNamePrefix,
      final ProcessingLogContext processingLogContext,
      final Optional<TrackedCallback> tracker
  ) {
    return createInner(
        format,
        schema,
        ksqlConfig,
        schemaRegistryClientFactory,
        loggerNamePrefix,
        processingLogContext,
        tracker
    );
  }
@Test
  public void shouldThrowOnMapKeyColumn() {
    // Given:
    schema = PersistenceSchema.from(
        ImmutableList.of(column(SqlTypes.map(SqlTypes.STRING, SqlTypes.STRING))),
        SerdeFeatures.of()
    );

    // When:
    final Exception e = assertThrows(
        KsqlException.class,
        () -> factory
            .create(format, schema, config, srClientFactory, LOGGER_PREFIX, processingLogCxt,
                Optional.empty())
    );

    // Then:
    assertThat(e.getMessage(), containsString(
        "Map keys, including types that contain maps, are not supported as they may lead "
            + "to unexpected behavior due to inconsistent serialization. Key column name: `foo`. "
            + "Column type: MAP<STRING, STRING>. "
            + "See https://github.com/confluentinc/ksql/issues/6621 for more."));
  }