KIP-932: Implement destroy APIs by ojasvajain · Pull Request #5406 · confluentinc/librdkafka

Ojasva Jain (ojasvajain) · 2026-04-16T05:24:01Z

This PR implements the following APIs related to share consumers:

void rd_kafka_share_destroy(rd_kafka_share_t *rkshare);
void rd_kafka_share_destroy_flags(rd_kafka_share_t *rkshare, int flags);

Note, handling of _DESTROY* in acknowledgement callback will be covered in a separate PR

Pending:

Handing of _DESTROY* in commit_sync
Broker decomission test cases
More test cases related to destroy API
Fix "Build configuration check" job getting stuck in destroy
Add missing test cases from this list
Add fatal error handling in close API
Evaluate RD_KAFKA_RESP_ERR__DESTROY check in rd_kafka_share_fetch_reply_op

Test Cases Added (so far):

  ┌─────┬──────────────────────────────────────────────────┬──────┬────────────────────────────────────────────────────────────────────────────────────────────────────────┐  
  │  #  │             Test (subtest function)              │ Type │                                              Description                                               │
  ├─────┼──────────────────────────────────────────────────┼──────┼────────────────────────────────────────────────────────────────────────────────────────────────────────┤  
  │     │                                                  │      │ Acks across commit_async + commit_sync with 5s ShareAck RTT delays, then destroys. Verifies destroy    │
  │ 1   │ test_destroy_with_cached_acks_and_delayed_broker │ mock │ completes within bounded time and that commit_sync returns REQUEST_TIMED_OUT for the still-pending     │  
  │     │                                                  │      │ partition.                                                                                             │  
  ├─────┼──────────────────────────────────────────────────┼──────┼────────────────────────────────────────────────────────────────────────────────────────────────────────┤
  │ 2   │ test_broker_decommission_with_commit_sync        │ mock │ Decommissions a broker mid-commit_sync (background thread moves leader + removes broker from           │  
  │     │                                                  │      │ metadata). Asserts per-partition err = __DESTROY_BROKER for the orphaned partition.                    │  
  ├─────┼──────────────────────────────────────────────────┼──────┼────────────────────────────────────────────────────────────────────────────────────────────────────────┤
  │ 3   │ test_broker_decommission_with_consume_batch      │ mock │ Decommissions a broker mid-consume_batch. Verifies the in-flight ShareFetch fails locally with         │  
  │     │                                                  │      │ __DESTROY_BROKER and the consumer keeps polling the surviving broker.                                  │  
  ├─────┼──────────────────────────────────────────────────┼──────┼────────────────────────────────────────────────────────────────────────────────────────────────────────┤
  │ 4   │ test_broker_decommission_during_close            │ mock │ Decommissions a broker mid-close. Verifies close completes despite the in-flight leave-ShareAck being  │  
  │     │                                                  │      │ killed.                                                                                                │  
  ├─────┼──────────────────────────────────────────────────┼──────┼────────────────────────────────────────────────────────────────────────────────────────────────────────┤
  │ 5   │ test_broker_decommission_with_commit_async       │ mock │ Decommissions a broker mid-commit_async. Verifies the async commit returns promptly with NULL error    │  
  │     │                                                  │      │ and the consumer survives.                                                                             │  
  ├─────┼──────────────────────────────────────────────────┼──────┼────────────────────────────────────────────────────────────────────────────────────────────────────────┤
  │ 6   │ test_destroy_during_rebalance                    │ mock │ Drives the cgrp into wait-incr-unassign-to-complete (heartbeat blocked by injected 10s RTT), then      │  
  │     │                                                  │      │ destroys. NO_CONSUMER_CLOSE must return in <2s; full close must complete in <15s.                      │  
  ├─────┼──────────────────────────────────────────────────┼──────┼────────────────────────────────────────────────────────────────────────────────────────────────────────┤
  │ 7   │ test_destroy_with_fatal_error                    │ mock │ Acks half, injects a 30s ShareAck delay, triggers fatal error, then destroys. Verifies the fatal-error │  
  │     │                                                  │      │  path (auto-promoted to NO_CONSUMER_CLOSE) returns in <2s regardless of caller flags.                  │  
  ├─────┼──────────────────────────────────────────────────┼──────┼────────────────────────────────────────────────────────────────────────────────────────────────────────┤
  │ 8   │ do_test_destroy_with_explicit_ack                │ real │ Consumes a batch with explicit-ack mode, acks all or half (parameterised), then destroys without       │  
  │     │                                                  │      │ commit. Verifies destroy handles pending acks correctly.                                               │  
  ├─────┼──────────────────────────────────────────────────┼──────┼────────────────────────────────────────────────────────────────────────────────────────────────────────┤
  │ 9   │ do_test_destroy_with_implicit_ack                │ real │ Consumes a batch with implicit-ack mode, never makes a follow-up poll (so implicit ack never fires),   │  
  │     │                                                  │      │ then destroys. Verifies destroy handles un-acked records gracefully.                                   │  
  ├─────┼──────────────────────────────────────────────────┼──────┼────────────────────────────────────────────────────────────────────────────────────────────────────────┤
  │ 10  │ do_test_destroy_with_subscribe_unsubscribe       │ real │ Creates a share consumer with various subscribe/unsubscribe combinations and destroys. Sanity check    │  
  │     │                                                  │      │ for state transitions through subscribe/unsubscribe.                                                   │  
  └─────┴──────────────────────────────────────────────────┴──────┴────────────────────────────────────────────────────────────────────────────────────────────────────────┘
                                                                                                                                                                              
  Notes:          
  - Each test runs with both destroy_flags=0 and RD_KAFKA_DESTROY_F_NO_CONSUMER_CLOSE variants.
  - Tests 2–5 each also run with both explicit_ack=true/false variants (except _with_consume_batch which is implicit only).                                                   
  - Test 8 also runs with both ack_half=true/false variants.                                                               
  - Mock tests live in main_0179_share_consumer_destroy_local; real-broker tests live in main_0179_share_consumer_destroy.

