You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "AndrewJSchofield (via GitHub)" <gi...@apache.org> on 2023/11/24 12:06:15 UTC
[PR] KAFKA-15890: Consumer.poll with long timeout unaware of assigned partitions [kafka]
AndrewJSchofield opened a new pull request, #14835:
URL: https://github.com/apache/kafka/pull/14835
In the new consumer, `Consumer.poll(Duration timeout)` blocks for the entire duration. If the consumer is joining a group and has not yet received its assignments, the poll begins before an assignment has yet been received. Because the poll is blocked, it does not notice when partitions are assigned, and it subsequently does not return any records. The old consumer only blocks for the duration of the heartbeat interval and loops for until the poll timeout has passed, and is thus able to check for assignments received.
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15890: Consumer.poll with long timeout unaware of assigned partitions [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on PR #14835:
URL: https://github.com/apache/kafka/pull/14835#issuecomment-1831559902
One of the test failures needs addressing. I'll investigate.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15890: Consumer.poll with long timeout unaware of assigned partitions [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #14835:
URL: https://github.com/apache/kafka/pull/14835#discussion_r1406356534
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/HeartbeatRequestManager.java:
##########
@@ -417,4 +394,129 @@ private void updateHeartbeatIntervalMs(final long heartbeatIntervalMs) {
this.heartbeatTimer.updateAndReset(heartbeatIntervalMs);
}
}
+
+ /**
+ * Builds the heartbeat requests correctly, ensuring that all information is sent according to
+ * the protocol, but subsequent requests do not send information which has not changed. This
+ * is important to ensure that reconciliation completes succesfully.
+ */
+ static class HeartbeatState {
+ private final SubscriptionState subscriptions;
+ private final MembershipManager membershipManager;
+ private final int rebalanceTimeoutMs;
+
+ // Fields of ConsumerHeartbeatRequest sent in the most recent request
+ private String sentInstanceId;
+ private int sentRebalanceTimeoutMs;
+ private TreeSet<String> sentSubscribedTopicNames;
+ // private String sentSubscribedTopicRegex;
+ private String sentServerAssignor;
+ private TreeSet<String> sentTopicPartitions;
+
+ public HeartbeatState(
+ final SubscriptionState subscriptions,
+ final MembershipManager membershipManager,
+ final int rebalanceTimeoutMs) {
+ this.subscriptions = subscriptions;
+ this.membershipManager = membershipManager;
+ this.rebalanceTimeoutMs = rebalanceTimeoutMs;
+ this.sentInstanceId = null;
+ this.sentRebalanceTimeoutMs = -1;
+ this.sentSubscribedTopicNames = null;
+ this.sentServerAssignor = null;
+ this.sentTopicPartitions = null;
+ }
+
+
+ public void reset() {
+ sentInstanceId = null;
+ sentRebalanceTimeoutMs = -1;
+ sentSubscribedTopicNames = null;
+ sentServerAssignor = null;
+ sentTopicPartitions = null;
+ }
+
+ public ConsumerGroupHeartbeatRequestData buildRequestData() {
+ ConsumerGroupHeartbeatRequestData data = new ConsumerGroupHeartbeatRequestData();
+
+ // GroupId - always sent
+ data.setGroupId(membershipManager.groupId());
+
+ // MemberId - always sent, after it has been received from the coordinator
+ if (membershipManager.memberId() != null) {
Review Comment:
I see what you mean. Actually, I think no check is needed. Initially, it's empty "" which causes the coordinator to create a UUID for the member ID. Subsequently, it's necessary to send the member ID on every request. I've just removed the check.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15890: Consumer.poll with long timeout unaware of assigned partitions [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #14835:
URL: https://github.com/apache/kafka/pull/14835#discussion_r1406402956
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/HeartbeatRequestManagerTest.java:
##########
@@ -332,6 +336,72 @@ public void testHeartbeatResponseOnErrorHandling(final Errors error, final boole
}
}
+ @Test
+ public void testHeartbeatState() {
+ // The initial ConsumerGroupHeartbeatRequest sets most fields to their initial empty values
+ ConsumerGroupHeartbeatRequestData data = heartbeatState.buildRequestData();
+ assertEquals(ConsumerTestBuilder.DEFAULT_GROUP_ID, data.groupId());
+ assertEquals("", data.memberId());
+ assertEquals(0, data.memberEpoch());
+ assertNull(data.instanceId());
+ assertEquals(ConsumerTestBuilder.DEFAULT_MAX_POLL_INTERVAL_MS, data.rebalanceTimeoutMs());
+ assertEquals(Collections.emptyList(), data.subscribedTopicNames());
+ assertNull(data.subscribedTopicRegex());
+ assertNull(data.serverAssignor());
+ assertEquals(Collections.emptyList(), data.topicPartitions());
+ membershipManager.onHeartbeatRequestSent();
+ assertEquals(MemberState.UNSUBSCRIBED, membershipManager.state());
+
+ // Mock a response from the group coordinator, that supplies the member ID and a new epoch
+ mockStableMember();
+ data = heartbeatState.buildRequestData();
+ assertEquals(ConsumerTestBuilder.DEFAULT_GROUP_ID, data.groupId());
+ assertEquals(memberId, data.memberId());
+ assertEquals(1, data.memberEpoch());
+ assertNull(data.instanceId());
+ assertEquals(-1, data.rebalanceTimeoutMs());
+ assertNull(data.subscribedTopicNames());
+ assertNull(data.subscribedTopicRegex());
+ assertNull(data.serverAssignor());
+ assertNull(data.topicPartitions());
+ membershipManager.onHeartbeatRequestSent();
+ assertEquals(MemberState.STABLE, membershipManager.state());
+
+ // Join the group and subscribe to a topic, but the response has not yet been received
+ String topic = "topic1";
+ subscriptions.subscribe(Collections.singleton(topic), Optional.empty());
+ membershipManager.onSubscriptionUpdated();
+ membershipManager.transitionToFenced(); // And indirect way of moving to JOINING state
Review Comment:
This is perfectly fine, but just for the record, it is not strictly needed. When the subscription changes (updated in the subscription state), it is picked up and sent out on the next heartbeat request, without requiring a transition to joining.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15890: Consumer.poll with long timeout unaware of assigned partitions [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #14835:
URL: https://github.com/apache/kafka/pull/14835#discussion_r1406297866
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/MembershipManagerImpl.java:
##########
@@ -1074,6 +1074,7 @@ boolean reconciliationInProgress() {
public void onUpdate(ClusterResource clusterResource) {
resolveMetadataForUnresolvedAssignment();
if (!assignmentReadyToReconcile.isEmpty()) {
+ transitionTo(MemberState.RECONCILING);
Review Comment:
Nice, this was definitely missing. Just wonder if it would be better to move the transition to reconciling to the reconcile function itself, to have it in a single place.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15890: Consumer.poll with long timeout unaware of assigned partitions [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #14835:
URL: https://github.com/apache/kafka/pull/14835#discussion_r1407452521
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/HeartbeatRequestManager.java:
##########
@@ -417,4 +394,128 @@ private void updateHeartbeatIntervalMs(final long heartbeatIntervalMs) {
this.heartbeatTimer.updateAndReset(heartbeatIntervalMs);
}
}
+
+ /**
+ * Builds the heartbeat requests correctly, ensuring that all information is sent according to
+ * the protocol, but subsequent requests do not send information which has not changed. This
+ * is important to ensure that reconciliation completes successfully.
+ */
+ static class HeartbeatState {
+ private final SubscriptionState subscriptions;
+ private final MembershipManager membershipManager;
+ private final int rebalanceTimeoutMs;
+ private final SentFields sentFields;
+
+ public HeartbeatState(
+ final SubscriptionState subscriptions,
+ final MembershipManager membershipManager,
+ final int rebalanceTimeoutMs) {
+ this.subscriptions = subscriptions;
+ this.membershipManager = membershipManager;
+ this.rebalanceTimeoutMs = rebalanceTimeoutMs;
+ this.sentFields = new SentFields();
+ }
+
+
+ public void reset() {
+ sentFields.reset();
+ }
+
+ public ConsumerGroupHeartbeatRequestData buildRequestData() {
+ ConsumerGroupHeartbeatRequestData data = new ConsumerGroupHeartbeatRequestData();
+
+ // GroupId - always sent
+ data.setGroupId(membershipManager.groupId());
+
+ // MemberId - always sent, empty until it has been received from the coordinator
+ data.setMemberId(membershipManager.memberId());
+
+ // MemberEpoch - always sent
+ data.setMemberEpoch(membershipManager.memberEpoch());
+
+ // InstanceId - sent if hasn't changed since the last heartbeat
+ membershipManager.groupInstanceId().ifPresent(groupInstanceId -> {
+ if (!groupInstanceId.equals(sentFields.instanceId)) {
+ data.setInstanceId(groupInstanceId);
+ sentFields.instanceId = groupInstanceId;
+ }
+ });
+
+ // RebalanceTimeoutMs - sent if hasn't changed since the last heartbeat
+ if (sentFields.rebalanceTimeoutMs != rebalanceTimeoutMs) {
+ data.setRebalanceTimeoutMs(rebalanceTimeoutMs);
+ sentFields.rebalanceTimeoutMs = rebalanceTimeoutMs;
+ }
+
+ if (!this.subscriptions.hasPatternSubscription()) {
+ // SubscribedTopicNames - sent if hasn't changed since the last heartbeat
+ TreeSet<String> subscribedTopicNames = new TreeSet<>(this.subscriptions.subscription());
+ if (!subscribedTopicNames.equals(sentFields.subscribedTopicNames)) {
+ data.setSubscribedTopicNames(new ArrayList<>(this.subscriptions.subscription()));
+ sentFields.subscribedTopicNames = subscribedTopicNames;
+ }
+ } else {
+ // SubscribedTopicRegex - sent if hasn't changed since the last heartbeat
+ // - not supported yet
+ }
+
+ // ServerAssignor - sent if hasn't changed since the last heartbeat
+ this.membershipManager.serverAssignor().ifPresent(serverAssignor -> {
+ if (!serverAssignor.equals(sentFields.serverAssignor)) {
+ data.setServerAssignor(serverAssignor);
+ sentFields.serverAssignor = serverAssignor;
+ }
+ });
+
+ // ClientAssignors - not supported yet
+
+ // TopicPartitions - sent if hasn't changed since the last heartbeat
+ TreeSet<String> assignedPartitions = new TreeSet<>(membershipManager.currentAssignment().stream()
+ .map(tp -> tp.topicId() + "-" + tp.partition()).collect(Collectors.toList()));
Review Comment:
`.collect(Collectors.toCollection(TreeSet::new)` and skip the intermediate list.
Also, you seem to avoid using `TopicIdPartition.toString` or just `TopicIdPartition.equals` here - is it to support topic renaming? If so, I think it's quite subtle and would deserve a little comment.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/HeartbeatRequestManager.java:
##########
@@ -128,13 +136,15 @@ public HeartbeatRequestManager(
final CoordinatorRequestManager coordinatorRequestManager,
final SubscriptionState subscriptions,
final MembershipManager membershipManager,
+ final HeartbeatState heartbeatState,
final HeartbeatRequestState heartbeatRequestState,
final BackgroundEventHandler backgroundEventHandler) {
this.logger = logContext.logger(this.getClass());
this.subscriptions = subscriptions;
Review Comment:
Looks like the subscription field is now unused.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/HeartbeatRequestManager.java:
##########
@@ -417,4 +394,128 @@ private void updateHeartbeatIntervalMs(final long heartbeatIntervalMs) {
this.heartbeatTimer.updateAndReset(heartbeatIntervalMs);
}
}
+
+ /**
+ * Builds the heartbeat requests correctly, ensuring that all information is sent according to
+ * the protocol, but subsequent requests do not send information which has not changed. This
+ * is important to ensure that reconciliation completes successfully.
+ */
+ static class HeartbeatState {
+ private final SubscriptionState subscriptions;
+ private final MembershipManager membershipManager;
+ private final int rebalanceTimeoutMs;
+ private final SentFields sentFields;
+
+ public HeartbeatState(
+ final SubscriptionState subscriptions,
+ final MembershipManager membershipManager,
+ final int rebalanceTimeoutMs) {
+ this.subscriptions = subscriptions;
+ this.membershipManager = membershipManager;
+ this.rebalanceTimeoutMs = rebalanceTimeoutMs;
+ this.sentFields = new SentFields();
+ }
+
+
+ public void reset() {
+ sentFields.reset();
+ }
+
+ public ConsumerGroupHeartbeatRequestData buildRequestData() {
+ ConsumerGroupHeartbeatRequestData data = new ConsumerGroupHeartbeatRequestData();
+
+ // GroupId - always sent
+ data.setGroupId(membershipManager.groupId());
+
+ // MemberId - always sent, empty until it has been received from the coordinator
+ data.setMemberId(membershipManager.memberId());
+
+ // MemberEpoch - always sent
+ data.setMemberEpoch(membershipManager.memberEpoch());
+
+ // InstanceId - sent if hasn't changed since the last heartbeat
Review Comment:
I think in this comment and the similar ones below, you wanted to say `send if it _has_ changed`
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15890: Consumer.poll with long timeout unaware of assigned partitions [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on PR #14835:
URL: https://github.com/apache/kafka/pull/14835#issuecomment-1831495002
@AndrewJSchofield
Looks like the test is failing
```
org.mockito.exceptions.misusing.WrongTypeOfReturnValue:
Long cannot be returned by poll()
poll() should return PollResult
***
If you're unsure why you're getting above error read on.
Due to the nature of the syntax above problem might occur because:
1. This exception *might* occur in wrongly written multi-threaded tests.
Please refer to Mockito FAQ on limitations of concurrency testing.
2. A spy is stubbed using when(spy.foo()).then() syntax. It is safer to stub spies -
- with doReturn|Throw() family of methods. More in javadocs for Mockito.spy() method.
```
Could you please have a look?
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15890: Consumer.poll with long timeout unaware of assigned partitions [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #14835:
URL: https://github.com/apache/kafka/pull/14835#discussion_r1411833816
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkThread.java:
##########
@@ -205,6 +205,26 @@ public void wakeup() {
networkClientDelegate.wakeup();
}
+ /**
+ * Returns the delay for which the application thread can safely wait before it should be responsive
+ * to results from the request managers. For example, the subscription state can change when heartbeats
+ * are sent, so blocking for longer than the heartbeat interval might mean the application thread is not
+ * responsive to changes.
+ *
+ * @return The maximum delay in milliseconds
+ */
+ public long maximumTimeToWait() {
+ final long currentTimeMs = time.milliseconds();
+ if (requestManagers == null) {
+ return MAX_POLL_TIMEOUT_MS;
+ }
+ return requestManagers.entries().stream()
+ .filter(Optional::isPresent)
+ .map(Optional::get)
+ .map(rm -> rm.maximumTimeToWait(currentTimeMs))
+ .reduce(Long.MAX_VALUE, Math::min);
+ }
+
Review Comment:
I'll take this concept and work on it in a follow-on PR.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15890: Consumer.poll with long timeout unaware of assigned partitions [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield closed pull request #14835: KAFKA-15890: Consumer.poll with long timeout unaware of assigned partitions
URL: https://github.com/apache/kafka/pull/14835
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15890: Consumer.poll with long timeout unaware of assigned partitions [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on code in PR #14835:
URL: https://github.com/apache/kafka/pull/14835#discussion_r1411218465
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -800,6 +811,11 @@ public void resetTimer() {
this.timer.reset(autoCommitInterval);
}
+ public long remainingMs(final long currentTimeMs) {
+ this.timer.update(currentTimeMs);
+ return this.timer.remainingMs();
+ }
+
Review Comment:
Example, this is updating the Timer from both threads.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/MembershipManagerImpl.java:
##########
@@ -1074,6 +1074,7 @@ boolean reconciliationInProgress() {
public void onUpdate(ClusterResource clusterResource) {
resolveMetadataForUnresolvedAssignment();
if (!assignmentReadyToReconcile.isEmpty()) {
+ transitionTo(MemberState.RECONCILING);
Review Comment:
OK, I think I'm running into this too. In my tests, the state machine was preemptively `RECONCILING` only to later find out the assignments were the same, after which it returned from `reconcile()` without resetting the state.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkThread.java:
##########
@@ -205,6 +205,26 @@ public void wakeup() {
networkClientDelegate.wakeup();
}
+ /**
+ * Returns the delay for which the application thread can safely wait before it should be responsive
+ * to results from the request managers. For example, the subscription state can change when heartbeats
+ * are sent, so blocking for longer than the heartbeat interval might mean the application thread is not
+ * responsive to changes.
+ *
+ * @return The maximum delay in milliseconds
+ */
+ public long maximumTimeToWait() {
+ final long currentTimeMs = time.milliseconds();
+ if (requestManagers == null) {
+ return MAX_POLL_TIMEOUT_MS;
+ }
+ return requestManagers.entries().stream()
+ .filter(Optional::isPresent)
+ .map(Optional::get)
+ .map(rm -> rm.maximumTimeToWait(currentTimeMs))
+ .reduce(Long.MAX_VALUE, Math::min);
+ }
+
Review Comment:
This will invoke the request managers directly from the application thread, right?
If so, I'm a little concerned by this approach because the implementation of `maximumTimeToWait` in each `RequestManager` will be executed by the application thread. In the implementations elsewhere in this PR we're reading and writing state in the request managers that is only intended to be read or written by the network I/O thread.
We need to be careful here 🤔
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/HeartbeatRequestManager.java:
##########
@@ -417,4 +394,127 @@ private void updateHeartbeatIntervalMs(final long heartbeatIntervalMs) {
this.heartbeatTimer.updateAndReset(heartbeatIntervalMs);
}
}
+
+ /**
+ * Builds the heartbeat requests correctly, ensuring that all information is sent according to
+ * the protocol, but subsequent requests do not send information which has not changed. This
+ * is important to ensure that reconciliation completes successfully.
+ */
+ static class HeartbeatState {
Review Comment:
Fair enough 😄
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkThread.java:
##########
@@ -205,6 +205,26 @@ public void wakeup() {
networkClientDelegate.wakeup();
}
+ /**
+ * Returns the delay for which the application thread can safely wait before it should be responsive
+ * to results from the request managers. For example, the subscription state can change when heartbeats
+ * are sent, so blocking for longer than the heartbeat interval might mean the application thread is not
+ * responsive to changes.
+ *
+ * @return The maximum delay in milliseconds
+ */
+ public long maximumTimeToWait() {
+ final long currentTimeMs = time.milliseconds();
+ if (requestManagers == null) {
+ return MAX_POLL_TIMEOUT_MS;
+ }
+ return requestManagers.entries().stream()
+ .filter(Optional::isPresent)
+ .map(Optional::get)
+ .map(rm -> rm.maximumTimeToWait(currentTimeMs))
+ .reduce(Long.MAX_VALUE, Math::min);
+ }
+
Review Comment:
This will invoke the request managers directly from the application thread, right?
If so, I'm a little concerned by this approach because the implementation of `maximumTimeToWait` in each `RequestManager` will be executed by the application thread. In the implementations elsewhere in this PR we're reading and writing state in the request managers that is only intended to be read or written by the network I/O thread.
We need to be careful here 🤔
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/HeartbeatRequestManager.java:
##########
@@ -417,4 +394,127 @@ private void updateHeartbeatIntervalMs(final long heartbeatIntervalMs) {
this.heartbeatTimer.updateAndReset(heartbeatIntervalMs);
}
}
+
+ /**
+ * Builds the heartbeat requests correctly, ensuring that all information is sent according to
+ * the protocol, but subsequent requests do not send information which has not changed. This
+ * is important to ensure that reconciliation completes successfully.
+ */
+ static class HeartbeatState {
Review Comment:
Fair enough 😄
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15890: Consumer.poll with long timeout unaware of assigned partitions [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on code in PR #14835:
URL: https://github.com/apache/kafka/pull/14835#discussion_r1411470319
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkThread.java:
##########
@@ -205,6 +205,26 @@ public void wakeup() {
networkClientDelegate.wakeup();
}
+ /**
+ * Returns the delay for which the application thread can safely wait before it should be responsive
+ * to results from the request managers. For example, the subscription state can change when heartbeats
+ * are sent, so blocking for longer than the heartbeat interval might mean the application thread is not
+ * responsive to changes.
+ *
+ * @return The maximum delay in milliseconds
+ */
+ public long maximumTimeToWait() {
+ final long currentTimeMs = time.milliseconds();
+ if (requestManagers == null) {
+ return MAX_POLL_TIMEOUT_MS;
+ }
+ return requestManagers.entries().stream()
+ .filter(Optional::isPresent)
+ .map(Optional::get)
+ .map(rm -> rm.maximumTimeToWait(currentTimeMs))
+ .reduce(Long.MAX_VALUE, Math::min);
+ }
+
Review Comment:
I'm wondering how "fresh" the "maximum time to wait" value needs to be...
Spitballing here, but... say we introduce a new instance variable in `ConsumerNetworkThread` that stores a cached value named, say, `notTotallyFreshMaxWaitCached`? We could then invoke `maximumTimeToWait()` from inside `runOnce()` and update `notTotallyFreshMaxWaitCached`, like so:
```java
private volatile long notTotallyFreshMaxWaitCached = MAX_POLL_TIMEOUT_MS;
void runOnce() {
// If there are errors processing any events, the error will be thrown immediately. This will have
// the effect of closing the background thread.
applicationEventProcessor.process();
final long currentTimeMs = time.milliseconds();
final long pollWaitTimeMs = requestManagers.entries().stream()
.filter(Optional::isPresent)
.map(Optional::get)
.map(rm -> rm.poll(currentTimeMs))
.map(networkClientDelegate::addAll)
.reduce(MAX_POLL_TIMEOUT_MS, Math::min);
networkClientDelegate.poll(pollWaitTimeMs, currentTimeMs);
notTotallyFreshMaxWaitCached = requestManagers.entries().stream()
.filter(Optional::isPresent)
.map(Optional::get)
.map(rm -> rm.maximumTimeToWait(currentTimeMs))
.reduce(Long.MAX_VALUE, Math::min);
}
public long maximumTimeToWait() {
if (requestManagers == null) {
return MAX_POLL_TIMEOUT_MS;
}
return notTotallyFreshMaxWaitCached;
}
```
(Or something like that.)
This way we still only touch the `RequestManager`s inside `runOnce()`, so we don't open overselves up to subtle concurrency issues.
Thoughts?
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15890: Consumer.poll with long timeout unaware of assigned partitions [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on code in PR #14835:
URL: https://github.com/apache/kafka/pull/14835#discussion_r1411470319
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkThread.java:
##########
@@ -205,6 +205,26 @@ public void wakeup() {
networkClientDelegate.wakeup();
}
+ /**
+ * Returns the delay for which the application thread can safely wait before it should be responsive
+ * to results from the request managers. For example, the subscription state can change when heartbeats
+ * are sent, so blocking for longer than the heartbeat interval might mean the application thread is not
+ * responsive to changes.
+ *
+ * @return The maximum delay in milliseconds
+ */
+ public long maximumTimeToWait() {
+ final long currentTimeMs = time.milliseconds();
+ if (requestManagers == null) {
+ return MAX_POLL_TIMEOUT_MS;
+ }
+ return requestManagers.entries().stream()
+ .filter(Optional::isPresent)
+ .map(Optional::get)
+ .map(rm -> rm.maximumTimeToWait(currentTimeMs))
+ .reduce(Long.MAX_VALUE, Math::min);
+ }
+
Review Comment:
I'm wondering how "fresh" the "maximum time to wait" value needs to be...
Spitballing here, but... say we introduce a new instance variable in `ConsumerNetworkThread` that stores a cached value named, say, `notTotallyFreshMaxWaitCached`? We could then invoke `maximumTimeToWait()` from inside `runOnce()` and update `notTotallyFreshMaxWaitCached`, like so:
```java
private volatile long notTotallyFreshMaxWaitCached = MAX_POLL_TIMEOUT_MS;
void runOnce() {
// If there are errors processing any events, the error will be thrown immediately. This will have
// the effect of closing the background thread.
applicationEventProcessor.process();
final long currentTimeMs = time.milliseconds();
final long pollWaitTimeMs = requestManagers.entries().stream()
.filter(Optional::isPresent)
.map(Optional::get)
.map(rm -> rm.poll(currentTimeMs))
.map(networkClientDelegate::addAll)
.reduce(MAX_POLL_TIMEOUT_MS, Math::min);
networkClientDelegate.poll(pollWaitTimeMs, currentTimeMs);
notTotallyFreshMaxWaitCached = requestManagers.entries().stream()
.filter(Optional::isPresent)
.map(Optional::get)
.map(rm -> rm.maximumTimeToWait(currentTimeMs))
.reduce(Long.MAX_VALUE, Math::min);
}
public long maximumTimeToWait() {
return notTotallyFreshMaxWaitCached;
}
```
(Or something like that.)
This way we still only touch the `RequestManager`s inside `runOnce()`, so we don't open overselves up to subtle concurrency issues.
Thoughts?
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15890: Consumer.poll with long timeout unaware of assigned partitions [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #14835:
URL: https://github.com/apache/kafka/pull/14835#discussion_r1406286116
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/HeartbeatRequestManager.java:
##########
@@ -417,4 +394,129 @@ private void updateHeartbeatIntervalMs(final long heartbeatIntervalMs) {
this.heartbeatTimer.updateAndReset(heartbeatIntervalMs);
}
}
+
+ /**
+ * Builds the heartbeat requests correctly, ensuring that all information is sent according to
+ * the protocol, but subsequent requests do not send information which has not changed. This
+ * is important to ensure that reconciliation completes succesfully.
Review Comment:
typo successfully
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15890: Consumer.poll with long timeout unaware of assigned partitions [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #14835:
URL: https://github.com/apache/kafka/pull/14835#discussion_r1406354482
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/MembershipManagerImpl.java:
##########
@@ -1074,6 +1074,7 @@ boolean reconciliationInProgress() {
public void onUpdate(ClusterResource clusterResource) {
resolveMetadataForUnresolvedAssignment();
if (!assignmentReadyToReconcile.isEmpty()) {
+ transitionTo(MemberState.RECONCILING);
Review Comment:
I think I'll leave this part up to you in the next revision.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15890: Consumer.poll with long timeout unaware of assigned partitions [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #14835:
URL: https://github.com/apache/kafka/pull/14835#discussion_r1411721121
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkThread.java:
##########
@@ -205,6 +205,26 @@ public void wakeup() {
networkClientDelegate.wakeup();
}
+ /**
+ * Returns the delay for which the application thread can safely wait before it should be responsive
+ * to results from the request managers. For example, the subscription state can change when heartbeats
+ * are sent, so blocking for longer than the heartbeat interval might mean the application thread is not
+ * responsive to changes.
+ *
+ * @return The maximum delay in milliseconds
+ */
+ public long maximumTimeToWait() {
+ final long currentTimeMs = time.milliseconds();
+ if (requestManagers == null) {
+ return MAX_POLL_TIMEOUT_MS;
+ }
+ return requestManagers.entries().stream()
+ .filter(Optional::isPresent)
+ .map(Optional::get)
+ .map(rm -> rm.maximumTimeToWait(currentTimeMs))
+ .reduce(Long.MAX_VALUE, Math::min);
+ }
+
Review Comment:
Something like this would work I think. The aim after all is to keep things responsive rather than an accurate regular tick.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15890: Consumer.poll with long timeout unaware of assigned partitions [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #14835:
URL: https://github.com/apache/kafka/pull/14835#discussion_r1411253921
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkThread.java:
##########
@@ -205,6 +205,26 @@ public void wakeup() {
networkClientDelegate.wakeup();
}
+ /**
+ * Returns the delay for which the application thread can safely wait before it should be responsive
+ * to results from the request managers. For example, the subscription state can change when heartbeats
+ * are sent, so blocking for longer than the heartbeat interval might mean the application thread is not
+ * responsive to changes.
+ *
+ * @return The maximum delay in milliseconds
+ */
+ public long maximumTimeToWait() {
+ final long currentTimeMs = time.milliseconds();
+ if (requestManagers == null) {
+ return MAX_POLL_TIMEOUT_MS;
+ }
+ return requestManagers.entries().stream()
+ .filter(Optional::isPresent)
+ .map(Optional::get)
+ .map(rm -> rm.maximumTimeToWait(currentTimeMs))
+ .reduce(Long.MAX_VALUE, Math::min);
+ }
+
Review Comment:
Yes, we do. In the previous commit, I inadvertently ended up with a ConcurrentModificationException for this very reason. This is why I scaled back the change to be only accessed the timer.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15890: Consumer.poll with long timeout unaware of assigned partitions [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #14835:
URL: https://github.com/apache/kafka/pull/14835#discussion_r1411254378
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -800,6 +811,11 @@ public void resetTimer() {
this.timer.reset(autoCommitInterval);
}
+ public long remainingMs(final long currentTimeMs) {
+ this.timer.update(currentTimeMs);
+ return this.timer.remainingMs();
+ }
+
Review Comment:
True. I think it's an atomic operation which is updating the timer to a more current value.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15890: Consumer.poll with long timeout unaware of assigned partitions [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #14835:
URL: https://github.com/apache/kafka/pull/14835#discussion_r1406699697
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/HeartbeatRequestManager.java:
##########
@@ -417,4 +394,127 @@ private void updateHeartbeatIntervalMs(final long heartbeatIntervalMs) {
this.heartbeatTimer.updateAndReset(heartbeatIntervalMs);
}
}
+
+ /**
+ * Builds the heartbeat requests correctly, ensuring that all information is sent according to
+ * the protocol, but subsequent requests do not send information which has not changed. This
+ * is important to ensure that reconciliation completes successfully.
+ */
+ static class HeartbeatState {
Review Comment:
Also, spotbugs will object if an inner class is non-static and does not use its `this` reference to the outer class.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15890: Consumer.poll with long timeout unaware of assigned partitions [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #14835:
URL: https://github.com/apache/kafka/pull/14835#discussion_r1406683119
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/HeartbeatRequestManager.java:
##########
@@ -417,4 +394,127 @@ private void updateHeartbeatIntervalMs(final long heartbeatIntervalMs) {
this.heartbeatTimer.updateAndReset(heartbeatIntervalMs);
}
}
+
+ /**
+ * Builds the heartbeat requests correctly, ensuring that all information is sent according to
+ * the protocol, but subsequent requests do not send information which has not changed. This
+ * is important to ensure that reconciliation completes successfully.
+ */
+ static class HeartbeatState {
Review Comment:
So that you can construct an independent instance of the class. It doesn't need a `this` and it's handy for testing to be able to make instances on demand.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15890: Consumer.poll with long timeout unaware of assigned partitions [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on PR #14835:
URL: https://github.com/apache/kafka/pull/14835#issuecomment-1829545443
> Thanks for the PR! prod code looks good to me, I just added three minor things that I spotted while going through the code. However, I have two requests:
> Can we add a test to check that we are interrupting the wait correctly in poll? I could imagine if we mock timeUntilNextPoll it shouldn't be too difficult.
Certainly. I'll do it.
> It seems we are solving two things in this PR - correctly build the heartbeat (only send the diff) and the fix to not miss something. We don't need to split the PR, but could you update the PR description?
Yes, I'll do that. We just needed both things to make the `kafka-console-consumer.sh` tool correctly join a stable group and start receiving records.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15890: Consumer.poll with long timeout unaware of assigned partitions [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on PR #14835:
URL: https://github.com/apache/kafka/pull/14835#issuecomment-1833320341
@AndrewJSchofield looks like there are related test failures
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15890: Consumer.poll with long timeout unaware of assigned partitions [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru merged PR #14835:
URL: https://github.com/apache/kafka/pull/14835
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15890: Consumer.poll with long timeout unaware of assigned partitions [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #14835:
URL: https://github.com/apache/kafka/pull/14835#discussion_r1406257828
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/HeartbeatRequestManager.java:
##########
@@ -417,4 +394,129 @@ private void updateHeartbeatIntervalMs(final long heartbeatIntervalMs) {
this.heartbeatTimer.updateAndReset(heartbeatIntervalMs);
}
}
+
+ /**
+ * Builds the heartbeat requests correctly, ensuring that all information is sent according to
+ * the protocol, but subsequent requests do not send information which has not changed. This
+ * is important to ensure that reconciliation completes succesfully.
+ */
+ static class HeartbeatState {
+ private final SubscriptionState subscriptions;
+ private final MembershipManager membershipManager;
+ private final int rebalanceTimeoutMs;
+
+ // Fields of ConsumerHeartbeatRequest sent in the most recent request
+ private String sentInstanceId;
+ private int sentRebalanceTimeoutMs;
+ private TreeSet<String> sentSubscribedTopicNames;
+ // private String sentSubscribedTopicRegex;
+ private String sentServerAssignor;
+ private TreeSet<String> sentTopicPartitions;
+
+ public HeartbeatState(
+ final SubscriptionState subscriptions,
+ final MembershipManager membershipManager,
+ final int rebalanceTimeoutMs) {
+ this.subscriptions = subscriptions;
+ this.membershipManager = membershipManager;
+ this.rebalanceTimeoutMs = rebalanceTimeoutMs;
+ this.sentInstanceId = null;
+ this.sentRebalanceTimeoutMs = -1;
+ this.sentSubscribedTopicNames = null;
+ this.sentServerAssignor = null;
+ this.sentTopicPartitions = null;
+ }
+
+
+ public void reset() {
+ sentInstanceId = null;
+ sentRebalanceTimeoutMs = -1;
+ sentSubscribedTopicNames = null;
+ sentServerAssignor = null;
+ sentTopicPartitions = null;
+ }
+
+ public ConsumerGroupHeartbeatRequestData buildRequestData() {
+ ConsumerGroupHeartbeatRequestData data = new ConsumerGroupHeartbeatRequestData();
+
+ // GroupId - always sent
+ data.setGroupId(membershipManager.groupId());
+
+ // MemberId - always sent, after it has been received from the coordinator
+ if (membershipManager.memberId() != null) {
Review Comment:
we should check for empty string (`memberId` defaults to empty string and is then updated when received from the coordinator)
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15890: Consumer.poll with long timeout unaware of assigned partitions [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #14835:
URL: https://github.com/apache/kafka/pull/14835#discussion_r1406686948
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/HeartbeatRequestManager.java:
##########
@@ -181,36 +191,19 @@ public NetworkClientDelegate.PollResult poll(long currentTimeMs) {
return new NetworkClientDelegate.PollResult(heartbeatRequestState.heartbeatIntervalMs, Collections.singletonList(request));
}
- private NetworkClientDelegate.UnsentRequest makeHeartbeatRequest() {
- // TODO: extract this logic for building the ConsumerGroupHeartbeatRequestData to a
- // stateful builder (HeartbeatState), that will keep the last data sent, and determine
- // the fields that changed and need to be included in the next HB (ex. check
- // subscriptionState changed from last sent to include assignment). It should also
- // ensure that all fields are sent on failure.
- ConsumerGroupHeartbeatRequestData data = new ConsumerGroupHeartbeatRequestData()
- .setGroupId(membershipManager.groupId())
- .setMemberEpoch(membershipManager.memberEpoch())
- .setRebalanceTimeoutMs(rebalanceTimeoutMs);
-
- if (membershipManager.memberId() != null) {
- data.setMemberId(membershipManager.memberId());
- }
-
- membershipManager.groupInstanceId().ifPresent(data::setInstanceId);
-
- if (this.subscriptions.hasPatternSubscription()) {
- // TODO: Pass the string to the GC if server side regex is used.
- } else {
- data.setSubscribedTopicNames(new ArrayList<>(this.subscriptions.subscription()));
- List<ConsumerGroupHeartbeatRequestData.TopicPartitions> topicPartitions =
- buildTopicPartitionsList(membershipManager.currentAssignment());
- data.setTopicPartitions(topicPartitions);
- }
-
- this.membershipManager.serverAssignor().ifPresent(data::setServerAssignor);
+ /**
+ * Returns the delay before the next network request for this request manager. Used to ensure that
+ * waiting in the application thread does not delay beyond the point that a result can be returned.
+ */
+ @Override
+ public long timeUntilNextPoll(long currentTimeMs) {
+ boolean heartbeatNow = membershipManager.shouldHeartbeatNow() && !heartbeatRequestState.requestInFlight();
+ return heartbeatNow ? 0L : heartbeatRequestState.nextHeartbeatMs(currentTimeMs);
+ }
+ private NetworkClientDelegate.UnsentRequest makeHeartbeatRequest() {
NetworkClientDelegate.UnsentRequest request = new NetworkClientDelegate.UnsentRequest(
- new ConsumerGroupHeartbeatRequest.Builder(data),
+ new ConsumerGroupHeartbeatRequest.Builder(this.heartbeatState.buildRequestData()),
Review Comment:
That style is used throughout this file.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15890: Consumer.poll with long timeout unaware of assigned partitions [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on PR #14835:
URL: https://github.com/apache/kafka/pull/14835#issuecomment-1833387661
@lucasbru I agree. The change has made one or two tests flaky. I will fix this.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15890: Consumer.poll with long timeout unaware of assigned partitions [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on code in PR #14835:
URL: https://github.com/apache/kafka/pull/14835#discussion_r1406656043
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -159,11 +159,19 @@ public NetworkClientDelegate.PollResult poll(final long currentTimeMs) {
return EMPTY;
List<NetworkClientDelegate.UnsentRequest> requests = pendingRequests.drain(currentTimeMs);
+ return new NetworkClientDelegate.PollResult(timeUntilNextPoll(currentTimeMs), requests);
+ }
+
+ /**
+ * Returns the delay before the next network request for this request manager. Used to ensure that
+ * waiting in the application thread does not delay beyond the point that a result can be returned.
+ */
+ @Override
+ public long timeUntilNextPoll(long currentTimeMs) {
// min of the remainingBackoffMs of all the request that are still backing off
- final long timeUntilNextPoll = Math.min(
- findMinTime(unsentOffsetCommitRequests(), currentTimeMs),
- findMinTime(unsentOffsetFetchRequests(), currentTimeMs));
- return new NetworkClientDelegate.PollResult(timeUntilNextPoll, requests);
+ return Math.min(
+ findMinTime(unsentOffsetCommitRequests(), currentTimeMs),
+ findMinTime(unsentOffsetFetchRequests(), currentTimeMs));
Review Comment:
Nit: leaving the indentation as is conforms to the existing 'four-spaces per tab' style.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/HeartbeatRequestManager.java:
##########
@@ -181,36 +191,19 @@ public NetworkClientDelegate.PollResult poll(long currentTimeMs) {
return new NetworkClientDelegate.PollResult(heartbeatRequestState.heartbeatIntervalMs, Collections.singletonList(request));
}
- private NetworkClientDelegate.UnsentRequest makeHeartbeatRequest() {
- // TODO: extract this logic for building the ConsumerGroupHeartbeatRequestData to a
- // stateful builder (HeartbeatState), that will keep the last data sent, and determine
- // the fields that changed and need to be included in the next HB (ex. check
- // subscriptionState changed from last sent to include assignment). It should also
- // ensure that all fields are sent on failure.
- ConsumerGroupHeartbeatRequestData data = new ConsumerGroupHeartbeatRequestData()
- .setGroupId(membershipManager.groupId())
- .setMemberEpoch(membershipManager.memberEpoch())
- .setRebalanceTimeoutMs(rebalanceTimeoutMs);
-
- if (membershipManager.memberId() != null) {
- data.setMemberId(membershipManager.memberId());
- }
-
- membershipManager.groupInstanceId().ifPresent(data::setInstanceId);
-
- if (this.subscriptions.hasPatternSubscription()) {
- // TODO: Pass the string to the GC if server side regex is used.
- } else {
- data.setSubscribedTopicNames(new ArrayList<>(this.subscriptions.subscription()));
- List<ConsumerGroupHeartbeatRequestData.TopicPartitions> topicPartitions =
- buildTopicPartitionsList(membershipManager.currentAssignment());
- data.setTopicPartitions(topicPartitions);
- }
-
- this.membershipManager.serverAssignor().ifPresent(data::setServerAssignor);
+ /**
+ * Returns the delay before the next network request for this request manager. Used to ensure that
+ * waiting in the application thread does not delay beyond the point that a result can be returned.
+ */
+ @Override
+ public long timeUntilNextPoll(long currentTimeMs) {
+ boolean heartbeatNow = membershipManager.shouldHeartbeatNow() && !heartbeatRequestState.requestInFlight();
+ return heartbeatNow ? 0L : heartbeatRequestState.nextHeartbeatMs(currentTimeMs);
+ }
+ private NetworkClientDelegate.UnsentRequest makeHeartbeatRequest() {
NetworkClientDelegate.UnsentRequest request = new NetworkClientDelegate.UnsentRequest(
- new ConsumerGroupHeartbeatRequest.Builder(data),
+ new ConsumerGroupHeartbeatRequest.Builder(this.heartbeatState.buildRequestData()),
Review Comment:
Nit: we got knocked in other reviews for unnecessarily qualifying with `this`.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/HeartbeatRequestManager.java:
##########
@@ -417,4 +394,127 @@ private void updateHeartbeatIntervalMs(final long heartbeatIntervalMs) {
this.heartbeatTimer.updateAndReset(heartbeatIntervalMs);
}
}
+
+ /**
+ * Builds the heartbeat requests correctly, ensuring that all information is sent according to
+ * the protocol, but subsequent requests do not send information which has not changed. This
+ * is important to ensure that reconciliation completes successfully.
+ */
+ static class HeartbeatState {
+ private final SubscriptionState subscriptions;
+ private final MembershipManager membershipManager;
+ private final int rebalanceTimeoutMs;
+
+ // Fields of ConsumerHeartbeatRequest sent in the most recent request
+ private String sentInstanceId;
+ private int sentRebalanceTimeoutMs;
+ private TreeSet<String> sentSubscribedTopicNames;
+ // private String sentSubscribedTopicRegex;
+ private String sentServerAssignor;
+ private TreeSet<String> sentTopicPartitions;
+
+ public HeartbeatState(
+ final SubscriptionState subscriptions,
+ final MembershipManager membershipManager,
+ final int rebalanceTimeoutMs) {
+ this.subscriptions = subscriptions;
+ this.membershipManager = membershipManager;
+ this.rebalanceTimeoutMs = rebalanceTimeoutMs;
+ this.sentInstanceId = null;
+ this.sentRebalanceTimeoutMs = -1;
+ this.sentSubscribedTopicNames = null;
+ this.sentServerAssignor = null;
+ this.sentTopicPartitions = null;
+ }
+
+
+ public void reset() {
+ sentInstanceId = null;
+ sentRebalanceTimeoutMs = -1;
+ sentSubscribedTopicNames = null;
+ sentServerAssignor = null;
+ sentTopicPartitions = null;
+ }
+
+ public ConsumerGroupHeartbeatRequestData buildRequestData() {
+ ConsumerGroupHeartbeatRequestData data = new ConsumerGroupHeartbeatRequestData();
+
+ // GroupId - always sent
+ data.setGroupId(membershipManager.groupId());
+
+ // MemberId - always sent, empty until it has been received from the coordinator
+ data.setMemberId(membershipManager.memberId());
+
+ // MemberEpoch - always sent
+ data.setMemberEpoch(membershipManager.memberEpoch());
+
+ // InstanceId - sent if hasn't changed since the last heartbeat
+ membershipManager.groupInstanceId().ifPresent(groupInstanceId -> {
+ if (!groupInstanceId.equals(sentInstanceId)) {
+ data.setInstanceId(groupInstanceId);
+ sentInstanceId = groupInstanceId;
+ }
+ });
+
+ // RebalanceTimeoutMs - sent if hasn't changed since the last heartbeat
+ if (sentRebalanceTimeoutMs != rebalanceTimeoutMs) {
+ data.setRebalanceTimeoutMs(rebalanceTimeoutMs);
+ sentRebalanceTimeoutMs = rebalanceTimeoutMs;
+ }
+
+ if (!this.subscriptions.hasPatternSubscription()) {
+ // SubscribedTopicNames - sent if hasn't changed since the last heartbeat
+ TreeSet<String> subscribedTopicNames = new TreeSet<>(this.subscriptions.subscription());
+ if (!subscribedTopicNames.equals(sentSubscribedTopicNames)) {
+ data.setSubscribedTopicNames(new ArrayList<>(this.subscriptions.subscription()));
+ sentSubscribedTopicNames = subscribedTopicNames;
+ }
+ } else {
+ // SubscribedTopicRegex - sent if hasn't changed since the last heartbeat
+ // - not supported yet
+ }
+
+ // ServerAssignor - sent if hasn't changed since the last heartbeat
+ this.membershipManager.serverAssignor().ifPresent(serverAssignor -> {
+ if (!serverAssignor.equals(sentServerAssignor)) {
+ data.setServerAssignor(serverAssignor);
+ sentServerAssignor = serverAssignor;
+ }
+ });
+
+ // ClientAssignors - not supported yet
+
+ // TopicPartitions - sent if hasn't changed since the last heartbeat
+ TreeSet<String> assignedPartitions = new TreeSet<>(membershipManager.currentAssignment().stream()
+ .map(tp -> tp.topicId() + "-" + tp.partition()).collect(Collectors.toList()));
+ if (!assignedPartitions.equals(sentTopicPartitions)) {
+ List<ConsumerGroupHeartbeatRequestData.TopicPartitions> topicPartitions =
+ buildTopicPartitionsList(membershipManager.currentAssignment());
+ data.setTopicPartitions(topicPartitions);
+ sentTopicPartitions = assignedPartitions;
+ }
+
+ return data;
+ }
+
+ private List<ConsumerGroupHeartbeatRequestData.TopicPartitions> buildTopicPartitionsList(Set<TopicIdPartition> topicIdPartitions) {
+ List<ConsumerGroupHeartbeatRequestData.TopicPartitions> result = new ArrayList<>();
+ Map<Uuid, List<Integer>> partitionsPerTopicId = new HashMap<>();
+ for (TopicIdPartition topicIdPartition : topicIdPartitions) {
+ Uuid topicId = topicIdPartition.topicId();
+ if (!partitionsPerTopicId.containsKey(topicId)) {
+ partitionsPerTopicId.put(topicId, new ArrayList<>());
+ }
+ partitionsPerTopicId.get(topicId).add(topicIdPartition.partition());
Review Comment:
```suggestion
partitionsPerTopicId.computeIfAbsent(topicId, __ -> new ArrayList<>()).add(partitionId);
```
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/HeartbeatRequestManager.java:
##########
@@ -417,4 +394,127 @@ private void updateHeartbeatIntervalMs(final long heartbeatIntervalMs) {
this.heartbeatTimer.updateAndReset(heartbeatIntervalMs);
}
}
+
+ /**
+ * Builds the heartbeat requests correctly, ensuring that all information is sent according to
+ * the protocol, but subsequent requests do not send information which has not changed. This
+ * is important to ensure that reconciliation completes successfully.
+ */
+ static class HeartbeatState {
+ private final SubscriptionState subscriptions;
+ private final MembershipManager membershipManager;
+ private final int rebalanceTimeoutMs;
+
+ // Fields of ConsumerHeartbeatRequest sent in the most recent request
+ private String sentInstanceId;
+ private int sentRebalanceTimeoutMs;
+ private TreeSet<String> sentSubscribedTopicNames;
+ // private String sentSubscribedTopicRegex;
+ private String sentServerAssignor;
+ private TreeSet<String> sentTopicPartitions;
Review Comment:
IMO, we should group all of the 'last sent' attributes in a separate little struct/class, just to keep their relation and their (re-)initializing logic together.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/HeartbeatRequestManager.java:
##########
@@ -417,4 +394,127 @@ private void updateHeartbeatIntervalMs(final long heartbeatIntervalMs) {
this.heartbeatTimer.updateAndReset(heartbeatIntervalMs);
}
}
+
+ /**
+ * Builds the heartbeat requests correctly, ensuring that all information is sent according to
+ * the protocol, but subsequent requests do not send information which has not changed. This
+ * is important to ensure that reconciliation completes successfully.
+ */
+ static class HeartbeatState {
Review Comment:
Any reason `HeartbeatState` needs to be `static`?
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15890: Consumer.poll with long timeout unaware of assigned partitions [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #14835:
URL: https://github.com/apache/kafka/pull/14835#discussion_r1406853731
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/HeartbeatRequestManager.java:
##########
@@ -417,4 +394,127 @@ private void updateHeartbeatIntervalMs(final long heartbeatIntervalMs) {
this.heartbeatTimer.updateAndReset(heartbeatIntervalMs);
}
}
+
+ /**
+ * Builds the heartbeat requests correctly, ensuring that all information is sent according to
+ * the protocol, but subsequent requests do not send information which has not changed. This
+ * is important to ensure that reconciliation completes successfully.
+ */
+ static class HeartbeatState {
+ private final SubscriptionState subscriptions;
+ private final MembershipManager membershipManager;
+ private final int rebalanceTimeoutMs;
+
+ // Fields of ConsumerHeartbeatRequest sent in the most recent request
+ private String sentInstanceId;
+ private int sentRebalanceTimeoutMs;
+ private TreeSet<String> sentSubscribedTopicNames;
+ // private String sentSubscribedTopicRegex;
+ private String sentServerAssignor;
+ private TreeSet<String> sentTopicPartitions;
Review Comment:
Done
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15890: Consumer.poll with long timeout unaware of assigned partitions [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on PR #14835:
URL: https://github.com/apache/kafka/pull/14835#issuecomment-1828148906
Thanks for the changes @AndrewJSchofield! LGTM.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org