Fix IllegalReferenceCountException exception to break the callback

[mattisonchao]

Motivation

Fix IllegalReferenceCountException exception to break the future callback.

The exception stack trace is as follows:

2023-07-27T00:43:58,769+0000 [mqtt-proxy-adapter-80-2] ERROR io.streamnative.pulsar.handlers.mqtt.adapter.AdapterChannel - [AdapterChannel] Proxy write to broker poc-use1-broker-1.poc-use1-broker-headless.o-xj8il.svc.cluster.local/240.240.0.4:1
883 message [!!!io.streamnative.pulsar.handlers.mqtt.adapter.MqttAdapterMessage@b720745=>io.netty.util.IllegalReferenceCountException:refCnt: 0!!!] failed.
java.util.concurrent.CompletionException: io.netty.channel.StacklessClosedChannelException
        at java.util.concurrent.CompletableFuture.encodeRelay(CompletableFuture.java:367) ~[?:?]
        at java.util.concurrent.CompletableFuture.completeRelay(CompletableFuture.java:376) ~[?:?]
        at java.util.concurrent.CompletableFuture$UniRelay.tryFire(CompletableFuture.java:1019) ~[?:?]
        at java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:506) ~[?:?]
        at java.util.concurrent.CompletableFuture.completeExceptionally(CompletableFuture.java:2088) ~[?:?]
        at io.streamnative.pulsar.handlers.mqtt.utils.FutureUtils.lambda$completableFuture$0(FutureUtils.java:32) ~[?:?]
        at io.netty.util.concurrent.DefaultPromise.notifyListener0(DefaultPromise.java:590) ~[io.netty-netty-common-4.1.93.Final.jar:4.1.93.Final]
        at io.netty.util.concurrent.DefaultPromise.notifyListenersNow(DefaultPromise.java:557) ~[io.netty-netty-common-4.1.93.Final.jar:4.1.93.Final]
        at io.netty.util.concurrent.DefaultPromise.notifyListeners(DefaultPromise.java:492) ~[io.netty-netty-common-4.1.93.Final.jar:4.1.93.Final]
        at io.netty.util.concurrent.DefaultPromise.setValue0(DefaultPromise.java:636) ~[io.netty-netty-common-4.1.93.Final.jar:4.1.93.Final]
        at io.netty.util.concurrent.DefaultPromise.setFailure0(DefaultPromise.java:629) ~[io.netty-netty-common-4.1.93.Final.jar:4.1.93.Final]
        at io.netty.util.concurrent.DefaultPromise.tryFailure(DefaultPromise.java:118) ~[io.netty-netty-common-4.1.93.Final.jar:4.1.93.Final]
        at io.netty.channel.AbstractChannel$AbstractUnsafe.safeSetFailure(AbstractChannel.java:999) ~[io.netty-netty-transport-4.1.93.Final.jar:4.1.93.Final]
        at io.netty.channel.AbstractChannel$AbstractUnsafe.write(AbstractChannel.java:860) ~[io.netty-netty-transport-4.1.93.Final.jar:4.1.93.Final]
        at io.netty.channel.DefaultChannelPipeline$HeadContext.write(DefaultChannelPipeline.java:1367) ~[io.netty-netty-transport-4.1.93.Final.jar:4.1.93.Final]
        at io.netty.channel.AbstractChannelHandlerContext.invokeWrite0(AbstractChannelHandlerContext.java:877) ~[io.netty-netty-transport-4.1.93.Final.jar:4.1.93.Final]
        at io.netty.channel.AbstractChannelHandlerContext.invokeWrite(AbstractChannelHandlerContext.java:863) ~[io.netty-netty-transport-4.1.93.Final.jar:4.1.93.Final]
        at io.netty.channel.AbstractChannelHandlerContext.write(AbstractChannelHandlerContext.java:968) ~[io.netty-netty-transport-4.1.93.Final.jar:4.1.93.Final]
        at io.netty.channel.AbstractChannelHandlerContext.write(AbstractChannelHandlerContext.java:856) ~[io.netty-netty-transport-4.1.93.Final.jar:4.1.93.Final]
        at io.netty.handler.codec.MessageToMessageEncoder.write(MessageToMessageEncoder.java:113) ~[io.netty-netty-codec-4.1.93.Final.jar:4.1.93.Final]
        at io.netty.channel.AbstractChannelHandlerContext.invokeWrite0(AbstractChannelHandlerContext.java:881) ~[io.netty-netty-transport-4.1.93.Final.jar:4.1.93.Final]
        at io.netty.channel.AbstractChannelHandlerContext.invokeWriteAndFlush(AbstractChannelHandlerContext.java:940) ~[io.netty-netty-transport-4.1.93.Final.jar:4.1.93.Final]
        at io.netty.channel.AbstractChannelHandlerContext$WriteTask.run(AbstractChannelHandlerContext.java:1247) ~[io.netty-netty-transport-4.1.93.Final.jar:4.1.93.Final]
        at io.netty.util.concurrent.AbstractEventExecutor.runTask(AbstractEventExecutor.java:174) ~[io.netty-netty-common-4.1.93.Final.jar:4.1.93.Final]
        at io.netty.util.concurrent.AbstractEventExecutor.safeExecute(AbstractEventExecutor.java:167) ~[io.netty-netty-common-4.1.93.Final.jar:4.1.93.Final]
        at io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:470) ~[io.netty-netty-common-4.1.93.Final.jar:4.1.93.Final]
        at io.netty.channel.epoll.EpollEventLoop.run(EpollEventLoop.java:403) ~[io.netty-netty-transport-classes-epoll-4.1.93.Final.jar:4.1.93.Final]
        at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:997) ~[io.netty-netty-common-4.1.93.Final.jar:4.1.93.Final]
        at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74) ~[io.netty-netty-common-4.1.93.Final.jar:4.1.93.Final]
        at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30) ~[io.netty-netty-common-4.1.93.Final.jar:4.1.93.Final]
        at java.lang.Thread.run(Thread.java:829) ~[?:?]
Caused by: io.netty.channel.StacklessClosedChannelException
        at io.netty.channel.AbstractChannel$AbstractUnsafe.write(Object, ChannelPromise)(Unknown Source) ~[io.netty-netty-transport-4.1.93.Final.jar:4.1.93.Final]

Modifications

• Make the log to warn level.
• Avoid print released messages.

Verifying this change

• Make sure that the change passes the CI checks.

Documentation

Check the box below.

Need to update docs?

• doc-required

  (If you need help on updating docs, create a doc issue)

• no-need-doc

  (Please explain why)

• doc

  (If this PR contains doc changes)