rgw: support kafka 4.x brokers

librdkafka submodule

Vendors librdkafka v2.12.1 as a submodule (default) and teaches the
notification teuthology task to set up Kafka 4.x KRaft brokers. The
distro-shipped librdkafka 1.6.1 cannot SCRAM-auth against Kafka 4.x
brokers (#75900); the bundled version fixes that. Build with
WITH_SYSTEM_RDKAFKA=ON to opt back into the distro library (floor 2.6.1).
follwinf changes needed with teh submodule:
* build kafka_stub only in case of system librdkafka
* use the same header location in submodule and system
* supress compilation warning to avoid -WError issues

2.12.1 matches Fedora(44)'s current librdkafka-devel, so the vendored copy
tracks a version that distro packagers have already vetted rather than
whichever release happens to be latest upstream.

The teuthology version matrix picks one of {3.9.2, 4.1, 4.2, 4.3} per run.
3.9.2 covers the last Zookeeper-based release for the migration cohort.
4.1/4.2/4.3 track Apache's current three supported minor lines and are
resolved to the latest patch at test time via dlcdn.apache.org/kafka/, so
new patches within a supported minor need no PR update. A new minor line
(4.4) requires renaming one YAML.
* new java version (17) is needed for kafka 4.2 and up (we wil use it on
  all version)
* kerberos task need to run before the kafka install task

After pulling this branch, fetch the new submodule:

    git submodule update --init src/librdkafka

Fixes: https://tracker.ceph.com/issues/77336

Signed-off-by: ShreeJejurikar <shreemj8@gmail.com>
Co-authored-by: Yuval Lifshitz <ylifshit@ibm.com>

.gitmodules

@@ -85,3 +85,6 @@
 [submodule "src/lss"]
 	path = src/lss
 	url = https://chromium.googlesource.com/linux-syscall-support
+[submodule "src/librdkafka"]
+	path = src/librdkafka
+	url = https://github.com/confluentinc/librdkafka.git

CMakeLists.txt

@@ -587,6 +587,7 @@ option(WITH_RADOSGW "RADOS Gateway is enabled" ON)
 option(WITH_RADOSGW_BEAST_OPENSSL "RADOS Gateway's Beast frontend uses OpenSSL" ON)
 option(WITH_RADOSGW_AMQP_ENDPOINT "RADOS Gateway's pubsub support for AMQP push endpoint" ON)
 option(WITH_RADOSGW_KAFKA_ENDPOINT "RADOS Gateway's pubsub support for Kafka push endpoint" ON)
+option(WITH_SYSTEM_RDKAFKA "build against system librdkafka instead of the bundled submodule" OFF)
 option(WITH_RADOSGW_LUA_PACKAGES "RADOS Gateway's support for dynamically adding lua packagess" ON)
 option(WITH_RADOSGW_DBSTORE "DBStore backend for RADOS Gateway" ON)
 option(WITH_RADOSGW_MOTR "CORTX-Motr backend for RADOS Gateway" OFF)

ceph.spec.in

@@ -125,6 +125,7 @@
 # not x86_64
 %bcond_with system_qat
 %endif
+%bcond_with system_rdkafka
 %if 0%{?fedora} || 0%{?suse_version} || 0%{?rhel} || 0%{?openEuler}
 %global weak_deps 1
 %endif
@@ -338,7 +339,11 @@ BuildRequires:  pkgconfig(yaml-cpp) >= 0.6
 BuildRequires:  pkgconfig(librabbitmq)
 %endif
 %if 0%{with kafka_endpoint}
-BuildRequires:  pkgconfig(rdkafka) >= 1.1.0
+%if 0%{with system_rdkafka}
+BuildRequires:  librdkafka-devel >= 2.11
+%else
+BuildRequires:  cyrus-sasl-devel
+%endif
 %endif
 %if 0%{with lua_packages}
 Requires:  lua-devel
@@ -1163,6 +1168,9 @@ Requires:	mailcap
 %if 0%{?weak_deps}
 Recommends:	gawk
 %endif
+%if 0%{with kafka_endpoint} && !0%{with system_rdkafka}
+Provides:	bundled(librdkafka) = 2.12.1
+%endif
 %description radosgw
 RADOS is a distributed object store used by the Ceph distributed
 storage system.  This package provides a REST gateway to the
@@ -1822,6 +1830,9 @@ cmake .. \
 %if 0%{with system_pmdk}
     -DWITH_SYSTEM_PMDK:BOOL=ON \
 %endif
+%if 0%{with system_rdkafka}
+    -DWITH_SYSTEM_RDKAFKA:BOOL=ON \
+%endif
 %if 0%{without jaeger}
     -DWITH_JAEGER:BOOL=OFF \
 %endif

debian/control

@@ -76,7 +76,8 @@ Build-Depends: automake,
                libgrpc++-dev,
                protobuf-compiler-grpc,
                libutf8proc-dev (>= 2.2.0),
-               librdkafka-dev (>= 1.1.0),
+               librdkafka-dev (>= 2.11) <pkg.ceph.system-rdkafka>,
+               libsasl2-dev <!pkg.ceph.system-rdkafka>,
                libthrift-dev (>= 0.13.0),
                libyaml-cpp-dev (>= 0.6),
                libzstd-dev <pkg.ceph.check>,

debian/rules

@@ -18,6 +18,9 @@ endif
 ifneq ($(filter pkg.ceph.crimson,$(DEB_BUILD_PROFILES)),)
   extraopts += -DWITH_CRIMSON=ON
 endif
+ifneq ($(filter pkg.ceph.system-rdkafka,$(DEB_BUILD_PROFILES)),)
+  extraopts += -DWITH_SYSTEM_RDKAFKA=ON
+endif
 
 extraopts += -DWITH_JAEGER=ON
 extraopts += -DWITH_SYSTEM_JERASURE=ON

doc/radosgw/notifications.rst

@@ -172,6 +172,10 @@ HTTP
 
 Kafka
 ~~~~~
+
+Supported broker versions: Kafka 3.9 and Kafka 4.x. Both Zookeeper-based
+(3.x) and KRaft (4.x) brokers are supported.
+
 After recovering from a broker failure, a persistent topic will try
 to resend all notifications in batches. If the topic is configured on
 the broker with a segment size smaller than our default (1MB), sending

qa/suites/rgw/notifications/tasks/kafka/0-install.yaml

@@ -10,9 +10,9 @@ overrides:
     ceph:
       extra_system_packages:
         rpm:
-        - java
+        - java-21-openjdk-headless
         deb:
-        - default-jre
+        - openjdk-21-jre-headless
   ceph:
     conf:
       global:

qa/suites/rgw/notifications/tasks/kafka/1-kerberos.yaml

@@ -0,0 +1,3 @@
+tasks:
+- kerberos:
+    client.0:

qa/suites/rgw/notifications/tasks/kafka/1-versions$/.qa

@@ -0,0 +1 @@
+../.qa/

qa/suites/rgw/notifications/tasks/kafka/1-versions$/3.9.2.yaml

@@ -0,0 +1,4 @@
+tasks:
+- kafka:
+    client.0:
+      kafka_version: 3.9.2

qa/suites/rgw/notifications/tasks/kafka/1-versions$/4.1.yaml

@@ -0,0 +1,4 @@
+tasks:
+- kafka:
+    client.0:
+      kafka_version: "4.1"

qa/suites/rgw/notifications/tasks/kafka/1-versions$/4.2.yaml

@@ -0,0 +1,4 @@
+tasks:
+- kafka:
+    client.0:
+      kafka_version: "4.2"

qa/suites/rgw/notifications/tasks/kafka/1-versions$/4.3.yaml

@@ -0,0 +1,4 @@
+tasks:
+- kafka:
+    client.0:
+      kafka_version: "4.3"

qa/suites/rgw/notifications/tasks/kafka/2-notification-tests.yaml

@@ -0,0 +1,5 @@
+tasks:
+- notification-tests:
+    client.0:
+      extra_attr: ["kafka_test", "kafka_security_test"]
+      rgw_server: client.0

qa/suites/rgw/notifications/tasks/kafka/test_kafka.yaml

@@ -1,10 +0,0 @@
-tasks:
-- kerberos:
-    client.0:
-- kafka:
-    client.0:
-      kafka_version: 3.9.2
-- notification-tests:
-    client.0:
-      extra_attr: ["kafka_test", "kafka_security_test"]
-      rgw_server: client.0

qa/tasks/kafka.py

@@ -3,8 +3,10 @@ Deploy and configure Kafka for Teuthology
 """
 import contextlib
 import logging
+import re
 import time
 import os
+import urllib.request
 
 from teuthology import misc as teuthology
 from teuthology import contextutil
@@ -26,8 +28,38 @@ def get_kafka_version(config):
             kafka_version = client_config.get('kafka_version')
     return kafka_version
 
+def is_kraft_mode(version_str):
+    """Kafka 4.0+ removed Zookeeper and uses KRaft consensus instead."""
+    return int(version_str.split('.')[0]) >= 4
+
+def resolve_kafka_version(version):
+    """
+    Resolve a minor version like "4.3" to the current patch (e.g. "4.3.1")
+    by querying Apache's live mirror. A fully-qualified version (three
+    dotted components, e.g. "3.9.2") is returned unchanged, so it can
+    still be used as a literal patch pin.
+
+    Raises RuntimeError if the minor line is not on dlcdn.apache.org.
+    """
+    if version.count('.') >= 2:
+        return version
+    with urllib.request.urlopen(
+            'https://dlcdn.apache.org/kafka/', timeout=30) as resp:
+        html = resp.read().decode()
+    pattern = re.compile(r'href="(' + re.escape(version) + r'\.\d+)/"')
+    matches = pattern.findall(html)
+    if not matches:
+        raise RuntimeError(
+            "Kafka {v}.x not found on dlcdn.apache.org "
+            "the minor line may have been dropped from Apache's "
+            "supported set.".format(v=version)
+        )
+    return max(matches, key=lambda v: tuple(int(p) for p in v.split('.')))
+
 kafka_prefix = 'kafka_2.13-'
 
+KRAFT_CONTROLLER_PORT = 9097
+
 def get_kafka_dir(ctx, config):
     kafka_version = get_kafka_version(config)
     current_version = kafka_prefix + kafka_version
@@ -55,6 +87,7 @@ def install_kafka(ctx, config):
         (remote,) = ctx.cluster.only(client).remotes.keys()
         test_dir=teuthology.get_testdir(ctx)
         current_version = get_kafka_version(config)
+        kraft = is_kraft_mode(current_version)
 
         kafka_file =  kafka_prefix + current_version + '.tgz'
 
@@ -104,7 +137,25 @@ def install_kafka(ctx, config):
         )
 
         client_kerberos = kerberos.get(client) if kerberos else None
-        broker_conf(ctx, client, kafka_dir, client_kerberos)
+        broker_conf(ctx, client, kafka_dir, client_kerberos, kraft)
+
+        if kraft:
+            # Kafka 4.x KRaft mode requires storage to be formatted before
+            # the broker starts. --standalone auto-bootstraps a single-node
+            # KRaft cluster.
+            uuid_result = remote.sh(
+                'cd {tdir}/bin && ./kafka-storage.sh random-uuid'.format(
+                    tdir=kafka_dir)
+            ).strip()
+            ctx.cluster.only(client).run(
+                args=[
+                    'cd', '{tdir}/bin'.format(tdir=kafka_dir), run.Raw('&&'),
+                    './kafka-storage.sh', 'format',
+                    '-t', uuid_result,
+                    '-c', '{tdir}/config/server.properties'.format(tdir=kafka_dir),
+                    '--standalone',
+                ],
+            )
 
     try:
         yield
@@ -126,20 +177,41 @@ def install_kafka(ctx, config):
             )
 
 
-def broker_conf(ctx, client, kafka_dir, kerberos):
+def broker_conf(ctx, client, kafka_dir, kerberos, kraft):
     """writing a custom server.properties config file"""
     (remote,) = ctx.cluster.only(client).remotes.keys()
     ip = remote.ip_address
+    if kraft:
+        identity = (
+            "node.id=1\n"
+            "process.roles=broker,controller\n"
+        )
+        controller_listener = ",CONTROLLER://0.0.0.0:{controller}".format(
+            controller=KRAFT_CONTROLLER_PORT)
+        controller_protocol = ",CONTROLLER:PLAINTEXT"
+        consensus_config = (
+            "controller.listener.names=CONTROLLER\n"
+            "controller.quorum.bootstrap.servers=localhost:{controller}\n"
+        ).format(controller=KRAFT_CONTROLLER_PORT)
+    else:
+        identity = "broker.id=0\n"
+        controller_listener = ""
+        controller_protocol = ""
+        consensus_config = (
+            "zookeeper.connect=localhost:2181\n"
+            "zookeeper.connection.timeout.ms=18000\n"
+        )
     conf = (
-        "broker.id=0\n"
+        identity +
         "listeners=PLAINTEXT://0.0.0.0:{plaintext},SSL://0.0.0.0:{ssl},"
         "SASL_SSL://0.0.0.0:{sasl_ssl},SASL_PLAINTEXT://0.0.0.0:{sasl_plaintext},"
-        "MTLS://0.0.0.0:{mtls}\n"
+        "MTLS://0.0.0.0:{mtls}" + controller_listener + "\n"
         "advertised.listeners=PLAINTEXT://{ip}:{plaintext},SSL://{ip}:{ssl},"
         "SASL_SSL://{ip}:{sasl_ssl},SASL_PLAINTEXT://{ip}:{sasl_plaintext},"
         "MTLS://{ip}:{mtls}\n"
         "listener.security.protocol.map=PLAINTEXT:PLAINTEXT,SSL:SSL,"
-        "SASL_SSL:SASL_SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,MTLS:SSL\n"
+        "SASL_SSL:SASL_SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,MTLS:SSL"
+        + controller_protocol + "\n"
         "inter.broker.listener.name=PLAINTEXT\n"
         "log.dirs={tdir}/data/kafka-logs\n"
         "num.network.threads=3\n"
@@ -155,8 +227,7 @@ def broker_conf(ctx, client, kafka_dir, kerberos):
         "log.retention.hours=168\n"
         "log.segment.bytes=1073741824\n"
         "log.retention.check.interval.ms=300000\n"
-        "zookeeper.connect=localhost:2181\n"
-        "zookeeper.connection.timeout.ms=18000\n"
+        + consensus_config +
         "group.initial.rebalance.delay.ms=0\n"
         # SSL configuration
         "ssl.keystore.location={tdir}/server.keystore.jks\n"
@@ -237,15 +308,17 @@ def broker_conf(ctx, client, kafka_dir, kerberos):
 @contextlib.contextmanager
 def run_kafka(ctx,config):
     """
