📢 แจ้งเตือน: confluentinc/librdkafka/pull/

Kaushik Raina (k-raina) · 2026-09-04T15:42:58Z

A consumer with enable.auto.commit=true could send an OffsetCommit for an assignment it had already lost. On a client-side session timeout, the member id is reset to empty, the assignment is marked lost, and its partitions are revoked.

The revoke-time auto-commit of those partitions was still sent because the assignment-lost flag was cleared inside rd_kafka_cgrp_unassign() and rd_kafka_cgrp_incremental_unassign(), before the removed partitions were served, defeating the guard that skips commits for a lost assignment.

The commit left with an empty member id and the previous generation. A broker rejects it with UNKNOWN_MEMBER_ID, or, for a static member (group.instance.id set), with FENCED_INSTANCE_ID, which librdkafka raises as a fatal error that stops the consumer.

Keep the assignment-lost flag set until the following rejoin applies a new assignment (assign()/incremental_assign() still clear it), so the offsets of a lost assignment are never committed. This restores the behavior documented in rd_kafka_cgrp_incremental_unassign()'s own comment.

Add test 0192 (static member, cooperative-sticky, mock) which asserts that no OffsetCommit is sent after a session timeout.

Happening since 1.6.0.

Alternative Approach

Add a check at offset commit i.e "subscribed, but member id is blank → refuse locally with _ASSIGNMENT_LOST."
Only impact of alternative: A commit made after being kicked returns _ASSIGNMENT_LOST instead of broker UNKNOWN_MEMBER_ID (dynamic) or a fatal FENCED_INSTANCE_ID (static). Apps matching on UNKNOWN_MEMBER_ID should handle both. This needs to be mentioned in ChangeLog

confluent-cla-assistant · 2026-09-04T15:43:09Z

🎉 All Contributor License Agreements have been signed. Ready to merge.
_{Please push an empty commit if you would like to re-run the checks to verify CLA status for all contributors.}

Copilot

🟡 Changes recommended

The new test stops mock request tracking before test_consumer_close(), which can miss OffsetCommit requests emitted during close and weaken the regression coverage.

Once you've addressed the issues Copilot identified, you can request another Copilot review.

Pull request overview

This PR fixes a classic consumer group edge case where auto-commit could incorrectly send an OffsetCommit after the consumer has already lost its assignment (e.g., on client-side session timeout), which can trigger fatal errors for static members (group.instance.id).

Changes:

Stop clearing the “assignment lost” flag during unassign() / incremental_unassign(), keeping it set until a new assignment is applied.
Add a new mock-based regression test (0192) for static membership + cooperative-sticky that asserts no OffsetCommit is sent after a client-side session timeout.
Document the fix in the changelog.

File summaries

File	Description
tests/test.c	Registers the new 0192 test in the test harness.
tests/CMakeLists.txt	Adds the new 0192 test source file to the build.
tests/0192-cgrp_static_session_timeout_mock.c	New mock test asserting no `OffsetCommit` is emitted after a session timeout for a static member.
src/rdkafka_cgrp.c	Keeps the assignment-lost flag set across unassign paths to prevent commits for lost assignments.
CHANGELOG.md	Notes the consumer fix and impacted versions/conditions.

Review details

Files reviewed: 5/5 changed files
Comments generated: 1
Review effort level: Lite

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.

+        offset_commit_cnt = test_mock_get_matching_request_cnt(
+            mcluster, is_offset_commit_request, NULL);
+
+        TEST_SAY("%" PRIusz
+                 " OffsetCommit request(s) sent after the "
+                 "session timeout\n",
+                 offset_commit_cnt);
+
+        rd_kafka_mock_stop_request_tracking(mcluster);
+
+        TEST_ASSERT(offset_commit_cnt == 0,
+                    "Expected no OffsetCommit for the lost assignment, "
+                    "but %" PRIusz
+                    " were sent. "
+                    "The member id is reset to \"\" by the session timeout "
+                    "while group.instance.id is still set, so a real broker "
+                    "answers FENCED_INSTANCE_ID and the consumer goes fatal.",
+                    offset_commit_cnt);
+
+        test_consumer_close(c);
+
+        rd_kafka_destroy(c);
+
+        test_mock_cluster_destroy(mcluster);


Emanuele Sabellico (emasab)

LGTM, Thanks Kaushik Raina (@k-raina) !

