@UtilityClass
@Slf4j
public class KafkaInitializer {

    private static final String EMP_KAFKA_BOOTSTRAP_SERVERS_PROPERTY_NAME = "kafka.bootstrap-servers=";
    private static final String AUTH_USERNAME = "kafka-test";
    private static final String AUTH_PASSWORD = "kafka-test-password";

    private static final Map<String, String> KAFKA_ENV_WITH_AUTHORIZATION =  Map.ofEntries(

            Map.entry("KAFKA_SASL_ENABLED_MECHANISMS", "SCRAM-SHA-256,PLAIN"),
            Map.entry("KAFKA_LISTENER_NAME_PLAINTEXT_SASL_ENABLED_MECHANISMS", "SCRAM-SHA-256"),
            Map.entry("KAFKA_LISTENER_NAME_BROKER_SASL_ENABLED_MECHANISMS", "SCRAM-SHA-256"),
            Map.entry("KAFKA_LISTENER_NAME_CONTROLLER_SASL_ENABLED_MECHANISMS", "PLAIN"),
            Map.entry("KAFKA_SASL_MECHANISM_INTER_BROKER_PROTOCOL", "SCRAM-SHA-256"),
            Map.entry("KAFKA_SASL_MECHANISM_CONTROLLER_PROTOCOL", "PLAIN"),

            Map.entry("KAFKA_LISTENER_SECURITY_PROTOCOL_MAP", "BROKER:SASL_PLAINTEXT,CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT"),

            Map.entry("KAFKA_LISTENER_NAME_PLAINTEXT_SCRAM-SHA-256_SASL_JAAS_CONFIG",
                      "org.apache.kafka.common.security.scram.ScramLoginModule required username='%s' password='%s';"
                              .formatted(AUTH_USERNAME, AUTH_PASSWORD)),

            Map.entry("KAFKA_LISTENER_NAME_BROKER_SCRAM-SHA-256_SASL_JAAS_CONFIG",
                      "org.apache.kafka.common.security.scram.ScramLoginModule required username='%s' password='%s';"
                              .formatted(AUTH_USERNAME, AUTH_PASSWORD)),

            Map.entry("KAFKA_LISTENER_NAME_CONTROLLER_PLAIN_SASL_JAAS_CONFIG",
                      "org.apache.kafka.common.security.plain.PlainLoginModule required username='%s' password='%s';"
                              .formatted(AUTH_USERNAME, AUTH_PASSWORD))
    );

    @Container
    private static final ConfluentKafkaContainer KAFKA_CONTAINER = new ConfluentKafkaContainer(
            DockerImageName.parse("confluentinc/cp-kafka:7.7.0")
    );

    public static class Initializer implements ApplicationContextInitializer<ConfigurableApplicationContext> {

        @Override
        public void initialize(@NotNull ConfigurableApplicationContext applicationContext) {
            try {
           
                KAFKA_CONTAINER.withEnv(KAFKA_ENV_WITH_AUTHORIZATION)
                        .start();


            KAFKA_CONTAINER.execInContainer(
                        "/usr/bin/kafka-storage", "format", "-t", " $CLUSTER_ID",
                        "--config", "/etc/kafka/kraft/server.properties",
                        "--add-scram", "SCRAM-SHA-256=[name=%s,password=%s]".formatted(AUTH_USERNAME, AUTH_PASSWORD),
                        "--ignore-formatted"
                );

                log.info("KafkaContainer for tests started");
            } catch (Throwable throwable) {
                log.error("KafkaContainer not started: {}", throwable.getMessage());
            }

            try {
                String propertyBootstrapServers = EMP_KAFKA_BOOTSTRAP_SERVERS_PROPERTY_NAME + KAFKA_CONTAINER.getBootstrapServers();
                TestPropertyValues.of(
                        propertyBootstrapServers
                ).applyTo(applicationContext.getEnvironment());
                log.info("TestPropertyValues of KafkaContainer: propertyBootstrapServers = {} was applied", propertyBootstrapServers);
            } catch (Throwable throwable) {
                log.error(
                        "TestPropertyValues of KafkaContainer don't apply for applicationContext: {}",
                        throwable.getMessage()
                );
                throw throwable;
            }
        }
    }
}

  if (authUsername != null && authPassword != null) {
            result.put(SECURITY_PROTOCOL_CONFIG, "SASL_PLAINTEXT");
            result.put(SASL_MECHANISM, "SCRAM-SHA-256");
            result.put(SASL_JAAS_CONFIG, String.format(SASL_JAAS_CONFIG_TEMPLATE, authUsername, authPassword));
        }

 private static final Map<String, String> KAFKA_ENV_WITH_AUTHORIZATION = Map.of(
            "KAFKA_LISTENER_SECURITY_PROTOCOL_MAP", "BROKER:PLAINTEXT,PLAINTEXT:SASL_PLAINTEXT",
            "KAFKA_LISTENER_NAME_PLAINTEXT_SASL_ENABLED_MECHANISMS", ScramMechanism.SCRAM_SHA_256.mechanismName(),
            "KAFKA_LISTENER_NAME_PLAINTEXT_SCRAM-SHA-256_SASL_JAAS_CONFIG",
            "org.apache.kafka.common.security.scram.ScramLoginModule required username='%s' password='%s';"
                    .formatted(AUTH_USERNAME, AUTH_PASSWORD)
    );

   @Container
    private static final KafkaContainer KAFKA_CONTAINER = new KafkaContainer(
            DockerImageName.parse("confluentinc/cp-kafka:7.7.0")
    );

    public static class Initializer implements ApplicationContextInitializer<ConfigurableApplicationContext> {

        @Override
        public void initialize(@NotNull ConfigurableApplicationContext applicationContext) {
            try {
      
                KAFKA_CONTAINER.withEnv(KAFKA_ENV_WITH_AUTHORIZATION).start();

                KAFKA_CONTAINER.execInContainer(
                        "kafka-configs", "--zookeeper", "localhost:2181", "--alter",
                        "--add-config", "SCRAM-SHA-256=[iterations=4096,password=%s]".formatted(AUTH_PASSWORD),
                        "--entity-type", "users", "--entity-name", AUTH_USERNAME
                );
                log.info("KafkaContainer for tests started");