-    This includes two parts:
-    1. Starting Zookeeper service
-    2. Starting Kafka service
+    Starts the Kafka broker. In Zookeeper mode (3.x) also starts Zookeeper
+    first; KRaft mode (4.x) embeds the controller in the broker process.
     """
     assert isinstance(config, dict)
-    log.info('Bringing up Zookeeper and Kafka services...')
+    kraft = is_kraft_mode(get_kafka_version(config))
+    log.info('Bringing up Kafka%s services...',
+             '' if kraft else ' and Zookeeper')
     for (client,_) in config.items():
         (remote,) = ctx.cluster.only(client).remotes.keys()
 
+        if not kraft:
             ctx.cluster.only(client).run(
                 args=['cd', '{tdir}/bin'.format(tdir=get_kafka_dir(ctx, config)), run.Raw('&&'),
                  './zookeeper-server-start.sh',
@@ -265,7 +338,8 @@ def run_kafka(ctx,config):
     try:
         yield
     finally:
-        log.info('Stopping Zookeeper and Kafka Services...')
+        log.info('Stopping Kafka%s services...',
+                 '' if kraft else ' and Zookeeper')
 
         for (client, _) in config.items():
             (remote,) = ctx.cluster.only(client).remotes.keys()
@@ -279,6 +353,7 @@ def run_kafka(ctx,config):
 
             time.sleep(5)
 
+            if not kraft:
                 ctx.cluster.only(client).run(
                     args=['cd', '{tdir}/bin'.format(tdir=get_kafka_dir(ctx, config)), run.Raw('&&'),
                      './zookeeper-server-stop.sh',
@@ -342,7 +417,12 @@ def task(ctx,config):
     tasks:
     - kafka:
         client.0:
-          kafka_version: 2.6.0
+          kafka_version: "4.3"
+
+    A minor version like "4.3" is resolved to the current patch at test
+    time via dlcdn.apache.org. A full version like "3.9.2" is used as-is.
+    Kafka 4.x is started in KRaft mode automatically; 3.x is started with
+    a colocated Zookeeper. Pick whichever version your suite needs.
     """
     assert config is None or isinstance(config, list) \
         or isinstance(config, dict), \