A consumer with enable.auto.commit=true could send an OffsetCommit for an assignment it had already lost. On a client-side session timeout, or when max.poll.interval.ms is exceeded, the member id is reset to empty, the assignment is marked lost, and its partitions are revoked. The revoke-time auto-commit of those partitions was still sent because the assignment-lost flag was cleared inside rd_kafka_cgrp_unassign() and rd_kafka_cgrp_incremental_unassign(), before the removed partitions were served, defeating the guard that skips commits for a lost assignment. The commit left with an empty member id and the previous generation. A broker rejects it with UNKNOWN_MEMBER_ID, or, for a static member (group.instance.id set), with FENCED_INSTANCE_ID, which librdkafka raises as a fatal error that stops the consumer. Keep the assignment-lost flag set until the following rejoin applies a new assignment (assign()/incremental_assign() still clear it), so the offsets of a lost assignment are never committed. This restores the behavior documented in rd_kafka_cgrp_incremental_unassign()'s own comment. Add test 0192 (static member, cooperative-sticky, mock) which asserts that no OffsetCommit is sent after a session timeout. Happening since 1.6.0.

…signment lost

Extend the #5585 entry with the second way a lost assignment was still auto-committed: a session timeout or max.poll.interval.ms expiry landing while a rebalance was already in progress left the assignment unmarked as lost, so the already-queued revoke committed as an ordinary one. Also list the fix in the v2.16.0 release summary. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

* Don't commit offsets of a lost assignment A consumer with enable.auto.commit=true could send an OffsetCommit for an assignment it had already lost. On a client-side session timeout, or when max.poll.interval.ms is exceeded, the member id is reset to empty, the assignment is marked lost, and its partitions are revoked. The revoke-time auto-commit of those partitions was still sent because the assignment-lost flag was cleared inside rd_kafka_cgrp_unassign() and rd_kafka_cgrp_incremental_unassign(), before the removed partitions were served, defeating the guard that skips commits for a lost assignment. The commit left with an empty member id and the previous generation. A broker rejects it with UNKNOWN_MEMBER_ID, or, for a static member (group.instance.id set), with FENCED_INSTANCE_ID, which librdkafka raises as a fatal error that stops the consumer. Keep the assignment-lost flag set until the following rejoin applies a new assignment (assign()/incremental_assign() still clear it), so the offsets of a lost assignment are never committed. This restores the behavior documented in rd_kafka_cgrp_incremental_unassign()'s own comment. Add test 0192 (static member, cooperative-sticky, mock) which asserts that no OffsetCommit is sent after a session timeout. Happening since 1.6.0. * Clear assignment lost flag when unassign completes * Add more tests * Add pre-existing missing check for unsubscribe as valid call after assignment lost * Style fix * Update change log

This reverts commit ad276a4.

