KAFKA-16827: Integrate kafka native-image with system tests (#16046)

This PR does following things

System tests should bring up Kafka broker in the native mode
System tests should run on Kafka broker in native mode
Extract out native build command so that it can be reused.
Allow system tests to run on Native Kafka broker using Docker mechanism

To run system tests by bringing up Kafka in native mode:
Pass kafka_mode as native in the ducktape globals:--globals '{\"kafka_mode\":\"native\"}'

Running system tests by bringing up kafka in native mode via docker mechanism
_DUCKTAPE_OPTIONS="--globals '{\"kafka_mode\":\"native\"}'" TC_PATHS="tests/kafkatest/tests/"  bash tests/docker/run_tests.sh

To only bring up ducker nodes to cater native kafka
bash tests/docker/ducker-ak up -m native

Reviewers: Manikumar Reddy <manikumar.reddy@gmail.com>

.gitignore

@@ -60,4 +60,5 @@ jmh-benchmarks/src/main/generated
 storage/kafka-tiered-storage/
 
 docker/test/report_*.html
+kafka.Kafka
 __pycache__

bin/kafka-run-class.sh

@@ -208,7 +208,7 @@ fi
 
 # JMX settings
 if [ -z "$KAFKA_JMX_OPTS" ]; then
-  KAFKA_JMX_OPTS="-Dcom.sun.management.jmxremote -Dcom.sun.management.jmxremote.authenticate=false  -Dcom.sun.management.jmxremote.ssl=false "
+  KAFKA_JMX_OPTS="-Dcom.sun.management.jmxremote=true -Dcom.sun.management.jmxremote.authenticate=false  -Dcom.sun.management.jmxremote.ssl=false "
 fi
 
 # JMX port to use
@@ -340,9 +340,16 @@ CLASSPATH=${CLASSPATH#:}
 # If Cygwin is detected, classpath is converted to Windows format.
 (( WINDOWS_OS_FORMAT )) && CLASSPATH=$(cygpath --path --mixed "${CLASSPATH}")
 
-# Launch mode
+# If KAFKA_MODE=native, it will bring up Kafka in the native mode.
-if [ "x$DAEMON_MODE" = "xtrue" ]; then
+# It expects the Kafka executable binary to be present at $base_dir/kafka.Kafka.
-  nohup "$JAVA" $KAFKA_HEAP_OPTS $KAFKA_JVM_PERFORMANCE_OPTS $KAFKA_GC_LOG_OPTS $KAFKA_JMX_OPTS $KAFKA_LOG4J_CMD_OPTS -cp "$CLASSPATH" $KAFKA_OPTS "$@" > "$CONSOLE_OUTPUT_FILE" 2>&1 < /dev/null &
+# This is specifically used to run system tests on native Kafka - by bringing up Kafka in the native mode.
+if [[ "x$KAFKA_MODE" == "xnative" ]] && [[ "$*" == *"kafka.Kafka"* ]]; then
+  exec $base_dir/kafka.Kafka start --config "$2"  $KAFKA_LOG4J_CMD_OPTS $KAFKA_JMX_OPTS $KAFKA_OPTS
 else
-  exec "$JAVA" $KAFKA_HEAP_OPTS $KAFKA_JVM_PERFORMANCE_OPTS $KAFKA_GC_LOG_OPTS $KAFKA_JMX_OPTS $KAFKA_LOG4J_CMD_OPTS -cp "$CLASSPATH" $KAFKA_OPTS "$@"
+  # Launch mode
+  if [ "x$DAEMON_MODE" = "xtrue" ]; then
+    nohup "$JAVA" $KAFKA_HEAP_OPTS $KAFKA_JVM_PERFORMANCE_OPTS $KAFKA_GC_LOG_OPTS $KAFKA_JMX_OPTS $KAFKA_LOG4J_CMD_OPTS -cp "$CLASSPATH" $KAFKA_OPTS "$@" > "$CONSOLE_OUTPUT_FILE" 2>&1 < /dev/null &
+  else
+    exec "$JAVA" $KAFKA_HEAP_OPTS $KAFKA_JVM_PERFORMANCE_OPTS $KAFKA_GC_LOG_OPTS $KAFKA_JMX_OPTS $KAFKA_LOG4J_CMD_OPTS -cp "$CLASSPATH" $KAFKA_OPTS "$@"
+  fi
 fi

core/src/main/scala/kafka/docker/KafkaDockerWrapper.scala

@@ -18,7 +18,7 @@ package kafka.docker
 
 import kafka.Kafka
 import kafka.tools.StorageTool
-import kafka.utils.Exit
+import kafka.utils.{Exit, Logging}
 import net.sourceforge.argparse4j.ArgumentParsers
 import net.sourceforge.argparse4j.impl.Arguments.store
 import net.sourceforge.argparse4j.inf.Namespace
@@ -26,7 +26,7 @@ import net.sourceforge.argparse4j.inf.Namespace
 import java.nio.charset.StandardCharsets
 import java.nio.file.{Files, Path, Paths, StandardCopyOption, StandardOpenOption}
 
-object KafkaDockerWrapper {
+object KafkaDockerWrapper extends Logging {
   def main(args: Array[String]): Unit = {
     val namespace = parseArguments(args)
     val command = namespace.getString("command")
@@ -48,6 +48,7 @@ object KafkaDockerWrapper {
         StorageTool.main(formatCmd)
       case "start" =>
         val configFile = namespace.getString("config")
+        info("Starting Kafka server in the native mode.")
         Kafka.main(Array(configFile))
       case _ =>
         throw new RuntimeException(s"Unknown operation $command. " +

docker/native/Dockerfile

@@ -20,40 +20,26 @@ ARG kafka_url
 WORKDIR /app
 
 ENV KAFKA_URL=$kafka_url
-COPY native-image-configs native-image-configs
+ENV NATIVE_IMAGE_PATH="native-image"
+ENV KAFKA_DIR="/app/kafka"
+ENV NATIVE_CONFIGS_DIR="/app/native-image-configs"
+ENV KAFKA_LIBS_DIR="$KAFKA_DIR/libs"
+ENV TARGET_PATH="$KAFKA_DIR/kafka.Kafka"
 
-RUN mkdir kafka; \
+COPY native-image-configs $NATIVE_CONFIGS_DIR
+COPY native_command.sh native_command.sh
+
+RUN mkdir $KAFKA_DIR; \
     microdnf install wget; \
     wget -nv -O kafka.tgz "$KAFKA_URL"; \
-    wget -nv -O kafka.tgz.asc "$kafka_url.asc"; \
+    wget -nv -O kafka.tgz.asc "$KAFKA_URL.asc"; \
-    tar xfz kafka.tgz -C kafka --strip-components 1; \
+    tar xfz kafka.tgz -C $KAFKA_DIR --strip-components 1; \
     wget -nv -O KEYS https://downloads.apache.org/kafka/KEYS; \
     gpg --import KEYS; \
     gpg --batch --verify kafka.tgz.asc kafka.tgz; \
     rm kafka.tgz ; \
-    cd kafka ;  \
 # Build the native-binary of the apache kafka using graalVM native-image.
-    native-image --no-fallback \
+    /app/native_command.sh $NATIVE_IMAGE_PATH $NATIVE_CONFIGS_DIR $KAFKA_LIBS_DIR $TARGET_PATH
-        --enable-http \
-        --enable-https \
-        --allow-incomplete-classpath \
-        --report-unsupported-elements-at-runtime \
-        --install-exit-handlers \
-        --enable-monitoring=jmxserver,jmxclient,heapdump,jvmstat \
-        -H:+ReportExceptionStackTraces \
-        -H:+EnableAllSecurityServices \
-        -H:EnableURLProtocols=http,https \
-        -H:AdditionalSecurityProviders=sun.security.jgss.SunProvider \
-        -H:ReflectionConfigurationFiles=/app/native-image-configs/reflect-config.json \
-        -H:JNIConfigurationFiles=/app/native-image-configs/jni-config.json \
-        -H:ResourceConfigurationFiles=/app/native-image-configs/resource-config.json \
-        -H:SerializationConfigurationFiles=/app/native-image-configs/serialization-config.json \
-        -H:PredefinedClassesConfigurationFiles=/app/native-image-configs/predefined-classes-config.json \
-        -H:DynamicProxyConfigurationFiles=/app/native-image-configs/proxy-config.json \
-        --verbose \
-        -march=compatibility \
-        -cp "libs/*" kafka.docker.KafkaDockerWrapper \
-        -o kafka.Kafka
 
 
 FROM alpine:latest

docker/native/native-image-configs/resource-config.json

@@ -25,13 +25,9 @@
   }, {
     "pattern":"\\Qkafka/kafka-version.properties\\E"
   }, {
-    "pattern":"\\Qlinux/amd64/libzstd-jni-1.5.5-6.so\\E"
+    "pattern":"\\Qlinux/amd64/libzstd-jni-1.5.6-3.so\\E"
   }, {
-    "pattern":"\\Qlinux/aarch64/libzstd-jni-1.5.5-6.so\\E"
+    "pattern":"\\Qlinux/aarch64/libzstd-jni-1.5.6-3.so\\E"
-  }, {
-    "pattern":"\\Qlinux/amd64/libzstd-jni-1.5.5-11.so\\E"
-  }, {
-    "pattern":"\\Qlinux/aarch64/libzstd-jni-1.5.5-11.so\\E"
   }, {
     "pattern":"\\Qnet/jpountz/util/linux/amd64/liblz4-java.so\\E"
   }, {

docker/native/native_command.sh

@@ -0,0 +1,43 @@
+#!/usr/bin/env bash
+
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+# $1 - The path of the GraalVM native-image. This binary is used to compile Java applications ahead-of-time into a standalone native binary.
+# $2 - The path of the directory that contains the native-image configuration files.
+# $3 - The path of the directory that contains the Apache Kafka libs.
+# $4 - The path of the resulting Kafka native binary after the build process.
+
+$1 --no-fallback \
+  --enable-http \
+  --enable-https \
+  --allow-incomplete-classpath \
+  --report-unsupported-elements-at-runtime \
+  --install-exit-handlers \
+  --enable-monitoring=jmxserver,jmxclient,heapdump,jvmstat \
+  -H:+ReportExceptionStackTraces \
+  -H:+EnableAllSecurityServices \
+  -H:EnableURLProtocols=http,https \
+  -H:AdditionalSecurityProviders=sun.security.jgss.SunProvider \
+  -H:ReflectionConfigurationFiles="$2"/reflect-config.json \
+  -H:JNIConfigurationFiles="$2"/jni-config.json \
+  -H:ResourceConfigurationFiles="$2"/resource-config.json \
+  -H:SerializationConfigurationFiles="$2"/serialization-config.json \
+  -H:PredefinedClassesConfigurationFiles="$2"/predefined-classes-config.json \
+  -H:DynamicProxyConfigurationFiles="$2"/proxy-config.json \
+  --verbose \
+  -march=compatibility \
+  -cp "$3/*" kafka.docker.KafkaDockerWrapper \
+  -o "$4"

tests/docker/Dockerfile

@@ -14,6 +14,34 @@
 # limitations under the License.
 
 ARG jdk_version=openjdk:8
+FROM $jdk_version AS build-native-image
+
+WORKDIR /build
+
+COPY native/ native
+
+ARG KAFKA_MODE
+ARG GRAALVM_URL="https://github.com/graalvm/graalvm-ce-builds/releases/download/jdk-21.0.1/graalvm-community-jdk-21.0.1_linux-aarch64_bin.tar.gz"
+
+ENV NATIVE_IMAGE_PATH="/build/graalvm/bin/native-image"
+ENV NATIVE_CONFIGS_DIR="/build/native/native-image-configs"
+ENV KAFKA_LIBS_DIR="/build/kafka/libs"
+ENV KAFKA_BIN_DIR="/build/kafka-binary"
+ENV TARGET_PATH="$KAFKA_BIN_DIR/kafka.Kafka"
+
+RUN mkdir $KAFKA_BIN_DIR
+
+RUN if [ "$KAFKA_MODE" = "native" ]; then \
+      apt update && apt install -y sudo build-essential libz-dev zlib1g-dev curl jq coreutils libffi-dev cmake pkg-config libfuse-dev && apt-get -y clean ; \
+      mkdir graalvm ; \
+      curl -L "$GRAALVM_URL" -o graalvm.tar.gz ; \
+      tar -xzf graalvm.tar.gz -C graalvm --strip-components=1 ; \
+      mkdir kafka ; \
+      tar xfz native/kafka.tgz -C kafka --strip-components 1 ; \
+      rm graalvm.tar.gz kafka.tgz ; \
+      /build/native/native_command.sh $NATIVE_IMAGE_PATH $NATIVE_CONFIGS_DIR $KAFKA_LIBS_DIR $TARGET_PATH ; \
+    fi
+
 FROM $jdk_version
 
 MAINTAINER Apache Kafka dev@kafka.apache.org
@@ -37,6 +65,7 @@ RUN apt update && apt install -y sudo git netcat iptables rsync unzip wget curl
 RUN python3 -m pip install -U pip==21.1.1;
 RUN pip3 install --upgrade cffi virtualenv pyasn1 boto3 pycrypto pywinrm ipaddress enum34 debugpy && pip3 install --upgrade "ducktape>0.8"
 
+COPY --from=build-native-image /build/kafka-binary/ /opt/kafka-binary/
 # Set up ssh
 COPY ./ssh-config /root/.ssh/config
 # NOTE: The paramiko library supports the PEM-format private key, but does not support the RFC4716 format.
@@ -107,7 +136,7 @@ ARG KIBOSH_VERSION="8841dd392e6fbf02986e2fb1f1ebf04df344b65a"
 ARG UID="1000"
 
 # Install Kibosh
-RUN apt-get install fuse
+RUN apt-get install fuse -y
 RUN cd /opt && git clone -q  https://github.com/confluentinc/kibosh.git && cd "/opt/kibosh" && git reset --hard $KIBOSH_VERSION && mkdir "/opt/kibosh/build" && cd "/opt/kibosh/build" && ../configure && make -j 2
 
 # Set up the ducker user.

tests/docker/ducker-ak

@@ -31,6 +31,9 @@ ducker_dir="$( cd "$( dirname "${BASH_SOURCE[0]}" )" && pwd )"
 # The absolute path to the root Kafka directory
 kafka_dir="$( cd "${ducker_dir}/../.." && pwd )"
 
+# The absolute path to the directory to hold the necessary files to construct a ducker image for executing Kafka in native mode.
+tmp_native_dir=${ducker_dir}/native
+
 # The memory consumption to allow during the docker build.
 # This does not include swap.
 docker_build_memory_limit="3200m"
@@ -47,6 +50,9 @@ default_jdk="openjdk:8"
 # The default ducker-ak image name.
 default_image_name="ducker-ak"
 
+# The default kafka server mode.
+default_kafka_mode="jvm"
+
 # Port to listen on when debugging
 debugpy_port=5678
 
@@ -247,6 +253,7 @@ ducker_build() {
         --build-arg "ducker_creator=${user_name}" \
         --build-arg "jdk_version=${jdk_version}" \
         --build-arg "UID=${UID}" \
+        --build-arg "KAFKA_MODE=${kafka_mode}" \
         -t "${image_name}" \
         -f "${ducker_dir}/Dockerfile" ${docker_args} -- .
     docker_status=$?
@@ -257,6 +264,7 @@ ducker_build() {
 $((${duration} % 60))s."
         echo "If this error is unexpected, consider running 'docker system prune -a' \
 to clear old images from your local cache."
+        cleanup_native_dir
         exit 1
     fi
     echo "Successfully built ${what} image in $((${duration} / 60))m \
@@ -305,6 +313,32 @@ setup_custom_ducktape() {
     must_do -v docker rm ducker01
 }
 
+cleanup_native_dir() {
+    if [ -d "${tmp_native_dir}" ]; then
+        echo "Deleting temporary native dir: ${tmp_native_dir}"
+        rm -rf "${tmp_native_dir}"
+    fi
+}
+
+prepare_native_dir() {
+    echo "Mode provided for system tests run: $kafka_mode"
+    must_pushd "${kafka_dir}"
+    cleanup_native_dir && mkdir "${tmp_native_dir}"
+
+    if [ "$kafka_mode" == "native" ]; then
+        kafka_tarball_filename=(core/build/distributions/kafka*SNAPSHOT.tgz)
+        if [ ! -e "${kafka_tarball_filename[0]}" ]; then
+            echo "Kafka tarball not present. Building Kafka tarball for native image."
+            ./gradlew clean releaseTarGz
+        fi
+
+        cp core/build/distributions/kafka*SNAPSHOT.tgz "${tmp_native_dir}"/kafka.tgz
+        cp -r docker/native/native-image-configs "${tmp_native_dir}"
+        cp docker/native/native_command.sh "${tmp_native_dir}"
+    fi
+    must_popd
+}
+
 ducker_up() {
     require_commands docker
     while [[ $# -ge 1 ]]; do
@@ -314,11 +348,13 @@ ducker_up() {
             -n|--num-nodes) set_once num_nodes "${2}" "number of nodes"; shift 2;;
             -j|--jdk) set_once jdk_version "${2}" "the OpenJDK base image"; shift 2;;
             -e|--expose-ports) set_once expose_ports "${2}" "the ports to expose"; shift 2;;
+            -m|--kafka_mode) set_once kafka_mode "${2}" "the mode in which kafka will run"; shift 2;;
             *) set_once image_name "${1}" "docker image name"; shift;;
         esac
     done
     [[ -n "${num_nodes}" ]] || num_nodes="${default_num_nodes}"
     [[ -n "${jdk_version}" ]] || jdk_version="${default_jdk}"
+    [[ -n "${kafka_mode}" ]] || kafka_mode="${default_kafka_mode}"
     [[ -n "${image_name}" ]] || image_name="${default_image_name}-${jdk_version/:/-}"
     [[ "${num_nodes}" =~ ^-?[0-9]+$ ]] || \
         die "ducker_up: the number of nodes must be an integer."
@@ -334,7 +370,9 @@ use only ${num_nodes}."
 
     docker ps >/dev/null || die "ducker_up: failed to run docker.  Please check that the daemon is started."
 
+    prepare_native_dir
     ducker_build "${image_name}"
+    cleanup_native_dir
 
     docker inspect --format='{{.Config.Labels}}' --type=image "${image_name}" | grep -q 'ducker.type'
     local docker_status=${PIPESTATUS[0]}
@@ -385,6 +423,10 @@ attempting to start new ones."
         [[ $? -ne 0 ]] && die "failed to append to the /etc/hosts file on ${node}"
     done
 
+    if [ "$kafka_mode" == "native" ]; then
+      docker exec --user=root ducker01 bash -c 'cp /opt/kafka-binary/kafka.Kafka /opt/kafka-dev/kafka.Kafka'
+    fi
+
     echo "ducker_up: added the latest entries to /etc/hosts on each node."
     generate_cluster_json_file "${num_nodes}" "${ducker_dir}/build/cluster.json"
     echo "ducker_up: successfully wrote ${ducker_dir}/build/cluster.json"

tests/docker/run_tests.sh

@@ -25,12 +25,21 @@ die() {
     exit 1
 }
 
+if [[ "$_DUCKTAPE_OPTIONS" == *"kafka_mode"* && "$_DUCKTAPE_OPTIONS" == *"native"* ]]; then
+    export KAFKA_MODE="native"
+else
+    export KAFKA_MODE="jvm"
+fi
+
 if [ "$REBUILD" == "t" ]; then
     ./gradlew clean systemTestLibs
+    if [ "$KAFKA_MODE" == "native" ]; then
+        ./gradlew clean releaseTarGz
+    fi
 fi
 
 if ${SCRIPT_DIR}/ducker-ak ssh | grep -q '(none)'; then
-    ${SCRIPT_DIR}/ducker-ak up -n "${KAFKA_NUM_CONTAINERS}" || die "ducker-ak up failed"
+    ${SCRIPT_DIR}/ducker-ak up -n "${KAFKA_NUM_CONTAINERS}" -m "${KAFKA_MODE}" || die "ducker-ak up failed"
 fi
 
 [[ -n ${_DUCKTAPE_OPTIONS} ]] && _DUCKTAPE_OPTIONS="-- ${_DUCKTAPE_OPTIONS}"

tests/kafkatest/services/kafka/kafka.py

@@ -806,7 +806,13 @@ class KafkaService(KafkaPathResolverMixin, JmxMixin, Service):
         return s
 
     def start_cmd(self, node):
-        cmd = "export JMX_PORT=%d; " % self.jmx_port
+        """
+        To bring up kafka using native image, pass following in ducktape options
+        --globals '{"kafka_mode": "native"}'
+        """
+        kafka_mode = self.context.globals.get("kafka_mode", "")
+        cmd = f"export KAFKA_MODE={kafka_mode}; "
+        cmd += "export JMX_PORT=%d; " % self.jmx_port
         cmd += "export KAFKA_LOG4J_OPTS=\"-Dlog4j.configuration=file:%s\"; " % self.LOG4J_CONFIG
         heap_kafka_opts = "-XX:+HeapDumpOnOutOfMemoryError -XX:HeapDumpPath=%s" % \
                           self.logs["kafka_heap_dump_file"]["path"]
@@ -926,7 +932,7 @@ class KafkaService(KafkaPathResolverMixin, JmxMixin, Service):
     def pids(self, node):
         """Return process ids associated with running processes on the given node."""
         try:
-            cmd = "jcmd | grep -e %s | awk '{print $1}'" % self.java_class_name()
+            cmd = "ps ax | grep -i %s | grep -v grep | awk '{print $1}'" % self.java_class_name()
             pid_arr = [pid for pid in node.account.ssh_capture(cmd, allow_fail=True, callback=int)]
             return pid_arr
         except (RemoteCommandError, ValueError) as e:
@@ -994,7 +1000,7 @@ class KafkaService(KafkaPathResolverMixin, JmxMixin, Service):
     def clean_node(self, node):
         JmxMixin.clean_node(self, node)
         self.security_config.clean_node(node)
-        node.account.kill_java_processes(self.java_class_name(),
+        node.account.kill_process(self.java_class_name(),
                                          clean_shutdown=False, allow_fail=True)
         node.account.ssh("sudo rm -rf -- %s" % KafkaService.PERSISTENT_ROOT, allow_fail=False)
 

tests/kafkatest/services/security/templates/jaas.conf

@@ -55,6 +55,7 @@ KafkaServer {
     useKeyTab=true
     storeKey=true
     keyTab="/mnt/security/keytab"
+    refreshKrb5Config=true
     principal="kafka/{{ node.account.hostname }}@EXAMPLE.COM";
 {% endif %}
 {% endif %}
@@ -81,6 +82,7 @@ Client {
     principal="zkclient@EXAMPLE.COM";
 {% else %}
    com.sun.security.auth.module.Krb5LoginModule required
+   refreshKrb5Config=true
    useKeyTab=true
    keyTab="/mnt/security/keytab"
    storeKey=true