@@ -355,6 +435,14 @@ def task(ctx,config):
     if isinstance(config, list):
         config = dict.fromkeys(config)
 
+    for client_config in config.values():
+        if client_config and 'kafka_version' in client_config:
+            raw = client_config['kafka_version']
+            resolved = resolve_kafka_version(raw)
+            if resolved != raw:
+                log.info("Kafka version resolved: %s -> %s", raw, resolved)
+            client_config['kafka_version'] = resolved
+
     ctx.kafka_dir = get_kafka_dir(ctx, config)
 
     log.debug('Kafka config is %s', config)

qa/tasks/kafka_failover.py

@@ -12,10 +12,21 @@ from teuthology.orchestra import run
 
 log = logging.getLogger(__name__)
 
+# This task currently supports Kafka 3.x (Zookeeper) only. KRaft (4.x)
+# requires a different multi-controller bootstrap for the two-broker HA
+# topology this task spins up. The suite yaml at
+# qa/suites/rgw/notifications/tasks/kafka_failover/test_kafka.yaml pins
+# kafka_version: 3.9.2; the runtime check below fails fast if anyone
+# changes that to a 4.x version without updating the task.
 def get_kafka_version(config):
     for client, client_config in config.items():
         if 'kafka_version' in client_config:
             kafka_version = client_config.get('kafka_version')
