[Bug] EventMeshGrpcConsumer client subscribe topic error

[mxsm]

Search before asking

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

Environment

Windows

EventMesh version

master

What happened

When I run CloudEventsAsyncSubscribe demo, An error occurs when using the EventMeshGrpcConsumer client to subscribe.

How to reproduce

run CloudEventsAsyncSubscribe demo

Debug logs

2023-06-29 21:41:12,128 ERROR [grpc-default-executor-1] SubStreamHandler(SubStreamHandler.java:109) - Received Server side error
io.grpc.StatusRuntimeException: CANCELLED: Failed to read message.
	at io.grpc.Status.asRuntimeException(Status.java:535) ~[grpc-api-1.43.2.jar:1.43.2]
	at io.grpc.stub.ClientCalls$StreamObserverToCallListenerAdapter.onClose(ClientCalls.java:479) ~[grpc-stub-1.43.2.jar:1.43.2]
	at io.grpc.internal.DelayedClientCall$DelayedListener$3.run(DelayedClientCall.java:463) ~[grpc-core-1.43.2.jar:1.43.2]
	at io.grpc.internal.DelayedClientCall$DelayedListener.delayOrExecute(DelayedClientCall.java:427) ~[grpc-core-1.43.2.jar:1.43.2]
	at io.grpc.internal.DelayedClientCall$DelayedListener.onClose(DelayedClientCall.java:460) ~[grpc-core-1.43.2.jar:1.43.2]
	at io.grpc.internal.ClientCallImpl.closeObserver(ClientCallImpl.java:562) ~[grpc-core-1.43.2.jar:1.43.2]
	at io.grpc.internal.ClientCallImpl.access$300(ClientCallImpl.java:70) ~[grpc-core-1.43.2.jar:1.43.2]
	at io.grpc.internal.ClientCallImpl$ClientStreamListenerImpl$1StreamClosed.runInternal(ClientCallImpl.java:743) ~[grpc-core-1.43.2.jar:1.43.2]
	at io.grpc.internal.ClientCallImpl$ClientStreamListenerImpl$1StreamClosed.runInContext(ClientCallImpl.java:722) ~[grpc-core-1.43.2.jar:1.43.2]
	at io.grpc.internal.ContextRunnable.run(ContextRunnable.java:37) ~[grpc-core-1.43.2.jar:1.43.2]
	at io.grpc.internal.SerializingExecutor.run(SerializingExecutor.java:133) ~[grpc-core-1.43.2.jar:1.43.2]
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) ~[?:1.8.0_202]
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) ~[?:1.8.0_202]
	at java.lang.Thread.run(Thread.java:748) [?:1.8.0_202]
Caused by: org.apache.eventmesh.common.exception.JsonException: deserialize json string to typeReference error
	at org.apache.eventmesh.common.utils.JsonUtils.parseTypeReferenceObject(JsonUtils.java:128) ~[classes/:?]
	at org.apache.eventmesh.client.grpc.util.EventMeshCloudEventBuilder.buildMessageFromEventMeshCloudEvent(EventMeshCloudEventBuilder.java:272) ~[classes/:?]
	at org.apache.eventmesh.client.grpc.consumer.SubStreamHandler$1.onNext(SubStreamHandler.java:73) ~[classes/:?]
	at org.apache.eventmesh.client.grpc.consumer.SubStreamHandler$1.onNext(SubStreamHandler.java:70) ~[classes/:?]
	at io.grpc.stub.ClientCalls$StreamObserverToCallListenerAdapter.onMessage(ClientCalls.java:466) ~[grpc-stub-1.43.2.jar:1.43.2]
	at io.grpc.internal.DelayedClientCall$DelayedListener.onMessage(DelayedClientCall.java:447) ~[grpc-core-1.43.2.jar:1.43.2]
	at io.grpc.internal.ClientCallImpl$ClientStreamListenerImpl$1MessagesAvailable.runInternal(ClientCallImpl.java:661) ~[grpc-core-1.43.2.jar:1.43.2]
	at io.grpc.internal.ClientCallImpl$ClientStreamListenerImpl$1MessagesAvailable.runInContext(ClientCallImpl.java:646) ~[grpc-core-1.43.2.jar:1.43.2]
	... 5 more
