创建事件总线的时候报错

[JasonXZh]

创建事件总线的时候报错，日志如下：

2023-12-03
03:47:55.258@@@@@@@@[reactor-http-server-kqueue-11]@@error@@o.a.r.e.a.p.d.r.r.RocketMQMetaService[52]@@create topic failed.
org.apache.rocketmq.client.exception.MQClientException: create new topic failed
For more information, please visit the url, http://rocketmq.apache.org/docs/faq/
at org.apache.rocketmq.client.impl.MQAdminImpl.createTopic(MQAdminImpl.java:134)
at org.apache.rocketmq.tools.admin.DefaultMQAdminExtImpl.createTopic(DefaultMQAdminExtImpl.java:1035)
at org.apache.rocketmq.tools.admin.DefaultMQAdminExt.createTopic(DefaultMQAdminExt.java:101)
at org.apache.rocketmq.tools.admin.DefaultMQAdminExt.createTopic(DefaultMQAdminExt.java:96)
at org.apache.rocketmq.eventbridge.adapter.persistence.data.rocketmq.repository.RocketMQMetaService.createTopic(RocketMQMetaService.java:50)
at org.apache.rocketmq.eventbridge.adapter.persistence.data.rocketmq.repository.RocketMQEventDataRepository.createEventBusPersistence(RocketMQEventDataRepository.java:62)
at org.apache.rocketmq.eventbridge.adapter.persistence.data.rocketmq.repository.RocketMQEventDataRepository$$FastClassBySpringCGLIB$$482fd7dd.invoke()
at org.springframework.cglib.proxy.MethodProxy.invoke(MethodProxy.java:204)
at org.springframework.aop.framework.CglibAopProxy$CglibMethodInvocation.invokeJoinpoint(CglibAopProxy.java:746)
at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:163)
at org.springframework.dao.support.PersistenceExceptionTranslationInterceptor.invoke(PersistenceExceptionTranslationInterceptor.java:139)
at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:185)
at org.springframework.aop.framework.CglibAopProxy$DynamicAdvisedInterceptor.intercept(CglibAopProxy.java:688)
at org.apache.rocketmq.eventbridge.adapter.persistence.data.rocketmq.repository.RocketMQEventDataRepository$$EnhancerBySpringCGLIB$$cef99113.createEventBusPersistence()
at org.apache.rocketmq.eventbridge.domain.model.bus.EventBusService.createEventBus(EventBusService.java:66)
at org.apache.rocketmq.eventbridge.domain.model.bus.EventBusService$$FastClassBySpringCGLIB$$86839a9d.invoke()
at org.springframework.cglib.proxy.MethodProxy.invoke(MethodProxy.java:204)
at org.springframework.aop.framework.CglibAopProxy$CglibMethodInvocation.invokeJoinpoint(CglibAopProxy.java:746)
at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:163)
at org.springframework.transaction.interceptor.TransactionAspectSupport.invokeWithinTransaction(TransactionAspectSupport.java:294)
at org.springframework.transaction.interceptor.TransactionInterceptor.invoke(TransactionInterceptor.java:98)
at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:185)
at org.springframework.aop.framework.CglibAopProxy$DynamicAdvisedInterceptor.intercept(CglibAopProxy.java:688)
at org.apache.rocketmq.eventbridge.domain.model.bus.EventBusService$$EnhancerBySpringCGLIB$$4462fba3.createEventBus()
at org.apache.rocketmq.eventbridge.adapter.api.controller.EventBusController.lambda$createEventBus$0(EventBusController.java:60)
at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.onNext(FluxMapFuseable.java:107)
at reactor.core.publisher.Operators$ScalarSubscription.request(Operators.java:1637)
at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.request(FluxMapFuseable.java:156)
at reactor.core.publisher.FluxPeekFuseable$PeekFuseableSubscriber.request(FluxPeekFuseable.java:138)
at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.request(FluxMapFuseable.java:156)
at org.springframework.http.server.reactive.ChannelSendOperator$WriteBarrier.onSubscribe(ChannelSendOperator.java:163)
at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.onSubscribe(FluxMapFuseable.java:90)
at reactor.core.publisher.FluxPeekFuseable$PeekFuseableSubscriber.onSubscribe(FluxPeekFuseable.java:172)
at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.onSubscribe(FluxMapFuseable.java:90)
at reactor.core.publisher.MonoCurrentContext.subscribe(MonoCurrentContext.java:33)
at reactor.core.publisher.MonoMapFuseable.subscribe(MonoMapFuseable.java:59)
at reactor.core.publisher.MonoPeekFuseable.subscribe(MonoPeekFuseable.java:74)
at reactor.core.publisher.FluxSourceMonoFuseable.subscribe(FluxSourceMonoFuseable.java:38)
at reactor.core.publisher.FluxMapFuseable.subscribe(FluxMapFuseable.java:63)
at org.springframework.http.server.reactive.ChannelSendOperator.subscribe(ChannelSendOperator.java:74)
at reactor.core.publisher.MonoOnErrorResume.subscribe(MonoOnErrorResume.java:44)
at reactor.core.publisher.MonoFlatMap$FlatMapMain.onNext(MonoFlatMap.java:150)
at reactor.core.publisher.Operators$MonoSubscriber.complete(Operators.java:1080)
at reactor.core.publisher.MonoFlatMap$FlatMapInner.onNext(MonoFlatMap.java:241)
at reactor.core.publisher.FluxOnErrorResume$ResumeSubscriber.onNext(FluxOnErrorResume.java:73)
at reactor.core.publisher.FluxPeekFuseable$PeekFuseableSubscriber.onNext(FluxPeekFuseable.java:198)
at reactor.core.publisher.FluxPeekFuseable$PeekFuseableSubscriber.onNext(FluxPeekFuseable.java:198)
at reactor.core.publisher.Operators$MonoSubscriber.complete(Operators.java:1080)
at reactor.core.publisher.MonoIgnoreThen$ThenAcceptInner.onNext(MonoIgnoreThen.java:290)
at reactor.core.publisher.Operators$MonoSubscriber.complete(Operators.java:1080)
at reactor.core.publisher.MonoFlatMap$FlatMapMain.onNext(MonoFlatMap.java:144)
at reactor.core.publisher.Operators$MonoSubscriber.complete(Operators.java:1080)
at reactor.core.publisher.MonoZip$ZipCoordinator.signal(MonoZip.java:241)
at reactor.core.publisher.MonoZip$ZipInner.onNext(MonoZip.java:323)
at reactor.core.publisher.FluxPeek$PeekSubscriber.onNext(FluxPeek.java:185)
at reactor.core.publisher.FluxDefaultIfEmpty$DefaultIfEmptySubscriber.onNext(FluxDefaultIfEmpty.java:92)
at reactor.core.publisher.FluxSwitchIfEmpty$SwitchIfEmptySubscriber.onNext(FluxSwitchIfEmpty.java:67)
at reactor.core.publisher.FluxOnErrorResume$ResumeSubscriber.onNext(FluxOnErrorResume.java:73)
at reactor.core.publisher.Operators$MonoSubscriber.complete(Operators.java:1080)
at reactor.core.publisher.MonoSingle$SingleSubscriber.onComplete(MonoSingle.java:171)
at reactor.core.publisher.FluxMap$MapSubscriber.onComplete(FluxMap.java:130)
at reactor.core.publisher.FluxFlatMap$FlatMapMain.checkTerminated(FluxFlatMap.java:773)
at reactor.core.publisher.FluxFlatMap$FlatMapMain.drainLoop(FluxFlatMap.java:543)
at reactor.core.publisher.FluxFlatMap$FlatMapMain.drain(FluxFlatMap.java:523)
at reactor.core.publisher.FluxFlatMap$FlatMapMain.onComplete(FluxFlatMap.java:409)
at reactor.core.publisher.DrainUtils.postCompleteDrain(DrainUtils.java:131)
at reactor.core.publisher.DrainUtils.postComplete(DrainUtils.java:186)
at reactor.core.publisher.FluxMapSignal$FluxMapSignalSubscriber.onComplete(FluxMapSignal.java:213)
at reactor.core.publisher.FluxPeek$PeekSubscriber.onComplete(FluxPeek.java:245)
at reactor.core.publisher.FluxMap$MapSubscriber.onComplete(FluxMap.java:130)
at reactor.core.publisher.FluxPeek$PeekSubscriber.onComplete(FluxPeek.java:245)
at reactor.core.publisher.FluxMap$MapSubscriber.onComplete(FluxMap.java:130)
at reactor.ipc.netty.channel.FluxReceive.terminateReceiver(FluxReceive.java:377)
at reactor.ipc.netty.channel.FluxReceive.drainReceiver(FluxReceive.java:202)
at reactor.ipc.netty.channel.FluxReceive.request(FluxReceive.java:110)
at reactor.core.publisher.FluxMap$MapSubscriber.request(FluxMap.java:149)
at reactor.core.publisher.FluxPeek$PeekSubscriber.request(FluxPeek.java:130)
at reactor.core.publisher.FluxMap$MapSubscriber.request(FluxMap.java:149)
at reactor.core.publisher.FluxPeek$PeekSubscriber.request(FluxPeek.java:130)
at reactor.core.publisher.FluxMapSignal$FluxMapSignalSubscriber.request(FluxMapSignal.java:225)
at reactor.core.publisher.FluxFlatMap$FlatMapMain.onSubscribe(FluxFlatMap.java:332)
at reactor.core.publisher.FluxMapSignal$FluxMapSignalSubscriber.onSubscribe(FluxMapSignal.java:115)
at reactor.core.publisher.FluxPeek$PeekSubscriber.onSubscribe(FluxPeek.java:163)
at reactor.core.publisher.FluxMap$MapSubscriber.onSubscribe(FluxMap.java:86)
at reactor.core.publisher.FluxPeek$PeekSubscriber.onSubscribe(FluxPeek.java:163)
at reactor.core.publisher.FluxMap$MapSubscriber.onSubscribe(FluxMap.java:86)
at reactor.ipc.netty.channel.FluxReceive.startReceiver(FluxReceive.java:273)
at reactor.ipc.netty.channel.FluxReceive.subscribe(FluxReceive.java:124)
at reactor.core.publisher.FluxMap.subscribe(FluxMap.java:62)
at reactor.ipc.netty.ByteBufFlux.subscribe(ByteBufFlux.java:242)
at reactor.core.publisher.FluxPeek.subscribe(FluxPeek.java:83)
at reactor.ipc.netty.ByteBufFlux.subscribe(ByteBufFlux.java:242)
at reactor.core.publisher.FluxMap.subscribe(FluxMap.java:62)
at reactor.core.publisher.FluxPeek.subscribe(FluxPeek.java:83)
at reactor.core.publisher.FluxMapSignal.subscribe(FluxMapSignal.java:69)
at reactor.core.publisher.FluxFlatMap.subscribe(FluxFlatMap.java:97)
at reactor.core.publisher.FluxMap.subscribe(FluxMap.java:62)
at reactor.core.publisher.MonoSingle.subscribe(MonoSingle.java:58)
at reactor.core.publisher.MonoOnErrorResume.subscribe(MonoOnErrorResume.java:44)
at reactor.core.publisher.MonoSwitchIfEmpty.subscribe(MonoSwitchIfEmpty.java:44)
at reactor.core.publisher.MonoDefaultIfEmpty.subscribe(MonoDefaultIfEmpty.java:37)
at reactor.core.publisher.MonoPeek.subscribe(MonoPeek.java:71)
at reactor.core.publisher.Mono.subscribe(Mono.java:3080)
at reactor.core.publisher.MonoZip.subscribe(MonoZip.java:128)
at reactor.core.publisher.MonoFlatMap.subscribe(MonoFlatMap.java:60)
at reactor.core.publisher.MonoDefer.subscribe(MonoDefer.java:52)
at reactor.core.publisher.MonoIgnoreThen$ThenIgnoreMain.drain(MonoIgnoreThen.java:148)
at reactor.core.publisher.MonoIgnoreThen.subscribe(MonoIgnoreThen.java:56)
at reactor.core.publisher.MonoPeekFuseable.subscribe(MonoPeekFuseable.java:74)
at reactor.core.publisher.MonoPeekFuseable.subscribe(MonoPeekFuseable.java:74)
at reactor.core.publisher.MonoOnErrorResume.subscribe(MonoOnErrorResume.java:44)
at reactor.core.publisher.MonoFlatMap$FlatMapMain.onNext(MonoFlatMap.java:150)
at reactor.core.publisher.FluxSwitchIfEmpty$SwitchIfEmptySubscriber.onNext(FluxSwitchIfEmpty.java:67)
at reactor.core.publisher.MonoNext$NextSubscriber.onNext(MonoNext.java:76)
at reactor.core.publisher.FluxConcatMap$ConcatMapImmediate.innerNext(FluxConcatMap.java:271)
at reactor.core.publisher.FluxConcatMap$ConcatMapInner.onNext(FluxConcatMap.java:803)
at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.onNext(FluxMapFuseable.java:115)
at reactor.core.publisher.Operators$ScalarSubscription.request(Operators.java:1637)
at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.request(FluxMapFuseable.java:156)
at reactor.core.publisher.Operators$MultiSubscriptionSubscriber.set(Operators.java:1451)
at reactor.core.publisher.Operators$MultiSubscriptionSubscriber.onSubscribe(Operators.java:1325)
at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.onSubscribe(FluxMapFuseable.java:90)
at reactor.core.publisher.MonoJust.subscribe(MonoJust.java:54)
at reactor.core.publisher.MonoMapFuseable.subscribe(MonoMapFuseable.java:59)
at reactor.core.publisher.Mono.subscribe(Mono.java:3080)
at reactor.core.publisher.FluxConcatMap$ConcatMapImmediate.drain(FluxConcatMap.java:418)
at reactor.core.publisher.FluxConcatMap$ConcatMapImmediate.onSubscribe(FluxConcatMap.java:210)
at reactor.core.publisher.FluxIterable.subscribe(FluxIterable.java:128)
at reactor.core.publisher.FluxIterable.subscribe(FluxIterable.java:61)
at reactor.core.publisher.FluxConcatMap.subscribe(FluxConcatMap.java:121)
at reactor.core.publisher.MonoNext.subscribe(MonoNext.java:40)
at reactor.core.publisher.MonoSwitchIfEmpty.subscribe(MonoSwitchIfEmpty.java:44)
at reactor.core.publisher.MonoFlatMap.subscribe(MonoFlatMap.java:60)
at reactor.core.publisher.MonoFlatMap.subscribe(MonoFlatMap.java:60)
at reactor.core.publisher.MonoDefer.subscribe(MonoDefer.java:52)
at reactor.core.publisher.MonoSubscriberContext.subscribe(MonoSubscriberContext.java:47)
at reactor.core.publisher.MonoDefer.subscribe(MonoDefer.java:52)
at reactor.core.publisher.MonoDefer.subscribe(MonoDefer.java:52)
at reactor.core.publisher.MonoOnErrorResume.subscribe(MonoOnErrorResume.java:44)
at reactor.core.publisher.MonoOnErrorResume.subscribe(MonoOnErrorResume.java:44)
at reactor.core.publisher.MonoOnErrorResume.subscribe(MonoOnErrorResume.java:44)
at reactor.core.publisher.Mono.subscribe(Mono.java:3080)
at reactor.core.publisher.MonoIgnoreThen$ThenIgnoreMain.drain(MonoIgnoreThen.java:167)
at reactor.core.publisher.MonoIgnoreThen.subscribe(MonoIgnoreThen.java:56)
at reactor.core.publisher.MonoPeekFuseable.subscribe(MonoPeekFuseable.java:70)
at reactor.core.publisher.MonoPeekTerminal.subscribe(MonoPeekTerminal.java:61)
at reactor.ipc.netty.channel.ChannelOperations.applyHandler(ChannelOperations.java:380)
at reactor.ipc.netty.http.server.HttpServerOperations.onHandlerStart(HttpServerOperations.java:397)
at io.netty.util.concurrent.AbstractEventExecutor.safeExecute(AbstractEventExecutor.java:163)
at io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:404)
at io.netty.channel.kqueue.KQueueEventLoop.run(KQueueEventLoop.java:273)
at io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:884)
at java.lang.Thread.run(Thread.java:745)
Caused by: org.apache.rocketmq.remoting.exception.RemotingSendRequestException: send request to <ocalhost/127.0.0.1:9876> failed
at org.apache.rocketmq.remoting.netty.NettyRemotingAbstract.invokeSyncImpl(NettyRemotingAbstract.java:440)
at org.apache.rocketmq.remoting.netty.NettyRemotingClient.invokeSync(NettyRemotingClient.java:377)
at org.apache.rocketmq.client.impl.MQClientAPIImpl.getTopicRouteInfoFromNameServer(MQClientAPIImpl.java:1367)
at org.apache.rocketmq.client.impl.MQClientAPIImpl.getTopicRouteInfoFromNameServer(MQClientAPIImpl.java:1357)
at org.apache.rocketmq.client.impl.MQAdminImpl.createTopic(MQAdminImpl.java:86)
... 152 common frames omitted
Caused by: io.netty.handler.codec.EncoderException: java.lang.NoSuchMethodError: com.alibaba.fastjson.JSON.toJSONString(Ljava/lang/Object;Z)Ljava/lang/String;
at io.netty.handler.codec.MessageToByteEncoder.write(MessageToByteEncoder.java:125)
at io.netty.channel.AbstractChannelHandlerContext.invokeWrite0(AbstractChannelHandlerContext.java:738)
at io.netty.channel.AbstractChannelHandlerContext.invokeWrite(AbstractChannelHandlerContext.java:730)
at io.netty.channel.AbstractChannelHandlerContext.write(AbstractChannelHandlerContext.java:816)
at io.netty.channel.AbstractChannelHandlerContext.write(AbstractChannelHandlerContext.java:723)
at io.netty.handler.timeout.IdleStateHandler.write(IdleStateHandler.java:302)
at io.netty.channel.AbstractChannelHandlerContext.invokeWrite0(AbstractChannelHandlerContext.java:738)
at io.netty.channel.AbstractChannelHandlerContext.invokeWrite(AbstractChannelHandlerContext.java:730)
at io.netty.channel.AbstractChannelHandlerContext.write(AbstractChannelHandlerContext.java:816)
at io.netty.channel.AbstractChannelHandlerContext.write(AbstractChannelHandlerContext.java:723)
at io.netty.channel.ChannelDuplexHandler.write(ChannelDuplexHandler.java:106)
at io.netty.channel.AbstractChannelHandlerContext.invokeWrite0(AbstractChannelHandlerContext.java:738)
at io.netty.channel.AbstractChannelHandlerContext.invokeWrite(AbstractChannelHandlerContext.java:730)
at io.netty.channel.AbstractChannelHandlerContext.access$1900(AbstractChannelHandlerContext.java:38)
at io.netty.channel.AbstractChannelHandlerContext$AbstractWriteTask.write(AbstractChannelHandlerContext.java:1081)
at io.netty.channel.AbstractChannelHandlerContext$WriteAndFlushTask.write(AbstractChannelHandlerContext.java:1128)
at io.netty.channel.AbstractChannelHandlerContext$AbstractWriteTask.run(AbstractChannelHandlerContext.java:1070)
at io.netty.util.concurrent.DefaultEventExecutor.run(DefaultEventExecutor.java:66)
... 2 common frames omitted
Caused by: java.lang.NoSuchMethodError: com.alibaba.fastjson.JSON.toJSONString(Ljava/lang/Object;Z)Ljava/lang/String;
at org.apache.rocketmq.remoting.protocol.RemotingSerializable.toJson(RemotingSerializable.java:34)
at org.apache.rocketmq.remoting.protocol.RemotingSerializable.encode(RemotingSerializable.java:26)
at org.apache.rocketmq.remoting.protocol.RemotingCommand.headerEncode(RemotingCommand.java:367)
at org.apache.rocketmq.remoting.protocol.RemotingCommand.encodeHeader(RemotingCommand.java:409)
at org.apache.rocketmq.remoting.protocol.RemotingCommand.encodeHeader(RemotingCommand.java:400)
at org.apache.rocketmq.remoting.netty.NettyEncoder.encode(NettyEncoder.java:38)
at org.apache.rocketmq.remoting.netty.NettyEncoder.encode(NettyEncoder.java:30)
at io.netty.handler.codec.MessageToByteEncoder.write(MessageToByteEncoder.java:107)
... 19 common frames omitted
2023-12-03
03:47:59.745@@@@@@@@[NettyClientSelector_1]@@INFO@@RocketmqRemoting[95]@@closeChannel: close the connection to remote address[127.0.0.1:9876] result: true

