You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@ratis.apache.org by "symious (via GitHub)" <gi...@apache.org> on 2026/03/06 06:52:45 UTC
[PR] RATIS-2421. Gracefully cancel stream after complete in GrpcLogAppender [ratis]
symious opened a new pull request, #1363:
URL: https://github.com/apache/ratis/pull/1363
## What changes were proposed in this pull request?
When onCompleted() is called, the client waits for the server response to finish the RPC and release the underlying resources.
However, in some cases the server response may never arrive, which leaves the RPC stream open and prevents resources from being released.
This ticket introduces a client-side cancel() after a short grace period following onCompleted(), to ensure the RPC stream is forcefully closed and resources are cleaned up.
## What is the link to the Apache JIRA
https://issues.apache.org/jira/browse/RATIS-2421
## How was this patch tested?
Test locally.
--
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: issues-unsubscribe@ratis.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] RATIS-2421. Gracefully cancel stream after complete in GrpcLogAppender [ratis]
Posted by "spacemonkd (via GitHub)" <gi...@apache.org>.
spacemonkd commented on PR #1363:
URL: https://github.com/apache/ratis/pull/1363#issuecomment-5021281973
@szetszwo this looks good to me, should we merge it? Your review would be great
--
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: issues-unsubscribe@ratis.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] RATIS-2421. Gracefully cancel stream after complete in GrpcLogAppender [ratis]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #1363:
URL: https://github.com/apache/ratis/pull/1363#issuecomment-5746509049
This PR has been marked as stale due to 60 days of inactivity. Please comment or remove the stale label to keep it open. Otherwise, it will be automatically closed in ~30 days.
--
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: issues-unsubscribe@ratis.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] RATIS-2421. Gracefully cancel stream after complete in GrpcLogAppender [ratis]
Posted by "symious (via GitHub)" <gi...@apache.org>.
symious commented on code in PR #1363:
URL: https://github.com/apache/ratis/pull/1363#discussion_r2915428267
##########
ratis-grpc/src/main/java/org/apache/ratis/grpc/server/GrpcLogAppender.java:
##########
@@ -57,10 +58,8 @@
import java.util.Optional;
import java.util.Queue;
import java.util.UUID;
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.TimeUnit;
+import java.util.concurrent.*;
Review Comment:
Fixed.
##########
ratis-grpc/src/main/java/org/apache/ratis/grpc/server/GrpcLogAppender.java:
##########
@@ -341,20 +340,37 @@ private boolean haveTooManyPendingRequests() {
}
static class StreamObservers {
- private final CallStreamObserver<AppendEntriesRequestProto> appendLog;
- private final CallStreamObserver<AppendEntriesRequestProto> heartbeat;
+ private final ClientCallStreamObserver<AppendEntriesRequestProto> appendLog;
+ private final ClientCallStreamObserver<AppendEntriesRequestProto> heartbeat;
private final TimeDuration waitForReady;
+ private final TimeDuration completeGracePeriod;
private volatile boolean running = true;
+ private final ScheduledExecutorService closer =
+ Executors.newSingleThreadScheduledExecutor(r -> {
+ Thread t = new Thread(r, "grpc-log-appender-stream-closer");
+ t.setDaemon(true);
+ return t;
+ });
+
+// private final AtomicBoolean closing = new AtomicBoolean(false);
Review Comment:
Fixed.
--
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: issues-unsubscribe@ratis.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] RATIS-2421. Gracefully cancel stream after complete in GrpcLogAppender [ratis]
Posted by "symious (via GitHub)" <gi...@apache.org>.
symious commented on code in PR #1363:
URL: https://github.com/apache/ratis/pull/1363#discussion_r2915437023
##########
ratis-grpc/src/main/java/org/apache/ratis/grpc/server/GrpcLogAppender.java:
##########
@@ -366,16 +382,73 @@ void onNext(AppendEntriesRequestProto proto)
while (!stream.isReady() && running) {
sleep(waitForReady, isHeartBeat);
}
- stream.onNext(proto);
+ try {
+ stream.onNext(proto);
+ } catch (Exception e) {
+ InterruptedIOException ioe =
+ new InterruptedIOException("Failed to send request via stream");
+ ioe.initCause(e);
+ throw ioe;
+ }
}
void stop() {
running = false;
}
void onCompleted() {
- appendLog.onCompleted();
- Optional.ofNullable(heartbeat).ifPresent(StreamObserver::onCompleted);
+ if (completed.compareAndSet(false, true)) {
+ completeStreamGracefully(appendLog, "appendLog");
+ Optional.ofNullable(heartbeat)
+ .ifPresent(s -> completeStreamGracefully(s, "heartbeat"));
+ }
+ final long delayMs = Math.max(1L, completeGracePeriod.toLong(TimeUnit.MILLISECONDS));
+ closer.schedule(this::cancelIfStillNeeded, delayMs, TimeUnit.MILLISECONDS);
+ }
+
+ void cancelNow(String reason, Throwable cause) {
+ if (cancelled.compareAndSet(false, true)) {
+ running = false;
+ cancelStream(appendLog, "appendLog", reason, cause);
+ Optional.ofNullable(heartbeat)
+ .ifPresent(s -> cancelStream(s, "heartbeat", reason, cause));
+ shutdownCloser();
+ }
+ }
+
+ private void cancelIfStillNeeded() {
+ if (cancelled.compareAndSet(false, true)) {
+ cancelStream(appendLog, "appendLog", "Stream completion timeout", null);
+ Optional.ofNullable(heartbeat)
+ .ifPresent(s -> cancelStream(s, "heartbeat", "Stream completion timeout", null));
+ }
+ shutdownCloser();
+ }
+
+ private void completeStreamGracefully(
+ ClientCallStreamObserver<AppendEntriesRequestProto> stream,
+ String name) {
+ try {
+ stream.onCompleted();
+ } catch (Exception e) {
+ LOG.warn("Failed to call onCompleted on {}", name, e);
+ }
+ }
+
+ private void cancelStream(
+ ClientCallStreamObserver<AppendEntriesRequestProto> stream,
Review Comment:
As a client, this one should always be a ClientCallStreamObserver.
--
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: issues-unsubscribe@ratis.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] RATIS-2421. Gracefully cancel stream after complete in GrpcLogAppender [ratis]
Posted by "devabhishekpal (via GitHub)" <gi...@apache.org>.
devabhishekpal commented on code in PR #1363:
URL: https://github.com/apache/ratis/pull/1363#discussion_r2914667294
##########
ratis-grpc/src/main/java/org/apache/ratis/grpc/server/GrpcLogAppender.java:
##########
@@ -366,16 +382,73 @@ void onNext(AppendEntriesRequestProto proto)
while (!stream.isReady() && running) {
sleep(waitForReady, isHeartBeat);
}
- stream.onNext(proto);
+ try {
+ stream.onNext(proto);
+ } catch (Exception e) {
+ InterruptedIOException ioe =
+ new InterruptedIOException("Failed to send request via stream");
+ ioe.initCause(e);
+ throw ioe;
+ }
}
void stop() {
running = false;
}
void onCompleted() {
- appendLog.onCompleted();
- Optional.ofNullable(heartbeat).ifPresent(StreamObserver::onCompleted);
+ if (completed.compareAndSet(false, true)) {
+ completeStreamGracefully(appendLog, "appendLog");
+ Optional.ofNullable(heartbeat)
+ .ifPresent(s -> completeStreamGracefully(s, "heartbeat"));
+ }
+ final long delayMs = Math.max(1L, completeGracePeriod.toLong(TimeUnit.MILLISECONDS));
+ closer.schedule(this::cancelIfStillNeeded, delayMs, TimeUnit.MILLISECONDS);
+ }
+
+ void cancelNow(String reason, Throwable cause) {
+ if (cancelled.compareAndSet(false, true)) {
+ running = false;
+ cancelStream(appendLog, "appendLog", reason, cause);
+ Optional.ofNullable(heartbeat)
+ .ifPresent(s -> cancelStream(s, "heartbeat", reason, cause));
+ shutdownCloser();
+ }
+ }
+
+ private void cancelIfStillNeeded() {
+ if (cancelled.compareAndSet(false, true)) {
+ cancelStream(appendLog, "appendLog", "Stream completion timeout", null);
+ Optional.ofNullable(heartbeat)
+ .ifPresent(s -> cancelStream(s, "heartbeat", "Stream completion timeout", null));
+ }
+ shutdownCloser();
+ }
+
+ private void completeStreamGracefully(
+ ClientCallStreamObserver<AppendEntriesRequestProto> stream,
+ String name) {
+ try {
+ stream.onCompleted();
+ } catch (Exception e) {
+ LOG.warn("Failed to call onCompleted on {}", name, e);
+ }
+ }
+
+ private void cancelStream(
+ ClientCallStreamObserver<AppendEntriesRequestProto> stream,
Review Comment:
We are doing explicit cast to `ClientCallStreamObserver<>`, but I am thinking can there be a case where `appendEntries()` can return only `CallStreamObserver`?
That is the method signature in GrpcServerProtocolClient.
In such a case this can throw error.
Better to accept CallStreamObserver<> and then do an instanceof check?
```
try {
if (stream instanceof ClientCallStreamObserver) {
((ClientCallStreamObserver<AppendEntriesRequestProto>) stream).cancel(reason, cause);
} else {
...handle cancel when it is not clientcallstreamobserver, something like stream.onError(...)?
}
} catch (Exception e) {
LOG.warn("Failed to cancel {}", name, e);
}
```
--
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: issues-unsubscribe@ratis.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] RATIS-2421. Gracefully cancel stream after complete in GrpcLogAppender [ratis]
Posted by "devabhishekpal (via GitHub)" <gi...@apache.org>.
devabhishekpal commented on code in PR #1363:
URL: https://github.com/apache/ratis/pull/1363#discussion_r2914612256
##########
ratis-grpc/src/main/java/org/apache/ratis/grpc/server/GrpcLogAppender.java:
##########
@@ -57,10 +58,8 @@
import java.util.Optional;
import java.util.Queue;
import java.util.UUID;
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.TimeUnit;
+import java.util.concurrent.*;
Review Comment:
Is a wildcard import required? I would prefer if we can import specific classes only.
--
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: issues-unsubscribe@ratis.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] RATIS-2421. Gracefully cancel stream after complete in GrpcLogAppender [ratis]
Posted by "devabhishekpal (via GitHub)" <gi...@apache.org>.
devabhishekpal commented on code in PR #1363:
URL: https://github.com/apache/ratis/pull/1363#discussion_r2914633600
##########
ratis-grpc/src/main/java/org/apache/ratis/grpc/server/GrpcLogAppender.java:
##########
@@ -366,16 +382,73 @@ void onNext(AppendEntriesRequestProto proto)
while (!stream.isReady() && running) {
sleep(waitForReady, isHeartBeat);
}
- stream.onNext(proto);
+ try {
+ stream.onNext(proto);
+ } catch (Exception e) {
+ InterruptedIOException ioe =
+ new InterruptedIOException("Failed to send request via stream");
+ ioe.initCause(e);
+ throw ioe;
+ }
}
void stop() {
running = false;
}
void onCompleted() {
- appendLog.onCompleted();
- Optional.ofNullable(heartbeat).ifPresent(StreamObserver::onCompleted);
+ if (completed.compareAndSet(false, true)) {
+ completeStreamGracefully(appendLog, "appendLog");
+ Optional.ofNullable(heartbeat)
+ .ifPresent(s -> completeStreamGracefully(s, "heartbeat"));
+ }
+ final long delayMs = Math.max(1L, completeGracePeriod.toLong(TimeUnit.MILLISECONDS));
+ closer.schedule(this::cancelIfStillNeeded, delayMs, TimeUnit.MILLISECONDS);
Review Comment:
This schedules `cancelIfStillNeeded()` outside the `completed.compareAndSet(...)` block. Repeated onCompleted() calls can queue redundant cancel tasks.
Can this be avoided?
Also I think if it’s invoked again after `shutdownCloser()` it may throw exception.
--
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: issues-unsubscribe@ratis.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] RATIS-2421. Gracefully cancel stream after complete in GrpcLogAppender [ratis]
Posted by "symious (via GitHub)" <gi...@apache.org>.
symious commented on code in PR #1363:
URL: https://github.com/apache/ratis/pull/1363#discussion_r2916251639
##########
ratis-grpc/src/main/java/org/apache/ratis/grpc/server/GrpcLogAppender.java:
##########
@@ -366,16 +382,73 @@ void onNext(AppendEntriesRequestProto proto)
while (!stream.isReady() && running) {
sleep(waitForReady, isHeartBeat);
}
- stream.onNext(proto);
+ try {
+ stream.onNext(proto);
+ } catch (Exception e) {
+ InterruptedIOException ioe =
+ new InterruptedIOException("Failed to send request via stream");
+ ioe.initCause(e);
+ throw ioe;
+ }
}
void stop() {
running = false;
}
void onCompleted() {
- appendLog.onCompleted();
- Optional.ofNullable(heartbeat).ifPresent(StreamObserver::onCompleted);
+ if (completed.compareAndSet(false, true)) {
+ completeStreamGracefully(appendLog, "appendLog");
+ Optional.ofNullable(heartbeat)
+ .ifPresent(s -> completeStreamGracefully(s, "heartbeat"));
+ }
+ final long delayMs = Math.max(1L, completeGracePeriod.toLong(TimeUnit.MILLISECONDS));
+ closer.schedule(this::cancelIfStillNeeded, delayMs, TimeUnit.MILLISECONDS);
Review Comment:
If onComplete() is invoked, the scheudler should be called and shutdown later, later requests should be rejected then.
--
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: issues-unsubscribe@ratis.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] RATIS-2421. Gracefully cancel stream after complete in GrpcLogAppender [ratis]
Posted by "spacemonkd (via GitHub)" <gi...@apache.org>.
spacemonkd commented on PR #1363:
URL: https://github.com/apache/ratis/pull/1363#issuecomment-4605037709
@szetszwo could you take a look at this 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: issues-unsubscribe@ratis.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] RATIS-2421. Gracefully cancel stream after complete in GrpcLogAppender [ratis]
Posted by "devabhishekpal (via GitHub)" <gi...@apache.org>.
devabhishekpal commented on code in PR #1363:
URL: https://github.com/apache/ratis/pull/1363#discussion_r2914613468
##########
ratis-grpc/src/main/java/org/apache/ratis/grpc/server/GrpcLogAppender.java:
##########
@@ -341,20 +340,37 @@ private boolean haveTooManyPendingRequests() {
}
static class StreamObservers {
- private final CallStreamObserver<AppendEntriesRequestProto> appendLog;
- private final CallStreamObserver<AppendEntriesRequestProto> heartbeat;
+ private final ClientCallStreamObserver<AppendEntriesRequestProto> appendLog;
+ private final ClientCallStreamObserver<AppendEntriesRequestProto> heartbeat;
private final TimeDuration waitForReady;
+ private final TimeDuration completeGracePeriod;
private volatile boolean running = true;
+ private final ScheduledExecutorService closer =
+ Executors.newSingleThreadScheduledExecutor(r -> {
+ Thread t = new Thread(r, "grpc-log-appender-stream-closer");
+ t.setDaemon(true);
+ return t;
+ });
+
+// private final AtomicBoolean closing = new AtomicBoolean(false);
Review Comment:
nit: can be removed?
--
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: issues-unsubscribe@ratis.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] RATIS-2421. Gracefully cancel stream after complete in GrpcLogAppender [ratis]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #1363:
URL: https://github.com/apache/ratis/pull/1363#issuecomment-4468705715
This PR has been marked as stale due to 60 days of inactivity. Please comment or remove the stale label to keep it open. Otherwise, it will be automatically closed in ~30 days.
--
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: issues-unsubscribe@ratis.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] RATIS-2421. Gracefully cancel stream after complete in GrpcLogAppender [ratis]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #1363:
URL: https://github.com/apache/ratis/pull/1363#issuecomment-6391552216
Thank you for your contribution. This PR is being closed due to inactivity. Please contact a maintainer if you would like to reopen it.
--
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: issues-unsubscribe@ratis.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] RATIS-2421. Gracefully cancel stream after complete in GrpcLogAppender [ratis]
Posted by "devabhishekpal (via GitHub)" <gi...@apache.org>.
devabhishekpal commented on code in PR #1363:
URL: https://github.com/apache/ratis/pull/1363#discussion_r2916235511
##########
ratis-grpc/src/main/java/org/apache/ratis/grpc/server/GrpcLogAppender.java:
##########
@@ -366,16 +382,73 @@ void onNext(AppendEntriesRequestProto proto)
while (!stream.isReady() && running) {
sleep(waitForReady, isHeartBeat);
}
- stream.onNext(proto);
+ try {
+ stream.onNext(proto);
+ } catch (Exception e) {
+ InterruptedIOException ioe =
+ new InterruptedIOException("Failed to send request via stream");
+ ioe.initCause(e);
+ throw ioe;
+ }
}
void stop() {
running = false;
}
void onCompleted() {
- appendLog.onCompleted();
- Optional.ofNullable(heartbeat).ifPresent(StreamObserver::onCompleted);
+ if (completed.compareAndSet(false, true)) {
+ completeStreamGracefully(appendLog, "appendLog");
+ Optional.ofNullable(heartbeat)
+ .ifPresent(s -> completeStreamGracefully(s, "heartbeat"));
+ }
+ final long delayMs = Math.max(1L, completeGracePeriod.toLong(TimeUnit.MILLISECONDS));
+ closer.schedule(this::cancelIfStillNeeded, delayMs, TimeUnit.MILLISECONDS);
Review Comment:
But if it is completed already aren't subsequent cancel calls queuing tasks which are no longer required?
--
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: issues-unsubscribe@ratis.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] RATIS-2421. Gracefully cancel stream after complete in GrpcLogAppender [ratis]
Posted by "symious (via GitHub)" <gi...@apache.org>.
symious commented on code in PR #1363:
URL: https://github.com/apache/ratis/pull/1363#discussion_r2915428748
##########
ratis-grpc/src/main/java/org/apache/ratis/grpc/server/GrpcLogAppender.java:
##########
@@ -366,16 +382,73 @@ void onNext(AppendEntriesRequestProto proto)
while (!stream.isReady() && running) {
sleep(waitForReady, isHeartBeat);
}
- stream.onNext(proto);
+ try {
+ stream.onNext(proto);
+ } catch (Exception e) {
+ InterruptedIOException ioe =
+ new InterruptedIOException("Failed to send request via stream");
+ ioe.initCause(e);
+ throw ioe;
+ }
}
void stop() {
running = false;
}
void onCompleted() {
- appendLog.onCompleted();
- Optional.ofNullable(heartbeat).ifPresent(StreamObserver::onCompleted);
+ if (completed.compareAndSet(false, true)) {
+ completeStreamGracefully(appendLog, "appendLog");
+ Optional.ofNullable(heartbeat)
+ .ifPresent(s -> completeStreamGracefully(s, "heartbeat"));
+ }
+ final long delayMs = Math.max(1L, completeGracePeriod.toLong(TimeUnit.MILLISECONDS));
+ closer.schedule(this::cancelIfStillNeeded, delayMs, TimeUnit.MILLISECONDS);
Review Comment:
The cancel and shutdown parts can be called multiple times.
--
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: issues-unsubscribe@ratis.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] RATIS-2421. Gracefully cancel stream after complete in GrpcLogAppender [ratis]
Posted by "symious (via GitHub)" <gi...@apache.org>.
symious commented on PR #1363:
URL: https://github.com/apache/ratis/pull/1363#issuecomment-4035670474
@devabhishekpal Thank you for the review. Updated and PTAL.
--
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: issues-unsubscribe@ratis.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] RATIS-2421. Gracefully cancel stream after complete in GrpcLogAppender [ratis]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] closed pull request #1363: RATIS-2421. Gracefully cancel stream after complete in GrpcLogAppender
URL: https://github.com/apache/ratis/pull/1363
--
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: issues-unsubscribe@ratis.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org