* [DOCS-39462]-Add a Schema Registry producer example librdkafka has no built-in Schema Registry serializer, so schema-first production in C is done with libserdes plus avro-c. Add an example showing that path, and point to it from the README, so C/C++ readers are not left with raw-bytes producers as the only reference. libserdes and avro-c are not librdkafka dependencies, so the example is deliberately left out of $(EXAMPLES) and built on request with 'make producer_schema_registry', matching how producer_cpp and rdkafka_consume_batch are already handled. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> * [DOCS-39462] Fix style and subject derivation in SR producer example Reformat producer_schema_registry.c with clang-format 18 so it passes the repo style check, and derive the Schema Registry subject from the `topic` variable with snprintf() instead of hardcoding "myTopic-value", so the two cannot drift apart. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> * troubleshoot * [DOCS-39462] Note the silent-failure mode of serdes_conf_new serdes_conf_new() returns NULL if a configuration property is rejected. Passing that NULL to serdes_new() does not crash: serdes_new() treats a NULL conf as "use defaults" and initializes an empty configuration with no Schema Registry URL, so the mistake only surfaces later as an opaque serdes_schema_add() failure. The example also passes (NULL, 0) for errstr, which suppresses the explanation entirely. Rather than add error handling that would clutter the example's main line, call out both points in a comment so readers carrying this into production know to pass a real buffer and check the result. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> * Security upgrades ahead of 2.16.0 release (#5598) * Bump bundled OpenSSL to 3.5.8 and libcurl to 8.22.0 for source/autoconf builds * TEMP: drop tag gate on x64/arm64 release-artifact builds to verify dep bump in CI * Revert "TEMP: drop tag gate on x64/arm64 release-artifact builds to verify dep bump in CI" This reverts commit bd573d7. * Update Changelog * [FIX] Decommission learned brokers when a KIP-1102 re-bootstrap sequence starts (#5600) * Tests for KIP-899 and KIP-1102 re-bootstrap cases * Test that a KIP-1102 re-bootstrap actually reaches a bootstrap broker The existing KIP-1102 cases only count "Starting re-bootstrap sequence" log lines. That line is emitted at the top of `rd_kafka_rebootstrap_tmr_cb()`, before any broker is added or connected, so it can't distinguish a working re-bootstrap from an inert one that starts and then keeps querying the very brokers that reported the client's metadata as stale. It also can't be distinguished by address: the harness passes every mock broker in `bootstrap.servers` and all of them are in Metadata, so "asked a bootstrap broker" and "asked a learned broker" are the same host:port on the wire. Add a 6th broker to the `REBOOTSTRAP_REQUIRED` variations and remove it from Metadata responses with `rd_kafka_mock_broker_remove_from_metadata()`. It keeps listening and stays in `bootstrap.servers`, so it is only ever reachable through a bootstrap (`RD_KAFKA_CONFIGURED`) broker object, never as a learned one. A Metadata request arriving there is therefore proof that the sequence resulted in a bootstrap broker being used. Replica assignment already skips brokers that aren't in metadata and clamps the replication factor to the eligible count, so the expected learned broker set is unchanged. Only the error-code variations get the assertion. The ERR__TRANSPORT ones are left at 5 brokers and untouched: there the mock closes the connections, so the client reaches a bootstrap broker through the "all brokers down" path regardless, and the assertion wouldn't be exercising the KIP-1102 path. Request tracking is started at the first action rather than at configuration time, since `rd_kafka_mock_start_request_tracking()` clears the request list and this discards the initial bootstrap requests, which may legitimately have gone to the bootstrap-only broker. The count is sampled from the after-action callback because the harness destroys the mock cluster before returning to the test. Both error-code variations currently fail, reporting 70 re-bootstrap sequences and 0 Metadata requests to the bootstrap-only broker. That is the defect being pinned down, not a test bug: with learned brokers up, `rd_kafka_broker_any_usable()` keeps returning one of them (weight_usable() gives RD_KAFKA_LEARNED a +1000 bonus), so `rd_kafka_connect_any()` is never reached and the re-added bootstrap brokers are never connected. * Fix KIP-1102 re-bootstrap never reaching a bootstrap broker Decommission learned brokers when a re-bootstrap sequence starts so that only the bootstrap servers are used until a Metadata response rebuilds the broker list. Previously the client kept asking the very (still connected) learned brokers that had reported its metadata as stale, since rd_kafka_broker_any_usable() weighs learned brokers higher and always found one usable, so a bootstrap broker was never actually reached. Queued messages on decommissioned brokers are handed back to their partitions and re-sent once new leaders are known from the next Metadata response. Admin requests in flight on a decommissioned broker now fail with RD_KAFKA_RESP_ERR__TRANSPORT (previously RD_KAFKA_RESP_ERR__DESTROY_BROKER) so callers retry instead of treating it as a hard failure. * Fix spurious re-bootstrap and mock cgrp crash from broker decommission rd_kafka_broker_decommission() forced its broker down without marking the failure as planned, so decommissioning several still-up learned brokers at once (as re-bootstrap now does) could itself cross the "all brokers down" threshold and spuriously start a second re-bootstrap sequence. Use rd_kafka_broker_planned_fail() so a decommission is never counted toward ALL_BROKERS_DOWN. Separately, decommissioning the current group coordinator immediately marks it dead and triggers a JoinGroup/SyncGroup retry on a new connection, which can arrive at the mock broker before it has processed the old connection's closure. The mock's per-member response tracking only tolerated one live request per member, asserting on the retry; discard the stale, now-unreachable response instead. * Change the 0151 expectations according to the new policy of removing learned brokers when re-bootstrapping * Bound the 0151 re-bootstrap count by the connect interval, not a measured count do_test_down_then_up_no_rebootstrap_loop asserted 7-12 re-bootstrap sequences, which failed branch-wide on the arm64 local-quick CI jobs with 20 or 21. The count is not a property of the configuration, so no observed range is stable. The subtest's premise was wrong: the attempts are not paced by `reconnect.backoff.ms`. A connect scheduled through rd_kafka_connect_any() ends up in RD_KAFKA_OP_CONNECT, which zeroes rkb_ts_reconnect to expedite the reconnect, so rd_kafka_broker_ reconnect_backoff() returns 0 and the backoff never ramps. The rate instead follows how many subsystems ask for a cluster connection while every broker is down: the sparse-connection keepalive in rd_kafka_1s_tmr_cb(), its deferred one-shot retry, and - the piece that only shows up in CI - ProducerID acquisition, since run_local_quick passes -P and enables idempotence. That third caller takes the rate from ~1/s to ~3/s, giving the 20-21 seen on CI. It reproduces anywhere with 'TESTS=0151 SUBTESTS=down_then_up_no_rebootstrap_loop ./test-runner -P', which yields 9 or 21 roughly half the time each. Assert the one bound that is actually guaranteed: sparse_connect_intvl, max(11, min(reconnect.backoff.ms / 2, 1000)) and so 50ms by default, gates rd_kafka_connect_any(), allowing at most 6000 / 50 = 120 sequences in the 6s window. A loop not gated by connection attempts exceeds that, so the assertion still catches what the subtest is for. Also correct the comment, which described the old pacing theory. * Flaky tests 0118 fix and revert 0151 min count * Test 0151: await a request is sent to the bootstrap-only broker * Avoid adding back the coordinator as learned broker when the response comes from a broker that is being decommissioned * Changelog entry * Upgrade considerations * Report ALL_BROKERS_DOWN at most once per reconnect.backoff.max.ms instead of on every re-bootstrap cycle With metadata.recovery.strategy=rebootstrap, the re-bootstrap timer callback resets every broker's rkb_down_reported flag so that all brokers are retried before the "all brokers down" condition can be declared again. During a sustained outage every retry fails, the threshold is crossed again on each cycle and ALL_BROKERS_DOWN was delivered to the application roughly once per reconnect interval for the whole duration of the outage. Track the time of the last report in rk_last_all_brokers_down_reported_ts and report again only when more than reconnect.backoff.max.ms has passed since it, so a long outage still surfaces periodically but not on every cycle. Reset the timestamp when any broker connection comes up, so a new outage is reported immediately regardless of how recently the previous report was made. The timestamp is compared and updated with a CAS, since the last broker down can be observed concurrently from several broker threads. The re-bootstrap itself is unchanged and still runs on every cycle. Test 0151 do_test_down_then_up_no_rebootstrap_loop now takes the outage duration, polls the handle while the broker is down so errors reach the error callback, and asserts that a 5s and a 15s outage deliver exactly outage / reconnect.backoff.max.ms + 1 errors (1 and 2) over the many re-bootstrap sequences they produce. Fixes #5546 * Don't skip FindCoordinator or Metadata responses from bootstrap brokers that are decommissioning. Fixes failure in 0143_exponential_backoff_mock * Additional changelog entry * test 0151: use atomics to prevent thread sync issues * fix: Don't commit offsets of a lost assignment (#5585) * Don't commit offsets of a lost assignment A consumer with enable.auto.commit=true could send an OffsetCommit for an assignment it had already lost. On a client-side session timeout, or when max.poll.interval.ms is exceeded, the member id is reset to empty, the assignment is marked lost, and its partitions are revoked. The revoke-time auto-commit of those partitions was still sent because the assignment-lost flag was cleared inside rd_kafka_cgrp_unassign() and rd_kafka_cgrp_incremental_unassign(), before the removed partitions were served, defeating the guard that skips commits for a lost assignment. The commit left with an empty member id and the previous generation. A broker rejects it with UNKNOWN_MEMBER_ID, or, for a static member (group.instance.id set), with FENCED_INSTANCE_ID, which librdkafka raises as a fatal error that stops the consumer. Keep the assignment-lost flag set until the following rejoin applies a new assignment (assign()/incremental_assign() still clear it), so the offsets of a lost assignment are never committed. This restores the behavior documented in rd_kafka_cgrp_incremental_unassign()'s own comment. Add test 0192 (static member, cooperative-sticky, mock) which asserts that no OffsetCommit is sent after a session timeout. Happening since 1.6.0. * Clear assignment lost flag when unassign completes * Add more tests * Add pre-existing missing check for unsubscribe as valid call after assignment lost * Style fix * Update change log * Bump version to v2.16.0 (#5607) * [DOCS-39462] Use libschemaregistry instead of deprecated libserdes libserdes is deprecated; Confluent's supported C/C++ Schema Registry serde library is libschemaregistry. Replace the C example built on libserdes + avro-c with a C++ example built on libschemaregistry + avro-cpp, modeled on libschemaregistry's own AvroProducer example, and update the Makefile target, .gitignore and docs links accordingly. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> * Revert "Security upgrades ahead of 2.16.0 release (#5598)" This reverts commit b3af96a. * Revert "[FIX] Decommission learned brokers when a KIP-1102 re-bootstrap sequence starts (#5600)" This reverts commit 130c66c. * Revert "fix: Don't commit offsets of a lost assignment (#5585)" This reverts commit ad276a4. * Revert "Bump version to v2.16.0 (#5607)" This reverts commit 2b7f72e. --------- Co-authored-by: Claude Opus 5 <noreply@anthropic.com> Co-authored-by: Ojasva Jain <ojain@confluent.io> Co-authored-by: Emanuele Sabellico <esabellico@confluent.io> Co-authored-by: Kaushik Raina <103954755+k-raina@users.noreply.github.com>

