#vulnerable code 
                                @Bean(name = KAFKA_BEAN_NAME, destroyMethod = "stop")
    @DependsOn("zookeeper")
    public GenericContainer kafka(
            KafkaStatusCheck kafkaStatusCheck,
            KafkaConfigurationProperties kafkaProperties,
            @Value("${embedded.zookeeper.containerZookeeperConnect}") String containerZookeeperConnect,
            ConfigurableEnvironment environment,
            Network network) {

        int kafkaInternalPort = kafkaProperties.getContainerBrokerPort(); // for access from other containers
        int kafkaExternalPort = kafkaProperties.getBrokerPort();  // for access from host
        int saslPlaintextKafkaExternalPort = kafkaProperties.getSaslPlaintextBrokerPort();
        // https://docs.confluent.io/current/installation/docker/docs/configuration.html search by KAFKA_ADVERTISED_LISTENERS

        log.info("Starting kafka broker. Docker image: {}", kafkaProperties.getDockerImage());

        GenericContainer kafka = new FixedHostPortGenericContainer<>(kafkaProperties.getDockerImage())
                .withLogConsumer(containerLogsConsumer(log))
                .withCreateContainerCmdModifier(cmd -> cmd.withHostName(KAFKA_HOST_NAME))
                .withCreateContainerCmdModifier(cmd -> cmd.withCapAdd(Capability.NET_ADMIN))
                .withEnv("KAFKA_ZOOKEEPER_CONNECT", containerZookeeperConnect)
                .withEnv("KAFKA_BROKER_ID", "-1")
                //see: https://stackoverflow.com/questions/41868161/kafka-in-kubernetes-cluster-how-to-publish-consume-messages-from-outside-of-kub
                //see: https://github.com/wurstmeister/kafka-docker/blob/master/README.md
                // order matters: external then internal since kafka.client.ClientUtils.getPlaintextBrokerEndPoints take first for simple consumers
                .withEnv("KAFKA_LISTENER_SECURITY_PROTOCOL_MAP",
                        "EXTERNAL_PLAINTEXT:PLAINTEXT," +
                        "EXTERNAL_SASL_PLAINTEXT:SASL_PLAINTEXT," +
                        "INTERNAL_PLAINTEXT:PLAINTEXT"
                )
                .withEnv("KAFKA_ADVERTISED_LISTENERS",
                        "EXTERNAL_PLAINTEXT://" + kafkaHost() + ":" + kafkaExternalPort + "," +
                        "EXTERNAL_SASL_PLAINTEXT://" + kafkaHost() + ":" + saslPlaintextKafkaExternalPort + "," +
                        "INTERNAL_PLAINTEXT://" + KAFKA_HOST_NAME + ":" + kafkaInternalPort
                )
                .withEnv("KAFKA_LISTENERS",
                        "EXTERNAL_PLAINTEXT://0.0.0.0:" + kafkaExternalPort + "," +
                        "EXTERNAL_SASL_PLAINTEXT://0.0.0.0:" + saslPlaintextKafkaExternalPort + "," +
                        "INTERNAL_PLAINTEXT://0.0.0.0:" + kafkaInternalPort
                )
                .withEnv("KAFKA_INTER_BROKER_LISTENER_NAME", "INTERNAL_PLAINTEXT")
                .withEnv("KAFKA_OFFSETS_TOPIC_NUM_PARTITIONS", "1")
                .withEnv("KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR", String.valueOf(kafkaProperties.getReplicationFactor()))
                .withEnv("KAFKA_TRANSACTION_STATE_LOG_MIN_ISR", "1")
                .withEnv("KAFKA_CONFLUENT_SUPPORT_METRICS_ENABLE", "false")
                .withEnv("KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR", "1")
                .withEnv("KAFKA_LOG_FLUSH_INTERVAL_MS", String.valueOf(kafkaProperties.getLogFlushIntervalMs()))
                .withEnv("KAFKA_REPLICA_SOCKET_TIMEOUT_MS", String.valueOf(kafkaProperties.getReplicaSocketTimeoutMs()))
                .withEnv("KAFKA_CONTROLLER_SOCKET_TIMEOUT_MS", String.valueOf(kafkaProperties.getControllerSocketTimeoutMs()))
                .withEnv("KAFKA_SASL_ENABLED_MECHANISMS", "PLAIN")
                .withEnv("ZOOKEEPER_SASL_ENABLED", "false")
                .withCopyFileToContainer(MountableFile.forClasspathResource("kafka_server_jaas.conf"), "/etc/kafka/kafka_server_jaas.conf")
                .withEnv("KAFKA_OPTS", "-Djava.security.auth.login.config=/etc/kafka/kafka_server_jaas.conf")
                .withExposedPorts(kafkaInternalPort, kafkaExternalPort, saslPlaintextKafkaExternalPort)
                .withFixedExposedPort(kafkaInternalPort, kafkaInternalPort)
                .withFixedExposedPort(kafkaExternalPort, kafkaExternalPort)
                .withFixedExposedPort(saslPlaintextKafkaExternalPort, saslPlaintextKafkaExternalPort)
                .withNetwork(network)
                .withNetworkAliases(KAFKA_HOST_NAME)
                .withExtraHost(KAFKA_HOST_NAME, "127.0.0.1")
                .waitingFor(kafkaStatusCheck)
                .withStartupTimeout(kafkaProperties.getTimeoutDuration());

        KafkaConfigurationProperties.FileSystemBind fileSystemBind = kafkaProperties.getFileSystemBind();
        if (fileSystemBind.isEnabled()) {
            String currentTimestamp = LocalDateTime.now().format(DateTimeFormatter.ofPattern("HH-mm-ss-nnnnnnnnn"));
            String dataFolder = fileSystemBind.getDataFolder();
            String kafkaData = Paths.get(dataFolder, currentTimestamp).toAbsolutePath().toString();
            log.info("Writing kafka data to: {}", kafkaData);

            kafka.withFileSystemBind(kafkaData, "/var/lib/kafka/data", BindMode.READ_WRITE);
        }

        startAndLogTime(kafka);
        registerKafkaEnvironment(kafka, environment, kafkaProperties);
        return kafka;
    } 
                            #location 18 
                            #vulnerability type RESOURCE_LEAK