+    if int(kafka_version.split('.')[0]) >= 4:
+        raise RuntimeError(
+            "kafka_failover task does not yet support Kafka 4.x (KRaft); "
+            "pin kafka_version to a 3.x release or extend this task."
+        )
     return kafka_version
 
 kafka_prefix = 'kafka_2.13-'

src/librdkafka

@@ -0,0 +1 @@
+Subproject commit e1db7eaa517f0a6438bc846a9c49ede73b9ea211

src/rgw/CMakeLists.txt

@@ -472,7 +472,35 @@ if(WITH_RADOSGW_AMQP_ENDPOINT)
   find_package(RabbitMQ REQUIRED)
 endif()
 if(WITH_RADOSGW_KAFKA_ENDPOINT)
-  find_package(RDKafka 1.1.0 REQUIRED)
+  set(rdkafka_min_version 2.11)
+  if(WITH_SYSTEM_RDKAFKA)
+    find_package(RDKafka ${rdkafka_min_version} REQUIRED)
+  elseif(EXISTS "${CMAKE_SOURCE_DIR}/src/librdkafka/CMakeLists.txt")
+    message(STATUS "Building librdkafka as submodule")
+    set(WITH_SSL ON)
+    set(WITH_SASL ON)
+    set(RDKAFKA_BUILD_EXAMPLES OFF)
+    set(RDKAFKA_BUILD_TESTS OFF)
+    set(RDKAFKA_BUILD_STATIC ON)
+    add_subdirectory(${CMAKE_SOURCE_DIR}/src/librdkafka
+                     ${CMAKE_BINARY_DIR}/src/librdkafka
+                     EXCLUDE_FROM_ALL)
+    target_compile_options(rdkafka PRIVATE -w)
+    if(NOT TARGET RDKafka::RDKafka)
+      add_library(RDKafka::RDKafka ALIAS rdkafka)
+    endif()
+    # create a compatibility symlink, so that rgw_a can include <librdkafka/rdkafka.h>
+    # both for the submodule and system librdkafka
+    set(rdkafka_compat_dir "${CMAKE_BINARY_DIR}/src/librdkafka-compat")
+    file(MAKE_DIRECTORY "${rdkafka_compat_dir}")
+    execute_process(COMMAND ${CMAKE_COMMAND} -E create_symlink
+      "${CMAKE_SOURCE_DIR}/src/librdkafka/src"
+      "${rdkafka_compat_dir}/librdkafka")
+    target_include_directories(rdkafka PUBLIC "$<BUILD_INTERFACE:${rdkafka_compat_dir}>")
+  else()
+    message(STATUS "src/librdkafka submodule not present; falling back to system librdkafka")
+    find_package(RDKafka ${rdkafka_min_version} REQUIRED)
+  endif()
 endif()
 
 target_link_libraries(rgw_a

src/test/CMakeLists.txt

@@ -85,8 +85,14 @@ if(WITH_RADOSGW)
     list(APPEND rgw_libs amqp_mock)
   endif()
   if(WITH_RADOSGW_KAFKA_ENDPOINT)
+    # kafka_stub provides fake rd_kafka_* symbols for tests; needed
+    # whenever we link against a dynamic system librdkafka (either
+    # opted-in via WITH_SYSTEM_RDKAFKA, or as a fallback when the
+    # bundled submodule is not checked out).
+    if(WITH_SYSTEM_RDKAFKA OR NOT EXISTS "${CMAKE_SOURCE_DIR}/src/librdkafka/CMakeLists.txt")
       list(APPEND rgw_libs kafka_stub)
     endif()
+  endif()
   add_subdirectory(rgw)
 endif(WITH_RADOSGW)
 if(WITH_RBD AND NOT WIN32)

src/test/rgw/CMakeLists.txt

@@ -21,11 +21,15 @@ if(WITH_RADOSGW_AMQP_ENDPOINT)
 endif()
 
 if(WITH_RADOSGW_KAFKA_ENDPOINT)
-  # kafka stub library
+  # kafka stub library — built whenever RGW links against a dynamic
+  # system librdkafka (WITH_SYSTEM_RDKAFKA=ON, or the fallback path
+  # when the bundled submodule is not checked out).
+  if(WITH_SYSTEM_RDKAFKA OR NOT EXISTS "${CMAKE_SOURCE_DIR}/src/librdkafka/CMakeLists.txt")
     set(kafka_stub_src
       kafka_stub.cc)
     add_library(kafka_stub STATIC ${kafka_stub_src})
   endif()
+endif()
 
 if(WITH_RADOSGW_LUA_PACKAGES)
   list(APPEND rgw_libs Boost::filesystem)

src/test/rgw/bucket_notification/README.rst

@@ -59,6 +59,33 @@ After running `vstart.sh`, Zookeeper, and Kafka services you're ready to run the
 
         BNTESTS_CONF=bntests.conf python -m pytest -s /path/to/ceph/src/test/rgw/bucket_notification/test_bn.py -v -m 'kafka_test'
 
+------------------------------
+Kafka 4.x setup (KRaft mode)
+------------------------------
+
+Kafka 4.0 removed Zookeeper. The broker uses its own KRaft consensus,
+so there is no separate Zookeeper process to start.
+
+First-time setup (per Kafka installation), format the storage::
+
+        KAFKA_CLUSTER_ID=$(bin/kafka-storage.sh random-uuid)
+        bin/kafka-storage.sh format -t $KAFKA_CLUSTER_ID -c config/server.properties --standalone
+
+The ``--standalone`` flag is required for single-node test setups.
+
+Then start the broker::
+
+        bin/kafka-server-start.sh config/server.properties
+
+For security tests, the same ``server.properties`` edits in the section
+below apply. The KRaft controller listener (e.g. ``CONTROLLER://:9097``)
+must remain in the ``listeners`` line on a port outside the 9092-9096
+range, since those are reserved by the security listeners. Do not set
+``controller.quorum.voters``; the ``--standalone`` flag passed to
+``kafka-storage.sh format`` writes the voter set into the bootstrap
+metadata automatically, and Kafka 4.1.1+ rejects the combination of
+``--standalone`` with an explicit ``controller.quorum.voters``.
+
 --------------------
 Kafka Security Tests
 --------------------