Copilot AI lite review requested due to automatic review settings September 4, 2026 15:42

Copilot started reviewing on behalf of Kaushik Raina (k-raina) September 4, 2026 15:43

View session

Kaushik Raina (k-raina) added the in progress label Sep 4, 2026

Kaushik Raina (k-raina) marked this pull request as ready for review September 4, 2026 15:44

Kaushik Raina (k-raina) requested a review from a team as a code owner September 4, 2026 15:44

Copilot AI reviewed Sep 4, 2026

View reviewed changes

airlock-confluentinc Bot force-pushed the fix/cgrp-empty-member-id-commit-on-lost branch from 2deb2ec to 95d6ecf Compare September 4, 2026 15:49

Kaushik Raina (k-raina) changed the title ~~[Fix] Don't commit offsets of a lost assignment~~ fix: Don't commit offsets of a lost assignment Sep 4, 2026

airlock-confluentinc Bot force-pushed the fix/cgrp-empty-member-id-commit-on-lost branch from dfc064d to cf9891d Compare September 17, 2026 12:08

Emanuele Sabellico (emasab) approved these changes Sep 21, 2026

View reviewed changes

Kaushik Raina (k-raina) added 6 commits September 21, 2026 14:47

Clear assignment lost flag when unassign completes

0afb69d

