Skip to content

RATIS-2426. Fix memory leak in ServerRequestStreamObserver #1368

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
ivandika3 wants to merge 14 commits into
base: master
Choose a base branch
from
Open

RATIS-2426. Fix memory leak in ServerRequestStreamObserver #1368

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
14 commits
Select commit Hold shift + click to select a range
aba97ef
Call onError in GrpcLogAppender#stop to prevent StreamObserver leak i…
ivandika3 Mar 9, 2026
d5c3b2e
Set previousOnNext and requestFuture to null to prevent leak
ivandika3 Mar 9, 2026
b0db797
Handle complete or cancel based on the Event
ivandika3 Mar 9, 2026
e36fe7c
Fix checkstyle
ivandika3 Mar 9, 2026
2a371c9
Push logic inside StreamObservers to prevent NPE
ivandika3 Mar 9, 2026
53e346b
Tidy up
ivandika3 Mar 9, 2026
653984b
Call onCompleted in the finally block
ivandika3 Mar 10, 2026
1275af6
Lock before completing
ivandika3 Mar 11, 2026
1678938
Null previousOnSet in handleError
ivandika3 Mar 11, 2026
e346a22
Add test
ivandika3 Mar 11, 2026
538e9f2
Add leak-detection counter to validate handleError cleanup
ivandika3 Mar 11, 2026
ad38528
Improve the injection naming
ivandika3 Mar 12, 2026
196d195
Don't make GrpcServerProtocolService public
ivandika3 Mar 12, 2026
10fae38
CodeInjectionForTesting to return previous value to allow revert
ivandika3 Mar 12, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions ratis-common/src/main/java/org/apache/ratis/util/CodeInjectionForTesting.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -50,9 +50,9 @@ public interface Code {
= new ConcurrentHashMap<>();

/** Put an injection point. */
public static void put(String injectionPoint, Code code) {
public static Code put(String injectionPoint, Code code) {
LOG.debug("put: {}, {}", injectionPoint, code);
INJECTION_POINTS.put(injectionPoint, code);
return INJECTION_POINTS.put(injectionPoint, code);
}

/** Execute the injected code, if there is any. */
Expand Down
62 changes: 50 additions & 12 deletions ratis-grpc/src/main/java/org/apache/ratis/grpc/server/GrpcLogAppender.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,7 @@
import org.apache.ratis.server.protocol.TermIndex;
import org.apache.ratis.server.raftlog.RaftLog;
import org.apache.ratis.server.util.ServerStringUtils;
import org.apache.ratis.thirdparty.io.grpc.Status;
import org.apache.ratis.thirdparty.io.grpc.StatusRuntimeException;
import org.apache.ratis.thirdparty.io.grpc.stub.CallStreamObserver;
import org.apache.ratis.thirdparty.io.grpc.stub.StreamObserver;
Expand Down Expand Up @@ -215,7 +216,7 @@ private void resetClient(AppendEntriesRequest request, Event event) {
try (AutoCloseableLock writeLock = lock.writeLock(caller, LOG::trace)) {
getClient().resetConnectBackoff();
if (appendLogRequestObserver != null) {
appendLogRequestObserver.stop();
appendLogRequestObserver.stop(event);
appendLogRequestObserver = null;
}
final int errorCount = replyState.process(event);
Expand Down Expand Up @@ -266,16 +267,23 @@ private boolean installSnapshot() {

@Override
public void run() throws IOException {
for(; isRunning(); mayWait()) {
//HB period is expired OR we have messages OR follower is behind with commit index
if (shouldSendAppendEntries() || isFollowerCommitBehindLastCommitIndex()) {
final boolean installingSnapshot = installSnapshot();
appendLog(installingSnapshot || haveTooManyPendingRequests());
try {
for (; isRunning(); mayWait()) {
//HB period is expired OR we have messages OR follower is behind with commit index
if (shouldSendAppendEntries() || isFollowerCommitBehindLastCommitIndex()) {
final boolean installingSnapshot = installSnapshot();
appendLog(installingSnapshot || haveTooManyPendingRequests());
}
getLeaderState().checkHealth(getFollower());
}
} finally {
try (AutoCloseableLock writeLock = lock.writeLock(caller, LOG::trace)) {
if (appendLogRequestObserver != null) {
appendLogRequestObserver.onCompleted();
appendLogRequestObserver = null;
}
}
getLeaderState().checkHealth(getFollower());
}

Optional.ofNullable(appendLogRequestObserver).ifPresent(StreamObservers::onCompleted);
}

public long getWaitTimeMs() {
Expand Down Expand Up @@ -366,16 +374,46 @@ void onNext(AppendEntriesRequestProto proto)
while (!stream.isReady() && running) {
sleep(waitForReady, isHeartBeat);
}
if (!running) {
return;
}
stream.onNext(proto);
}

void stop() {
void stop(Event event) {
running = false;
if (event == Event.COMPLETE) {
onCompleted();
} else {
cancelStream("stop due to " + event);
Copy link
Copy Markdown
Contributor

@szetszwo szetszwo Apr 27, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We have the following call sequence:

  • AppendLogResponseHandler.onError(..) / onCompleted() -> resetClient(..) -> stop()

In the gRPC example,

  • onCompleted() implementation will response onCompleted() but
  • onError(Throwable t) only calls logger.log(Level.WARNING, ...)

So, I think stop() should not call cancelStream(..)

}
}

void onCompleted() {
appendLog.onCompleted();
Optional.ofNullable(heartbeat).ifPresent(StreamObserver::onCompleted);
try {
appendLog.onCompleted();
} catch (Exception e) {
LOG.debug("Failed to complete appendLog stream", e);
}
try {
Optional.ofNullable(heartbeat).ifPresent(StreamObserver::onCompleted);
} catch (Exception e) {
LOG.debug("Failed to complete heartbeat stream", e);
}
}

void cancelStream(String reason) {
try {
appendLog.onError(new StatusRuntimeException(Status.CANCELLED.withDescription(reason)));
} catch (Exception e) {
LOG.debug("Failed to cancel appendLog stream", e);
}
try {
Optional.ofNullable(heartbeat).ifPresent((hb) ->
hb.onError(new StatusRuntimeException(Status.CANCELLED.withDescription(reason))));
} catch (Exception e) {
LOG.debug("Failed to cancel heartbeat stream", e);
}
}
}

Expand Down
11 changes: 11 additions & 0 deletions ratis-grpc/src/main/java/org/apache/ratis/grpc/server/GrpcServerProtocolService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,8 @@
import org.apache.ratis.proto.RaftProtos.*;
import org.apache.ratis.proto.grpc.RaftServerProtocolServiceGrpc.RaftServerProtocolServiceImplBase;
import org.apache.ratis.util.BatchLogger;
import org.apache.ratis.util.CodeInjectionForTesting;
import org.apache.ratis.util.JavaUtils;
import org.apache.ratis.util.MemoizedSupplier;
import org.apache.ratis.util.ProtoUtils;
import org.slf4j.Logger;
Expand All @@ -46,6 +48,9 @@
class GrpcServerProtocolService extends RaftServerProtocolServiceImplBase {
public static final Logger LOG = LoggerFactory.getLogger(GrpcServerProtocolService.class);

public static final String GRPC_SERVER_HANDLE_ERROR =
JavaUtils.getClassSimpleName(GrpcServerProtocolService.class) + ".handleError";

private enum BatchLogKey implements BatchLogger.Key {
COMPLETED_REQUEST,
COMPLETED_REPLY
Expand Down Expand Up @@ -114,7 +119,9 @@ StatusRuntimeException wrapException(Throwable e, REQUEST request) {
private void handleError(Throwable e, REQUEST request) {
GrpcUtil.warn(LOG, () -> getId() + ": Failed " + op + " request " + requestToString(request), e);
if (isClosed.compareAndSet(false, true)) {
previousOnNext.set(null);
responseObserver.onError(wrapException(e, request));
CodeInjectionForTesting.execute(GRPC_SERVER_HANDLE_ERROR, getId(), null, previousOnNext.get());
}
}

Expand Down Expand Up @@ -172,18 +179,22 @@ public void onCompleted() {
BatchLogger.print(BatchLogKey.COMPLETED_REQUEST, getName(),
suffix -> LOG.info("{}: Completed {}, lastRequest: {} {}",
getId(), op, getPreviousRequestString(), suffix));
previousOnNext.set(null);
requestFuture.get().thenAccept(reply -> {
BatchLogger.print(BatchLogKey.COMPLETED_REPLY, getName(),
suffix -> LOG.info("{}: Completed {}, lastReply: {} {}",
getId(), op, ProtoUtils.shortDebugString(reply), suffix));
responseObserver.onCompleted();
});
requestFuture.set(null);
}
}
@Override
public void onError(Throwable t) {
GrpcUtil.warn(LOG, () -> getId() + ": "+ op + " onError, lastRequest: " + getPreviousRequestString(), t);
if (isClosed.compareAndSet(false, true)) {
previousOnNext.set(null);
requestFuture.set(null);
Status status = Status.fromThrowable(t);
if (status != null && status.getCode() != Status.Code.CANCELLED) {
responseObserver.onCompleted();
Expand Down
Loading
Loading