ShareGroupHeartBeat API #5299
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -2993,6 +2993,229 @@ rd_kafka_mock_handle_ConsumerGroupHeartbeat(rd_kafka_mock_connection_t *mconn, | |
| return -1; | ||
| } | ||
|
|
||
| /** | ||
| * @brief Helper to write assignment TopicPartitions to ShareGroupHeartbeat | ||
| * response. | ||
| */ | ||
| static void rd_kafka_mock_handle_ShareGroupHeartbeat_write_TopicPartitions( | ||
| rd_kafka_buf_t *resp, | ||
| rd_kafka_topic_partition_list_t *assignment) { | ||
| const rd_kafka_topic_partition_field_t fields[] = { | ||
| RD_KAFKA_TOPIC_PARTITION_FIELD_PARTITION, | ||
| RD_KAFKA_TOPIC_PARTITION_FIELD_END}; | ||
|
|
||
| rd_kafka_topic_partition_list_sort_by_topic_id(assignment); | ||
| rd_kafka_buf_write_topic_partitions( | ||
| resp, assignment, rd_false /* don't skip invalid offsets */, | ||
| rd_false /* any offset */, rd_true /* use topic id */, | ||
| rd_false /* don't use topic name */, fields); | ||
| } | ||
|
|
||
| /** | ||
| * @brief Handle ShareGroupHeartbeat request (API Key 76). | ||
| */ | ||
| static int | ||
| rd_kafka_mock_handle_ShareGroupHeartbeat(rd_kafka_mock_connection_t *mconn, | ||
| rd_kafka_buf_t *rkbuf) { | ||
| const rd_bool_t log_decode_errors = rd_true; | ||
| rd_kafka_mock_cluster_t *mcluster = mconn->broker->cluster; | ||
| rd_kafka_buf_t *resp; | ||
| rd_kafkap_str_t GroupId, MemberId, RackId; | ||
| rd_kafkap_str_t *SubscribedTopicNames = NULL; | ||
| int32_t MemberEpoch = 0, SubscribedTopicNamesCnt; | ||
| int32_t i; | ||
| rd_kafka_resp_err_t err = RD_KAFKA_RESP_ERR_NO_ERROR; | ||
| rd_kafka_mock_sharegroup_t *mshgrp = NULL; | ||
| rd_kafka_mock_sharegroup_member_t *member = NULL; | ||
| rd_bool_t assignment_changed = rd_false; | ||
|
|
||
| resp = rd_kafka_mock_buf_new_response(rkbuf); | ||
|
|
||
| /* Inject Error */ | ||
| err = rd_kafka_mock_next_request_error(mconn, resp); | ||
| if (err) | ||
| goto build_response; | ||
|
|
||
| /* GroupId */ | ||
| rd_kafka_buf_read_str(rkbuf, &GroupId); | ||
|
|
||
| /* Coordinator check */ | ||
| { | ||
| rd_kafka_mock_broker_t *mrkb; | ||
|
|
||
| mrkb = rd_kafka_mock_cluster_get_coord( | ||
| mcluster, RD_KAFKA_COORD_GROUP, &GroupId); | ||
|
|
||
| if (!mrkb) | ||
| err = RD_KAFKA_RESP_ERR_COORDINATOR_NOT_AVAILABLE; | ||
| else if (mrkb != mconn->broker) | ||
| err = RD_KAFKA_RESP_ERR_NOT_COORDINATOR; | ||
| } | ||
|
|
||
| if (err) | ||
| goto build_response; | ||
|
|
||
| /* MemberId */ | ||
| rd_kafka_buf_read_str(rkbuf, &MemberId); | ||
|
|
||
| /* MemberEpoch */ | ||
| rd_kafka_buf_read_i32(rkbuf, &MemberEpoch); | ||
|
|
||
| /* RackId (nullable) */ | ||
| rd_kafka_buf_read_str(rkbuf, &RackId); | ||
|
|
||
| /* SubscribedTopicNames array (nullable) */ | ||
| rd_kafka_buf_read_arraycnt(rkbuf, &SubscribedTopicNamesCnt, | ||
| RD_KAFKAP_TOPICS_MAX); | ||
| if (SubscribedTopicNamesCnt >= 0) { | ||
| SubscribedTopicNames = rd_calloc( | ||
| SubscribedTopicNamesCnt > 0 ? SubscribedTopicNamesCnt : 1, | ||
| sizeof(rd_kafkap_str_t)); | ||
| for (i = 0; i < SubscribedTopicNamesCnt; i++) { | ||
| rd_kafka_buf_read_str(rkbuf, &SubscribedTopicNames[i]); | ||
| } | ||
| } | ||
|
|
||
| { | ||
| mtx_lock(&mcluster->lock); | ||
|
|
||
| mshgrp = rd_kafka_mock_sharegroup_get(mcluster, &GroupId); | ||
|
|
||
| if (MemberEpoch == -1) { | ||
| /* LEAVE: Member wants to leave */ | ||
| member = rd_kafka_mock_sharegroup_member_find( | ||
| mshgrp, &MemberId); | ||
| if (member) { | ||
| rd_kafka_mock_sharegroup_member_destroy(mshgrp, | ||
| member); | ||
| member = NULL; | ||
| assignment_changed = rd_true; | ||
| } | ||
|
|
||
| } else if (MemberEpoch == 0) { | ||
| /* JOIN: New member wants to join */ | ||
| member = rd_kafka_mock_sharegroup_member_get( | ||
| mshgrp, &MemberId, MemberEpoch, mconn); | ||
|
|
||
| if (member) { | ||
| if (rd_kafka_mock_sharegroup_member_subscribed_topic_names_set( | ||
| member, SubscribedTopicNames, | ||
| SubscribedTopicNamesCnt)) { | ||
| assignment_changed = rd_true; | ||
| } else { | ||
| /* New member always triggers | ||
| * recalculation */ | ||
| assignment_changed = rd_true; | ||
| } | ||
| MemberEpoch = member->member_epoch; | ||
| } else { | ||
| err = RD_KAFKA_RESP_ERR_UNKNOWN_MEMBER_ID; | ||
| } | ||
|
|
||
| } else { | ||
| /* HEARTBEAT: Existing member heartbeat */ | ||
| member = rd_kafka_mock_sharegroup_member_find( | ||
| mshgrp, &MemberId); | ||
| if (!member) { | ||
| err = RD_KAFKA_RESP_ERR_UNKNOWN_MEMBER_ID; | ||
| } else if (MemberEpoch > member->member_epoch) { | ||
| /* Client epoch is ahead of server - indicates | ||
| * a bug or stale coordinator. */ | ||
| err = RD_KAFKA_RESP_ERR_FENCED_MEMBER_EPOCH; | ||
|
There was a problem hiding this comment. Does this need to be
Member
Author
There was a problem hiding this comment. I checked the java implementation, in that they are returning |
||
| } else if (MemberEpoch < member->member_epoch) { | ||
| /* Client epoch is behind. Allow if it matches | ||
| * the previous epoch (response with bumped | ||
| * epoch may have been lost). Otherwise fence. | ||
| */ | ||
| if (MemberEpoch != | ||
| member->previous_member_epoch) { | ||
| err = | ||
| RD_KAFKA_RESP_ERR_FENCED_MEMBER_EPOCH; | ||
| } else { | ||
| /* Accept previous epoch - client is | ||
| * catching up */ | ||
| member->conn = mconn; | ||
| MemberEpoch = member->member_epoch; | ||
| } | ||
| } else { | ||
| /* Epoch matches - normal heartbeat */ | ||
| /* Check for subscription changes */ | ||
| if (SubscribedTopicNamesCnt >= 0 && | ||
| rd_kafka_mock_sharegroup_member_subscribed_topic_names_set( | ||
| member, SubscribedTopicNames, | ||
| SubscribedTopicNamesCnt)) { | ||
| assignment_changed = rd_true; | ||
| } | ||
| member->conn = mconn; | ||
| MemberEpoch = member->member_epoch; | ||
| } | ||
|
Comment on lines
+3116
to
+3155
There was a problem hiding this comment. ShareGroupHeartbeat processing does not mark the member as active on successful heartbeats (it updates member->conn/MemberEpoch but never updates ts_last_activity). This will cause members to time out even while heartbeating (breaking session-timeout semantics and likely making tests flaky). Call rd_kafka_mock_sharegroup_member_active() whenever a heartbeat is accepted (including the “catching up” previous-epoch path). |
||
| } | ||
|
|
||
| /* Recalculate assignments if needed */ | ||
| if (assignment_changed && mshgrp->member_cnt > 0) { | ||
| rd_kafka_mock_sharegroup_assignment_recalculate(mshgrp); | ||
| if (member) | ||
| MemberEpoch = member->member_epoch; | ||
| } | ||
|
|
||
| mtx_unlock(&mcluster->lock); | ||
| } | ||
|
|
||
| build_response: | ||
|
|
||
| /* ThrottleTimeMs */ | ||
| rd_kafka_buf_write_i32(resp, 0); | ||
|
|
||
| /* ErrorCode */ | ||
| rd_kafka_buf_write_i16(resp, err); | ||
|
|
||
| /* ErrorMessage */ | ||
| if (err) | ||
| rd_kafka_buf_write_str(resp, rd_kafka_err2str(err), -1); | ||
| else | ||
| rd_kafka_buf_write_str(resp, NULL, -1); | ||
|
|
||
| /* MemberId */ | ||
| if (!err && member) | ||
| rd_kafka_buf_write_str(resp, member->id, -1); | ||
| else | ||
| rd_kafka_buf_write_str(resp, NULL, -1); | ||
|
|
||
| /* MemberEpoch */ | ||
| rd_kafka_buf_write_i32(resp, MemberEpoch); | ||
|
|
||
| /* HeartbeatIntervalMs */ | ||
| if (mshgrp) | ||
| rd_kafka_buf_write_i32(resp, mshgrp->heartbeat_interval_ms); | ||
| else | ||
| rd_kafka_buf_write_i32(resp, 5000); | ||
|
|
||
| /* Assignment */ | ||
| if (!err && member && member->assignment) { | ||
| /* Send assignment even if empty (cnt == 0). | ||
| * Null (-1) means "no change", while an empty assignment | ||
| * means "you have 0 partitions". */ | ||
| rd_kafka_buf_write_i8(resp, 1); | ||
| rd_kafka_mock_handle_ShareGroupHeartbeat_write_TopicPartitions( | ||
| resp, member->assignment); | ||
| rd_kafka_buf_write_tags_empty(resp); | ||
| } else { | ||
| rd_kafka_buf_write_i8(resp, -1); | ||
| } | ||
|
|
||
| rd_kafka_buf_write_tags_empty(resp); | ||
|
|
||
| rd_kafka_mock_connection_send_response(mconn, resp); | ||
|
|
||
| RD_IF_FREE(SubscribedTopicNames, rd_free); | ||
| return 0; | ||
|
|
||
| err_parse: | ||
| RD_IF_FREE(SubscribedTopicNames, rd_free); | ||
| rd_kafka_buf_destroy(resp); | ||
| return -1; | ||
| } | ||
|
|
||
| /** | ||
| * @brief Default request handlers | ||
| */ | ||
|
|
@@ -3025,6 +3248,8 @@ const struct rd_kafka_mock_api_handler | |
| {2, 2, -1, rd_kafka_mock_handle_OffsetForLeaderEpoch}, | ||
| [RD_KAFKAP_ConsumerGroupHeartbeat] = | ||
| {1, 1, 1, rd_kafka_mock_handle_ConsumerGroupHeartbeat}, | ||
| [RD_KAFKAP_ShareGroupHeartbeat] = | ||
| {1, 1, 1, rd_kafka_mock_handle_ShareGroupHeartbeat}, | ||
| [RD_KAFKAP_GetTelemetrySubscriptions] = | ||
| {0, 0, 0, rd_kafka_mock_handle_GetTelemetrySubscriptions}, | ||
| [RD_KAFKAP_PushTelemetry] = {0, 0, 0, | ||
|
|
||
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
On JOIN (MemberEpoch == 0) this calls rd_kafka_mock_sharegroup_member_subscribed_topic_names_set() even when SubscribedTopicNamesCnt is -1 (nullable array). Passing -1 through leads to rd_list_new(SubscribedTopicNamesCnt, ...) with a negative size. Only call the setter when SubscribedTopicNamesCnt >= 0, and treat -1 as “no change / not provided”.