[Bug] When listens for remote nacos offset configuration, ConnectorRecordPartition deserialization fails. #4366
Description
Search before asking
- I had searched in the issues and found no similar issues.
Environment
Mac
EventMesh version
master
What happened
In the NacosConfigService.class file, at line 166, there is an issue with remote Nacos configuration information in the offset persistence plugin. This issue occurs because the JacksonUtils provided by Nacos does not have a configured KeySerializer. As a result, there is a deserialization error when attempting to deserialize the Map<ConnectorRecordPartition, RecordOffset> configuration item.
How to reproduce
Steps to reproduce the issue:
- Start RocketMQ.
- Start EventMesh Runtime.
- Start EventMesh Connector for RocketMQ.
- Send a few test messages to RocketMQ. At this point, you will see red error messages appearing in the console.
2023-08-16 16:19:39,805 ERROR [nacos.client.cachedata.internal.notifier] CacheData(CacheData.java:351) - [fixed-127.0.0.1_8848] [notify-error] dataId=rocketmqSource, group=rocketmqSource, md5=26694c29b57fb50b18373544dbb1fa1b, listener=org.apache.eventmesh.openconnect.offsetmgmt.nacos.NacosConfigService$1@42e2ba57
com.alibaba.nacos.api.exception.runtime.NacosDeserializationException: errCode: 101, errMsg: Nacos deserialize for class [org.apache.eventmesh.openconnect.offsetmgmt.nacos.NacosConfigService$1$1] failed, cause error[Cannot find a (Map) Key deserializer for type [simple type, class org.apache.eventmesh.openconnect.offsetmgmt.api.storage.ConnectorRecordPartition]
at [Source: (String)"{"ConnectorRecordPartition{connectorName='rocketmqSource'} RecordPartition{partition={queueId=5, topic=TopicTest, brokerName=broker-a}}":{"offset":{"queueOffset":"6"}},"ConnectorRecordPartition{connectorName='rocketmqSource'} RecordPartition{partition={queueId=6, topic=TopicTest, brokerName=broker-a}}":{"offset":{"queueOffset":"6"}},"ConnectorRecordPartition{connectorName='rocketmqSource'} RecordPartition{partition={queueId=7, topic=TopicTest, brokerName=broker-a}}":{"offset":{"queueOffset":"5"}"[truncated 2012 chars]; line: 1, column: 1]].
at com.alibaba.nacos.common.utils.JacksonUtils.toObj(JacksonUtils.java:198) ~[nacos-client-2.2.1.jar:?]
at org.apache.eventmesh.openconnect.offsetmgmt.nacos.NacosConfigService$1.receiveConfigInfo(NacosConfigService.java:166) ~[eventmesh-openconnect-offsetmgmt-nacos-1.9.0-release.jar:1.9.0-release]
at com.alibaba.nacos.client.config.impl.CacheData.lambda$safeNotifyListener$1(CacheData.java:334) ~[nacos-client-2.2.1.jar:?]
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) [?:1.8.0_351]
at java.util.concurrent.FutureTask.run(FutureTask.java:266) [?:1.8.0_351]
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) [?:1.8.0_351]
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) [?:1.8.0_351]
at java.lang.Thread.run(Thread.java:750) [?:1.8.0_351]
Caused by: com.fasterxml.jackson.databind.exc.InvalidDefinitionException: Cannot find a (Map) Key deserializer for type [simple type, class org.apache.eventmesh.openconnect.offsetmgmt.api.storage.ConnectorRecordPartition]
at [Source: (String)"{"ConnectorRecordPartition{connectorName='rocketmqSource'} RecordPartition{partition={queueId=5, topic=TopicTest, brokerName=broker-a}}":{"offset":{"queueOffset":"6"}},"ConnectorRecordPartition{connectorName='rocketmqSource'} RecordPartition{partition={queueId=6, topic=TopicTest, brokerName=broker-a}}":{"offset":{"queueOffset":"6"}},"ConnectorRecordPartition{connectorName='rocketmqSource'} RecordPartition{partition={queueId=7, topic=TopicTest, brokerName=broker-a}}":{"offset":{"queueOffset":"5"}"[truncated 2012 chars]; line: 1, column: 1]
at com.fasterxml.jackson.databind.exc.InvalidDefinitionException.from(InvalidDefinitionException.java:67) ~[jackson-databind-2.13.0.jar:2.13.0]
at com.fasterxml.jackson.databind.DeserializationContext.reportBadDefinition(DeserializationContext.java:1904) ~[jackson-databind-2.13.0.jar:2.13.0]
at com.fasterxml.jackson.databind.deser.DeserializerCache._handleUnknownKeyDeserializer(DeserializerCache.java:603) ~[jackson-databind-2.13.0.jar:2.13.0]
at com.fasterxml.jackson.databind.deser.DeserializerCache.findKeyDeserializer(DeserializerCache.java:168) ~[jackson-databind-2.13.0.jar:2.13.0]
at com.fasterxml.jackson.databind.DeserializationContext.findKeyDeserializer(DeserializationContext.java:669) ~[jackson-databind-2.13.0.jar:2.13.0]
at com.fasterxml.jackson.databind.deser.std.MapDeserializer.createContextual(MapDeserializer.java:302) ~[jackson-databind-2.13.0.jar:2.13.0]
at com.fasterxml.jackson.databind.DeserializationContext.handleSecondaryContextualization(DeserializationContext.java:856) ~[jackson-databind-2.13.0.jar:2.13.0]
at com.fasterxml.jackson.databind.DeserializationContext.findRootValueDeserializer(DeserializationContext.java:647) ~[jackson-databind-2.13.0.jar:2.13.0]
at com.fasterxml.jackson.databind.ObjectMapper._findRootDeserializer(ObjectMapper.java:4806) ~[jackson-databind-2.13.0.jar:2.13.0]
at com.fasterxml.jackson.databind.ObjectMapper._readMapAndClose(ObjectMapper.java:4676) ~[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 com.alibaba.nacos.common.utils.JacksonUtils.toObj(JacksonUtils.java:196) ~[nacos-client-2.2.1.jar:?]
... 7 more
Debug logs
No response
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 *