feat: Kafka admin client by labrenbe · Pull Request #1045 · stackabletech/kafka-operator

labrenbe · 2026-10-16T08:45:17Z

Part of stackabletech/issues#888.

Stacked on #1028.

Description

New crate stackable-kafka-admin with the requests for describing, creating and deleting topics, adding partitions, reading and changing topic configs, and listing partition reassignments.
Plaintext, TLS and mutual TLS connections to the bootstrap servers, with the certificates read from files on every new connection, so rotated certificates are picked up without a restart.
One connection at a time: it is reused across requests, replaced after idling, reconnected after a failed request, and a request that hits a connection the broker already closed is sent once more.
API versions are pinned to what Kafka 3.9.2 and 4.3.1 both support and checked against the versions the broker advertises.
Errors distinguish a request the broker rejected, with the Kafka error code, from a connection problem, so the caller can report them differently.
A shared crate carries the contract between operator and agent: how the operator configures the agent's connection to Kafka through environment variables.
Unit tests run the client against a mock broker on a local socket.

Out-of-scope

Everything the agent does not need: ACLs, consumer groups, quotas, broker configs, producing and consuming, explicit replica placement, and partition reassignments beyond listing the running ones.
ZooKeeper clusters. Requests go to a bootstrap broker, which in KRaft mode forwards them to the controller.
SASL authentication, including Kerberos. Only anonymous, TLS and mutual TLS are supported.
API version negotiation. The versions are pinned to what Kafka 3.9.2 and 4.3.1 both support and checked against the broker.
Connection management beyond one connection with one request at a time and a single retry on a closed connection: no pooling, no backoff, no metrics.

Definition of Done Checklist

Not all of these items are applicable to all PRs, the author should update this template to only leave the boxes in that are relevant
Please make sure all these things are done and tick the boxes

Author

Changes are OpenShift compatible
CRD changes approved
CRD documentation for all fields, following the style guide.
Helm chart can be installed and deployed operator works
Integration tests passed (for non trivial changes)
Changes need to be "offline" compatible
Links to generated (nightly) docs added
Release note snippet added

Reviewer

Code contains useful comments
Code contains useful logging statements
(Integration-)Test cases added
Documentation added or updated. Follows the style guide.
Changelog updated
Cargo.toml only contains references to git tags (not specific commits or branches)

Acceptance

Feature Tracker has been updated
Proper release label has been added
Links to generated (nightly) docs added
Release note snippet added
Add type/deprecation label & add to the deprecation schedule
Add type/experimental label & add to the experimental features tracker

NickLarsenNZ · 2026-10-21T11:14:00Z

 indoc = "2.0"
+kafka-protocol = { version = "0.18", default-features = false, features = ["broker", "client"] }
 rstest = "0.26"
+rustls = { version = "0.23", default-features = false, features = ["logging", "ring", "std", "tls12"] }


Note to self, check if we enable the LogTracer (to get the logs). In 0.24, rustls will enable tracing so we won't need a log bridge for that.

NickLarsenNZ · 2026-10-21T11:19:13Z

+        if self.host.contains(':') {
+            write!(f, "[{}]:{}", self.host, self.port)
+        } else {
+            write!(f, "{}:{}", self.host, self.port)
+        }


nit: We should avoid positional (unnamed) interpolations.

Something like this reads nicer:

Suggested change

if self.host.contains(':') {

write!(f, "[{}]:{}", self.host, self.port)

} else {

write!(f, "{}:{}", self.host, self.port)

}

let Self { host, port } = self;

if self.host.contains(':') {

write!(f, "[{host}]:{port}")

} else {

write!(f, "{host}:{port}")

}

To be even more sure, we could check by parsing the ipv6 address instead of looking for :.

Changed in 28efc85.

…afka admin client

xeniape · 2026-10-26T13:29:18Z

+        describe_topic_configs_result(response, topic)
+    }
+
+    /// `None` uses the broker default for the number of partitions or the replication factor.


This comment is unclear what it refers to. It should describe the function?

Improved in b00fe65

(The comment was referring to the functions behaviour when the optional replication factor and partitions are None)

xeniape · 2026-10-26T13:44:43Z

