Fix IllegalReferenceCountException exception to break the callback (#…

…1041)
streamnative · Jul 27, 2023 · 99c0414 · 99c0414
1 parent 91086d3
commit 99c0414
Showing 1 changed file with 3 additions and 1 deletion.
diff --git a/mqtt-impl/src/main/java/io/streamnative/pulsar/handlers/mqtt/adapter/AdapterChannel.java b/mqtt-impl/src/main/java/io/streamnative/pulsar/handlers/mqtt/adapter/AdapterChannel.java
@@ -40,6 +40,7 @@ public AdapterChannel(MQTTProxyAdapter adapter,
 
  public CompletableFuture<Void> writeAndFlush(final MqttAdapterMessage adapterMsg) {
  checkArgument(StringUtils.isNotBlank(adapterMsg.getClientId()), "clientId is blank");
+ final String clientId = adapterMsg.getClientId();
  adapterMsg.setEncodeType(MqttAdapterMessage.EncodeType.ADAPTER_MESSAGE);
  CompletableFuture<Void> future = channelFuture.thenCompose(channel -> {
  if (!channel.isActive()) {
@@ -49,7 +50,8 @@ public CompletableFuture<Void> writeAndFlush(final MqttAdapterMessage adapterMsg
  return FutureUtils.completableFuture(channel.writeAndFlush(adapterMsg));
  });
  future.exceptionally(ex -> {
- log.error("[AdapterChannel] Proxy write to broker {} message {} failed.", broker, adapterMsg, ex);
+ log.warn("[AdapterChannel][{}] Proxy write to broker {} failed."
+ + " error message: {}", clientId, broker, ex.getMessage());
  return null;
  });
  return future;