You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "ShivsundarR (via GitHub)" <gi...@apache.org> on 2024/09/09 14:26:16 UTC
[PR] KAFKA-17502: Modified commitSync() and close() handling in clients [kafka]
ShivsundarR opened a new pull request, #17136:
URL: https://github.com/apache/kafka/pull/17136
*What*
Currently the code in `ShareConsumeRequestManager` works on the basis that there can only be one `commitSync()`/`close()` at a time. But there is a chance these calls timeout on the application thread, but are still sent later on the background thread. This will mean the incoming commitSync()/close() will not be processed, resulting in possible loss of acknowledgements.
To cover this case, we will now have a list of `AcknowledgeRequestStates` to store the `commitSyncs()` and a separate requestState to store the `close()`. This queue will be processed one by one until its empty. For `close()`, we are still assuming there can only be one active `close()` at a time.
--
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-17502: Modified commitSync() and close() handling in clients [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #17136:
URL: https://github.com/apache/kafka/pull/17136#discussion_r1755099465
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareConsumeRequestManager.java:
##########
@@ -1011,33 +1036,47 @@ public void completeIfEmpty() {
}
}
- static class Pair<V> {
+ static class Tuple<V> {
private V asyncRequest;
- private V syncRequest;
+ private Queue<V> syncRequestQueue;
+ private V closeRequest;
- public Pair(V asyncRequest, V syncRequest) {
+ public Tuple(V asyncRequest, Queue<V> syncRequestQueue, V closeRequest) {
this.asyncRequest = asyncRequest;
- this.syncRequest = syncRequest;
+ this.syncRequestQueue = syncRequestQueue;
+ this.closeRequest = closeRequest;
}
public void setAsyncRequest(V asyncRequest) {
this.asyncRequest = asyncRequest;
}
- public void setSyncRequest(V second) {
- this.syncRequest = second;
+ public void setSyncRequestQueue(Queue<V> syncRequestQueue) {
Review Comment:
That's fine. It just seemed weird forcing the caller to make and set the queue.
--
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-17502: Modified commitSync() and close() handling in clients [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #17136:
URL: https://github.com/apache/kafka/pull/17136#discussion_r1754436434
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareConsumeRequestManager.java:
##########
@@ -84,10 +86,9 @@ public class ShareConsumeRequestManager implements RequestManager, MemberStateLi
private Uuid memberId;
private boolean fetchMoreRecords = false;
private final Map<TopicIdPartition, Acknowledgements> fetchAcknowledgementsMap;
- private final Map<Integer, Pair<AcknowledgeRequestState>> acknowledgeRequestStates;
+ private final Map<Integer, Tuple<AcknowledgeRequestState>> acknowledgeRequestStates;
private final long retryBackoffMs;
private final long retryBackoffMaxMs;
- private boolean closing = false;
private final CompletableFuture<Void> closeFuture;
Review Comment:
It seems to me that the handling of `closeFuture` is flawed. An `AcknowledgeRequestState` is node-specific, but the future is associated with the entire request manager. It's only valid to complete the future once all close acks have been completed, not just the first one.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareConsumeRequestManager.java:
##########
@@ -1011,33 +1036,47 @@ public void completeIfEmpty() {
}
}
- static class Pair<V> {
+ static class Tuple<V> {
private V asyncRequest;
- private V syncRequest;
+ private Queue<V> syncRequestQueue;
+ private V closeRequest;
- public Pair(V asyncRequest, V syncRequest) {
+ public Tuple(V asyncRequest, Queue<V> syncRequestQueue, V closeRequest) {
this.asyncRequest = asyncRequest;
- this.syncRequest = syncRequest;
+ this.syncRequestQueue = syncRequestQueue;
+ this.closeRequest = closeRequest;
}
public void setAsyncRequest(V asyncRequest) {
this.asyncRequest = asyncRequest;
}
- public void setSyncRequest(V second) {
- this.syncRequest = second;
+ public void setSyncRequestQueue(Queue<V> syncRequestQueue) {
Review Comment:
You could encapsulate this queue entirely in this class. For example, `addSyncRequest` could create the queue if it's not yet been created.
--
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-17502: Modified commitSync() and close() handling in clients [kafka]
Posted by "omkreddy (via GitHub)" <gi...@apache.org>.
omkreddy merged PR #17136:
URL: https://github.com/apache/kafka/pull/17136
--
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-17502: Modified commitSync() and close() handling in clients [kafka]
Posted by "ShivsundarR (via GitHub)" <gi...@apache.org>.
ShivsundarR commented on code in PR #17136:
URL: https://github.com/apache/kafka/pull/17136#discussion_r1754897975
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareConsumeRequestManager.java:
##########
@@ -1011,33 +1036,47 @@ public void completeIfEmpty() {
}
}
- static class Pair<V> {
+ static class Tuple<V> {
private V asyncRequest;
- private V syncRequest;
+ private Queue<V> syncRequestQueue;
+ private V closeRequest;
- public Pair(V asyncRequest, V syncRequest) {
+ public Tuple(V asyncRequest, Queue<V> syncRequestQueue, V closeRequest) {
this.asyncRequest = asyncRequest;
- this.syncRequest = syncRequest;
+ this.syncRequestQueue = syncRequestQueue;
+ this.closeRequest = closeRequest;
}
public void setAsyncRequest(V asyncRequest) {
this.asyncRequest = asyncRequest;
}
- public void setSyncRequest(V second) {
- this.syncRequest = second;
+ public void setSyncRequestQueue(Queue<V> syncRequestQueue) {
Review Comment:
Yeah makes sense, I have updated the addSyncRequest() method. Still the getSyncRequestQueue() method exists though as we need the queue contents when we process the acknowledgements. Here will renaming the method work, or is it fine as it is?
--
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-17502: Modified commitSync() and close() handling in clients [kafka]
Posted by "ShivsundarR (via GitHub)" <gi...@apache.org>.
ShivsundarR commented on code in PR #17136:
URL: https://github.com/apache/kafka/pull/17136#discussion_r1754905117
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareConsumeRequestManager.java:
##########
@@ -84,10 +86,9 @@ public class ShareConsumeRequestManager implements RequestManager, MemberStateLi
private Uuid memberId;
private boolean fetchMoreRecords = false;
private final Map<TopicIdPartition, Acknowledgements> fetchAcknowledgementsMap;
- private final Map<Integer, Pair<AcknowledgeRequestState>> acknowledgeRequestStates;
+ private final Map<Integer, Tuple<AcknowledgeRequestState>> acknowledgeRequestStates;
private final long retryBackoffMs;
private final long retryBackoffMaxMs;
- private boolean closing = false;
private final CompletableFuture<Void> closeFuture;
Review Comment:
Yes, thanks, I had missed this. I am now put an extra variable to indicate if a requestState has completed processing, and then accordingly waited till all the nodes are finished and then completed the `closeFuture`.
--
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-17502: Modified commitSync() and close() handling in clients [kafka]
Posted by "ShivsundarR (via GitHub)" <gi...@apache.org>.
ShivsundarR commented on code in PR #17136:
URL: https://github.com/apache/kafka/pull/17136#discussion_r1754897975
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareConsumeRequestManager.java:
##########
@@ -1011,33 +1036,47 @@ public void completeIfEmpty() {
}
}
- static class Pair<V> {
+ static class Tuple<V> {
private V asyncRequest;
- private V syncRequest;
+ private Queue<V> syncRequestQueue;
+ private V closeRequest;
- public Pair(V asyncRequest, V syncRequest) {
+ public Tuple(V asyncRequest, Queue<V> syncRequestQueue, V closeRequest) {
this.asyncRequest = asyncRequest;
- this.syncRequest = syncRequest;
+ this.syncRequestQueue = syncRequestQueue;
+ this.closeRequest = closeRequest;
}
public void setAsyncRequest(V asyncRequest) {
this.asyncRequest = asyncRequest;
}
- public void setSyncRequest(V second) {
- this.syncRequest = second;
+ public void setSyncRequestQueue(Queue<V> syncRequestQueue) {
Review Comment:
Yeah makes sense, I have updated the addSyncRequest() method. Still the getSyncRequestQueue() method exists though as we need the queue contents when we process the acknowledgements.
--
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-17502: Modified commitSync() and close() handling in clients [kafka]
Posted by "ShivsundarR (via GitHub)" <gi...@apache.org>.
ShivsundarR closed pull request #17136: KAFKA-17502: Modified commitSync() and close() handling in clients
URL: https://github.com/apache/kafka/pull/17136
--
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