confluent-cla-assistant · 2026-04-16T05:24:13Z

🎉 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.}

1. src/rdkafka_broker.c: Add rd_kafka_broker_share_fetch_session_clear() to broker thread exit path — prevents assertion failure in rd_kafka_broker_destroy_final when toppars_to_forget is non-NULL after a broker is decommissioned during close. 2. tests/0178-share_consumer_close.c: - Uncomment test_close_with_broker_down_is_fatal_cb - Add delayed_down_state struct + delayed_down_cb background thread - Add test_close_with_broker_decommission: validates close() completes when a broker goes down mid-close - Register test in main_0178_share_consumer_close_local Note: test passes on both branches (with/without rko_op_cb fix) because set_down triggers __STATE (not __DESTROY) — decommission is too slow to race ahead of the leave op processing. Needs rd_kafka_mock_broker_remove_from_metadata (PR #5406) to decouple metadata removal from connection drop for a deterministic test.

Pranav Rathi (pranavrth)

Added comments for the implementation part. I will go through the tests in another pass.

Pranav Rathi (pranavrth) · 2026-05-12T13:03:06Z

         * Step 1: Dispatch acknowledgement callbacks.
         * Per-partition errors were set by the broker thread.
         */
        rd_kafka_share_dispatch_ack_callbacks(


Just to remember - Let's add ack callback tests as well.

Pranav Rathi (pranavrth) · 2026-05-12T14:02:20Z

 */
 rd_kafka_resp_err_t
 rd_kafka_share_consumer_closed_err(rd_kafka_share_t *rkshare);
-


…n fixes

Ojasva Jain (ojasvajain) · 2026-05-14T07:17:01Z

-        test_ack_cb_state_t state;
+        int consumed              = 0;
+        int attempts              = 0;
+        test_ack_cb_state_t state = {0};


This issue was causing the tests fail on the feature branch. It should be fixed now.

Ojasva Jain (ojasvajain) · 2026-05-14T07:17:08Z

-        ack_cb_state_t state;
+        int consumed         = 0;
+        int attempts         = 0;
+        ack_cb_state_t state = {0};


Same as above

Ojasva Jain (ojasvajain) · 2026-05-14T07:18:20Z

        while ((rcvd < 4 || state.callback_cnt < 1) && attempts-- > 0) {
                size_t batch_rcvd = 0;
-                err = rd_kafka_share_consume_batch(rkshare, 2000, batch,
+                err = rd_kafka_share_consume_batch(rkshare, 2000, batch + rcvd,


If msgs don't get redelivered msgs in one consume_batch call, some msgs might leak. Observed this issue in one of the runs.

Pranav Rathi (pranavrth) · 2026-05-15T11:37:00Z

        }
 }
+
+void rd_kafka_share_acks_clear_during_broker_decommission(


Add test case for this flow.

Pranav Rathi (pranavrth) · 2026-05-15T11:42:18Z

+         * apply the result to the in-flight commit_sync request (copies
+         * err into the cgrp's result list, decrements awaiting count,
+         * completes the sync if this was the last broker outstanding). */
+        if (rkb->rkb_pending_commit_sync.sync_ack_details) {


We need to dispatch callbacks as well.

Pranav Rathi (pranavrth) · 2026-05-15T11:43:12Z

        }
 }
+
+void rd_kafka_share_acks_clear_during_broker_decommission(


add broker id as well in the logs.

Pranav Rathi (pranavrth) · 2026-05-15T11:53:10Z

                 * (buf callback init -> INVALID_RECORD_STATE -> per-
                 * partition err or top-level err). */
-                if (ack_details) {
+                if (ack_details && rko_orig->rko_err) {


Move it to before dispatch ack_details (step 1) part.

Add test as well for this.

Pranav Rathi (pranavrth) · 2026-05-15T11:56:21Z

+                rd_kafka_share_commit_sync_apply_result(rk, rkcg, ack_details);
                rd_kafka_dbg(
                    rk, CGRP, "SHARE",
                    "Commit sync reply from broker %s: %s, "


We can change wording for this.

Pranav Rathi (pranavrth) · 2026-05-15T12:30:12Z

+        rd_atomic32_add(&rkb->termination_in_progress, 1);
+
        if (RD_KAFKA_IS_SHARE_CONSUMER(rk) &&
            rkb->rkb_source == RD_KAFKA_LEARNED) {


Let's verify this is correct.

Pranav Rathi (pranavrth) · 2026-05-15T12:37:54Z

                                     rd_bool_t should_fetch,
                                     rd_bool_t should_leave);

+void rd_kafka_share_commit_sync_apply_result(rd_kafka_t *rk,


Check if this should be moved to any other file?

Pranav Rathi (pranavrth) · 2026-05-15T12:53:32Z

+        int receipt_cnt;
+        int receipt_capacity;
+        int callback_invocations;
+        mtx_t lock;


Check if it is needed.

Pranav Rathi (pranavrth) · 2026-05-15T13:22:03Z

+                }
+                attempts++;
+        }
+        TEST_ASSERT(rcvd >= TEST_MSGS,


Pranav Rathi (pranavrth) · 2026-05-15T13:27:35Z

+         *
+         *   surviving_partition: TEST_MSGS/2 offsets, NO_ERROR
+         *   target_partition:    TEST_MSGS/2 offsets, __DESTROY_BROKER */
+        {


Ojasva Jain (ojasvajain) added the share consumer label Apr 16, 2026

Ojasva Jain (ojasvajain) marked this pull request as ready for review April 16, 2026 19:29

Ojasva Jain (ojasvajain) requested a review from a team as a code owner April 16, 2026 19:29

airlock-confluentinc Bot force-pushed the dev_kip-932_close_only branch from 214b852 to 70c29ba Compare April 18, 2026 19:26

airlock-confluentinc Bot force-pushed the dev_kip-932_destroy branch from 60cafef to db6b4d8 Compare April 20, 2026 09:50

Pranav Rathi (pranavrth) mentioned this pull request Apr 20, 2026

[KIP-932] : Implement Share consumer interface with poll API confluentinc/confluent-kafka-python#2217

Merged

airlock-confluentinc Bot force-pushed the dev_kip-932_destroy branch from db6b4d8 to 5cc3e10 Compare April 20, 2026 14:29

airlock-confluentinc Bot force-pushed the dev_kip-932_close_only branch 2 times, most recently from 97d35a7 to 105eb74 Compare April 26, 2026 18:17

airlock-confluentinc Bot force-pushed the dev_kip-932_destroy branch from 024e756 to bf9fb74 Compare April 27, 2026 09:32

airlock-confluentinc Bot force-pushed the dev_kip-932_close_only branch 5 times, most recently from a32c996 to 7b88cec Compare May 4, 2026 14:03

airlock-confluentinc Bot force-pushed the dev_kip-932_destroy branch from 0b3197d to 9059cbc Compare May 5, 2026 04:27

Base automatically changed from dev_kip-932_close_only to dev_kip-932_queues-for-kafka May 5, 2026 05:45

airlock-confluentinc Bot force-pushed the dev_kip-932_destroy branch from dff116d to 716df74 Compare May 8, 2026 06:57

Ojasva Jain (ojasvajain) changed the base branch from dev_kip-932_queues-for-kafka to dev_kip-932_acknowledgement_callback May 8, 2026 06:57

airlock-confluentinc Bot force-pushed the dev_kip-932_destroy branch from 716df74 to fdbac55 Compare May 8, 2026 10:02

airlock-confluentinc Bot force-pushed the dev_kip-932_acknowledgement_callback branch from 76eb457 to daa5c73 Compare May 8, 2026 15:46

airlock-confluentinc Bot force-pushed the dev_kip-932_destroy branch from e26cdc3 to b3426df Compare May 9, 2026 10:38

Base automatically changed from dev_kip-932_acknowledgement_callback to dev_kip-932_queues-for-kafka May 11, 2026 15:08

airlock-confluentinc Bot force-pushed the dev_kip-932_destroy branch from 92f947f to 59453f2 Compare May 12, 2026 05:43

Pranav Rathi (pranavrth) requested changes May 12, 2026

View reviewed changes

Ojasva Jain (ojasvajain) added 2 commits May 13, 2026 15:02

KIP-932: destroy apis + broker decommission + tests + leader-migratio…

fbf90d1

…n fixes

address comments

36bea0e

airlock-confluentinc Bot force-pushed the dev_kip-932_destroy branch from 59453f2 to 36bea0e Compare May 13, 2026 17:51

Ojasva Jain (ojasvajain) added 9 commits May 13, 2026 23:23

fix yaml formatting, remove extra file

f470089

address nits

a09ac30

fix comment in shell script

f344de9

initialize ack_cb_state with clean mem

44856c7

fix style

7843d28

initialize ack_cb_state with {0} in 0173

564b9f5

style fixes and increase timeout for commit_sync in 0179

8b392a3

minor fixes related to possible leaking of rkmessages

992ed9a

increase socket timeout and broker delay in 0179

f807a12

Ojasva Jain (ojasvajain) commented May 14, 2026

View reviewed changes

Ojasva Jain (ojasvajain) added 5 commits May 14, 2026 13:30

tune timeouts in 0179

1b7c6aa

fix decommission delay in 0179

b14d8ba

add background cpu/mem tracking

a285f37

remove mem/cpu debugging, enable debugging in 0179

bf0c223

slight impl fixes + ack cb assertions in test cases

975c747

Pranav Rathi (pranavrth) requested changes May 15, 2026

View reviewed changes

Conversation

Ojasva Jain (ojasvajain) commented Apr 16, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

confluent-cla-assistant Bot commented Apr 16, 2026

Uh oh!

Pranav Rathi (pranavrth) left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

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!

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!

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!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

Ojasva Jain (ojasvajain) commented Apr 16, 2026 •

edited

Loading