Context
io.projectreactor.rabbitmq:reactor.rabbitmq:1.5.6com.rabbitmq:amqp-client:5.18.0org.springframework:spring-webflux:6.0.8org.springframework.boot:spring-boot-starter-parent:3.0.6
Issue
Calling method:
reactor.rabbitmq.Sender#declareExchange(reactor.rabbitmq.ExchangeSpecification, reactor.rabbitmq.ResourceManagementOptions)
throws the following exception:
reactor.core.Exceptions$ErrorCallbackNotImplemented: java.lang.UnsupportedOperationException
Caused by: java.lang.UnsupportedOperationException: null
at reactor.rabbitmq.ChannelProxy.getConnection(ChannelProxy.java:69)
at reactor.rabbitmq.ChannelProxy.asyncCompletableRpc(ChannelProxy.java:553)
at reactor.rabbitmq.Sender.lambda$declareExchange$16(Sender.java:547)
at reactor.core.publisher.FluxMap$MapSubscriber.onNext(FluxMap.java:106)
at reactor.core.publisher.Operators$MonoSubscriber.complete(Operators.java:1839)
at reactor.core.publisher.MonoCacheTime.subscribeOrReturn(MonoCacheTime.java:151)
at reactor.core.publisher.Mono.subscribe(Mono.java:4470)
at reactor.core.publisher.Mono.subscribeWith(Mono.java:4551)
at reactor.core.publisher.Mono.subscribe(Mono.java:4452)
at reactor.core.publisher.Mono.subscribe(Mono.java:4388)
at reactor.core.publisher.Mono.subscribe(Mono.java:4335)
at <my.package.Clazz.callingMethod(Clazz.java:xxx)>
at reactor.core.scheduler.SchedulerTask.call(SchedulerTask.java:68)
at reactor.core.scheduler.SchedulerTask.call(SchedulerTask.java:28)
at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
at java.base/java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:304)
at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136)
at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635)
at java.base/java.lang.Thread.run(Thread.java:833)
Notes
Same is true for almost every methods which calls asyncCompletableRpc under the hood.
My understanding is that DD agent seems to proxify that ChannelProxy, and then somehow calls the ChannelProxy.getConnection method directly, which is purposefully throwing an exception:
@Override
public Connection getConnection() {
throw new UnsupportedOperationException();
}Indeed, as stated in ChannelProxy documentation, it's sole purpose is to wrap a delegate Channel to simply ensure an actual channel is always open:
A Channel proxy that re-open the underlying channel if necessary.
This class is used only for resource management in Sender, so only Channel.asyncCompletableRpc(Method) checks whether the underlying Channel is closed or not, and re-open it if necessary. All other methods are not supported.
Idea
Proxification
I am far from being an expert, but it seems like this is related to how DD agent proxify and eventually calls proxified method itself instead of leaving proxified methods calls methods themselves.
Here, we can see that during call to datadog.trace.bootstrap.CallDepthThreadLocalMap#incrementCallDepth, the Class<?> argument is interface com.rabbitmq.client.Channel instead of being the proxified/actual reactor.rabbitmq.ChannelProxy class.
This might or might not be a lead, but further from that, I have no clue.
Wrong bytecode
I might be looking at it from the wrong angle, without the correct bytecode, but when debugging, I have the following stacktrace (only when using DD agent):
getConnection:69, ChannelProxy (reactor.rabbitmq)
asyncCompletableRpc:553, ChannelProxy (reactor.rabbitmq)
lambda$declareExchange$16:547, Sender (reactor.rabbitmq)
apply:-1, Sender$$Lambda$2696 (reactor.rabbitmq)
But when I look at reactor.rabbitmq.ChannelProxy.asyncCompletableRpc:553, it reads:
@Override
public CompletableFuture<Command> asyncCompletableRpc(Method method) throws IOException {
if (!delegate.isOpen()) {
openNewChannelIfNecessary();
}
return delegate.asyncCompletableRpc(method);
}getConnection:69, ChannelProxy (reactor.rabbitmq) is never called, only call which changes when using DD agent is that datadog.trace.bootstrap.CallDepthThreadLocalMap#incrementCallDepth is called during delegate.isOpen() part.
Maybe this leads to a connection attempt by agent to do some measurements?
Context
io.projectreactor.rabbitmq:reactor.rabbitmq:1.5.6com.rabbitmq:amqp-client:5.18.0org.springframework:spring-webflux:6.0.8org.springframework.boot:spring-boot-starter-parent:3.0.6Issue
Calling method:
reactor.rabbitmq.Sender#declareExchange(reactor.rabbitmq.ExchangeSpecification, reactor.rabbitmq.ResourceManagementOptions)throws the following exception:
Notes
Same is true for almost every methods which calls
asyncCompletableRpcunder the hood.My understanding is that DD agent seems to proxify that
ChannelProxy, and then somehow calls theChannelProxy.getConnectionmethod directly, which is purposefully throwing an exception:Indeed, as stated in
ChannelProxydocumentation, it's sole purpose is to wrap a delegateChannelto simply ensure an actual channel is always open:Idea
Proxification
I am far from being an expert, but it seems like this is related to how DD agent proxify and eventually calls proxified method itself instead of leaving proxified methods calls methods themselves.
Here, we can see that during call to
datadog.trace.bootstrap.CallDepthThreadLocalMap#incrementCallDepth, theClass<?>argument isinterface com.rabbitmq.client.Channelinstead of being the proxified/actualreactor.rabbitmq.ChannelProxyclass.This might or might not be a lead, but further from that, I have no clue.
Wrong bytecode
I might be looking at it from the wrong angle, without the correct bytecode, but when debugging, I have the following stacktrace (only when using DD agent):
But when I look at
reactor.rabbitmq.ChannelProxy.asyncCompletableRpc:553, it reads:getConnection:69, ChannelProxy (reactor.rabbitmq)is never called, only call which changes when using DD agent is thatdatadog.trace.bootstrap.CallDepthThreadLocalMap#incrementCallDepthis called duringdelegate.isOpen()part.Maybe this leads to a connection attempt by agent to do some measurements?