[KIP-848] Fixed not adhering to HB interval when auto commit interval was less than HB interval (#5114)

Author: pranavrth

src/rdkafka_cgrp.c

@@ -6399,43 +6399,49 @@ static void rd_kafka_cgrp_consumer_assignment_done(rd_kafka_cgrp_t *rkcg) {
                 rd_kafka_cgrp_consumer_incr_unassign_done(rkcg);
                 break;
 
-        case RD_KAFKA_CGRP_JOIN_STATE_STEADY:
-                rd_kafka_cgrp_consumer_expedite_next_heartbeat(
-                    rkcg, "back to steady state");
-
-                if (rkcg->rkcg_rebalance_rejoin) {
-                        rkcg->rkcg_rebalance_rejoin = rd_false;
-                        rd_kafka_cgrp_rejoin(
-                            rkcg,
-                            "rejoining group to redistribute "
-                            "previously owned partitions to other "
-                            "group members");
-                        break;
-                }
-
-                /* FALLTHRU */
+        case RD_KAFKA_CGRP_JOIN_STATE_INIT:
+        case RD_KAFKA_CGRP_JOIN_STATE_STEADY: {
 
-        case RD_KAFKA_CGRP_JOIN_STATE_INIT: {
-                rd_bool_t still_in_group = rd_true;
+                rd_bool_t not_in_group = rd_false;
                 /*
                  * There maybe a case when there are no assignments are
                  * assigned to this consumer. In this case, while terminating
                  * the consumer can be in STEADY or INIT state and won't go
                  * to intermediate state. In this scenario, last leave call is
                  * done from here.
                  */
-                still_in_group &= !rd_kafka_cgrp_leave_maybe(rkcg);
+                not_in_group |= rd_kafka_cgrp_leave_maybe(rkcg);
 
                 /* Check if cgrp is trying to terminate, which is safe to do
                  * in these two states. Otherwise we'll need to wait for
                  * the current state to decommission. */
-                still_in_group &= !rd_kafka_cgrp_try_terminate(rkcg);
+                not_in_group |= rd_kafka_cgrp_try_terminate(rkcg);
+
+                if (not_in_group)
+                        break;
 
-                if (still_in_group)
+                if (rkcg->rkcg_join_state == RD_KAFKA_CGRP_JOIN_STATE_INIT) {
                         rd_kafka_cgrp_consumer_expedite_next_heartbeat(
-                            rkcg, "back to init state");
+                            rkcg, "Assignment Done: in init state");
+                } else if (rkcg->rkcg_rebalance_rejoin) {
+                        /* No need to expedite the HB here as it's being
+                         * expedited in the rejoin call.*/
+                        rkcg->rkcg_rebalance_rejoin = rd_false;
+                        rd_kafka_cgrp_rejoin(
+                            rkcg,
+                            "Assignment Done: rejoining group to redistribute "
+                            "previously owned partitions to other "
+                            "group members");
+                } else if (rkcg->rkcg_consumer_flags &
+                           RD_KAFKA_CGRP_CONSUMER_F_WAIT_ACK) {
+                        rd_kafka_cgrp_consumer_expedite_next_heartbeat(
+                            rkcg,
+                            "Assignment Done: in steady state, waiting for "
+                            "ack");
+                }
                 break;
         }
+
         default:
                 break;
         }

tests/0147-consumer_group_consumer_mock.c

@@ -805,6 +805,59 @@ static void do_test_metadata_unknown_topic_id_tests(void) {
         }
 }
 
+static void do_test_adherence_to_hb_interval(void) {
+        rd_kafka_mock_cluster_t *mcluster;
+        const char *bootstraps;
+        rd_kafka_topic_partition_list_t *subscription;
+        rd_kafka_t *c;
+        const char *topic =
+            test_mk_topic_name("do_test_adherence_to_hb_interval", 1);
+        rd_kafka_conf_t *conf;
+        size_t heartbeat_request_count = 0;
+
+        SUB_TEST_QUICK("do_test_adherence_to_hb_interval");
+
+        mcluster = test_mock_cluster_new(1, &bootstraps);
+        rd_kafka_mock_set_group_consumer_heartbeat_interval_ms(mcluster, 1000);
+        rd_kafka_mock_topic_create(mcluster, topic, 3, 1);
+
+        test_conf_init(&conf, NULL, 0);
+        test_conf_set(conf, "bootstrap.servers", bootstraps);
+        test_conf_set(conf, "auto.offset.reset", "earliest");
+        test_conf_set(conf, "auto.commit.interval.ms", "100");
+        c = test_create_consumer(topic, NULL, conf, NULL);
+
+        subscription = rd_kafka_topic_partition_list_new(1);
+        rd_kafka_topic_partition_list_add(subscription, topic,
+                                          RD_KAFKA_PARTITION_UA);
+
+        rd_kafka_mock_start_request_tracking(mcluster);
+        TEST_SAY("Subscribing to topic\n");
+        TEST_CALL_ERR__(rd_kafka_subscribe(c, subscription));
+        rd_kafka_topic_partition_list_destroy(subscription);
+
+        TEST_SAY("Subscription done, waiting for heartbeats\n");
+
+        rd_sleep(2); /* Sleep to ensure that some HB are sent */
+
+        heartbeat_request_count = test_mock_get_matching_request_cnt(
+            mcluster, is_heartbeat_request, NULL);
+        TEST_SAY("Heartbeat request count: %zu\n", heartbeat_request_count);
+
+        /* Assert that we received the expected number of heartbeats */
+        TEST_ASSERT(heartbeat_request_count >= 3 &&
+                        heartbeat_request_count <= 5,
+                    "Expected between 3 and 5 heartbeats, got %zu",
+                    heartbeat_request_count);
+
+        rd_kafka_mock_stop_request_tracking(mcluster);
+
+        rd_kafka_destroy(c);
+        test_mock_cluster_destroy(mcluster);
+
+        SUB_TEST_PASS();
+}
+
 int main_0147_consumer_group_consumer_mock(int argc, char **argv) {
         TEST_SKIP_MOCK_CLUSTER(0);
 
@@ -821,5 +874,7 @@ int main_0147_consumer_group_consumer_mock(int argc, char **argv) {
 
         do_test_metadata_unknown_topic_id_tests();
 
+        do_test_adherence_to_hb_interval();
+
         return 0;
 }

tests/test.c

@@ -7423,7 +7423,7 @@ rd_kafka_mock_cluster_t *test_mock_cluster_new(int broker_cnt,
  *        received by mock cluster \p mcluster, matching
  *        function \p match , called with opaque \p opaque .
  */
-static size_t test_mock_get_matching_request_cnt(
+size_t test_mock_get_matching_request_cnt(
     rd_kafka_mock_cluster_t *mcluster,
     rd_bool_t (*match)(rd_kafka_mock_request_t *request, void *opaque),
     void *opaque) {

tests/test.h

@@ -877,6 +877,12 @@ rd_kafka_resp_err_t test_delete_all_test_topics(int timeout_ms);
 void test_mock_cluster_destroy(rd_kafka_mock_cluster_t *mcluster);
 rd_kafka_mock_cluster_t *test_mock_cluster_new(int broker_cnt,
                                                const char **bootstraps);
+
+size_t test_mock_get_matching_request_cnt(
+    rd_kafka_mock_cluster_t *mcluster,
+    rd_bool_t (*match)(rd_kafka_mock_request_t *request, void *opaque),
+    void *opaque);
+
 size_t test_mock_wait_matching_requests(
     rd_kafka_mock_cluster_t *mcluster,
     size_t num,