"""
import contextlib
import logging
+import re
import time
import os
+import urllib.request
from teuthology import misc as teuthology
from teuthology import contextutil
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
(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'
)
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
)
-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"
"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"
@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('&&'),
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'])
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), \
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)
projects / ceph.git / commitdiff