Add support for kafka 3 with zookeeper support (adding images for 3.0.1, 3.1.1 and 3.2.0)

[lunarfs]

• What is it good for Support for Kafka 3.0.1, 3.1.1 and 3.2.0 with the regular ZooKeeper setup.

• What I did @jack12816 did some approach of getting to kafka 3.0.0 I have worked with his commits as basis for getting to 3.0.1,2.1.1 and 3.2.0 the PR is inspired by https://github.com/wurstmeister/kafka-docker/pull/688 and https://github.com/wurstmeister/kafka-docker/pull/691 I had to change some Kafka connection options (--zookeeper is deprecated since multiple versions, and since Kafka 3.0 it was removed -- replaced by the --bootstrap-server option. I did a fix for some tests for the new version/outputs I have added code to report the no longer supported KAFKA_ADVERTISED_HOST_NAME KAFKA_ADVERTISED_PORT and KAFKA_PORT and KAFKA_HOST_NAME (see https://github.com/apache/kafka/pull/10872 for more information). I have updated README.md to reflect that < 3.0 no longer supports KAFKA_ADVERTISED_HOST_NAME etc. So i have done some changes to start-kafka.sh to reflect the removed options.. it is a syntactic nightmare of if then else.. and there might be some edge cases i have not yet covered.. feel free to comment

• A picture of a cute animal (not mandatory but encouraged)

[jimbogithub]

@lunarfs Like #707 I'm a +1 for adding 3.0.1 support (3.1.0 doesn't have the bug fix I need). Could you add that as well?

[lunarfs]

@lunarfs Like #707 I'm a +1 for adding 3.0.1 support (3.1.0 doesn't have the bug fix I need). Could you add that as well?

I would be happy to supply 3.0.1 support, but I think I would like @wurstmeister to comment a bit on the format of how the approach to the 3.x is done in this PR, is it ok to do it like this or should we do some other approach with the tests. are there new corner cases we need to test.

[lunarfs]

@lunarfs Like #707 I'm a +1 for adding 3.0.1 support (3.1.0 doesn't have the bug fix I need). Could you add that as well?

added support for 3.0.1, 3.1.1 and 3.2.0

[jimbogithub]

@lunarfs I'm testing this out in my local environment. Looks pretty good except for a couple of issues if I have KAFKA_CREATE_TOPICS enabled:

1. KAFKA_PORT is used by create-topics.sh in order to poll for Kafka to be ready, hence I've had to patch out the startup reject of this setting.
2. create-topics.sh also requires BROKER_LIST for the 3.x style CONNECT_OPTS. This is easily added to the docker-compose environment, so probably just needs to be documented.

So far so good after making those changes.

[jimbogithub]

@lunarfs I'm testing this out in my local environment. Looks pretty good except for a couple of issues if I have KAFKA_CREATE_TOPICS enabled:

1. KAFKA_PORT is used by create-topics.sh in order to poll for Kafka to be ready, hence I've had to patch out the startup reject of this setting.
2. create-topics.sh also requires BROKER_LIST for the 3.x style CONNECT_OPTS. This is easily added to the docker-compose environment, so probably just needs to be documented.

So far so good after making those changes.

Maybe BROKER_LIST could be examined to determine the port instead of re-enabling KAFKA_PORT?

[lunarfs]

@lunarfs I'm testing this out in my local environment. Looks pretty good except for a couple of issues if I have KAFKA_CREATE_TOPICS enabled:

1. KAFKA_PORT is used by create-topics.sh in order to poll for Kafka to be ready, hence I've had to patch out the startup reject of this setting.
2. create-topics.sh also requires BROKER_LIST for the 3.x style CONNECT_OPTS. This is easily added to the docker-compose environment, so probably just needs to be documented.

So far so good after making those changes.

Maybe BROKER_LIST could be examined to determine the port instead of re-enabling KAFKA_PORT?

@jimbogithub thanks, this was a nice catch, I did not properly modify the tests to catch this scenario.. I have modified tests such that i could capture the scenario. I think the BROKER_LIST is not a valid source for this, but the KAFKA_LISTNERS has the internal port captured, so getting it from there should be fine I think.. and then we also do not need to give extra params in the compose file as KAFKA_LISTNERS is already available in create-topics.sh. For the test container however BROKER_LIST is the better option, so if KAFKA_LISTNERS is not available we will use BROKER_LIST.