[Jashinck]

Use RocketMQ 4.9.x version please

[JasonXZh]

Use RocketMQ 4.9.x version please

I have changed to RocketMQ 4.9.4, but got the same error.

[Jashinck]

Use RocketMQ 4.9.x version please

I have changed to RocketMQ 4.9.4, but got the same error.

After confirm deploy RocketMQ 4.9.x version in local, then make sure JSON relply version no conflict by marven

[JasonXZh]

Use RocketMQ 4.9.x version please

I have changed to RocketMQ 4.9.4, but got the same error.

After confirm deploy RocketMQ 4.9.x version in local, then make sure JSON relply version no conflict by marven

I use the eventbridge downloaded from here:
https://www.apache.org/dyn/closer.cgi?path=rocketmq/rocketmq-eventbridge/1.0.0/rocketmq-eventbridge-1.0.0-bin-release.zip

I found The fastjson version is 2.0.2 in the rocketmq-eventbridge.jar, but the fastjson used in the rocketmq is 1.2.x.
Is this the reason for the error ?

[Jashinck]

Use RocketMQ 4.9.x version please

I have changed to RocketMQ 4.9.4, but got the same error.

After confirm deploy RocketMQ 4.9.x version in local, then make sure JSON relply version no conflict by marven

I use the eventbridge downloaded from here: https://www.apache.org/dyn/closer.cgi?path=rocketmq/rocketmq-eventbridge/1.0.0/rocketmq-eventbridge-1.0.0-bin-release.zip

I found The fastjson version is 2.0.2 in the rocketmq-eventbridge.jar, but the fastjson used in the rocketmq is 1.2.x. Is this the reason for the error ?

run ther master branch code please

[JasonXZh]

Use RocketMQ 4.9.x version please

I have changed to RocketMQ 4.9.4, but got the same error.

After confirm deploy RocketMQ 4.9.x version in local, then make sure JSON relply version no conflict by marven

I use the eventbridge downloaded from here: https://www.apache.org/dyn/closer.cgi?path=rocketmq/rocketmq-eventbridge/1.0.0/rocketmq-eventbridge-1.0.0-bin-release.zip
I found The fastjson version is 2.0.2 in the rocketmq-eventbridge.jar, but the fastjson used in the rocketmq is 1.2.x. Is this the reason for the error ?

run ther master branch code please

I have packaged eventbridge from the master branch, and I see it used fastjson-1.2.83.jar.
So I chose rocketmq 5.1.4 which used fastjson-1.2.83.jar.
But I still got the same error...