diff --git a/README.md b/README.md index a44b2946d..6d1f5c489 100644 --- a/README.md +++ b/README.md @@ -23,6 +23,14 @@ For example, set the following parameters to true in broker.conf enableLmq = true enableMultiDispatch = true ``` + +### Deployment Modes +RocketMQ-MQTT supports two deployment modes: + ++ Cluster-Ready Mode (Meta Enabled, Default): This is the default and full-featured mode. It relies on a meta service (based on RocketMQ's KV storage) for dynamic configuration, cluster node discovery, and advanced features like Retain Messages and Will Messages. This mode is suitable for production and high-availability environments. + ++ Standalone Mode (Meta Disabled): This is a simplified mode designed for single-node deployments or scenarios where advanced features are not required. It operates without any dependency on the meta service, making configuration and deployment much simpler. In this mode, features like Retain Messages and Will Messages are disabled. + ### Build Requirements The current project requires JDK 1.8.x. When building on MAC arm64 the recommended JDK 1.8 must be based on 386 architecture or use Maven flag `-Dos.arch=x86_64` when building with Maven. @@ -43,15 +51,17 @@ cd conf ``` Some important configuration items in the **service.conf** configuration file -| **Config Key** | **Instruction** | -|-----------------------|---------------------------------------------------------------| -| username | used for auth | -| secretKey | used for auth | -| NAMESRV_ADDR | specify namesrv address | -| eventNotifyRetryTopic | notify event retry topic | -| clientRetryTopic | client retry topic | -| metaAddr | meta all nodes ip:port. Same as membersAddress in meta.config | - +| **Config Key** | **Instruction** | +|-----------------------|----------------------------------------------------------------------------------------------------------------------------------| +| username | used for auth | +| secretKey | used for auth | +| NAMESRV_ADDR | specify namesrv address | +| eventNotifyRetryTopic | notify event retry topic | +| clientRetryTopic | client retry topic | +| enableMetaModule | (Optional) Switch for meta module. Defaults to true. Set to `false` to enable Standalone Mode. | +| metaAddr | meta all nodes ip:port. Same as membersAddress in meta.config. **Required when `enableMetaModule` is `true`.** | +| staticFirstTopics | A comma-separated list of all first-level MQTT topics. E.g., `chat,iot`. **Required when `enableMetaModule` is `false`.** | +| localAddress | The address for the server's internal RPC loopback. `127.0.0.1` is recommended. **Required when `enableMetaModule` is `false`.** | And some configuration items in the **meta.conf** configuration file @@ -80,6 +90,7 @@ sh mqadmin updateKvConfig -s LMQ -k ALL_FIRST_TOPICS -v {topic1,topic2} -n {name sh mqadmin updateKvConfig -s LMQ -k {topic} -v {topic/+} -n {namesrv} ``` 6. Start Process +Note: In standalone mode, Only mqtt.sh is needed to start. ```shell cd bin sh meta.sh start diff --git a/mqtt-common/src/main/java/org/apache/rocketmq/mqtt/common/model/MqttRetainException.java b/mqtt-common/src/main/java/org/apache/rocketmq/mqtt/common/model/MqttRetainException.java new file mode 100644 index 000000000..523e8be9e --- /dev/null +++ b/mqtt-common/src/main/java/org/apache/rocketmq/mqtt/common/model/MqttRetainException.java @@ -0,0 +1,26 @@ +/* + * 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.rocketmq.mqtt.common.model; + +public class MqttRetainException extends RuntimeException { + public MqttRetainException() { } + + public MqttRetainException(String message) { + super(message); + } +} \ No newline at end of file diff --git a/mqtt-cs/src/main/java/org/apache/rocketmq/mqtt/cs/protocol/mqtt/MqttPacketDispatcher.java b/mqtt-cs/src/main/java/org/apache/rocketmq/mqtt/cs/protocol/mqtt/MqttPacketDispatcher.java index d6563f236..4bf640e5a 100644 --- a/mqtt-cs/src/main/java/org/apache/rocketmq/mqtt/cs/protocol/mqtt/MqttPacketDispatcher.java +++ b/mqtt-cs/src/main/java/org/apache/rocketmq/mqtt/cs/protocol/mqtt/MqttPacketDispatcher.java @@ -23,6 +23,7 @@ import io.netty.channel.SimpleChannelInboundHandler; import io.netty.handler.codec.mqtt.MqttConnectMessage; import io.netty.handler.codec.mqtt.MqttMessage; +import io.netty.handler.codec.mqtt.MqttMessageType; import io.netty.handler.codec.mqtt.MqttPubAckMessage; import io.netty.handler.codec.mqtt.MqttPublishMessage; import io.netty.handler.codec.mqtt.MqttSubscribeMessage; @@ -32,9 +33,11 @@ import org.apache.rocketmq.mqtt.common.hook.UpstreamHookManager; import org.apache.rocketmq.mqtt.common.model.MqttMessageUpContext; import org.apache.rocketmq.mqtt.common.util.HostInfo; +import org.apache.rocketmq.mqtt.cs.channel.ChannelCloseFrom; import org.apache.rocketmq.mqtt.cs.channel.ChannelDecodeException; import org.apache.rocketmq.mqtt.cs.channel.ChannelException; import org.apache.rocketmq.mqtt.cs.channel.ChannelInfo; +import org.apache.rocketmq.mqtt.cs.channel.ChannelManager; import org.apache.rocketmq.mqtt.cs.protocol.mqtt.handler.MqttConnectHandler; import org.apache.rocketmq.mqtt.cs.protocol.mqtt.handler.MqttDisconnectHandler; import org.apache.rocketmq.mqtt.cs.protocol.mqtt.handler.MqttPingHandler; @@ -90,6 +93,9 @@ public class MqttPacketDispatcher extends SimpleChannelInboundHandler @Resource private ChannelManager channelManager; + @Resource + private ServiceConf serviceConf; + @Resource private SessionLoop sessionLoop; @@ -111,15 +115,21 @@ public void doHandler(ChannelHandlerContext ctx, MqttConnectMessage connectMessa sessionLoop.loadSession(ChannelInfo.getClientId(channel), channel); // save will message - WillMessage willMessage = null; if (variableHeader.isWillFlag()) { + if (!serviceConf.isEnableMetaModule()) { + String clientId = ChannelInfo.getClientId(channel); + logger.error("Client [{}] trying to set a Will Message, but the meta module is disabled. Connection refused.", clientId); + channelManager.closeConnect(channel, ChannelCloseFrom.SERVER, "Will Message feature is disabled"); + return; + } + if (payload.willTopic() == null || payload.willMessageInBytes() == null) { logger.error("Will message and will topic can not be empty"); channelManager.closeConnect(channel, ChannelCloseFrom.SERVER, "Will message and will topic can not be empty"); return; } - willMessage = new WillMessage(payload.willTopic(), payload.willMessageInBytes(), variableHeader.isWillRetain(), variableHeader.willQos()); + WillMessage willMessage = new WillMessage(payload.willTopic(), payload.willMessageInBytes(), variableHeader.isWillRetain(), variableHeader.willQos()); willLoop.addWillMessage(channel, willMessage); } diff --git a/mqtt-cs/src/main/java/org/apache/rocketmq/mqtt/cs/session/loop/WillLoop.java b/mqtt-cs/src/main/java/org/apache/rocketmq/mqtt/cs/session/loop/WillLoop.java index a1dd337a6..2726a3ded 100644 --- a/mqtt-cs/src/main/java/org/apache/rocketmq/mqtt/cs/session/loop/WillLoop.java +++ b/mqtt-cs/src/main/java/org/apache/rocketmq/mqtt/cs/session/loop/WillLoop.java @@ -31,6 +31,7 @@ import org.apache.rocketmq.mqtt.cs.channel.ChannelInfo; import org.apache.rocketmq.mqtt.cs.config.WillLoopConf; import org.apache.rocketmq.mqtt.cs.session.infly.MqttMsgId; +import org.apache.rocketmq.mqtt.ds.config.ServiceConf; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.stereotype.Component; @@ -63,12 +64,21 @@ public class WillLoop { @Resource private WillMsgSender willMsgSender; + @Resource + private ServiceConf serviceConf; + public void setWillLoopConf(WillLoopConf willLoopConf) { this.willLoopConf = willLoopConf; } @PostConstruct public void init() { + + if (!serviceConf.isEnableMetaModule()) { + logger.info("Meta module is disabled, WillLoop will not be initialized."); + return; + } + aliveService.scheduleWithFixedDelay(() -> csLoop(), 15 * 1000, 10 * 1000, TimeUnit.MILLISECONDS); aliveService.scheduleWithFixedDelay(() -> masterLoop(), 10 * 1000, 10 * 1000, TimeUnit.MILLISECONDS); @@ -239,6 +249,11 @@ private void _handleShutDownCS(String ip, AtomicInteger loopNum) { } public void closeConnect(Channel channel, String clientId, String reason) { + + if (!serviceConf.isEnableMetaModule()) { + return; + } + String ip = IpUtil.getLocalAddressCompatible(); String willKey = ip + Constants.CTRL_1 + clientId; CompletableFuture willMessageFuture = willMsgPersistManager.get(willKey); diff --git a/mqtt-ds/src/main/java/org/apache/rocketmq/mqtt/ds/config/ServiceConf.java b/mqtt-ds/src/main/java/org/apache/rocketmq/mqtt/ds/config/ServiceConf.java index 337a3bc50..984af0ce1 100644 --- a/mqtt-ds/src/main/java/org/apache/rocketmq/mqtt/ds/config/ServiceConf.java +++ b/mqtt-ds/src/main/java/org/apache/rocketmq/mqtt/ds/config/ServiceConf.java @@ -46,6 +46,10 @@ public class ServiceConf { private String metaAddr; private long retainMsgExpire = 3 * 24 * 60 * 60 * 1000L; + private boolean enableMetaModule = true; + private String staticFirstTopics; + private String localAddress; + @PostConstruct public void init() throws IOException { ClassPathResource classPathResource = new ClassPathResource(CONF_FILE_NAME); @@ -62,8 +66,17 @@ public void init() throws IOException { if (StringUtils.isBlank(eventNotifyRetryTopic)) { throw new RemoteException("eventNotifyRetryTopic is blank"); } - if (StringUtils.isBlank(metaAddr)) { - throw new RemoteException("metaAddr is blank"); + if (this.enableMetaModule) { + if (StringUtils.isBlank(metaAddr)) { + throw new IOException("metaAddr is blank when meta module is enabled"); + } + } else { + if (StringUtils.isBlank(staticFirstTopics)) { + throw new IOException("When meta module is disabled, staticFirstTopics cannot be blank."); + } + if (StringUtils.isBlank(localAddress)) { + throw new IOException("When meta module is disabled, localAddress cannnot be blank."); + } } } @@ -158,4 +171,28 @@ public long getRetainMsgExpire() { public void setRetainMsgExpire(long retainMsgExpire) { this.retainMsgExpire = retainMsgExpire; } + + public String getStaticFirstTopics() { + return staticFirstTopics; + } + + public void setStaticFirstTopics(String staticFirstTopics) { + this.staticFirstTopics = staticFirstTopics; + } + + public boolean isEnableMetaModule() { + return enableMetaModule; + } + + public void setEnableMetaModule(boolean enableMetaModule) { + this.enableMetaModule = enableMetaModule; + } + + public String getLocalAddress() { + return localAddress; + } + + public void setLocalAddress(String localAddress) { + this.localAddress = localAddress; + } } diff --git a/mqtt-ds/src/main/java/org/apache/rocketmq/mqtt/ds/meta/MetaRpcClient.java b/mqtt-ds/src/main/java/org/apache/rocketmq/mqtt/ds/meta/MetaRpcClient.java index e32e73579..f0dc2bd6f 100644 --- a/mqtt-ds/src/main/java/org/apache/rocketmq/mqtt/ds/meta/MetaRpcClient.java +++ b/mqtt-ds/src/main/java/org/apache/rocketmq/mqtt/ds/meta/MetaRpcClient.java @@ -59,6 +59,12 @@ public class MetaRpcClient { @PostConstruct public void init() throws InterruptedException, TimeoutException { + + if (!serviceConf.isEnableMetaModule()) { + logger.info("Meta module is disabled, MetaRpcClient will not be initialized."); + return; + } + initRpcServer(); cliClientService = new CliClientServiceImpl(); cliClientService.init(new CliOptions()); diff --git a/mqtt-ds/src/main/java/org/apache/rocketmq/mqtt/ds/notify/NotifyManager.java b/mqtt-ds/src/main/java/org/apache/rocketmq/mqtt/ds/notify/NotifyManager.java index e027fa563..6cb8b9b74 100644 --- a/mqtt-ds/src/main/java/org/apache/rocketmq/mqtt/ds/notify/NotifyManager.java +++ b/mqtt-ds/src/main/java/org/apache/rocketmq/mqtt/ds/notify/NotifyManager.java @@ -73,7 +73,6 @@ public class NotifyManager { private NettyRemotingClient remotingClient; private DefaultMQProducer defaultMQProducer; - @Resource private ServiceConf serviceConf; @@ -85,6 +84,7 @@ public class NotifyManager { @PostConstruct public void init() throws MQClientException { + defaultMQPushConsumer = MqFactory.buildDefaultMQPushConsumer(dispatcherConsumerGroup, serviceConf.getProperties(), new Dispatcher()); defaultMQPushConsumer.setPullInterval(1); defaultMQPushConsumer.setConsumeMessageBatchMaxSize(64); @@ -94,28 +94,57 @@ public void init() throws MQClientException { defaultMQProducer = MqFactory.buildDefaultMQProducer(MixAll.CID_RMQ_SYS_PREFIX + "NotifyRetrySend", serviceConf.getProperties()); + NettyClientConfig config = new NettyClientConfig(); + remotingClient = new NettyRemotingClient(config); + try { - defaultMQPushConsumer.start(); defaultMQProducer.start(); + remotingClient.start(); } catch (Exception e) { - logger.error("", e); + logger.error("Fatal: NotifyManager failed to start producer or remoting client.", e); + throw new MQClientException("Failed to start producer or remoting client", e); } - scheduler = new ScheduledThreadPoolExecutor(1, new ThreadFactoryImpl("Refresh_Notify_Topic_")); - scheduler.scheduleWithFixedDelay(() -> { - try { - refresh(); - } catch (Exception e) { - logger.error("", e); + if (serviceConf.isEnableMetaModule()) { + logger.info("Meta module is enabled. Topics will be discovered dynamically."); + scheduler = new ScheduledThreadPoolExecutor(1, new ThreadFactoryImpl("Refresh_Notify_Topic_")); + scheduler.scheduleWithFixedDelay(() -> { + try { + refresh(); + } catch (Exception e) { + logger.error("Error during scheduled topic refresh", e); + } + }, 0, 5, TimeUnit.SECONDS); + } else { + logger.info("Meta module is disabled. Subscribing to statically configured topics."); + String staticTopicsConfig = serviceConf.getStaticFirstTopics(); + + if (StringUtils.isNotBlank(staticTopicsConfig)) { + String[] topicArray = staticTopicsConfig.split(","); + for (String topic : topicArray) { + String trimmedTopic = topic.trim(); + if (StringUtils.isNotBlank(trimmedTopic)) { + subscribe(trimmedTopic); + this.topics.add(trimmedTopic); + logger.info("Successfully configured subscription for static topic: {}", trimmedTopic); + } + } + } else { + logger.warn("Meta module is disabled, but 'mqtt.static.first.topics' is not configured. NotifyManager may not receive any application messages."); } - }, 0, 5, TimeUnit.SECONDS); + } - NettyClientConfig config = new NettyClientConfig(); - remotingClient = new NettyRemotingClient(config); - remotingClient.start(); + try { + defaultMQPushConsumer.start(); + logger.info("NotifyManager's DefaultMQPushConsumer started successfully."); + } catch (Exception e) { + logger.error("Fatal: NotifyManager failed to start the core message consumer.", e); + throw new MQClientException("Failed to start the core message consumer", e); + } } private void refresh() throws MQClientException { + Set tmp = metaPersistManager.getAllFirstTopics(); logger.info("Notify Manager is refreshing, all first topic is " + tmp); @@ -214,7 +243,18 @@ private void setPubTopic(MessageEvent messageEvent, MessageExt message) { public void notifyMessage(Set messageEvents) throws MQBrokerException, RemotingException, InterruptedException, MQClientException { - Set connectorNodes = metaPersistManager.getConnectNodeSet(); + Set connectorNodes; + if (serviceConf.isEnableMetaModule()) { + connectorNodes = metaPersistManager.getConnectNodeSet(); + } else { + String localAddress = serviceConf.getLocalAddress(); + if (StringUtils.isBlank(localAddress)) { + logger.error("Meta module is disabled, but 'mqtt.local.address' is not configured. Cannot notify messages."); + return; + } + connectorNodes = java.util.Collections.singleton(localAddress); + } + if (connectorNodes == null || connectorNodes.isEmpty()) { throw new RemotingException("No Connect Nodes"); } @@ -250,12 +290,14 @@ public void notifyMessage(Set messageEvents) throws } protected boolean doNotify(String node, Set messageEvents) { - Set connectorNodes = metaPersistManager.getConnectNodeSet(); - if (connectorNodes == null || connectorNodes.isEmpty()) { - return false; - } - if (!connectorNodes.contains(node)) { - return true; + if (serviceConf.isEnableMetaModule()) { + Set connectorNodes = metaPersistManager.getConnectNodeSet(); + if (connectorNodes == null || connectorNodes.isEmpty()) { + return false; + } + if (!connectorNodes.contains(node)) { + return true; + } } try { RemotingCommand eventCommand = createMsgEventCommand(messageEvents); diff --git a/mqtt-ds/src/main/java/org/apache/rocketmq/mqtt/ds/upstream/processor/PublishProcessor.java b/mqtt-ds/src/main/java/org/apache/rocketmq/mqtt/ds/upstream/processor/PublishProcessor.java index 8de2726ed..7256a3df3 100644 --- a/mqtt-ds/src/main/java/org/apache/rocketmq/mqtt/ds/upstream/processor/PublishProcessor.java +++ b/mqtt-ds/src/main/java/org/apache/rocketmq/mqtt/ds/upstream/processor/PublishProcessor.java @@ -29,10 +29,12 @@ import org.apache.rocketmq.mqtt.common.hook.HookResult; import org.apache.rocketmq.mqtt.common.model.Message; import org.apache.rocketmq.mqtt.common.model.MqttMessageUpContext; +import org.apache.rocketmq.mqtt.common.model.MqttRetainException; import org.apache.rocketmq.mqtt.common.model.MqttTopic; import org.apache.rocketmq.mqtt.common.model.StoreResult; import org.apache.rocketmq.mqtt.common.util.MessageUtil; import org.apache.rocketmq.mqtt.common.util.TopicUtils; +import org.apache.rocketmq.mqtt.ds.config.ServiceConf; import org.apache.rocketmq.mqtt.ds.meta.FirstTopicManager; import org.apache.rocketmq.mqtt.ds.meta.WildcardManager; import org.apache.rocketmq.mqtt.ds.upstream.UpstreamProcessor; @@ -43,11 +45,16 @@ import javax.annotation.Resource; import java.util.Set; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionException; @Component public class PublishProcessor implements UpstreamProcessor, WillMsgSender { private static Logger logger = LoggerFactory.getLogger(PublishProcessor.class); + + @Resource + private ServiceConf serviceConf; + @Resource private LmqQueueStore lmqQueueStore; @@ -62,13 +69,57 @@ public class PublishProcessor implements UpstreamProcessor, WillMsgSender { @Override public CompletableFuture process(MqttMessageUpContext context, MqttMessage mqttMessage) { - CompletableFuture r = put(context, mqttMessage); - return r.thenCompose(storeResult -> HookResult.newHookResult(HookResult.SUCCESS, null, - JSON.toJSONBytes(storeResult))); + CompletableFuture storeFuture = put(context, mqttMessage); + + return storeFuture + .thenCompose(storeResult -> { + return HookResult.newHookResult(HookResult.SUCCESS, null, JSON.toJSONBytes(storeResult)); + }) + .exceptionally(throwable -> { + Throwable cause = throwable.getCause() != null ? throwable.getCause() : throwable; + + if (cause instanceof MqttRetainException) { + logger.warn("A defined business rejection occurred: {}", cause.getMessage()); + return new HookResult(HookResult.FAIL, cause.getMessage(), null); + } + logger.error("An unexpected error will be propagated up.", cause); + throw new CompletionException(cause); + }); } public CompletableFuture put(MqttMessageUpContext context, MqttMessage mqttMessage) { MqttPublishMessage mqttPublishMessage = (MqttPublishMessage) mqttMessage; + + if (mqttPublishMessage.fixedHeader().isRetain()) { + if (!serviceConf.isEnableMetaModule()) { + logger.error("Client [{}] on topic [{}] tried to publish a Retain Message, but the meta module is disabled. Rejecting request.", + context.getClientId(), mqttPublishMessage.variableHeader().topicName()); + + MqttRetainException exception = new MqttRetainException("Retain Message feature is disabled."); + + CompletableFuture failedFuture = new CompletableFuture<>(); + failedFuture.completeExceptionally(exception); + return failedFuture; + } else { + MqttPublishMessage retainedMqttPublishMessage = mqttPublishMessage.copy(); + //Change the retained flag of message that will send MQ is 0 + mqttPublishMessage = MessageUtil.removeRetainedFlag(mqttPublishMessage); + //Keep the retained flag of message that will store meta + Message metaMessage = MessageUtil.toMessage(retainedMqttPublishMessage); + String msgIdForMeta = MessageClientIDSetter.createUniqID(); + long bornTimeForMeta = System.currentTimeMillis(); + metaMessage.setMsgId(msgIdForMeta); + metaMessage.setBornTimestamp(bornTimeForMeta); + metaMessage.setEmpty(ByteBufUtil.getBytes(retainedMqttPublishMessage.content()).length == 0); + CompletableFuture storeRetainedFuture = retainedPersistManager.storeRetainedMessage(TopicUtils.normalizeTopic(metaMessage.getOriginTopic()), metaMessage); + storeRetainedFuture.whenComplete((res, throwable) -> { + if (throwable != null) { + logger.error("Store topic:{} retained message error.{}", metaMessage.getOriginTopic(), throwable); + } + }); + } + } + boolean isEmpty = false; //deal empty payload if (ByteBufUtil.getBytes(mqttPublishMessage.content()).length == 0) { @@ -84,23 +135,6 @@ public CompletableFuture put(MqttMessageUpContext context, MqttMess Set queueNames = wildcardManager.matchQueueSetByMsgTopic(pubTopic, context.getNamespace()); //According to topic to find queue long bornTime = System.currentTimeMillis(); - if (mqttPublishMessage.fixedHeader().isRetain()) { - MqttPublishMessage retainedMqttPublishMessage = mqttPublishMessage.copy(); - //Change the retained flag of message that will send MQ is 0 - mqttPublishMessage = MessageUtil.removeRetainedFlag(mqttPublishMessage); - //Keep the retained flag of message that will store meta - Message metaMessage = MessageUtil.toMessage(retainedMqttPublishMessage); - metaMessage.setMsgId(msgId); - metaMessage.setBornTimestamp(bornTime); - metaMessage.setEmpty(isEmpty); - CompletableFuture storeRetainedFuture = retainedPersistManager.storeRetainedMessage(TopicUtils.normalizeTopic(metaMessage.getOriginTopic()), metaMessage); - storeRetainedFuture.whenComplete((res, throwable) -> { - if (throwable != null) { - logger.error("Store topic:{} retained message error.{}", metaMessage.getOriginTopic(), throwable); - } - }); - } - Message message = MessageUtil.toMessage(mqttPublishMessage); message.setMsgId(msgId); message.setBornTimestamp(bornTime);