Add support for Kafka 3.0.0.

.travis.yml

@@ -8,7 +8,7 @@ services:
 # This version will be also tagged as 'latest'
 env:
   global:
-    - LATEST="2.13-2.7.1"
+    - LATEST="2.13-3.0.0"
 
 # Build recommended versions based on: http://kafka.apache.org/downloads
 matrix:
@@ -29,6 +29,8 @@ matrix:
     env: KAFKA_VERSION=2.6.0
   - scala: 2.13
     env: KAFKA_VERSION=2.7.1
+  - scala: 2.13
+    env: KAFKA_VERSION=3.0.0
 
 install:
   - docker --version

CHANGELOG.md

@@ -3,6 +3,11 @@ 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.
 
+11-Oct-2021
+-----------
+
+- Add support for Kafka `3.0.0`
+
 19-July-2021
 ----------
 

Dockerfile

@@ -1,6 +1,6 @@
 FROM azul/zulu-openjdk-alpine:8u292-8.54.0.21
 
-ARG kafka_version=2.7.1
+ARG kafka_version=3.0.0
 ARG scala_version=2.13
 ARG glibc_version=2.31-r0
 ARG vcs_ref=unspecified

README.md

@@ -17,6 +17,7 @@ Tags and releases
 
 All versions of the image are built from the same set of scripts with only minor variations (i.e. certain features are not supported on older versions). The version format mirrors the Kafka format, `<scala version>-<kafka version>`. Initially, all images are built with the recommended version of scala documented on [http://kafka.apache.org/downloads](http://kafka.apache.org/downloads). Available tags are:
 
+- `2.13-3.0.0`
 - `2.13-2.7.1`
 - `2.13-2.6.0`
 - `2.12-2.5.0`

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
@@ -26,7 +29,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
 
@@ -96,6 +99,16 @@ if [[ -z "$KAFKA_ADVERTISED_HOST_NAME$KAFKA_LISTENERS" ]]; then
     export KAFKA_ADVERTISED_HOST_NAME="$HOSTNAME_VALUE"
 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
 echo "" >> "$KAFKA_HOME/config/server.properties"
 

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.10/test.create-topics.kafka.sh

@@ -8,27 +8,47 @@ testCreateTopics() {
 	# TOPICS array contains the topic name to create / validate
 	# CLEANUP array contains the expected cleanup policy configuration for the topic
 	TOPICS[0]="default-$NOW"
-	CLEANUP[0]=""
+	CONFIG[0]=""
 
 	TOPICS[1]="compact-$NOW"
-	CLEANUP[1]="compact,compression.type=snappy"
+	CONFIG[1]="cleanup.policy=compact,compression.type=snappy"
 
 	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-configs.sh --zookeeper "$KAFKA_ZOOKEEPER_CONNECT" --entity-type topics --entity-name "$TOPIC" --describe | grep 'Configs for topic' | awk -F'cleanup.policy=' '{print $2}')
+		# 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/docker-compose.yml

@@ -49,7 +49,7 @@ services:
     image: confluentinc/cp-kafkacat:5.0.0
     environment:
        - BROKER_LIST
-       - KAFKA_VERSION=${KAFKA_VERSION-2.7.1}
+       - KAFKA_VERSION=${KAFKA_VERSION-3.0.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