[Bug] When listens for remote nacos offset configuration, ConnectorRecordPartition deserialization fails.

[yanrongzhen]

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:

1. Start RocketMQ.
2. Start EventMesh Runtime.
3. Start EventMesh Connector for RocketMQ.
4. 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 *

[xwm1992]

@yanrongzhen I have fixed this bug, and will submit pr later.