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

.travis.yml

@@ -8,7 +8,7 @@ services:
 # This version will be also tagged as 'latest'
 env:
   global:
-    - LATEST="2.13-2.8.1"
+    - LATEST="2.13-3.2.0"
 
 # Build recommended versions based on: http://kafka.apache.org/downloads
 matrix:
@@ -29,6 +29,12 @@ matrix:
     env: KAFKA_VERSION=2.7.2
   - scala: 2.13
     env: KAFKA_VERSION=2.8.1
+  - scala: 2.13
+    env: KAFKA_VERSION=3.0.1
+  - scala: 2.13
+    env: KAFKA_VERSION=3.1.1
+  - scala: 2.13
+    env: KAFKA_VERSION=3.2.0
 
 # Upgrade Docker Engine so we can use buildx
 before_install:

CHANGELOG.md

@@ -2,7 +2,10 @@ Changelog
 =========
 
 Kafka features are not tied to a specific kafka-docker version (ideally all changes will be merged into all branches). Therefore, this changelog will track changes to the image by date.
+12-May-2022
+-----------
 
+- Add support for Kafka `3.1.0`
 09-Apr-2022
 ----------
 

Dockerfile

@@ -1,6 +1,6 @@
 FROM openjdk:11-jre-slim
 
-ARG kafka_version=2.8.1
+ARG kafka_version=3.1.0
 ARG scala_version=2.13
 ARG vcs_ref=unspecified
 ARG build_date=unspecified

broker-list.sh

@@ -1,5 +1,6 @@
 #!/bin/bash
 
 CONTAINERS=$(docker ps | grep 9092 | awk '{print $1}')
-BROKERS=$(for CONTAINER in ${CONTAINERS}; do docker port "$CONTAINER" 9092 | sed -e "s/0.0.0.0:/$HOST_IP:/g"; done)
+# We are just interested in ipv4 addresses, so ignore ipv6 addresses
+BROKERS=$(for CONTAINER in ${CONTAINERS}; do docker port "$CONTAINER" 9092 | grep '0.0.0.0' | sed -e "s/0.0.0.0:/$HOST_IP:/g"; done)
 echo "${BROKERS//$'\n'/,}"

create-topics.sh