[jimbogithub]

The topic creation is still problematic. It's not working with the example config:

HOSTNAME_COMMAND: curl http://169.254.169.254/latest/meta-data/public-hostname
KAFKA_ADVERTISED_LISTENERS: INSIDE://:9092,OUTSIDE://_{HOSTNAME_COMMAND}:9094
KAFKA_LISTENERS: INSIDE://:9092,OUTSIDE://:9094
KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: INSIDE:PLAINTEXT,OUTSIDE:PLAINTEXT
KAFKA_INTER_BROKER_LISTENER_NAME: INSIDE

I use very similar config myself and it results in:

kafka0_1     | Exception in thread "main" org.apache.kafka.common.KafkaException: Failed to create new KafkaAdminClient
kafka0_1     | 	at org.apache.kafka.clients.admin.KafkaAdminClient.createInternal(KafkaAdminClient.java:538)
kafka0_1     | 	at org.apache.kafka.clients.admin.KafkaAdminClient.createInternal(KafkaAdminClient.java:470)
kafka0_1     | 	at org.apache.kafka.clients.admin.Admin.create(Admin.java:133)
kafka0_1     | 	at kafka.admin.TopicCommand$TopicService$.createAdminClient(TopicCommand.scala:205)
kafka0_1     | 	at kafka.admin.TopicCommand$TopicService$.apply(TopicCommand.scala:209)
kafka0_1     | 	at kafka.admin.TopicCommand$.main(TopicCommand.scala:50)
kafka0_1     | 	at kafka.admin.TopicCommand.main(TopicCommand.scala)
kafka0_1     | Caused by: org.apache.kafka.common.config.ConfigException: Invalid url in bootstrap.servers: OUTSIDE
kafka0_1     | 	at org.apache.kafka.clients.ClientUtils.parseAndValidateAddresses(ClientUtils.java:59)
kafka0_1     | 	at org.apache.kafka.clients.ClientUtils.parseAndValidateAddresses(ClientUtils.java:48)
kafka0_1     | 	at org.apache.kafka.clients.admin.KafkaAdminClient.createInternal(KafkaAdminClient.java:490)

Maybe prefer BROKER_LIST when available so that it can always be overridden? Or look for KAFKA_INTER_BROKER_LISTENER_NAME to help find the internal address in the KAFKA_LISTENERS?

[dswiecki]

@lunarfs @jimbogithub are there available any preview docker images of this change?

[jimbogithub]

@dswiecki I have an image at jimbodock/kafka:2.13-3.1.1. It is slightly stripped down (removed the embedded Docker which is only needed for the CI tests). It is based on the build at the time of https://github.com/wurstmeister/kafka-docker/pull/709#issuecomment-1171028830 so supports KAFKA_PORT and BROKER_LIST to be able to create topics.

[mattathompson]

🙏 Excited for this.

[lunarfs]

The topic creation is still problematic. It's not working with the example config:

HOSTNAME_COMMAND: curl http://169.254.169.254/latest/meta-data/public-hostname
KAFKA_ADVERTISED_LISTENERS: INSIDE://:9092,OUTSIDE://_{HOSTNAME_COMMAND}:9094
KAFKA_LISTENERS: INSIDE://:9092,OUTSIDE://:9094
KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: INSIDE:PLAINTEXT,OUTSIDE:PLAINTEXT
KAFKA_INTER_BROKER_LISTENER_NAME: INSIDE

I use very similar config myself and it results in:

