You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@eventmesh.apache.org by "9997766 (via GitHub)" <gi...@apache.org> on 2024/03/27 02:47:27 UTC
[I] [Bug]grpc SubStreamHandler exception (eventmesh)
9997766 opened a new issue, #4804:
URL: https://github.com/apache/eventmesh/issues/4804
### Search before asking
- [X] I had searched in the [issues](https://github.com/apache/eventmesh/issues?q=is%3Aissue) 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<T> extends Thread implements Serializable {
...
public SubStreamHandler(final ConsumerServiceStub consumerAsyncClient, final EventMeshGrpcClientConfig clientConfig,
final ReceiveMsgHook<T> 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
```Java
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](https://www.apache.org/foundation/policies/conduct) *
--
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@eventmesh.apache.org.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@eventmesh.apache.org
For additional commands, e-mail: issues-help@eventmesh.apache.org
Re: [I] [Bug]grpc SubStreamHandler exception (eventmesh)
Posted by "Pil0tXia (via GitHub)" <gi...@apache.org>.
Pil0tXia commented on issue #4804:
URL: https://github.com/apache/eventmesh/issues/4804#issuecomment-2029464999
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.
--
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@eventmesh.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@eventmesh.apache.org
For additional commands, e-mail: issues-help@eventmesh.apache.org
Re: [I] [Bug]grpc SubStreamHandler exception (eventmesh)
Posted by "9997766 (via GitHub)" <gi...@apache.org>.
9997766 commented on issue #4804:
URL: https://github.com/apache/eventmesh/issues/4804#issuecomment-2047042401
no problem. but i am bot familar to develop with github。you submit a request to me ?
--
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@eventmesh.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@eventmesh.apache.org
For additional commands, e-mail: issues-help@eventmesh.apache.org
Re: [I] [Bug]grpc SubStreamHandler exception (eventmesh)
Posted by "9997766 (via GitHub)" <gi...@apache.org>.
9997766 commented on issue #4804:
URL: https://github.com/apache/eventmesh/issues/4804#issuecomment-2047075819
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)
--
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@eventmesh.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@eventmesh.apache.org
For additional commands, e-mail: issues-help@eventmesh.apache.org
Re: [I] [Bug]grpc SubStreamHandler exception (eventmesh)
Posted by "9997766 (via GitHub)" <gi...@apache.org>.
9997766 commented on issue #4804:
URL: https://github.com/apache/eventmesh/issues/4804#issuecomment-2029353621
when will this problem be solved?
--
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@eventmesh.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@eventmesh.apache.org
For additional commands, e-mail: issues-help@eventmesh.apache.org
Re: [I] [Bug]grpc SubStreamHandler exception (eventmesh)
Posted by "Pil0tXia (via GitHub)" <gi...@apache.org>.
Pil0tXia commented on issue #4804:
URL: https://github.com/apache/eventmesh/issues/4804#issuecomment-2047151599
@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.
--
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@eventmesh.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@eventmesh.apache.org
For additional commands, e-mail: issues-help@eventmesh.apache.org
Re: [I] [Bug]grpc SubStreamHandler exception (eventmesh)
Posted by "9997766 (via GitHub)" <gi...@apache.org>.
9997766 commented on issue #4804:
URL: https://github.com/apache/eventmesh/issues/4804#issuecomment-2033370944
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: ***@***.***>
--
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@eventmesh.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@eventmesh.apache.org
For additional commands, e-mail: issues-help@eventmesh.apache.org
Re: [I] [Bug] grpc SubStreamHandler exception (eventmesh)
Posted by "mxsm (via GitHub)" <gi...@apache.org>.
mxsm closed issue #4804: [Bug] grpc SubStreamHandler exception
URL: https://github.com/apache/eventmesh/issues/4804
--
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@eventmesh.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@eventmesh.apache.org
For additional commands, e-mail: issues-help@eventmesh.apache.org
Re: [I] [Bug]grpc SubStreamHandler exception (eventmesh)
Posted by "9997766 (via GitHub)" <gi...@apache.org>.
9997766 commented on issue #4804:
URL: https://github.com/apache/eventmesh/issues/4804#issuecomment-2033367795
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: ***@***.***>
--
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@eventmesh.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@eventmesh.apache.org
For additional commands, e-mail: issues-help@eventmesh.apache.org
Re: [I] [Bug]grpc SubStreamHandler exception (eventmesh)
Posted by "Pil0tXia (via GitHub)" <gi...@apache.org>.
Pil0tXia commented on issue #4804:
URL: https://github.com/apache/eventmesh/issues/4804#issuecomment-2031182267
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.
After calling the `close()` method, the `SubStreamHandler` instance will never run again unless the `EventMeshGrpcConsumer` is restarted, which is also a problem.
Do you have a better idea please?
--
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@eventmesh.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@eventmesh.apache.org
For additional commands, e-mail: issues-help@eventmesh.apache.org
Re: [I] [Bug]grpc SubStreamHandler exception (eventmesh)
Posted by "Pil0tXia (via GitHub)" <gi...@apache.org>.
Pil0tXia commented on issue #4804:
URL: https://github.com/apache/eventmesh/issues/4804#issuecomment-2033700885
@9997766 Thanks. You may check PR https://github.com/apache/eventmesh/pull/4807.
--
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@eventmesh.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@eventmesh.apache.org
For additional commands, e-mail: issues-help@eventmesh.apache.org
Re: [I] [Bug]grpc SubStreamHandler exception (eventmesh)
Posted by "9997766 (via GitHub)" <gi...@apache.org>.
9997766 commented on issue #4804:
URL: https://github.com/apache/eventmesh/issues/4804#issuecomment-2046910719
> 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?
--
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@eventmesh.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@eventmesh.apache.org
For additional commands, e-mail: issues-help@eventmesh.apache.org
Re: [I] [Bug]grpc SubStreamHandler exception (eventmesh)
Posted by "Pil0tXia (via GitHub)" <gi...@apache.org>.
Pil0tXia commented on issue #4804:
URL: https://github.com/apache/eventmesh/issues/4804#issuecomment-2047008939
@9997766 We need another reviewer to merge this PR.
@pandaapo May you please help?
--
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@eventmesh.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@eventmesh.apache.org
For additional commands, e-mail: issues-help@eventmesh.apache.org