[Bug] grpc SubStreamHandler exception

[9997766]

Search before asking

• I had searched in the issues and found no similar issues.

Environment

Linux

EventMesh version

1.10.0

What happened

There is a problem with the class of SubStreamHandler. grpc is two-way communication. I have integrated grpc pub and sub functions in the project.
This code throws an exception, and every 30 seconds,

private void senderOnNext(final CloudEvent subscription) {
try {
synchronized (sender) {
sender.onNext(subscription);
}
} catch (Exception e) {
log.error("StreamObserver Error onNext", e);
}
}

I analyzed the reason, because when the exception is thrown, you should call onCompleted to close the sender, but that is not done here, resulting in
grpc thinks the connection is still there and keeps throwing exceptions in a loop.

I tried the method below，it solved the problem。
Will you change code quickly？ I am using the project，but this problem stop me。

@slf4j
public class SubStreamHandler extends Thread implements Serializable {

...
public SubStreamHandler(final ConsumerServiceStub consumerAsyncClient, final EventMeshGrpcClientConfig clientConfig,
final ReceiveMsgHook listener) {
this.consumerAsyncClient = consumerAsyncClient;
this.clientConfig = clientConfig;
this.listener = listener;
}

public void sendSubscription(final CloudEvent subscription) {
    synchronized (this) {
        if (this.sender == null) {
            this.sender = consumerAsyncClient.subscribeStream(createReceiver());
        }
    }
    senderOnNext(subscription);
}

private StreamObserver<CloudEvent> createReceiver() {
    return new StreamObserver<CloudEvent>() {

        @Override
        public void onNext(final CloudEvent message) {
            T msg = EventMeshCloudEventBuilder.buildMessageFromEventMeshCloudEvent(message, listener.getProtocolType());
            if (msg instanceof Set) {
                log.info("Received message from Server:{}", message);
            } else {
                log.info("Received message from Server.|seq={}|uniqueId={}|",
                        EventMeshCloudEventUtils.getSeqNum(message), EventMeshCloudEventUtils.getUniqueId(message));
                CloudEvent streamReply = null;
                try {
                    Optional<T> reply = listener.handle(msg);
                    if (reply.isPresent()) {
                        streamReply = buildReplyMessage(message, reply.get());
                    }
                } catch (Exception e) {
                    log.error("Error in handling reply message.|seq={}|uniqueId={}|",
                            EventMeshCloudEventUtils.getSeqNum(message), EventMeshCloudEventUtils.getUniqueId(message), e);
                }
                if (streamReply != null) {
                    log.info("Sending reply message to Server.|seq={}|uniqueId={}|",
                            EventMeshCloudEventUtils.getSeqNum(streamReply), EventMeshCloudEventUtils.getUniqueId(streamReply));
                    senderOnNext(streamReply);
                }
            }
        }

        @Override
        public void onError(final Throwable t) {
            log.error("Received Server side error", t);
            **close();**
        }

        @Override
        public void onCompleted() {
            log.info("Finished receiving messages from server.");
            close();
        }
    };
}

private CloudEvent buildReplyMessage(final CloudEvent reqMessage, final T replyMessage) {
    final CloudEvent cloudEvent = EventMeshCloudEventBuilder.buildEventMeshCloudEvent(replyMessage,
            clientConfig, listener.getProtocolType());

    return CloudEvent.newBuilder(cloudEvent).putAllAttributes(reqMessage.getAttributesMap()).putAllAttributes(cloudEvent.getAttributesMap())
            .putAttributes(ProtocolKey.DATA_CONTENT_TYPE,
                    CloudEventAttributeValue.newBuilder().setCeString(EventMeshDataContentType.JSON.getCode()).build())
            // Indicate that it is a subscription response
            .putAttributes(ProtocolKey.SUB_MESSAGE_TYPE, CloudEventAttributeValue.newBuilder().setCeString(ProtocolKey.SUB_REPLY_MESSAGE).build())
            .build();
}

@Override
public void run() {
    try {
        latch.await();
    } catch (InterruptedException e) {
        log.error("SubStreamHandler Thread interrupted", e);
        Thread.currentThread().interrupt();
    }
}

public void close() {
    if (this.sender != null) {
        senderOnComplete();
    }

    latch.countDown();

    log.info("SubStreamHandler closed.");
}

private void senderOnNext(final CloudEvent subscription) {
    try {
        synchronized (sender) {
            sender.onNext(subscription);
        }
    } catch (Exception e) {
        log.error("StreamObserver Error onNext", e);
        **close();**
    }
}

private void senderOnComplete() {
    try {
        synchronized (sender) {
            sender.onCompleted();
            **sender=null;**
        }
    } catch (Exception e) {
        log.error("StreamObserver Error onComplete", e);
    }
}

}

How to reproduce

