lucasbru commented on code in PR #15511:
URL: https://github.com/apache/kafka/pull/15511#discussion_r1523523053
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/HeartbeatRequestManager.java:
##########
@@ -566,18 +567,18 @@ public ConsumerGroupHeartbeatRequestData
buildRequestData() {
// ClientAssignors - not supported yet
- // TopicPartitions - only sent if it has changed since the last
heartbeat. Note that
- // the string consists of just the topic ID and the partitions.
When an assignment is
- // received, we might not yet know the topic name, and then it is
learnt subsequently
- // by a metadata update.
- TreeSet<String> assignedPartitions =
membershipManager.currentAssignment().entrySet().stream()
- .map(entry -> entry.getKey() + "-" + entry.getValue())
- .collect(Collectors.toCollection(TreeSet::new));
- if (!assignedPartitions.equals(sentFields.topicPartitions)) {
+ // TopicPartitions - sent with the first heartbeat after a new
assignment from the server was
+ // reconciled. This is ensured by resending the topic partitions
whenever the local assignment,
+ // including its local epoch is changed (although the local epoch
is not sent in the heartbeat).
+ LocalAssignment local = membershipManager.currentAssignment();
+ if (local == null) {
+ data.setTopicPartitions(Collections.emptyList());
+ sentFields.topicPartitions = null;
+ } else if (!local.equals(sentFields.topicPartitions)) {
Review Comment:
Renamed the field
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