kafka0_1     | Exception in thread "main" org.apache.kafka.common.KafkaException: Failed to create new KafkaAdminClient
kafka0_1     | 	at org.apache.kafka.clients.admin.KafkaAdminClient.createInternal(KafkaAdminClient.java:538)
kafka0_1     | 	at org.apache.kafka.clients.admin.KafkaAdminClient.createInternal(KafkaAdminClient.java:470)
kafka0_1     | 	at org.apache.kafka.clients.admin.Admin.create(Admin.java:133)
kafka0_1     | 	at kafka.admin.TopicCommand$TopicService$.createAdminClient(TopicCommand.scala:205)
kafka0_1     | 	at kafka.admin.TopicCommand$TopicService$.apply(TopicCommand.scala:209)
kafka0_1     | 	at kafka.admin.TopicCommand$.main(TopicCommand.scala:50)
kafka0_1     | 	at kafka.admin.TopicCommand.main(TopicCommand.scala)
kafka0_1     | Caused by: org.apache.kafka.common.config.ConfigException: Invalid url in bootstrap.servers: OUTSIDE
kafka0_1     | 	at org.apache.kafka.clients.ClientUtils.parseAndValidateAddresses(ClientUtils.java:59)
kafka0_1     | 	at org.apache.kafka.clients.ClientUtils.parseAndValidateAddresses(ClientUtils.java:48)
kafka0_1     | 	at org.apache.kafka.clients.admin.KafkaAdminClient.createInternal(KafkaAdminClient.java:490)

Maybe prefer BROKER_LIST when available so that it can always be overridden? Or look for KAFKA_INTER_BROKER_LISTENER_NAME to help find the internal address in the KAFKA_LISTENERS?

OK, I will take a look at this one of the followin days

[jozefbarcin]

:pray: Excited for this.

[nise-wg2]

🙏 Looking forward to this!

[jimbogithub]

The topic creation is still problematic. It's not working with the example config:

HOSTNAME_COMMAND: curl http://169.254.169.254/latest/meta-data/public-hostname
KAFKA_ADVERTISED_LISTENERS: INSIDE://:9092,OUTSIDE://_{HOSTNAME_COMMAND}:9094
KAFKA_LISTENERS: INSIDE://:9092,OUTSIDE://:9094
KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: INSIDE:PLAINTEXT,OUTSIDE:PLAINTEXT
KAFKA_INTER_BROKER_LISTENER_NAME: INSIDE

I use very similar config myself and it results in:

kafka0_1     | Exception in thread "main" org.apache.kafka.common.KafkaException: Failed to create new KafkaAdminClient
kafka0_1     | 	at org.apache.kafka.clients.admin.KafkaAdminClient.createInternal(KafkaAdminClient.java:538)
kafka0_1     | 	at org.apache.kafka.clients.admin.KafkaAdminClient.createInternal(KafkaAdminClient.java:470)
kafka0_1     | 	at org.apache.kafka.clients.admin.Admin.create(Admin.java:133)
kafka0_1     | 	at kafka.admin.TopicCommand$TopicService$.createAdminClient(TopicCommand.scala:205)
kafka0_1     | 	at kafka.admin.TopicCommand$TopicService$.apply(TopicCommand.scala:209)
kafka0_1     | 	at kafka.admin.TopicCommand$.main(TopicCommand.scala:50)
kafka0_1     | 	at kafka.admin.TopicCommand.main(TopicCommand.scala)
kafka0_1     | Caused by: org.apache.kafka.common.config.ConfigException: Invalid url in bootstrap.servers: OUTSIDE
kafka0_1     | 	at org.apache.kafka.clients.ClientUtils.parseAndValidateAddresses(ClientUtils.java:59)
kafka0_1     | 	at org.apache.kafka.clients.ClientUtils.parseAndValidateAddresses(ClientUtils.java:48)
kafka0_1     | 	at org.apache.kafka.clients.admin.KafkaAdminClient.createInternal(KafkaAdminClient.java:490)

Maybe prefer BROKER_LIST when available so that it can always be overridden? Or look for KAFKA_INTER_BROKER_LISTENER_NAME to help find the internal address in the KAFKA_LISTENERS?

OK, I will take a look at this one of the followin days

@lunarfs I've added a PR to your PR that resolves the above by simply making the KAFKA_LISTENERS parsing smarter. If you can ripple that through to here then I suspect this is good to go.

[jimbogithub]

Preview available here: jimbodock/kafka:2.13-3.3.1. (with the embedded docker cli stripped out)

[jozefbarcin]

Preview available here: jimbodock/kafka:2.13-3.3.1. (with the embedded docker cli stripped out)

Works fine for me