Add more tests

b6764c5

Add pre-existing missing check for unsubscribe as valid call after as…

62b11a8

…signment lost

Style fix

bf3e165

Update change log

e352abc

airlock-confluentinc Bot force-pushed the fix/cgrp-empty-member-id-commit-on-lost branch from 97391bb to e352abc Compare September 21, 2026 09:33

Kaushik Raina (k-raina) enabled auto-merge (squash) September 21, 2026 12:04

Emanuele Sabellico (emasab) approved these changes Sep 21, 2026

View reviewed changes

Kaushik Raina (k-raina) merged commit edff8fc into master Sep 21, 2026
4 checks passed

Kaushik Raina (k-raina) deleted the fix/cgrp-empty-member-id-commit-on-lost branch September 21, 2026 12:50

airlock-confluentinc Bot pushed a commit that referenced this pull request Sep 21, 2026

Revert "fix: Don't commit offsets of a lost assignment (#5585)"

a5b71ff

This reverts commit ad276a4.

Navigation Menu

Repository navigation

fix: Don't commit offsets of a lost assignment - #5585

fix: Don't commit offsets of a lost assignment#5585
Kaushik Raina (k-raina) merged 6 commits into
masterfrom
fix/cgrp-empty-member-id-commit-on-lost

Kaushik Raina (k-raina) commented Sep 4, 2026 •

edited

Loading

Uh oh!

confluent-cla-assistant Bot commented Sep 4, 2026

Uh oh!

Copilot AI left a comment

Uh oh!

Emanuele Sabellico (emasab) left a comment

Uh oh!

Uh oh!

Reviewers

Uh oh!

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

Conversation

Kaushik Raina (k-raina) commented Sep 4, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

confluent-cla-assistant Bot commented Sep 4, 2026

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

🟡 Changes recommended

Uh oh!

Emanuele Sabellico (emasab) left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Reviewers

Uh oh!

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

Kaushik Raina (k-raina) commented Sep 4, 2026 •

edited

Loading