+        .context(MissingResultSnafu { api_key, topic })
+}
+
+fn check(api_key: i16, topic: &str, error_code: i16, message: Option<StrBytes>) -> Result<()> {


Would prefer better naming, what gets checked?

Renamed this and other helpers in b00fe65.

xeniape · 2026-10-26T13:45:41Z

+}
+
+// Used instead of [0] to correctly handle empty results & take ownership
+fn first<T>(results: Vec<T>, api_key: i16, topic: &str) -> Result<T> {


Here also, would prefer better naming, I assume it something like return_first_result?

Renamed to first_result in b00fe65.

xeniape · 2026-10-26T13:47:57Z

+    // Without the details, e.g. if `DESCRIBE_CONFIGS` is not allowed, the topic was still created.
+    check(
+        CreateTopicsRequest::KEY,
+        topic,
+        result.topic_config_error_code,
+        None,
+    )?;


I don't understand this part. So the topic gets created but the user still sees an error but also without an error message? So how would they understand what went wrong while the topic is still there now?

The error is surfaced to the user in the KafkaTopic condition as KafkaRejected with a message like the broker rejected the CreateTopics request for the topic "<topic>" with TopicAuthorizationFailed. We check the response from Kafka for an error code which then gets returned to the caller as the Error::Kafka variant.

xeniape · 2026-10-26T13:56:35Z

+const METADATA_VERSION: i16 = 12;
+const CREATE_TOPICS_VERSION: i16 = 7;
+const DELETE_TOPICS_VERSION: i16 = 6;
+const CREATE_PARTITIONS_VERSION: i16 = 3;
+const DESCRIBE_CONFIGS_VERSION: i16 = 4;
+const INCREMENTAL_ALTER_CONFIGS_VERSION: i16 = 1;
+const LIST_PARTITION_REASSIGNMENTS_VERSION: i16 = 0;
+
+/// Must stay below the connection's request timeout.
+const BROKER_TIMEOUT_MS: i32 = 25_000;
+const _: () = assert!((BROKER_TIMEOUT_MS as u128) < connection::REQUEST_TIMEOUT.as_millis());
+
+/// Must stay below the broker's `connections.max.idle.ms` (10 minutes by default).
+const CONNECTION_MAX_IDLE: Duration = Duration::from_secs(5 * 60);
+
+/// Lets the broker apply `num.partitions` or `default.replication.factor` when creating a topic.
+const BROKER_DEFAULT: i32 = -1;
+
+const RESOURCE_TYPE_TOPIC: i8 = 2;
+const CONFIG_OPERATION_SET: i8 = 0;
+const CONFIG_OPERATION_DELETE: i8 = 1;


I don't understand all those numbers here, could you add some documentation to that please? Why and how were those numbers chosen? If someone needs to maintain this they need some guidance I would say.

Added documentation on the source for the constants in 56b9894.

add kafka admin client managing topics, topic config and partition count

c26dc42

labrenbe self-assigned this Oct 16, 2026

labrenbe changed the base branch from feat/kafka-agent to main October 16, 2026 08:51

labrenbe changed the base branch from main to feat/kafka-agent October 16, 2026 08:51

labrenbe added 5 commits October 16, 2026 14:17

add support for list_partition_reassignments

87a5479

replace kafka_error with semantic error checks in the kafka admin client

798619e

remove is retryable checks for kafka errors

d0a7abb

simplify kafka error handling

95581f0

add a flag for topic already exists errors

69ed88e

NickLarsenNZ reviewed Oct 21, 2026

View reviewed changes

labrenbe added 3 commits October 21, 2026 15:15

Merge branch 'feat/kafka-agent' into feat/kafka-admin

8241cb3

use BrokerAddress in more places and remove awaiting_response

58e3393

fix flaky unit test

198e800

adwk67 reviewed Oct 21, 2026

View reviewed changes

Comment thread rust/kafka-admin/src/connection.rs Resolved

Comment thread Cargo.toml Resolved

Comment thread rust/kafka-admin/src/client.rs Resolved

labrenbe added 2 commits October 23, 2026 15:42

return the created topic and all configs with their source from the k…

668119f

…afka admin client

Merge branch 'feat/kafka-agent' into feat/kafka-admin

dec1799

labrenbe marked this pull request as ready for review October 23, 2026 18:14

labrenbe added this to Stackable Engineering Oct 23, 2026

labrenbe moved this to Development: In Review in Stackable Engineering Oct 23, 2026

labrenbe added 2 commits October 23, 2026 20:53

Merge branch 'feat/kafka-agent' into feat/kafka-admin

75eb776

Merge branch 'feat/kafka-agent' into feat/kafka-admin

b95fb0a

xeniape reviewed Oct 26, 2026

View reviewed changes

Comment thread rust/kafka-admin/src/client.rs Resolved

labrenbe added 2 commits October 27, 2026 09:13

Merge branch 'feat/kafka-agent' into feat/kafka-admin

e0e76ed

Merge branch 'feat/kafka-agent' into feat/kafka-admin

51e863a

labrenbe added 15 commits October 27, 2026 14:56

Merge branch 'feat/kafka-agent' into feat/kafka-admin

9549489

Merge branch 'feat/kafka-agent' into feat/kafka-admin

e7d2299

return only topic-level config and no source

150886f

document source of magic kafka numbers

56b9894

use ApiKey enum

1aee2b0

Merge branch 'feat/kafka-agent' into feat/kafka-admin

b435ce4

Merge branch 'feat/kafka-agent' into feat/kafka-admin

2f23562

Merge branch 'feat/kafka-agent' into feat/kafka-admin

cb10888

retry broken connection requests immediatly with a new connection

a326d59

Merge branch 'feat/kafka-agent' into feat/kafka-admin

048ff91

improve broker address Display impl

28efc85

build the tls config once instead of per-server

82f8950

keep the connection if only the version check fails

1937b53

rename helpers and doc comments to be more descriptive

b00fe65

Merge branch 'feat/kafka-agent' into feat/kafka-admin

8ec0341

labrenbe mentioned this pull request Oct 30, 2026

feat: KafkaTopic controller #1050

Open

20 tasks

Conversation

labrenbe commented Oct 16, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Description

Out-of-scope

Definition of Done Checklist

Author

Reviewer

Acceptance

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

xeniape Oct 26, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

labrenbe Oct 29, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Reviewers

Uh oh!

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

labrenbe commented Oct 16, 2026 •

edited

Loading

xeniape Oct 26, 2026 •

edited

Loading

labrenbe Oct 29, 2026 •

edited

Loading