@Override
    protected void containerIsStarting(InspectContainerResponse containerInfo) {
        super.containerIsStarting(containerInfo);

        List<String> advertisedListeners = new ArrayList<>();
        advertisedListeners.add(getBootstrapServers());
        advertisedListeners.add(brokerAdvertisedListener(containerInfo));

        List<Supplier<String>> listenersToTransform = new ArrayList<>(getContainerDef().listeners);
        for (int i = 0; i < listenersToTransform.size(); i++) {
            Supplier<String> listenerSupplier = listenersToTransform.get(i);
            String protocol = String.format("%s-%d", PROTOCOL_PREFIX, i);
            String listener = listenerSupplier.get();
            String listenerProtocol = String.format("%s://%s", protocol, listener);
            advertisedListeners.add(listenerProtocol);
        }

        String kafkaAdvertisedListeners = String.join(",", advertisedListeners);

        String command = "#!/bin/bash\n";
        // exporting KAFKA_ADVERTISED_LISTENERS with the container hostname
        command += String.format("export KAFKA_ADVERTISED_LISTENERS=%s\n", kafkaAdvertisedListeners);

        if (!this.kraftEnabled || isLessThanCP740()) {
            // Optimization: skip the checks
            command += "echo '' > /etc/confluent/docker/ensure \n";
        }

        if (this.kraftEnabled) {
            command += commandKraft();
        } else if (this.externalZookeeperConnect == null) {
            command += commandZookeeper();
        }

        // Run the original command
        command += "/etc/confluent/docker/run \n";
        copyFileToContainer(Transferable.of(command, 0777), STARTER_SCRIPT);
    }
@deprecated use {@link #KafkaContainer(DockerImageName)} instead
modules/kafka/src/main/java/org/testcontainers/containers/KafkaContainer.java