From: ShreeJejurikar Date: Sat, 20 Jun 2026 10:58:49 +0000 (+0530) Subject: rgw: support kafka 4.x brokers X-Git-Url: http://git-server-git.apps.pok.os.sepia.ceph.com/?a=commitdiff_plain;h=efe54523a45ef1ba1f824dd55705a42a00fbd73e;p=ceph.git 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 Co-authored-by: Yuval Lifshitz --- diff --git a/.gitmodules b/.gitmodules index 58bd31d06bf..b0473d24fff 100644 --- a/.gitmodules +++ b/.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 diff --git a/CMakeLists.txt b/CMakeLists.txt index 43ce6ef4147..ca29f4ddae1 100644 --- a/CMakeLists.txt +++ b/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) diff --git a/ceph.spec.in b/ceph.spec.in index e7032069af6..027250be3ee 100644 --- a/ceph.spec.in +++ b/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 diff --git a/debian/control b/debian/control index b91fd483239..2f243909db9 100644 --- a/debian/control +++ b/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) , + libsasl2-dev , libthrift-dev (>= 0.13.0), libyaml-cpp-dev (>= 0.6), libzstd-dev , diff --git a/debian/rules b/debian/rules index ce9a07a7f65..c4af2e2fe00 100755 --- a/debian/rules +++ b/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 diff --git a/doc/radosgw/notifications.rst b/doc/radosgw/notifications.rst index 2587501a5f0..0a86cff37a7 100644 --- a/doc/radosgw/notifications.rst +++ b/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 diff --git a/qa/suites/rgw/notifications/tasks/kafka/0-install.yaml b/qa/suites/rgw/notifications/tasks/kafka/0-install.yaml index 5c83d5c0d23..a65eb91834b 100644 --- a/qa/suites/rgw/notifications/tasks/kafka/0-install.yaml +++ b/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: diff --git a/qa/suites/rgw/notifications/tasks/kafka/1-kerberos.yaml b/qa/suites/rgw/notifications/tasks/kafka/1-kerberos.yaml new file mode 100644 index 00000000000..9fda9db6cc5 --- /dev/null +++ b/qa/suites/rgw/notifications/tasks/kafka/1-kerberos.yaml @@ -0,0 +1,3 @@ +tasks: +- kerberos: + client.0: diff --git a/qa/suites/rgw/notifications/tasks/kafka/1-versions$/.qa b/qa/suites/rgw/notifications/tasks/kafka/1-versions$/.qa new file mode 120000 index 00000000000..a602a0353e7 --- /dev/null +++ b/qa/suites/rgw/notifications/tasks/kafka/1-versions$/.qa @@ -0,0 +1 @@ +../.qa/ \ No newline at end of file diff --git a/qa/suites/rgw/notifications/tasks/kafka/1-versions$/3.9.2.yaml b/qa/suites/rgw/notifications/tasks/kafka/1-versions$/3.9.2.yaml new file mode 100644 index 00000000000..ff22b3add8c --- /dev/null +++ b/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 diff --git a/qa/suites/rgw/notifications/tasks/kafka/1-versions$/4.1.yaml b/qa/suites/rgw/notifications/tasks/kafka/1-versions$/4.1.yaml new file mode 100644 index 00000000000..991a8d9634f --- /dev/null +++ b/qa/suites/rgw/notifications/tasks/kafka/1-versions$/4.1.yaml @@ -0,0 +1,4 @@ +tasks: +- kafka: + client.0: + kafka_version: "4.1" diff --git a/qa/suites/rgw/notifications/tasks/kafka/1-versions$/4.2.yaml b/qa/suites/rgw/notifications/tasks/kafka/1-versions$/4.2.yaml new file mode 100644 index 00000000000..404354b49de --- /dev/null +++ b/qa/suites/rgw/notifications/tasks/kafka/1-versions$/4.2.yaml @@ -0,0 +1,4 @@ +tasks: +- kafka: + client.0: + kafka_version: "4.2" diff --git a/qa/suites/rgw/notifications/tasks/kafka/1-versions$/4.3.yaml b/qa/suites/rgw/notifications/tasks/kafka/1-versions$/4.3.yaml new file mode 100644 index 00000000000..a3debf0fe0b --- /dev/null +++ b/qa/suites/rgw/notifications/tasks/kafka/1-versions$/4.3.yaml @@ -0,0 +1,4 @@ +tasks: +- kafka: + client.0: + kafka_version: "4.3" diff --git a/qa/suites/rgw/notifications/tasks/kafka/2-notification-tests.yaml b/qa/suites/rgw/notifications/tasks/kafka/2-notification-tests.yaml new file mode 100644 index 00000000000..0e2bbe28b0b --- /dev/null +++ b/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 \ No newline at end of file diff --git a/qa/suites/rgw/notifications/tasks/kafka/test_kafka.yaml b/qa/suites/rgw/notifications/tasks/kafka/test_kafka.yaml deleted file mode 100644 index 915b93199b2..00000000000 --- a/qa/suites/rgw/notifications/tasks/kafka/test_kafka.yaml +++ /dev/null @@ -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 diff --git a/qa/tasks/kafka.py b/qa/tasks/kafka.py index 0d2f095826d..3116b380e71 100644 --- a/qa/tasks/kafka.py +++ b/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,22 +308,24 @@ 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() - ctx.cluster.only(client).run( - args=['cd', '{tdir}/bin'.format(tdir=get_kafka_dir(ctx, config)), run.Raw('&&'), - './zookeeper-server-start.sh', - '{tir}/config/zookeeper.properties'.format(tir=get_kafka_dir(ctx, config)), - run.Raw('&'), 'exit' - ], - ) + 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', + '{tir}/config/zookeeper.properties'.format(tir=get_kafka_dir(ctx, config)), + run.Raw('&'), 'exit' + ], + ) ctx.cluster.only(client).run( args=['cd', '{tdir}/bin'.format(tdir=get_kafka_dir(ctx, config)), run.Raw('&&'), @@ -265,28 +338,30 @@ 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() ctx.cluster.only(client).run( args=['cd', '{tdir}/bin'.format(tdir=get_kafka_dir(ctx, config)), run.Raw('&&'), - './kafka-server-stop.sh', + './kafka-server-stop.sh', '{tir}/config/server.properties'.format(tir=get_kafka_dir(ctx, config)), ], ) time.sleep(5) - ctx.cluster.only(client).run( - args=['cd', '{tdir}/bin'.format(tdir=get_kafka_dir(ctx, config)), run.Raw('&&'), - './zookeeper-server-stop.sh', - '{tir}/config/zookeeper.properties'.format(tir=get_kafka_dir(ctx, config)), - ], - ) + 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', + '{tir}/config/zookeeper.properties'.format(tir=get_kafka_dir(ctx, config)), + ], + ) - time.sleep(5) + time.sleep(5) ctx.cluster.only(client).run(args=['killall', '-9', 'java']) @@ -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) diff --git a/qa/tasks/kafka_failover.py b/qa/tasks/kafka_failover.py index 2658217beef..4c3870e34de 100644 --- a/qa/tasks/kafka_failover.py +++ b/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-' diff --git a/src/librdkafka b/src/librdkafka new file mode 160000 index 00000000000..e1db7eaa517 --- /dev/null +++ b/src/librdkafka @@ -0,0 +1 @@ +Subproject commit e1db7eaa517f0a6438bc846a9c49ede73b9ea211 diff --git a/src/rgw/CMakeLists.txt b/src/rgw/CMakeLists.txt index 875c5b21397..872d3a8bc13 100644 --- a/src/rgw/CMakeLists.txt +++ b/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 + # 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 "$") + 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 diff --git a/src/test/CMakeLists.txt b/src/test/CMakeLists.txt index c5da3310582..2df929008e6 100644 --- a/src/test/CMakeLists.txt +++ b/src/test/CMakeLists.txt @@ -85,7 +85,13 @@ if(WITH_RADOSGW) list(APPEND rgw_libs amqp_mock) endif() if(WITH_RADOSGW_KAFKA_ENDPOINT) - list(APPEND rgw_libs kafka_stub) + # 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) diff --git a/src/test/rgw/CMakeLists.txt b/src/test/rgw/CMakeLists.txt index 04791cd5cda..c8d9d605f31 100644 --- a/src/test/rgw/CMakeLists.txt +++ b/src/test/rgw/CMakeLists.txt @@ -21,10 +21,14 @@ if(WITH_RADOSGW_AMQP_ENDPOINT) endif() if(WITH_RADOSGW_KAFKA_ENDPOINT) - # kafka stub library - set(kafka_stub_src - kafka_stub.cc) - add_library(kafka_stub STATIC ${kafka_stub_src}) + # 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) diff --git a/src/test/rgw/bucket_notification/README.rst b/src/test/rgw/bucket_notification/README.rst index 59b3f5818e1..9373c5509b6 100644 --- a/src/test/rgw/bucket_notification/README.rst +++ b/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 --------------------