Fix NPE cased by inflating message (#1045)

streamnative · Jul 27, 2023 · a1ebce4 · a1ebce4
1 parent 9aed8bf
commit a1ebce4
Show file tree

Hide file tree

Showing 5 changed files with 33 additions and 2 deletions.
diff --git a/mqtt-impl/src/main/java/io/streamnative/pulsar/handlers/mqtt/Connection.java b/mqtt-impl/src/main/java/io/streamnative/pulsar/handlers/mqtt/Connection.java
@@ -266,7 +266,11 @@ public void sendConnAck() {
  builder.responseInformation(resInformation.value());
  }
  MqttAck connAck = builder.build();
- sendAck(connAck).thenAccept(__ -> assignState(CONNECT_ACK, ESTABLISHED));
+ assignState(CONNECT_ACK, ESTABLISHED);
+ sendAck(connAck).exceptionally(ex -> {
+ assignState(ESTABLISHED, DISCONNECTED);
+ return null;
+ });
  if (log.isDebugEnabled()) {
  log.debug("The CONNECT message has been processed. CId={}", clientId);
  }

diff --git a/mqtt-impl/src/main/java/io/streamnative/pulsar/handlers/mqtt/MQTTCommonInboundHandler.java b/mqtt-impl/src/main/java/io/streamnative/pulsar/handlers/mqtt/MQTTCommonInboundHandler.java
@@ -50,7 +50,7 @@ public void channelRead(ChannelHandlerContext ctx, Object message) {
  checkArgument(message instanceof MqttAdapterMessage);
  MqttAdapterMessage adapterMsg = (MqttAdapterMessage) message;
  MqttMessage mqttMessage = adapterMsg.getMqttMessage();
- ProtocolMethodProcessor processor = processors.computeIfAbsent(adapterMsg.getClientId(), key -> {
+ final ProtocolMethodProcessor processor = processors.computeIfAbsent(adapterMsg.getClientId(), key -> {
  MQTTBrokerProtocolMethodProcessor p = new MQTTBrokerProtocolMethodProcessor(mqttService, ctx);
  CompletableFuture<Void> inactiveFuture = p.getInactiveFuture();
  inactiveFuture.whenComplete((id, ex) -> {
@@ -65,6 +65,21 @@ public void channelRead(ChannelHandlerContext ctx, Object message) {
  log.debug("Inbound handler read message : type={}, clientId : {} adapter encodeType : {}", messageType,
  adapterMsg.getClientId(), adapterMsg.getEncodeType());
  }
+ if (messageType != MqttMessageType.CONNECT && !processor.connectionEstablished()) {
+ if (adapterMsg.fromProxy()) {
+ if (log.isDebugEnabled()) {
+ log.debug("[{}] Ignore inflating message from proxy. message:{}",
+ ctx.channel().remoteAddress(), mqttMessage);
+ }
+ ReferenceCountUtil.safeRelease(mqttMessage); // release publish packet
+ return;
+ } else {
+ log.warn("[{}] Receive message before connect. message:{}",
+ ctx.channel().remoteAddress(), mqttMessage);
+ ReferenceCountUtil.safeRelease(mqttMessage); // release publish packet
+ ctx.channel().close();
+ }
+ }
  switch (messageType) {
  case CONNECT:
  processor.processConnect(adapterMsg);

diff --git a/mqtt-impl/src/main/java/io/streamnative/pulsar/handlers/mqtt/ProtocolMethodProcessor.java b/mqtt-impl/src/main/java/io/streamnative/pulsar/handlers/mqtt/ProtocolMethodProcessor.java
@@ -43,4 +43,6 @@ public interface ProtocolMethodProcessor {
  void processPingReq(MqttAdapterMessage msg);
 
  void processAuthReq(MqttAdapterMessage msg);
+
+ boolean connectionEstablished();
 }
diff --git a/...ain/java/io/streamnative/pulsar/handlers/mqtt/proxy/MQTTProxyProtocolMethodProcessor.java b/...ain/java/io/streamnative/pulsar/handlers/mqtt/proxy/MQTTProxyProtocolMethodProcessor.java
@@ -229,6 +229,11 @@ public void processPingReq(final MqttAdapterMessage msg) {
  });
  }
 
+ @Override
+ public boolean connectionEstablished() {
+ return connection != null && connection.getState() == Connection.ConnectionState.ESTABLISHED;
+ }
+
  @Override
  public void processDisconnect(final MqttAdapterMessage msg) {
  if (isDisconnected.compareAndSet(false, true)) {

diff --git a/.../java/io/streamnative/pulsar/handlers/mqtt/support/MQTTBrokerProtocolMethodProcessor.java b/.../java/io/streamnative/pulsar/handlers/mqtt/support/MQTTBrokerProtocolMethodProcessor.java
@@ -321,6 +321,11 @@ public void processPingReq(MqttAdapterMessage adapter) {
  connection.send(pingResp());
  }
 
+ @Override
+ public boolean connectionEstablished() {
+ return connection != null && connection.getState() == Connection.ConnectionState.ESTABLISHED;
+ }
+
  @Override
  public void processSubscribe(MqttAdapterMessage adapter) {
  MqttSubscribeMessage msg = (MqttSubscribeMessage) adapter.getMqttMessage();