diff --git a/.gitignore b/.gitignore index 7dfe61c38cccf..015df8ead8327 100644 --- a/.gitignore +++ b/.gitignore @@ -60,4 +60,5 @@ jmh-benchmarks/src/main/generated storage/kafka-tiered-storage/ docker/test/report_*.html +kafka.Kafka __pycache__ diff --git a/bin/kafka-run-class.sh b/bin/kafka-run-class.sh index 4e63ff1fb972a..e1d9e540495d5 100755 --- a/bin/kafka-run-class.sh +++ b/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 [ "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 & +# If KAFKA_MODE=native, it will bring up Kafka in the native mode. +# It expects the Kafka executable binary to be present at $base_dir/kafka.Kafka. +# 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 diff --git a/core/src/main/scala/kafka/docker/KafkaDockerWrapper.scala b/core/src/main/scala/kafka/docker/KafkaDockerWrapper.scala index eadbdb8aefc53..8db498db12738 100644 --- a/core/src/main/scala/kafka/docker/KafkaDockerWrapper.scala +++ b/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. " + diff --git a/docker/native/Dockerfile b/docker/native/Dockerfile index 54db837170bc1..36104b76edca2 100644 --- a/docker/native/Dockerfile +++ b/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"; \ - tar xfz kafka.tgz -C kafka --strip-components 1; \ + wget -nv -O kafka.tgz.asc "$KAFKA_URL.asc"; \ + 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 \ - --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 + /app/native_command.sh $NATIVE_IMAGE_PATH $NATIVE_CONFIGS_DIR $KAFKA_LIBS_DIR $TARGET_PATH FROM alpine:latest diff --git a/docker/native/native-image-configs/resource-config.json b/docker/native/native-image-configs/resource-config.json index d3fb2031ee6ec..121f24b63722e 100644 --- a/docker/native/native-image-configs/resource-config.json +++ b/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/amd64/libzstd-jni-1.5.5-11.so\\E" - }, { - "pattern":"\\Qlinux/aarch64/libzstd-jni-1.5.5-11.so\\E" + "pattern":"\\Qlinux/aarch64/libzstd-jni-1.5.6-3.so\\E" }, { "pattern":"\\Qnet/jpountz/util/linux/amd64/liblz4-java.so\\E" }, { diff --git a/docker/native/native_command.sh b/docker/native/native_command.sh new file mode 100755 index 0000000000000..6d2ef5efa5cbc --- /dev/null +++ b/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" diff --git a/tests/docker/Dockerfile b/tests/docker/Dockerfile index 25ccfb7c5e31f..9d6342a68ebc5 100644 --- a/tests/docker/Dockerfile +++ b/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. diff --git a/tests/docker/ducker-ak b/tests/docker/ducker-ak index 4add3172022fd..21f0f64219469 100755 --- a/tests/docker/ducker-ak +++ b/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" diff --git a/tests/docker/run_tests.sh b/tests/docker/run_tests.sh index 0128fd6897ad9..3285e9ab92007 100755 --- a/tests/docker/run_tests.sh +++ b/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}" diff --git a/tests/kafkatest/services/kafka/kafka.py b/tests/kafkatest/services/kafka/kafka.py index 7e7f9e104f7e5..5f245b1005363 100644 --- a/tests/kafkatest/services/kafka/kafka.py +++ b/tests/kafkatest/services/kafka/kafka.py @@ -806,7 +806,13 @@ def render_configs(self, configs): 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 @@ def run_features_command(self, op, new_version): 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 @@ def thread_dump(self, node): 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) diff --git a/tests/kafkatest/services/security/templates/jaas.conf b/tests/kafkatest/services/security/templates/jaas.conf index e4fe72cd86088..709dcacd8921c 100644 --- a/tests/kafkatest/services/security/templates/jaas.conf +++ b/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