root
/
kafka
mirror of https://github.com/apache/kafka.git
1
0
Fork 0
Code Issues Actions 3 Packages Projects Releases Wiki Activity

MINOR: simplify consumer logic (#15519) 

For static member, the `group.instance.id` cannot change.

Reviewers: Chia-Ping Tsai <chia7712@gmail.com>, Lianet Magrans <lianetmr@gmail.com>, David Jacot <david.jacot@gmail.com>
This commit is contained in:
Matthias J. Sax 2024-03-13 21:52:25 -07:00 committed by GitHub
parent 722967a2b7
commit d88a97adef
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
1 changed files with 2 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
clients/src/main/java/org/apache/kafka/clients/consumer/internals/HeartbeatRequestManager.java
View File

@ -529,14 +529,8 @@ public class HeartbeatRequestManager implements RequestManager {
// MemberEpoch - always sent
data.setMemberEpoch(membershipManager.memberEpoch());
// InstanceId - only sent if has changed since the last heartbeat
// Always send when leaving the group as a static member
membershipManager.groupInstanceId().ifPresent(groupInstanceId -> {
if (!groupInstanceId.equals(sentFields.instanceId) || membershipManager.memberEpoch() == ConsumerGroupHeartbeatRequest.LEAVE_GROUP_STATIC_MEMBER_EPOCH) {
data.setInstanceId(groupInstanceId);
sentFields.instanceId = groupInstanceId;
}
});
// InstanceId - set if present
membershipManager.groupInstanceId().ifPresent(data::setInstanceId);
// RebalanceTimeoutMs - only sent if has changed since the last heartbeat
if (sentFields.rebalanceTimeoutMs != rebalanceTimeoutMs) {
@ -593,7 +587,6 @@ public class HeartbeatRequestManager implements RequestManager {
// Fields of ConsumerHeartbeatRequest sent in the most recent request
static class SentFields {
private String instanceId = null;
private int rebalanceTimeoutMs = -1;
private TreeSet<String> subscribedTopicNames = null;
private String serverAssignor = null;
@ -602,7 +595,6 @@ public class HeartbeatRequestManager implements RequestManager {
SentFields() {}
void reset() {
instanceId = null;
rebalanceTimeoutMs = -1;
subscribedTopicNames = null;
serverAssignor = null;