Caused by: com.fasterxml.jackson.databind.exc.MismatchedInputException: Cannot deserialize value of type `java.util.HashMap<java.lang.String,java.lang.String>` from Array value (token `JsonToken.START_ARRAY`)
 at [Source: (String)"[{"topic":"TEST-TOPIC-GRPC-ASYNC","mode":"CLUSTERING","type":"ASYNC"}]"; line: 1, column: 1]
	at com.fasterxml.jackson.databind.exc.MismatchedInputException.from(MismatchedInputException.java:59) ~[jackson-databind-2.13.0.jar:2.13.0]
	at com.fasterxml.jackson.databind.DeserializationContext.reportInputMismatch(DeserializationContext.java:1741) ~[jackson-databind-2.13.0.jar:2.13.0]
	at com.fasterxml.jackson.databind.DeserializationContext.handleUnexpectedToken(DeserializationContext.java:1515) ~[jackson-databind-2.13.0.jar:2.13.0]
	at com.fasterxml.jackson.databind.deser.std.StdDeserializer._deserializeFromArray(StdDeserializer.java:222) ~[jackson-databind-2.13.0.jar:2.13.0]
	at com.fasterxml.jackson.databind.deser.std.MapDeserializer.deserialize(MapDeserializer.java:447) ~[jackson-databind-2.13.0.jar:2.13.0]
	at com.fasterxml.jackson.databind.deser.std.MapDeserializer.deserialize(MapDeserializer.java:32) ~[jackson-databind-2.13.0.jar:2.13.0]
	at com.fasterxml.jackson.databind.deser.DefaultDeserializationContext.readRootValue(DefaultDeserializationContext.java:322) ~[jackson-databind-2.13.0.jar:2.13.0]
	at com.fasterxml.jackson.databind.ObjectMapper._readMapAndClose(ObjectMapper.java:4675) ~[jackson-databind-2.13.0.jar:2.13.0]
	at com.fasterxml.jackson.databind.ObjectMapper.readValue(ObjectMapper.java:3630) ~[jackson-databind-2.13.0.jar:2.13.0]
	at com.fasterxml.jackson.databind.ObjectMapper.readValue(ObjectMapper.java:3613) ~[jackson-databind-2.13.0.jar:2.13.0]
	at org.apache.eventmesh.common.utils.JsonUtils.parseTypeReferenceObject(JsonUtils.java:126) ~[classes/:?]
	at org.apache.eventmesh.client.grpc.util.EventMeshCloudEventBuilder.buildMessageFromEventMeshCloudEvent(EventMeshCloudEventBuilder.java:272) ~[classes/:?]
	at org.apache.eventmesh.client.grpc.consumer.SubStreamHandler$1.onNext(SubStreamHandler.java:73) ~[classes/:?]
	at org.apache.eventmesh.client.grpc.consumer.SubStreamHandler$1.onNext(SubStreamHandler.java:70) ~[classes/:?]
	at io.grpc.stub.ClientCalls$StreamObserverToCallListenerAdapter.onMessage(ClientCalls.java:466) ~[grpc-stub-1.43.2.jar:1.43.2]
	at io.grpc.internal.DelayedClientCall$DelayedListener.onMessage(DelayedClientCall.java:447) ~[grpc-core-1.43.2.jar:1.43.2]
	at io.grpc.internal.ClientCallImpl$ClientStreamListenerImpl$1MessagesAvailable.runInternal(ClientCallImpl.java:661) ~[grpc-core-1.43.2.jar:1.43.2]
	at io.grpc.internal.ClientCallImpl$ClientStreamListenerImpl$1MessagesAvailable.runInContext(ClientCallImpl.java:646) ~[grpc-core-1.43.2.jar:1.43.2]
	... 5 more
2023-06-29 21:41:12,129 INFO  [grpc-default-executor-1] SubStreamHandler(SubStreamHandler.java:164) - SubStreamHandler closed.

Are you willing to submit PR?

• Yes I am willing to submit a PR!