rd_kafka_assign() hangs forever when closing consumer

[aiquestion]

Description

we enconter a issue when closing the consumer in Rust SDK, it hangs calling rd_kafka_assign() in rebalance_cb.
code is like (using C code)

rd_kafka_conf_set_events(conf, RD_KAFKA_EVENT_REBALANCE|RD_KAFKA_EVENT_OFFSET_COMMIT|RD_KAFKA_EVENT_ERROR|RD_KAFKA_EVENT_OAUTHBEARER_TOKEN_REFRESH);
rd_kafka_poll_set_consumer(rk);
rd_kafka_queue_t * queue = rd_kafka_queue_get_consumer(rk);
// should poll here, but we don't do poll to reproduce the bug

// close consumer
rd_kafka_consumer_close_queue(rk, queue);
while (!rd_kafka_consumer_closed(rk)) {
                rd_kafka_event_t *rkev;
                rkev = rd_kafka_queue_poll(queue, 60 * 1000);
                fprintf(stderr, "%% Get event %s \n", rd_kafka_event_name(rkev));
                if (rkev) {
                        if (rd_kafka_event_type(rkev) == RD_KAFKA_EVENT_REBALANCE) {
                                rebalance_cb_event(rk, rkev);
                        }
                }
}

static void rebalance_cb_event(rd_kafka_t *rk,
                               rd_kafka_event_t* rkev) {
        rd_kafka_error_t *error     = NULL;
        rd_kafka_resp_err_t ret_err = RD_KAFKA_RESP_ERR_NO_ERROR;
        switch (rd_kafka_event_error(rkev)) {
        case RD_KAFKA_RESP_ERR__ASSIGN_PARTITIONS:
                print_partition_list(stderr, rd_kafka_topic_partition_list_new(0));
                if (!strcmp(rd_kafka_rebalance_protocol(rk), "COOPERATIVE")) {
                        error = rd_kafka_incremental_assign(
                            rk, rd_kafka_topic_partition_list_new(0));
                }
                else {
                        ret_err = rd_kafka_assign(
                            rk, rd_kafka_topic_partition_list_new(0));
                        printf(stderr, "assign ret: %s\n",
                               rd_kafka_err2str(ret_err));
                        sleep(3);
                }
                break;
        case RD_KAFKA_RESP_ERR__REVOKE_PARTITIONS:
                print_partition_list(stderr, rd_kafka_topic_partition_list_new(0));
                if (!strcmp(rd_kafka_rebalance_protocol(rk), "COOPERATIVE")) {
                        error = rd_kafka_incremental_unassign(rk, rd_kafka_topic_partition_list_new(0));
                } else {
                        ret_err  = rd_kafka_assign(rk, NULL);
                        wait_eof = 0;
                }
                break;
        default:
                rd_kafka_assign(rk, NULL);
                break;
        }
        if (error) {
                fprintf(stderr, "incremental assign failure: %s\n",
                        rd_kafka_error_string(error));
                rd_kafka_error_destroy(error);
        } else if (ret_err) {
                fprintf(stderr, "assign failure: %s\n",
                        rd_kafka_err2str(ret_err));
        }
}

In some cases the program will never exist, and the call stack hangs in rd_kafka_assign

  * frame #0: 0x00000001916019ec libsystem_kernel.dylib`__psynch_cvwait + 8
    frame #1: 0x000000019163f55c libsystem_pthread.dylib`_pthread_cond_wait + 1228
    frame #2: 0x00000001003ee258 rdkafka_complex_consumer_example`cnd_wait + 12
    frame #3: 0x0000000100397c14 rdkafka_complex_consumer_example`rd_kafka_q_pop_serve + 620
    frame #4: 0x000000010039a764 rdkafka_complex_consumer_example`rd_kafka_op_req + 148
    frame #5: 0x00000001003ce394 rdkafka_complex_consumer_example`rd_kafka_assign + 80
    frame #6: 0x00000001003686f0 rdkafka_complex_consumer_example`main [inlined] rebalance_cb_event(rk=0x000000014e010a00, rkev=<unavailable>) at rdkafka_complex_consumer_example.c:175:35 [opt]
    frame #7: 0x0000000100368634 rdkafka_complex_consumer_example`main(argc=<unavailable>, argv=<unavailable>) at rdkafka_complex_consumer_example.c:619:33 [opt]

After investigation, it seems that:

• consumer start and join/sync group success.
• tigger a rebalance event, since user didn't poll, it will not be handled
• rd_kafka_consumer_close_queue to trigger a RD_KAFKA_OP_TERMINATE in cgrp
• when closing rd_kafka_queue_poll will be called, and rebalance event handled
• RD_KAFKA_OP_ASSIGN send into cgrp_op queue
• RD_KAFKA_OP_TERMINATE handled, and exit the rd_kafka_q_serve
• rd_kafka_cgrp_serve called rd_kafka_q_purge(rkcg->rkcg_ops); to purge the RD_KAFKA_OP_ASSIGN event
• main thread never get reply from RD_KAFKA_OP_ASSIGN, so it hangs forever

// in rdkafka.c thread_main:
                int cnt = rd_kafka_q_serve(rk->rk_ops, timeout_ms, 0,
                                 RD_KAFKA_Q_CB_CALLBACK, NULL, NULL);
                if (rk->rk_cgrp) /* FIXME: move to timer-triggered */
                        rd_kafka_cgrp_serve(rk->rk_cgrp);

a proper fix is to handle all rkcg->rkcg_ops before rd_kafka_q_purge(rkcg->rkcg_ops);
PR

How to reproduce

It's a random case, i managed to re-produce it after i modify many code:

• change heart beat response to always error ( so we are not in the group)
• change rd_kafka_q_serve in main thread to handle 1 op at a time
• disable rd_kafka_cgrp_serve before RD_KAFKA_OP_GET_REBALANCE_PROTOCOL called so RD_KAFKA_OP_ASSIGN can be in queue first and handle after rd_kafka_cgrp_serve called.
  aiquestion@1bfc1e4

build rdkafka_complex_consumer_example.c and run it. it will hang. after apply the fix in PR, it can close successfully.

IMPORTANT: Always try to reproduce the issue on the latest released version (see https://github.com/confluentinc/librdkafka/releases), if it can't be reproduced on the latest version the issue has been fixed.

Checklist

IMPORTANT: We will close issues where the checklist has not been completed.

Please provide the following information:

• librdkafka version (release number or git tag): <REPLACE with e.g., v0.10.5 or a git sha. NOT "latest" or "current">
• Apache Kafka version: <REPLACE with e.g., 0.10.2.3>
• librdkafka client configuration: <REPLACE with e.g., message.timeout.ms=123, auto.reset.offset=earliest, ..>
• Operating system: <REPLACE with e.g., Centos 5 (x64)>
• Provide logs (with debug=.. as necessary) from librdkafka
• Provide broker log excerpts
• Critical issue