From 3e4ac80ddc282e9fb55dfb9404ab836f60964066 Mon Sep 17 00:00:00 2001 From: Justin Bertram Date: Fri, 28 Aug 2026 11:06:55 -0500 Subject: [PATCH] ARTEMIS-6217 mitigate spurious shutdown MQTT logs When the broker is processing MQTT packets (especially those related to QoS 2 message flows) a shutdown can cause the broker to log spurious messages, including several different stack-traces that are effectively harmless, but can alarm users. The broker should detect these and avoid logging them. Co-Authored-By: Claude --- .../protocol/mqtt/MQTTConnectionManager.java | 9 ++++++ .../core/protocol/mqtt/MQTTLogger.java | 9 ++---- .../protocol/mqtt/MQTTProtocolHandler.java | 10 +++++-- .../protocol/mqtt/MQTTPublishManager.java | 13 ++++++-- .../protocol/mqtt/MQTTSessionCallback.java | 9 ++++++ .../ActiveMQIDGeneratorStoppedException.java | 30 +++++++++++++++++++ .../core/server/ActiveMQMessageBundle.java | 3 +- .../impl/BatchIDGeneratorUnitTest.java | 15 ++++++---- 8 files changed, 81 insertions(+), 17 deletions(-) create mode 100644 artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/ActiveMQIDGeneratorStoppedException.java diff --git a/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTConnectionManager.java b/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTConnectionManager.java index 92f113af29fe..7e7e809ed0ab 100644 --- a/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTConnectionManager.java +++ b/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTConnectionManager.java @@ -16,15 +16,20 @@ */ package org.apache.activemq.artemis.core.protocol.mqtt; +import java.lang.invoke.MethodHandles; + import io.netty.buffer.ByteBufAllocator; import io.netty.handler.codec.mqtt.MqttConnectMessage; import io.netty.handler.codec.mqtt.MqttProperties; import io.netty.handler.codec.mqtt.MqttVersion; import org.apache.activemq.artemis.api.core.client.ActiveMQClient; +import org.apache.activemq.artemis.core.persistence.impl.journal.ActiveMQIDGeneratorStoppedException; import org.apache.activemq.artemis.core.server.ActiveMQServer; import org.apache.activemq.artemis.core.server.ServerSession; import org.apache.activemq.artemis.core.server.impl.ServerSessionImpl; import org.apache.activemq.artemis.utils.UUIDGenerator; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import static io.netty.handler.codec.mqtt.MqttProperties.MqttPropertyType.ASSIGNED_CLIENT_IDENTIFIER; import static io.netty.handler.codec.mqtt.MqttProperties.MqttPropertyType.AUTHENTICATION_METHOD; @@ -41,6 +46,8 @@ */ public class MQTTConnectionManager { + private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass()); + private MQTTSession session; public MQTTConnectionManager(MQTTSession session) { @@ -194,6 +201,8 @@ synchronized void disconnect(boolean failure) { try { session.stop(failure); session.getConnection().destroy(); + } catch (ActiveMQIDGeneratorStoppedException ignored) { + logger.debug("Unable to cleanly disconnect MQTT client {} because the storage manager is stopping", session.getState().getClientId(), ignored); } catch (Exception e) { MQTTLogger.LOGGER.errorDisconnectingClient(e); } finally { diff --git a/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTLogger.java b/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTLogger.java index 915886d7e543..e326510cbcfa 100644 --- a/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTLogger.java +++ b/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTLogger.java @@ -24,7 +24,7 @@ /** * Logger Codes 830000 - 839999 */ -@LogBundle(projectCode = "AMQ", regexID = "83[0-9]{4}") +@LogBundle(projectCode = "AMQ", regexID = "83[0-9]{4}", retiredIDs = {834015}) public interface MQTTLogger { MQTTLogger LOGGER = BundleFactory.newBundle(MQTTLogger.class, MQTTLogger.class.getPackage().getName()); @@ -77,11 +77,8 @@ public interface MQTTLogger { @LogMessage(id = 834013, value = "Invalid MQTT session state message. Will not load this state into memory.", level = LogMessage.Level.WARN) void errorDeserializingStateMessage(Exception e); - @LogMessage(id = 834014, value = "MQTT client {} sent PUBREC for packet {}, but acknowledgement failed. Internal consumer {} not found. Internal session is {}.", level = LogMessage.Level.WARN) - void failedToAckMessageConsumerNotFound(String clientId, int packetId, long consumerId, String closed); - - @LogMessage(id = 834015, value = "Unable to handle MQTT packet [{}] from {}. Internal session is closed.", level = LogMessage.Level.ERROR) - void internalSessionClosed(String packet, String clientId); + @LogMessage(id = 834014, value = "MQTT client {} sent PUBREC for packet {}, but acknowledgement failed. Internal consumer {} not found.", level = LogMessage.Level.WARN) + void failedToAckMessageConsumerNotFound(String clientId, int packetId, long consumerId); @LogMessage(id = 834016, value = "Storage operation failed. Error code: {}; message: {}", level = LogMessage.Level.ERROR) void storageOperationError(int errorCode, String errorMessage); diff --git a/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTProtocolHandler.java b/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTProtocolHandler.java index 5b42455a1681..8fdfa4c4ad6e 100644 --- a/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTProtocolHandler.java +++ b/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTProtocolHandler.java @@ -43,6 +43,7 @@ import io.netty.util.CharsetUtil; import io.netty.util.ReferenceCountUtil; import org.apache.activemq.artemis.api.core.ActiveMQSecurityException; +import org.apache.activemq.artemis.api.core.ActiveMQShutdownException; import org.apache.activemq.artemis.api.core.Pair; import org.apache.activemq.artemis.core.io.IOCallback; import org.apache.activemq.artemis.core.persistence.OperationContext; @@ -126,7 +127,8 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) { } if (session.getServerSession() != null && session.getServerSession().isClosed()) { - MQTTLogger.LOGGER.internalSessionClosed(MQTTUtil.getMessageForLogging(message, session.getVersion()), session.getState().getClientId()); + // the client sent a packet after its session was closed (e.g. during shutdown or disconnect) + logger.debug("Unable to handle MQTT packet [{}] from {}. Internal session is closed.", MQTTUtil.getMessageForLogging(message, session.getVersion()), session.getState().getClientId()); if (session.getVersion() == MQTTVersion.MQTT_5) { sendDisconnect(MQTTReasonCodes.IMPLEMENTATION_SPECIFIC_ERROR); } @@ -214,7 +216,11 @@ public void act(MqttMessage message) { disconnect(true); } } catch (Exception e) { - MQTTLogger.LOGGER.errorProcessingPacket(session.getState().getClientId(), MQTTUtil.getMessageForLogging(message, session.getVersion()), e.getMessage(), e); + if (e instanceof ActiveMQShutdownException ignored) { + logger.debug("Unable to process MQTT packet for client {} because the broker is shutting down; packet: {}", session.getState().getClientId(), MQTTUtil.getMessageForLogging(message, session.getVersion()), ignored); + } else { + MQTTLogger.LOGGER.errorProcessingPacket(session.getState().getClientId(), MQTTUtil.getMessageForLogging(message, session.getVersion()), e.getMessage(), e); + } if (session.getVersion() == MQTTVersion.MQTT_5) { sendDisconnect(MQTTReasonCodes.IMPLEMENTATION_SPECIFIC_ERROR); } diff --git a/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTPublishManager.java b/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTPublishManager.java index 8c337426f295..188c6530d958 100644 --- a/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTPublishManager.java +++ b/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTPublishManager.java @@ -38,6 +38,7 @@ import org.apache.activemq.artemis.api.core.Message; import org.apache.activemq.artemis.api.core.RoutingType; import org.apache.activemq.artemis.api.core.SimpleString; +import org.apache.activemq.artemis.core.persistence.impl.journal.ActiveMQIDGeneratorStoppedException; import org.apache.activemq.artemis.core.protocol.mqtt.exceptions.DisconnectException; import org.apache.activemq.artemis.core.server.ServerConsumer; import org.apache.activemq.artemis.core.server.ServerProducer; @@ -338,7 +339,11 @@ private void acknowledgeDelivery(int packetId, boolean needsPubRel) throws Excep if (delivery != null) { ServerConsumer consumer = session.getServerSession().locateConsumer(delivery.getConsumerId()); if (consumer == null) { - MQTTLogger.LOGGER.failedToAckMessageConsumerNotFound(state.getClientId(), packetId, delivery.getConsumerId(), session.getServerSession().isClosed() ? "closed" : "not closed"); + if (session.getServerSession().isClosed()) { + logger.debug("MQTT client {} sent an acknowledgement for packet {}, but internal consumer {} was not found because the session is closed.", state.getClientId(), packetId, delivery.getConsumerId()); + } else { + MQTTLogger.LOGGER.failedToAckMessageConsumerNotFound(state.getClientId(), packetId, delivery.getConsumerId()); + } sendAcknowledgementReply(packetId, MQTTReasonCodes.PACKET_IDENTIFIER_NOT_FOUND, needsPubRel); return; } @@ -360,7 +365,11 @@ private void acknowledgeDelivery(int packetId, boolean needsPubRel) throws Excep if (tx != null) { tx.rollback(); } - MQTTLogger.LOGGER.failedToAckMessage(state.getClientId(), e.getMessage()); + if (e instanceof ActiveMQIDGeneratorStoppedException ignored) { + logger.debug("MQTT client {} failed to acknowledge message because the storage manager is stopping", state.getClientId(), ignored); + } else { + MQTTLogger.LOGGER.failedToAckMessage(state.getClientId(), e.getMessage()); + } sendAcknowledgementReply(packetId, MQTTReasonCodes.PACKET_IDENTIFIER_NOT_FOUND, needsPubRel); } } diff --git a/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTSessionCallback.java b/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTSessionCallback.java index 5cd349748490..58c6b55901ea 100644 --- a/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTSessionCallback.java +++ b/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTSessionCallback.java @@ -16,14 +16,21 @@ */ package org.apache.activemq.artemis.core.protocol.mqtt; +import java.lang.invoke.MethodHandles; + import org.apache.activemq.artemis.api.core.SimpleString; +import org.apache.activemq.artemis.core.persistence.impl.journal.ActiveMQIDGeneratorStoppedException; import org.apache.activemq.artemis.core.server.MessageReference; import org.apache.activemq.artemis.core.server.ServerConsumer; import org.apache.activemq.artemis.spi.core.protocol.SessionCallback; import org.apache.activemq.artemis.spi.core.remoting.ReadyListener; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; public class MQTTSessionCallback implements SessionCallback { + private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass()); + private final MQTTSession session; private final MQTTConnection connection; private final int defaultMaximumInFlightPublishMessages; @@ -50,6 +57,8 @@ public int sendMessage(MessageReference ref, int deliveryCount) { try { session.getMqttPublishManager().publishToClient(ref.getMessage().toCore(), consumer); + } catch (ActiveMQIDGeneratorStoppedException ignored) { + logger.debug("Unable to send message to MQTT client because the storage manager is stopping; consumer: {}; message: {}", consumer, ref, ignored); } catch (Exception e) { MQTTLogger.LOGGER.unableToSendMessage(session.getState().getClientId(), ref, e); } diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/ActiveMQIDGeneratorStoppedException.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/ActiveMQIDGeneratorStoppedException.java new file mode 100644 index 000000000000..73296d8f8bcc --- /dev/null +++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/ActiveMQIDGeneratorStoppedException.java @@ -0,0 +1,30 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.activemq.artemis.core.persistence.impl.journal; + +/** + * Thrown by an {@link org.apache.activemq.artemis.utils.IDGenerator} when an ID is requested after the generator has + * been stopped. + */ +public class ActiveMQIDGeneratorStoppedException extends RuntimeException { + + private static final long serialVersionUID = 8328635365036357836L; + + public ActiveMQIDGeneratorStoppedException(String message) { + super(message); + } +} diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQMessageBundle.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQMessageBundle.java index 35cf342bc07c..9ab509ae6a74 100644 --- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQMessageBundle.java +++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQMessageBundle.java @@ -48,6 +48,7 @@ import org.apache.activemq.artemis.api.core.SimpleString; import org.apache.activemq.artemis.api.core.TransportConfiguration; import org.apache.activemq.artemis.core.io.SequentialFile; +import org.apache.activemq.artemis.core.persistence.impl.journal.ActiveMQIDGeneratorStoppedException; import org.apache.activemq.artemis.core.postoffice.Binding; import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.ReplicationSyncFileMessage; import org.apache.activemq.artemis.core.security.CheckType; @@ -533,7 +534,7 @@ IllegalStateException invalidRoutingTypeUpdate(String queueName, IllegalArgumentException positivePowerOfTwo(String name, Number val); @Message(id = 229257, value = "IDGenerator has been stopped") - RuntimeException idGeneratorStopped(); + ActiveMQIDGeneratorStoppedException idGeneratorStopped(); @Message(id = 229258, value = "Invalid cluster bridge message! No queue IDs defined in the property {}") ActiveMQIllegalStateException noQueueIdsDefined(SimpleString idsHeaderName); diff --git a/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/persistence/impl/BatchIDGeneratorUnitTest.java b/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/persistence/impl/BatchIDGeneratorUnitTest.java index c889c46075d2..7f2a05d53762 100644 --- a/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/persistence/impl/BatchIDGeneratorUnitTest.java +++ b/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/persistence/impl/BatchIDGeneratorUnitTest.java @@ -16,11 +16,6 @@ */ package org.apache.activemq.artemis.tests.unit.core.persistence.impl; -import static org.junit.jupiter.api.Assertions.assertEquals; -import static org.junit.jupiter.api.Assertions.assertFalse; -import static org.junit.jupiter.api.Assertions.assertThrowsExactly; -import static org.junit.jupiter.api.Assertions.assertTrue; - import java.io.File; import java.util.ArrayList; import java.util.List; @@ -33,12 +28,19 @@ import org.apache.activemq.artemis.core.journal.RecordInfo; import org.apache.activemq.artemis.core.journal.impl.JournalImpl; import org.apache.activemq.artemis.core.persistence.StorageManager; +import org.apache.activemq.artemis.core.persistence.impl.journal.ActiveMQIDGeneratorStoppedException; import org.apache.activemq.artemis.core.persistence.impl.journal.BatchingIDGenerator; import org.apache.activemq.artemis.core.persistence.impl.journal.JournalRecordIds; import org.apache.activemq.artemis.core.persistence.impl.nullpm.NullStorageManager; import org.apache.activemq.artemis.tests.util.ActiveMQTestBase; import org.junit.jupiter.api.Test; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertThrowsExactly; +import static org.junit.jupiter.api.Assertions.assertTrue; + public class BatchIDGeneratorUnitTest extends ActiveMQTestBase { @Test @@ -120,7 +122,8 @@ public void testSequence() throws Exception { } private void validateStoppedGenerator(BatchingIDGenerator stoppedGenerator) { - assertThrowsExactly(RuntimeException.class, stoppedGenerator::generateID); + assertThrowsExactly(ActiveMQIDGeneratorStoppedException.class, stoppedGenerator::generateID); + assertThrows(RuntimeException.class, stoppedGenerator::generateID); } protected void loadIDs(final Journal journal, final BatchingIDGenerator batch) throws Exception {