when exception happen ，this method will cause endless loop。
for example，grpc time out。
private void senderOnNext(final CloudEvent subscription) {
try {
synchronized (sender) {
sender.onNext(subscription);
}
} catch (Exception e) {
log.error("StreamObserver Error onNext", e);
}
}

Debug logs

no

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct *

[9997766]

when will this problem be solved?

[Pil0tXia]

Since you checked the Yes I am willing to submit a PR! box, we assumed you would submit a PR to fix the problem. I will try to reproduce this issue.

[Pil0tXia]

After calling the close() method, the SubStreamHandler instance will never run again unless the EventMeshGrpcConsumer re-subscribes.

@9997766 Do you have an idea for this issue?

[9997766]

NO， reply as follws At 2024-04-02 14:35:11, "Pil0tXia" ***@***.***> wrote: If you set the sender to null in the senderOnComplete() method, each subsequent message will require the sender to be recreated in sendSubscription(), increasing the performance overhead. 1、 if dont set sender to null, you will receive a halfclose exception; 2、 and， there is still a endless exception problem; After calling the close() method, the SubStreamHandler instance will never run again unless the EventMeshGrpcConsumer is restarted, which is also a problem. 3、so this is another problem，i think you will find a method to resolve it; Do you have a better idea please? — Reply to this email directly, view it on GitHub, or unsubscribe. You are receiving this because you authored the thread.Message ID: ***@***.***>

[9997766]

ok, it is easy to reproduce this problem; you can run in debug mode,and put the point on the receive logic, hold it, stop it to run, about 1 minute ,release it ,let it run continuely, it will show the problem； it will show a endless exception，and print the exception about every 30s； the code i show in the issue，have already modify the problem； At 2024-04-01 17:21:06, "Pil0tXia" ***@***.***> wrote: Since you checked the Yes I am willing to submit a PR! box, we assumed you would submit a PR to fix the problem. I will try to reproduce this issue. — Reply to this email directly, view it on GitHub, or unsubscribe. You are receiving this because you authored the thread.Message ID: ***@***.***>

[Pil0tXia]

@9997766 Thanks. You may check PR #4807.

[9997766]

Linux

I readed the code，you solved the problem。 But the main branche, do not have the 4807 code yet。 when will 4807 be merged to the main branch?

[Pil0tXia]

@9997766 We need another reviewer to merge this PR.

@pandaapo May you please help?

[9997766]

no problem. but i am not so familar with developping with github。you submit a merge request to me ?

[9997766]

and i find another problem
i use nginx to loadbalance the grpc ports on 2 eventmeshs, the jdk in the application always will report the follwing error.
it seems eventmesh have reset the grpc connection, and the client receive this error;
I'm not sure about the specific impact at the moment, but I keep encountering this error, which is causing some doubts about the stability of the EventMesh application system.

2024-04-10 17:33:49.374 ERROR 18488 [] [grpc-default-executor-0] org.apache.eventmesh.client.grpc.consumer.SubStreamHandler Received Server side error

io.grpc.StatusRuntimeException: INTERNAL: RST_STREAM closed stream. HTTP/2 error code: INTERNAL_ERROR
at io.grpc.Status.asRuntimeException(Status.java:535)
at io.grpc.stub.ClientCalls$StreamObserverToCallListenerAdapter.onClose(ClientCalls.java:479)
at io.grpc.internal.DelayedClientCall$DelayedListener$3.run(DelayedClientCall.java:463)
at io.grpc.internal.DelayedClientCall$DelayedListener.delayOrExecute(DelayedClientCall.java:427)
at io.grpc.internal.DelayedClientCall$DelayedListener.onClose(DelayedClientCall.java:460)
at io.grpc.internal.ClientCallImpl.closeObserver(ClientCallImpl.java:562)
at io.grpc.internal.ClientCallImpl.access$300(ClientCallImpl.java:70)
at io.grpc.internal.ClientCallImpl$ClientStreamListenerImpl$1StreamClosed.runInternal(ClientCallImpl.java:743)
at io.grpc.internal.ClientCallImpl$ClientStreamListenerImpl$1StreamClosed.runInContext(ClientCallImpl.java:722)
at io.grpc.internal.ContextRunnable.run(ContextRunnable.java:37)
at io.grpc.internal.SerializingExecutor.run(SerializingExecutor.java:133)
at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1144)
at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:642)
at java.base/java.lang.Thread.run(Thread.java:1583)

[Pil0tXia]

@9997766

i am not so familar with developping with github

You may refer to this guide: https://eventmesh.apache.org/community/contribute/contribute/

We need another reviewer to merge this PR.

I mean that at least two committers are required to merge a PR, and #4807 is waiting for one more committer's approval to be merged into master.

and i find another problem

It is better to submit this problem to a new issue, since merging #4807 will close this issue and lose tracking.

It seems that this log is reported by SDK, and you have to check Runtime for detailed reason.