@@ -33,6 +33,14 @@ if [[ "$MAJOR_VERSION" == "0" && "$MINOR_VERSION" -gt "9" ]] || [[ "$MAJOR_VERSI
     KAFKA_0_10_OPTS="--if-not-exists"
 fi
 
+# since 3.0.0 there is no --zookeeper option anymore, so we have to use the
+# --bootstrap-server option with a random broker
+if [[ "$MAJOR_VERSION" -ge "3" ]]; then
+    CONNECT_OPTS="--bootstrap-server $(echo "${BROKER_LIST}" | cut -d ',' -f1)"
+else
+    CONNECT_OPTS="--zookeeper ${KAFKA_ZOOKEEPER_CONNECT}"
+fi
+
 # Expected format:
 #   name:partitions:replicas:cleanup.policy
 IFS="${KAFKA_CREATE_TOPICS_SEPARATOR-,}"; for topicToCreate in $KAFKA_CREATE_TOPICS; do
@@ -45,7 +53,7 @@ IFS="${KAFKA_CREATE_TOPICS_SEPARATOR-,}"; for topicToCreate in $KAFKA_CREATE_TOP
 
     COMMAND="JMX_PORT='' ${KAFKA_HOME}/bin/kafka-topics.sh \\
 		--create \\
-		--zookeeper ${KAFKA_ZOOKEEPER_CONNECT} \\
+		${CONNECT_OPTS} \\
 		--topic ${topicConfig[0]} \\
 		--partitions ${topicConfig[1]} \\
 		--replication-factor ${topicConfig[2]} \\

start-kafka.sh

@@ -1,5 +1,8 @@
 #!/bin/bash -e
 
+# shellcheck disable=SC1091
+source /usr/bin/versions.sh
+
 # Allow specific kafka versions to perform any unique bootstrap operations
 OVERRIDE_FILE="/opt/overrides/${KAFKA_VERSION}.sh"
 if [[ -x "$OVERRIDE_FILE" ]]; then
@@ -10,9 +13,32 @@ fi
 # Store original IFS config, so we can restore it at various stages
 ORIG_IFS=$IFS
 
-if [[ -z "$KAFKA_ZOOKEEPER_CONNECT" ]]; then
-    echo "ERROR: missing mandatory config: KAFKA_ZOOKEEPER_CONNECT"
-    exit 1
+if [[ "$MAJOR_VERSION" -ge "3" ]]; then
+    if [[ -z "BROKER_LIST" ]]; then
+        echo "ERROR: missing mandatory config: BROKER_LIST"
+        exit 1
+    fi
+    if [[ -v KAFKA_ADVERTISED_HOST_NAME ]]; then
+        echo "ERROR: KAFKA_ADVERTISED_HOST_NAME is removed as of kafka 3, remove KAFKA_ADVERTISED_HOST_NAME=$KAFKA_ADVERTISED_HOST_NAME from your config"
+        exit 1
+    fi
+    if [[ -v KAFKA_ADVERTISED_PORT ]]; then
+        echo "ERROR: KAFKA_ADVERTISED_PORT is removed as of kafka 3, remove KAFKA_ADVERTISED_PORT=$KAFKA_ADVERTISED_PORT from your config"
+        exit 1
+    fi
+    if [[ -v KAFKA_PORT ]]; then
+        echo "ERROR: KAFKA_PORT is removed as of kafka 3, remove KAFKA_PORT=$KAFKA_PORT from your config"
+        exit 1
+    fi
+    if [[ -v KAFKA_HOST_NAME ]]; then
+        echo "ERROR: KAFKA_HOST_NAME is removed as of kafka 3, remove KAFKA_HOST_NAME=$KAFKA_HOST_NAME from your config"
+        exit 1
+    fi
+else
+    if [[ -z "$KAFKA_ZOOKEEPER_CONNECT" ]]; then
+        echo "ERROR: missing mandatory config: KAFKA_ZOOKEEPER_CONNECT"
+        exit 1
+    fi
 fi
 
 if [[ -z "$KAFKA_PORT" ]]; then
@@ -26,7 +52,7 @@ if [[ -z "$KAFKA_ADVERTISED_PORT" && \
   -z "$KAFKA_LISTENERS" && \
   -z "$KAFKA_ADVERTISED_LISTENERS" && \
   -S /var/run/docker.sock ]]; then
-    KAFKA_ADVERTISED_PORT=$(docker port "$(hostname)" $KAFKA_PORT | sed -r 's/.*:(.*)/\1/g' | head -n1) 
+    KAFKA_ADVERTISED_PORT=$(docker port "$(hostname)" $KAFKA_PORT | sed -r 's/.*:(.*)/\1/g' | head -n1)
     export KAFKA_ADVERTISED_PORT
 fi
 
@@ -81,19 +107,30 @@ if [[ -n "$RACK_COMMAND" && -z "$KAFKA_BROKER_RACK" ]]; then
 fi
 
 # Try and configure minimal settings or exit with error if there isn't enough information
-if [[ -z "$KAFKA_ADVERTISED_HOST_NAME$KAFKA_LISTENERS" ]]; then
-    if [[ -n "$KAFKA_ADVERTISED_LISTENERS" ]]; then
-        echo "ERROR: Missing environment variable KAFKA_LISTENERS. Must be specified when using KAFKA_ADVERTISED_LISTENERS"
-        exit 1
-    elif [[ -z "$HOSTNAME_VALUE" ]]; then
-        echo "ERROR: No listener or advertised hostname configuration provided in environment."
-        echo "       Please define KAFKA_LISTENERS / (deprecated) KAFKA_ADVERTISED_HOST_NAME"
-        exit 1
-    fi
+if [[ "$MAJOR_VERSION" -lt "3" ]]; then
+    if [[ -z "$KAFKA_ADVERTISED_HOST_NAME$KAFKA_LISTENERS" ]]; then
+        if [[ -n "$KAFKA_ADVERTISED_LISTENERS" ]]; then
+            echo "ERROR: Missing environment variable KAFKA_LISTENERS. Must be specified when using KAFKA_ADVERTISED_LISTENERS"
+            exit 1
+        elif [[ -z "$HOSTNAME_VALUE" ]]; then
+            echo "ERROR: No listener or advertised hostname configuration provided in environment."
+            echo "       Please define KAFKA_LISTENERS / (deprecated) KAFKA_ADVERTISED_HOST_NAME"
+            exit 1
+        fi
 
-    # Maintain existing behaviour
-    # If HOSTNAME_COMMAND is provided, set that to the advertised.host.name value if listeners are not defined.
-    export KAFKA_ADVERTISED_HOST_NAME="$HOSTNAME_VALUE"
+        # Maintain existing behaviour
+        # If HOSTNAME_COMMAND is provided, set that to the advertised.host.name value if listeners are not defined.
+        export KAFKA_ADVERTISED_HOST_NAME="$HOSTNAME_VALUE"
+    fi
+fi
+# `advertised.port` and `advertised.host.name` are removed with Kafka 3.0.0
+# See: https://github.com/apache/kafka/pull/10872
+if [[ "$MAJOR_VERSION" -ge "3" ]]; then
+    if [ -z "$KAFKA_ADVERTISED_LISTENERS" ]; then
+        if [ -n "${KAFKA_ADVERTISED_HOST_NAME}" ]; then
+            export KAFKA_ADVERTISED_LISTENERS="PLAINTEXT://${KAFKA_ADVERTISED_HOST_NAME}:${KAFKA_PORT}"
+        fi
+    fi
 fi
 
 #Issue newline to config file in case there is not one already

test/0.0/test.create-topics-custom-separator.kafka.sh

@@ -20,13 +20,25 @@ testCreateTopicsCustomSeparator() {
 
 	create-topics.sh
 
+	# shellcheck disable=SC1091
+	source "/usr/bin/versions.sh"
+
+	# since 3.0.0 there is no --zookeeper option anymore, so we have to use the
+	# --bootstrap-server option with a random broker
+	if [[ "$MAJOR_VERSION" -ge "3" ]]; then
+		CONNECT_OPTS="--bootstrap-server $(echo "${BROKER_LIST}" | cut -d ',' -f1)"
+	else
+		CONNECT_OPTS="--zookeeper ${KAFKA_ZOOKEEPER_CONNECT}"
+	fi
+
 	# Loop through each array, validate that topic exists
 	for i in "${!TOPICS[@]}"; do
 		TOPIC=${TOPICS[i]}
 
 		echo "Validating topic '$TOPIC'"
 
-		EXISTS=$(/opt/kafka/bin/kafka-topics.sh --zookeeper "$KAFKA_ZOOKEEPER_CONNECT" --list --topic "$TOPIC")
+		# shellcheck disable=SC2086
+		EXISTS=$(/opt/kafka/bin/kafka-topics.sh ${CONNECT_OPTS} --list --topic "$TOPIC")
 		if [[ "$EXISTS" != "$TOPIC" ]]; then
 			echo "$TOPIC topic not created"
 			return 1

test/0.0/test.start-kafka-advertised-host.kafka.sh

@@ -16,5 +16,13 @@ testAdvertisedHost() {
 	assertAbsent 'advertised.listeners'
 	assertAbsent 'listeners'
 }
+source "/usr/bin/versions.sh"
 
-testAdvertisedHost
+# since 3.0.0 there is no --zookeeper option anymore, so we have to use the
+# --bootstrap-server option with a random broker
+if [[ "$MAJOR_VERSION" -ge "3" ]]; then
+    echo "with kafka from version 3.0.0 'advertised.port' and 'advertised.host.name' are removed with Kafka 3.0.0"
+    echo "See: https://github.com/apache/kafka/pull/10872"
+else
+    testAdvertisedHost
+fi

test/0.0/test.start-kafka-bug-312-kafka-env.kafka.sh

@@ -21,5 +21,13 @@ testKafkaEnv() {
 
     echo " > Set KAFKA_OPTS=$KAFKA_OPTS"
 }
+source "/usr/bin/versions.sh"
 
-testKafkaEnv
+# since 3.0.0 there is no --zookeeper option anymore, so we have to use the
+# --bootstrap-server option with a random broker
+if [[ "$MAJOR_VERSION" -ge "3" ]]; then
+    echo "this thes is obsolete with kafka from version 3.0.0 'advertised.host.name' are removed with Kafka 3.0.0"
+    echo "See: https://github.com/apache/kafka/pull/10872"
+else
+    testKafkaEnv
+fi

test/0.0/test.start-kafka-bug-313-kafka-opts.kafka.sh

@@ -19,5 +19,13 @@ testKafkaOpts() {
 
     echo "  > Set KAFKA_OPTS=$KAFKA_OPTS"
 }
+source "/usr/bin/versions.sh"
 
-testKafkaOpts
+# since 3.0.0 there is no --zookeeper option anymore, so we have to use the
+# --bootstrap-server option with a random broker
+if [[ "$MAJOR_VERSION" -ge "3" ]]; then
+    echo "this thes is obsolete with kafka from version 3.0.0 'advertised.host.name' are removed with Kafka 3.0.0"
+    echo "See: https://github.com/apache/kafka/pull/10872"
+else
+    testKafkaOpts
+fi

test/0.0/test.start-kafka-host-name.kafka.sh

@@ -14,5 +14,13 @@ testHostnameCommand() {
 	assertAbsent 'advertised.listeners'
 	assertAbsent 'listeners'
 }
+source "/usr/bin/versions.sh"
 
-testHostnameCommand
+# since 3.0.0 there is no --zookeeper option anymore, so we have to use the
+# --bootstrap-server option with a random broker
+if [[ "$MAJOR_VERSION" -ge "3" ]]; then
+    echo "this thes is obsolete with kafka from version 3.0.0 'advertised.host.name' are removed with Kafka 3.0.0"
+    echo "See: https://github.com/apache/kafka/pull/10872"
+else
+    testHostnameCommand
+fi

test/0.0/test.start-kafka-log4j-config.kafka.sh

@@ -4,7 +4,14 @@ source test.functions
 
 testLog4jConfig() {
 	# Given Log4j overrides are provided
-	export KAFKA_ADVERTISED_HOST_NAME="testhost"
+	source "/usr/bin/versions.sh"
+
+    # since 3.0.0 there is no --zookeeper option anymore, so we have to use the
+    # --bootstrap-server option with a random broker
+    if [[ "$MAJOR_VERSION" -lt "3" ]]; then
+        export KAFKA_ADVERTISED_HOST_NAME="testhost"
+    fi
+
 	export LOG4J_LOGGER_KAFKA=DEBUG
 
 	# When the script is invoked
@@ -13,5 +20,4 @@ testLog4jConfig() {
 	# Then the configuration file is correct
 	assertExpectedLog4jConfig "log4j.logger.kafka=DEBUG"
 }
-
-testLog4jConfig
+testLog4jConfig

test/0.0/test.start-kafka-restart.kafka.sh

@@ -14,5 +14,13 @@ testRestart() {
 	assertExpectedConfig 'advertised.host.name=testhost'
 	assertAbsent 'listeners'
 }
+source "/usr/bin/versions.sh"
 
-testRestart
+# since 3.0.0 there is no --zookeeper option anymore, so we have to use the
+# --bootstrap-server option with a random broker
+if [[ "$MAJOR_VERSION" -ge "3" ]]; then
+    echo "this thes is obsolete with kafka from version 3.0.0 'advertised.host.name' are removed with Kafka 3.0.0"
+    echo "See: https://github.com/apache/kafka/pull/10872"
+else
+    testRestart
+fi

test/0.10/test.create-topics.kafka.sh

@@ -6,29 +6,51 @@ testCreateTopics() {
 	NOW=$(date +%s)
 
 	# TOPICS array contains the topic name to create / validate
-	# CLEANUP array contains the expected cleanup policy configuration for the topic
+	# CONFIG array contains the expected cleanup policy configuration for the topic
 	TOPICS[0]="default-$NOW"
-	CLEANUP[0]=""
+	CONFIG[0]=""
 
 	TOPICS[1]="compact-$NOW"
-	CLEANUP[1]="compression.type=snappy,cleanup.policy=compact"
+	CONFIG[1]=$(echo -e "compression.type=snappy\ncleanup.policy=compact" | sort \
+			| tr '\n' ',' \
+			| sed 's/,$//')
 
 	KAFKA_CREATE_TOPICS="${TOPICS[0]}:1:1,${TOPICS[1]}:2:1:compact --config=compression.type=snappy" create-topics.sh
 
-	# Loop through each array, validate that topic exists, and correct cleanup policy is set
+	# shellcheck disable=SC1091
+	source "/usr/bin/versions.sh"
+
+	# since 3.0.0 there is no --zookeeper option anymore, so we have to use the
+	# --bootstrap-server option with a random broker
+	if [[ "$MAJOR_VERSION" -ge "3" ]]; then
+		CONNECT_OPTS="--bootstrap-server $(echo "${BROKER_LIST}" | cut -d ',' -f1)"
+	else
+		CONNECT_OPTS="--zookeeper ${KAFKA_ZOOKEEPER_CONNECT}"
+	fi
+
+	# Loop through each array, validate that topic exists, and correct configuration is set
 	for i in "${!TOPICS[@]}"; do
 		TOPIC=${TOPICS[i]}
 
 		echo "Validating topic '$TOPIC'"
 
-		EXISTS=$(/opt/kafka/bin/kafka-topics.sh --zookeeper "$KAFKA_ZOOKEEPER_CONNECT" --list --topic "$TOPIC")
-		POLICY=$(/opt/kafka/bin/kafka-topics.sh --zookeeper "$KAFKA_ZOOKEEPER_CONNECT" --describe --topic "$TOPIC" | awk -F'Configs:' '{print $2}' | xargs)
+		# shellcheck disable=SC2086
+		EXISTS=$(/opt/kafka/bin/kafka-topics.sh ${CONNECT_OPTS} --list --topic "$TOPIC")
+		# shellcheck disable=SC2086
+		ACTUAL_CONFIG=$(/opt/kafka/bin/kafka-configs.sh ${CONNECT_OPTS} --entity-type topics --entity-name "$TOPIC" --describe \
+			| cut -d'{' -f1 \
+			| grep -oE '(compression.type|cleanup.policy)=([^ ,]+)' \
+			| sort \
+			| tr '\n' ',' \
+			| sed 's/,$//')
 
-		RESULT="$EXISTS:$POLICY"
-		EXPECTED="$TOPIC:${CLEANUP[i]}"
+		RESULT="$EXISTS:$ACTUAL_CONFIG"
+		EXPECTED="$TOPIC:${CONFIG[i]}"
 
 		if [[ "$RESULT" != "$EXPECTED" ]]; then
-			echo "$TOPIC topic not configured correctly: '$RESULT'"
+			echo "$TOPIC topic not configured correctly:"
+			echo "    Actual: '$RESULT'"
+			echo "  Expected: '$EXPECTED'"
 			return 1
 		fi
 	done

test/Readme.md

@@ -18,7 +18,7 @@ Run selected tests
 ### Kafka
 
 ```
-docker-compose run --rm kafkatest <testname pattern>
+docker-compose run -e BROKER_LIST=$(../internal-broker-list.sh) --rm kafkatest <testname pattern>
 ```
 
 ### Kafkacat

test/docker-compose.yml

@@ -48,8 +48,7 @@ services:
   kafkacattest:
     image: confluentinc/cp-kafkacat:5.0.0
     environment:
-       - BROKER_LIST
-       - KAFKA_VERSION=${KAFKA_VERSION-2.8.1}
+       - KAFKA_VERSION=${KAFKA_VERSION-3.1.0}
     volumes:
       - .:/tests
     working_dir: /tests

test/runAllTests.sh

@@ -7,12 +7,12 @@ echo "BROKER_LIST=$BROKER_LIST"
 
 runAll() {
   # Tests that require kafka
-  docker-compose run --rm kafkatest
+  docker-compose run -e BROKER_LIST="${BROKER_LIST}" --rm kafkatest
 
   RESULT=$?
   if [[ $RESULT -eq 0 ]]; then
     # Tests that require kafkacat
-    docker-compose run --rm kafkacattest
+    docker-compose run -e BROKER_LIST="${BROKER_LIST}" --rm kafkacattest
     RESULT=$?
   fi