#fixed code
                                @Bean(name = KAFKA_BEAN_NAME, destroyMethod = "stop")
    @DependsOn("zookeeper")
    public GenericContainer kafka(
            KafkaStatusCheck kafkaStatusCheck,
            KafkaConfigurationProperties kafkaProperties,
            @Value("${embedded.zookeeper.containerZookeeperConnect}") String containerZookeeperConnect,
            ConfigurableEnvironment environment,
            Network network) {

        int kafkaInternalPort = kafkaProperties.getContainerBrokerPort(); // for access from other containers
        int kafkaExternalPort = kafkaProperties.getBrokerPort();  // for access from host
        int saslPlaintextKafkaExternalPort = kafkaProperties.getSaslPlaintextBrokerPort();
        // https://docs.confluent.io/current/installation/docker/docs/configuration.html search by KAFKA_ADVERTISED_LISTENERS

        String dockerImageVersion = kafkaProperties.getDockerImageVersion();
        log.info("Starting kafka broker. Docker image version: {}", dockerImageVersion);

        KafkaContainer kafka = new KafkaContainer(dockerImageVersion)
        {
            @Override
            public String getBootstrapServers() {
                super.getBootstrapServers();
                return "EXTERNAL_PLAINTEXT://" + getHost() + ":" + getMappedPort(kafkaExternalPort) + "," +
                        "EXTERNAL_SASL_PLAINTEXT://" + getHost() + ":" + getMappedPort(saslPlaintextKafkaExternalPort) + "," +
                        "INTERNAL_PLAINTEXT://" + KAFKA_HOST_NAME + ":" + kafkaInternalPort;
            }
        }
                .withLogConsumer(containerLogsConsumer(log))
                .withCreateContainerCmdModifier(cmd -> cmd.withHostName(KAFKA_HOST_NAME))
                .withCreateContainerCmdModifier(cmd -> cmd.withCapAdd(Capability.NET_ADMIN))
                .withExternalZookeeper(containerZookeeperConnect)
                .withEnv("KAFKA_BROKER_ID", "-1")
                //see: https://stackoverflow.com/questions/41868161/kafka-in-kubernetes-cluster-how-to-publish-consume-messages-from-outside-of-kub
                //see: https://github.com/wurstmeister/kafka-docker/blob/master/README.md
                // order matters: external then internal since kafka.client.ClientUtils.getPlaintextBrokerEndPoints take first for simple consumers
                .withEnv("KAFKA_LISTENER_SECURITY_PROTOCOL_MAP",
                        "EXTERNAL_PLAINTEXT:PLAINTEXT," +
                        "EXTERNAL_SASL_PLAINTEXT:SASL_PLAINTEXT," +
                        "INTERNAL_PLAINTEXT:PLAINTEXT," +
                        "BROKER:PLAINTEXT"
                )
                .withEnv("KAFKA_LISTENERS",
                        "EXTERNAL_PLAINTEXT://0.0.0.0:" + kafkaExternalPort + "," +
                        "EXTERNAL_SASL_PLAINTEXT://0.0.0.0:" + saslPlaintextKafkaExternalPort + "," +
                        "INTERNAL_PLAINTEXT://0.0.0.0:" + kafkaInternalPort + "," +
                        "BROKER://0.0.0.0:9092"
                )
                .withEnv("KAFKA_INTER_BROKER_LISTENER_NAME", "INTERNAL_PLAINTEXT")
                .withEnv("KAFKA_OFFSETS_TOPIC_NUM_PARTITIONS", "1")
                .withEnv("KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR", String.valueOf(kafkaProperties.getReplicationFactor()))
                .withEnv("KAFKA_TRANSACTION_STATE_LOG_MIN_ISR", "1")
                .withEnv("KAFKA_CONFLUENT_SUPPORT_METRICS_ENABLE", "false")
                .withEnv("KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR", "1")
                .withEnv("KAFKA_LOG_FLUSH_INTERVAL_MS", String.valueOf(kafkaProperties.getLogFlushIntervalMs()))
                .withEnv("KAFKA_REPLICA_SOCKET_TIMEOUT_MS", String.valueOf(kafkaProperties.getReplicaSocketTimeoutMs()))
                .withEnv("KAFKA_CONTROLLER_SOCKET_TIMEOUT_MS", String.valueOf(kafkaProperties.getControllerSocketTimeoutMs()))
                .withEnv("KAFKA_SASL_ENABLED_MECHANISMS", "PLAIN")
                .withEnv("ZOOKEEPER_SASL_ENABLED", "false")
                .withCopyFileToContainer(MountableFile.forClasspathResource("kafka_server_jaas.conf"), "/etc/kafka/kafka_server_jaas.conf")
                .withEnv("KAFKA_OPTS", "-Djava.security.auth.login.config=/etc/kafka/kafka_server_jaas.conf")
                .withExposedPorts(kafkaInternalPort, kafkaExternalPort, saslPlaintextKafkaExternalPort, KAFKA_PORT)
                .withNetwork(network)
                .withNetworkAliases(KAFKA_HOST_NAME)
                .withExtraHost(KAFKA_HOST_NAME, "127.0.0.1")
                .waitingFor(kafkaStatusCheck)
                .withStartupTimeout(kafkaProperties.getTimeoutDuration());

        KafkaConfigurationProperties.FileSystemBind fileSystemBind = kafkaProperties.getFileSystemBind();
        if (fileSystemBind.isEnabled()) {
            String currentTimestamp = LocalDateTime.now().format(DateTimeFormatter.ofPattern("HH-mm-ss-nnnnnnnnn"));
            String dataFolder = fileSystemBind.getDataFolder();
            String kafkaData = Paths.get(dataFolder, currentTimestamp).toAbsolutePath().toString();
            log.info("Writing kafka data to: {}", kafkaData);

            kafka.withFileSystemBind(kafkaData, "/var/lib/kafka/data", BindMode.READ_WRITE);
        }

        startAndLogTime(kafka);
        registerKafkaEnvironment(kafka, environment, kafkaProperties);
        return kafka;
    }