[namikmesic-hub]

Summary

KafkaClient::connect built an idempotent producer for every client, and librdkafka requests a producer id 500 ms after creation. A broker authorizes that as IdempotentWrite on the cluster or Write on a topic, which a service that only consumes lacks.

The producer is now built at the first publish, on a blocking thread, and shared by every clone. Instead, connect builds the consumer-type metadata client that ping probes with, so a connection setting librdkafka refuses still fails at startup. A client therefore keeps one idle librdkafka client, as before. The close in shutdown waits for a producer build in flight and refuses a later one.

Tests

kafka_consume_only_mock: a_client_that_only_consumes_sends_no_init_producer_id, red before the fix with one InitProducerId, and a_client_closed_before_any_publish_builds_no_producer.

client::tests: connect_builds_no_producer_and_concurrent_clones_share_one, a_publish_after_shutdown_builds_no_producer, a_failed_producer_build_leaves_the_cell_empty_for_the_next_publish, rejected_tls_material_fails_connect, a_probe_reports_a_refused_configuration, a_producer_build_blocked_in_librdkafka_leaves_the_runtime_free, shutdown_waits_for_a_producer_build_in_flight.

kafka_msk_iam_integration: msk_iam_connect_rejects_tls_material_librdkafka_refuses, msk_iam_ping_times_out_against_a_non_routable_broker, and msk_iam_generate_oauth_token_does_not_panic now closes the client after its publish.

Test plan

• cargo nextest run --features <kafka set> on the mock, MSK IAM and Docker-backed Kafka suites, --lib, and --no-default-features
• cargo fmt -- --check and cargo clippy -D warnings on every CI feature set, the nine --lib sets included

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.

📢 Thoughts on this report? Let us know!

[zannis]

Building the producer at the first publish is the right fix, and the mock-cluster test that counts InitProducerId after the 500 ms timer is a good guard. I think the change can be smaller: build the ping client at connect (which also keeps config errors failing at startup), and drop the builds_in_flight lock, which OnceCell already makes unnecessary. Details inline, plus one docs wording fix.

[zannis]

With the producer built lazily, a missing write permission only shows up at a consumer's first republish. That includes a handler timeout or panic, both of which become Retry. A consumer that can publish (RetryStrategy::Republish, the default on owned topologies, or a declared DLQ) needs Write on the target topic, plus IdempotentWrite on brokers before 2.8. Without it, InitProducerId fails with CLUSTER_AUTHORIZATION_FAILED, librdkafka marks the producer fatal, and the cell keeps it for the client's lifetime. Then:

• Concurrent path: the partition's commits stop at the first retried record.
• PerRecord: the member reconnects in a loop on that record.
• Sequenced topics: the next commit moves past the retried message, so it's dropped and not counted.
• DLQ: every reject is lost.

Could the consumer check this at startup when its config can publish, and fail with an error naming the missing permission? Two options:

1. Call producer() at start, wait about 500 ms plus one round trip, and check client().fatal_error(). It's simple, but it's a negative-only signal.
2. Describe the topic and the DLQ with authorized operations included (KIP-430) and check for WRITE. The answer is definite and needs only Describe, but rdkafka 0.39 doesn't wrap it, so it would go through rdkafka-sys FFI.

InPlace with no DLQ, and broadcast, would skip the check.

[zannis]

Moving this to a follow-up PR: a startup check that a consumer able to publish (Republish retries or a DLQ) is allowed to write. Not blocking this one.

[zannis]

The second commit covers the three review points: the metadata client is built at connect again, so refused TLS/SASL settings fail at startup; the build lock is gone; and the docs sentence is narrowed. The startup write-permission check moves to a follow-up. That hazard predates this change: on main the idempotent producer already went fatal shortly after connect.