diff --git a/.travis.yml b/.travis.yml index f3bcb1db58..34d38c2c54 100644 --- a/.travis.yml +++ b/.travis.yml @@ -33,3 +33,4 @@ script: after_success: - mvn clean install -Pit-test - mvn sonar:sonar -Psonar-apache + - bash <(curl -s https://codecov.io/bash) || echo 'Codecov failed to upload' diff --git a/README.md b/README.md index a47efbcea3..9b28078f93 100644 --- a/README.md +++ b/README.md @@ -1,5 +1,6 @@ ## Apache RocketMQ [![Build Status](https://travis-ci.org/apache/rocketmq.svg?branch=master)](https://travis-ci.org/apache/rocketmq) [![Coverage Status](https://coveralls.io/repos/github/apache/rocketmq/badge.svg?branch=master)](https://coveralls.io/github/apache/rocketmq?branch=master) +[![CodeCov](https://codecov.io/gh/apache/rocketmq/branch/master/graph/badge.svg)](https://codecov.io/gh/apache/rocketmq) [![Maven Central](https://maven-badges.herokuapp.com/maven-central/org.apache.rocketmq/rocketmq-all/badge.svg)](http://search.maven.org/#search%7Cga%7C1%7Corg.apache.rocketmq) [![GitHub release](https://img.shields.io/badge/release-download-orange.svg)](https://rocketmq.apache.org/dowloading/releases) [![License](https://img.shields.io/badge/license-Apache%202-4EB1BA.svg)](https://www.apache.org/licenses/LICENSE-2.0.html) diff --git a/acl/pom.xml b/acl/pom.xml index a455361887..bb91ac8988 100644 --- a/acl/pom.xml +++ b/acl/pom.xml @@ -13,7 +13,7 @@ org.apache.rocketmq rocketmq-all - 4.7.0 + 4.7.1-SNAPSHOT rocketmq-acl rocketmq-acl ${project.version} diff --git a/broker/pom.xml b/broker/pom.xml index 50144f01f2..275758f928 100644 --- a/broker/pom.xml +++ b/broker/pom.xml @@ -13,7 +13,7 @@ org.apache.rocketmq rocketmq-all - 4.7.0 + 4.7.1-SNAPSHOT 4.0.0 diff --git a/broker/src/main/java/org/apache/rocketmq/broker/BrokerController.java b/broker/src/main/java/org/apache/rocketmq/broker/BrokerController.java index 85009d620f..194f2850fa 100644 --- a/broker/src/main/java/org/apache/rocketmq/broker/BrokerController.java +++ b/broker/src/main/java/org/apache/rocketmq/broker/BrokerController.java @@ -1240,6 +1240,7 @@ public class BrokerController { } } - - + public ExecutorService getSendMessageExecutor() { + return sendMessageExecutor; + } } diff --git a/broker/src/main/java/org/apache/rocketmq/broker/client/net/Broker2Client.java b/broker/src/main/java/org/apache/rocketmq/broker/client/net/Broker2Client.java index 4c409f2254..4d28cd819f 100644 --- a/broker/src/main/java/org/apache/rocketmq/broker/client/net/Broker2Client.java +++ b/broker/src/main/java/org/apache/rocketmq/broker/client/net/Broker2Client.java @@ -70,7 +70,8 @@ public class Broker2Client { try { this.brokerController.getRemotingServer().invokeOneway(channel, request, 10); } catch (Exception e) { - log.error("Check transaction failed because invoke producer exception. group={}, msgId={}", group, messageExt.getMsgId(), e.getMessage()); + log.error("Check transaction failed because invoke producer exception. group={}, msgId={}, error={}", + group, messageExt.getMsgId(), e.toString()); } } @@ -96,7 +97,7 @@ public class Broker2Client { try { this.brokerController.getRemotingServer().invokeOneway(channel, request, 10); } catch (Exception e) { - log.error("notifyConsumerIdsChanged exception, " + consumerGroup, e.getMessage()); + log.error("notifyConsumerIdsChanged exception. group={}, error={}", consumerGroup, e.toString()); } } @@ -185,14 +186,14 @@ public class Broker2Client { log.info("[reset-offset] reset offset success. topic={}, group={}, clientId={}", topic, group, entry.getValue().getClientId()); } catch (Exception e) { - log.error("[reset-offset] reset offset exception. topic={}, group={}", - new Object[] {topic, group}, e); + log.error("[reset-offset] reset offset exception. topic={}, group={} ,error={}", + topic, group, e.toString()); } } else { response.setCode(ResponseCode.SYSTEM_ERROR); response.setRemark("the client does not support this feature. version=" + MQVersion.getVersionDesc(version)); - log.warn("[reset-offset] the client does not support this feature. version={}", + log.warn("[reset-offset] the client does not support this feature. channel={}, version={}", RemotingHelper.parseChannelRemoteAddr(entry.getKey()), MQVersion.getVersionDesc(version)); return response; } @@ -253,7 +254,7 @@ public class Broker2Client { result.setCode(ResponseCode.SYSTEM_ERROR); result.setRemark("the client does not support this feature. version=" + MQVersion.getVersionDesc(version)); - log.warn("[get-consumer-status] the client does not support this feature. version={}", + log.warn("[get-consumer-status] the client does not support this feature. channel={}, version={}", RemotingHelper.parseChannelRemoteAddr(entry.getKey()), MQVersion.getVersionDesc(version)); return result; } else if (UtilAll.isBlank(originClientId) || originClientId.equals(clientId)) { @@ -279,8 +280,8 @@ public class Broker2Client { } } catch (Exception e) { log.error( - "[get-consumer-status] get consumer status exception. topic={}, group={}, offset={}", - new Object[] {topic, group}, e); + "[get-consumer-status] get consumer status exception. topic={}, group={}, error={}", + topic, group, e.toString()); } if (!UtilAll.isBlank(originClientId) && originClientId.equals(clientId)) { diff --git a/broker/src/main/java/org/apache/rocketmq/broker/filter/ConsumerFilterManager.java b/broker/src/main/java/org/apache/rocketmq/broker/filter/ConsumerFilterManager.java index e9c5286bb1..5ccabb9f6d 100644 --- a/broker/src/main/java/org/apache/rocketmq/broker/filter/ConsumerFilterManager.java +++ b/broker/src/main/java/org/apache/rocketmq/broker/filter/ConsumerFilterManager.java @@ -47,7 +47,7 @@ public class ConsumerFilterManager extends ConfigManager { private static final long MS_24_HOUR = 24 * 3600 * 1000; private ConcurrentMap - filterDataByTopic = new ConcurrentHashMap(256); + filterDataByTopic = new ConcurrentHashMap(256); private transient BrokerController brokerController; private transient BloomFilter bloomFilter; diff --git a/broker/src/main/java/org/apache/rocketmq/broker/processor/AdminBrokerProcessor.java b/broker/src/main/java/org/apache/rocketmq/broker/processor/AdminBrokerProcessor.java index fb7aeceda9..eb81183205 100644 --- a/broker/src/main/java/org/apache/rocketmq/broker/processor/AdminBrokerProcessor.java +++ b/broker/src/main/java/org/apache/rocketmq/broker/processor/AdminBrokerProcessor.java @@ -299,7 +299,9 @@ public class AdminBrokerProcessor extends AsyncNettyRequestProcessor implements this.brokerController.getTopicConfigManager().deleteTopicConfig(requestHeader.getTopic()); this.brokerController.getMessageStore() .cleanUnusedTopic(this.brokerController.getTopicConfigManager().getTopicConfigTable().keySet()); - + if (this.brokerController.getBrokerConfig().isAutoDeleteUnusedStats()) { + this.brokerController.getBrokerStatsManager().onTopicDeleted(requestHeader.getTopic()); + } response.setCode(ResponseCode.SUCCESS); response.setRemark(null); return response; @@ -715,6 +717,9 @@ public class AdminBrokerProcessor extends AsyncNettyRequestProcessor implements this.brokerController.getSubscriptionGroupManager().deleteSubscriptionGroupConfig(requestHeader.getGroupName()); + if (this.brokerController.getBrokerConfig().isAutoDeleteUnusedStats()) { + this.brokerController.getBrokerStatsManager().onGroupDeleted(requestHeader.getGroupName()); + } response.setCode(ResponseCode.SUCCESS); response.setRemark(null); return response; diff --git a/broker/src/main/java/org/apache/rocketmq/broker/processor/SendMessageProcessor.java b/broker/src/main/java/org/apache/rocketmq/broker/processor/SendMessageProcessor.java index 801d886c43..20b5cfaae8 100644 --- a/broker/src/main/java/org/apache/rocketmq/broker/processor/SendMessageProcessor.java +++ b/broker/src/main/java/org/apache/rocketmq/broker/processor/SendMessageProcessor.java @@ -79,7 +79,7 @@ public class SendMessageProcessor extends AbstractSendMessageProcessor implement @Override public void asyncProcessRequest(ChannelHandlerContext ctx, RemotingCommand request, RemotingResponseCallback responseCallback) throws Exception { - asyncProcessRequest(ctx, request).thenAccept(responseCallback::callback); + asyncProcessRequest(ctx, request).thenAcceptAsync(responseCallback::callback, this.brokerController.getSendMessageExecutor()); } public CompletableFuture asyncProcessRequest(ChannelHandlerContext ctx, @@ -281,6 +281,10 @@ public class SendMessageProcessor extends AbstractSendMessageProcessor implement msgInner.setBornHost(ctx.channel().remoteAddress()); msgInner.setStoreHost(this.getStoreHost()); msgInner.setReconsumeTimes(requestHeader.getReconsumeTimes() == null ? 0 : requestHeader.getReconsumeTimes()); + String clusterName = this.brokerController.getBrokerConfig().getBrokerClusterName(); + MessageAccessor.putProperty(msgInner, MessageConst.PROPERTY_CLUSTER, clusterName); + msgInner.setPropertiesString(MessageDecoder.messageProperties2String(msgInner.getProperties())); + CompletableFuture putMessageResult = null; Map origProps = MessageDecoder.string2messageProperties(requestHeader.getProperties()); String transFlag = origProps.get(MessageConst.PROPERTY_TRANSACTION_PREPARED); diff --git a/client/pom.xml b/client/pom.xml index cf1479ddfd..b73b6b697b 100644 --- a/client/pom.xml +++ b/client/pom.xml @@ -19,7 +19,7 @@ org.apache.rocketmq rocketmq-all - 4.7.0 + 4.7.1-SNAPSHOT 4.0.0 diff --git a/client/src/main/java/org/apache/rocketmq/client/consumer/DefaultLitePullConsumer.java b/client/src/main/java/org/apache/rocketmq/client/consumer/DefaultLitePullConsumer.java index ef76cfd4d5..6718eb55fd 100644 --- a/client/src/main/java/org/apache/rocketmq/client/consumer/DefaultLitePullConsumer.java +++ b/client/src/main/java/org/apache/rocketmq/client/consumer/DefaultLitePullConsumer.java @@ -279,7 +279,7 @@ public class DefaultLitePullConsumer extends ClientConfig implements LitePullCon @Override public Long committed(MessageQueue messageQueue) throws MQClientException { - return this.defaultLitePullConsumerImpl.committed(messageQueue); + return this.defaultLitePullConsumerImpl.committed(queueWithNamespace(messageQueue)); } @Override @@ -289,12 +289,12 @@ public class DefaultLitePullConsumer extends ClientConfig implements LitePullCon @Override public void seekToBegin(MessageQueue messageQueue) throws MQClientException { - this.defaultLitePullConsumerImpl.seekToBegin(messageQueue); + this.defaultLitePullConsumerImpl.seekToBegin(queueWithNamespace(messageQueue)); } @Override public void seekToEnd(MessageQueue messageQueue) throws MQClientException { - this.defaultLitePullConsumerImpl.seekToEnd(messageQueue); + this.defaultLitePullConsumerImpl.seekToEnd(queueWithNamespace(messageQueue)); } @Override diff --git a/client/src/main/java/org/apache/rocketmq/client/consumer/listener/MessageListenerOrderly.java b/client/src/main/java/org/apache/rocketmq/client/consumer/listener/MessageListenerOrderly.java index d148df5c67..419169d76a 100644 --- a/client/src/main/java/org/apache/rocketmq/client/consumer/listener/MessageListenerOrderly.java +++ b/client/src/main/java/org/apache/rocketmq/client/consumer/listener/MessageListenerOrderly.java @@ -20,8 +20,7 @@ import java.util.List; import org.apache.rocketmq.common.message.MessageExt; /** - * A MessageListenerConcurrently object is used to receive asynchronously delivered messages orderly.one queue,one - * thread + * A MessageListenerOrderly object is used to receive messages orderly. One queue by one thread */ public interface MessageListenerOrderly extends MessageListener { /** diff --git a/client/src/main/java/org/apache/rocketmq/client/impl/MQClientAPIImpl.java b/client/src/main/java/org/apache/rocketmq/client/impl/MQClientAPIImpl.java index c7d5a06dfd..073e3679be 100644 --- a/client/src/main/java/org/apache/rocketmq/client/impl/MQClientAPIImpl.java +++ b/client/src/main/java/org/apache/rocketmq/client/impl/MQClientAPIImpl.java @@ -643,72 +643,64 @@ public class MQClientAPIImpl { final Message msg, final RemotingCommand response ) throws MQBrokerException, RemotingCommandException { + SendStatus sendStatus; switch (response.getCode()) { - case ResponseCode.FLUSH_DISK_TIMEOUT: - case ResponseCode.FLUSH_SLAVE_TIMEOUT: + case ResponseCode.FLUSH_DISK_TIMEOUT: { + sendStatus = SendStatus.FLUSH_DISK_TIMEOUT; + break; + } + case ResponseCode.FLUSH_SLAVE_TIMEOUT: { + sendStatus = SendStatus.FLUSH_SLAVE_TIMEOUT; + break; + } case ResponseCode.SLAVE_NOT_AVAILABLE: { + sendStatus = SendStatus.SLAVE_NOT_AVAILABLE; + break; } case ResponseCode.SUCCESS: { - SendStatus sendStatus = SendStatus.SEND_OK; - switch (response.getCode()) { - case ResponseCode.FLUSH_DISK_TIMEOUT: - sendStatus = SendStatus.FLUSH_DISK_TIMEOUT; - break; - case ResponseCode.FLUSH_SLAVE_TIMEOUT: - sendStatus = SendStatus.FLUSH_SLAVE_TIMEOUT; - break; - case ResponseCode.SLAVE_NOT_AVAILABLE: - sendStatus = SendStatus.SLAVE_NOT_AVAILABLE; - break; - case ResponseCode.SUCCESS: - sendStatus = SendStatus.SEND_OK; - break; - default: - assert false; - break; - } - - SendMessageResponseHeader responseHeader = - (SendMessageResponseHeader) response.decodeCommandCustomHeader(SendMessageResponseHeader.class); - - //If namespace not null , reset Topic without namespace. - String topic = msg.getTopic(); - if (StringUtils.isNotEmpty(this.clientConfig.getNamespace())) { - topic = NamespaceUtil.withoutNamespace(topic, this.clientConfig.getNamespace()); - } - - MessageQueue messageQueue = new MessageQueue(topic, brokerName, responseHeader.getQueueId()); - - String uniqMsgId = MessageClientIDSetter.getUniqID(msg); - if (msg instanceof MessageBatch) { - StringBuilder sb = new StringBuilder(); - for (Message message : (MessageBatch) msg) { - sb.append(sb.length() == 0 ? "" : ",").append(MessageClientIDSetter.getUniqID(message)); - } - uniqMsgId = sb.toString(); - } - SendResult sendResult = new SendResult(sendStatus, - uniqMsgId, - responseHeader.getMsgId(), messageQueue, responseHeader.getQueueOffset()); - sendResult.setTransactionId(responseHeader.getTransactionId()); - String regionId = response.getExtFields().get(MessageConst.PROPERTY_MSG_REGION); - String traceOn = response.getExtFields().get(MessageConst.PROPERTY_TRACE_SWITCH); - if (regionId == null || regionId.isEmpty()) { - regionId = MixAll.DEFAULT_TRACE_REGION_ID; - } - if (traceOn != null && traceOn.equals("false")) { - sendResult.setTraceOn(false); - } else { - sendResult.setTraceOn(true); - } - sendResult.setRegionId(regionId); - return sendResult; - } - default: + sendStatus = SendStatus.SEND_OK; break; + } + default: { + throw new MQBrokerException(response.getCode(), response.getRemark()); + } } - throw new MQBrokerException(response.getCode(), response.getRemark()); + SendMessageResponseHeader responseHeader = + (SendMessageResponseHeader) response.decodeCommandCustomHeader(SendMessageResponseHeader.class); + + //If namespace not null , reset Topic without namespace. + String topic = msg.getTopic(); + if (StringUtils.isNotEmpty(this.clientConfig.getNamespace())) { + topic = NamespaceUtil.withoutNamespace(topic, this.clientConfig.getNamespace()); + } + + MessageQueue messageQueue = new MessageQueue(topic, brokerName, responseHeader.getQueueId()); + + String uniqMsgId = MessageClientIDSetter.getUniqID(msg); + if (msg instanceof MessageBatch) { + StringBuilder sb = new StringBuilder(); + for (Message message : (MessageBatch) msg) { + sb.append(sb.length() == 0 ? "" : ",").append(MessageClientIDSetter.getUniqID(message)); + } + uniqMsgId = sb.toString(); + } + SendResult sendResult = new SendResult(sendStatus, + uniqMsgId, + responseHeader.getMsgId(), messageQueue, responseHeader.getQueueOffset()); + sendResult.setTransactionId(responseHeader.getTransactionId()); + String regionId = response.getExtFields().get(MessageConst.PROPERTY_MSG_REGION); + String traceOn = response.getExtFields().get(MessageConst.PROPERTY_TRACE_SWITCH); + if (regionId == null || regionId.isEmpty()) { + regionId = MixAll.DEFAULT_TRACE_REGION_ID; + } + if (traceOn != null && traceOn.equals("false")) { + sendResult.setTraceOn(false); + } else { + sendResult.setTraceOn(true); + } + sendResult.setRegionId(regionId); + return sendResult; } public PullResult pullMessage( @@ -1366,7 +1358,7 @@ public class MQClientAPIImpl { GetRouteInfoRequestHeader requestHeader = new GetRouteInfoRequestHeader(); requestHeader.setTopic(topic); - RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.GET_ROUTEINTO_BY_TOPIC, requestHeader); + RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.GET_ROUTEINFO_BY_TOPIC, requestHeader); RemotingCommand response = this.remotingClient.invokeSync(null, request, timeoutMillis); assert response != null; diff --git a/client/src/main/java/org/apache/rocketmq/client/impl/consumer/ConsumeMessageOrderlyService.java b/client/src/main/java/org/apache/rocketmq/client/impl/consumer/ConsumeMessageOrderlyService.java index e440bd98a9..f65567b685 100644 --- a/client/src/main/java/org/apache/rocketmq/client/impl/consumer/ConsumeMessageOrderlyService.java +++ b/client/src/main/java/org/apache/rocketmq/client/impl/consumer/ConsumeMessageOrderlyService.java @@ -452,7 +452,7 @@ public class ConsumeMessageOrderlyService implements ConsumeMessageService { final int consumeBatchSize = ConsumeMessageOrderlyService.this.defaultMQPushConsumer.getConsumeMessageBatchMaxSize(); - List msgs = this.processQueue.takeMessags(consumeBatchSize); + List msgs = this.processQueue.takeMessages(consumeBatchSize); defaultMQPushConsumerImpl.resetRetryAndNamespace(msgs, defaultMQPushConsumer.getConsumerGroup()); if (!msgs.isEmpty()) { final ConsumeOrderlyContext context = new ConsumeOrderlyContext(this.messageQueue); diff --git a/client/src/main/java/org/apache/rocketmq/client/impl/consumer/DefaultLitePullConsumerImpl.java b/client/src/main/java/org/apache/rocketmq/client/impl/consumer/DefaultLitePullConsumerImpl.java index 8ad7a6b44b..f54078fcfd 100644 --- a/client/src/main/java/org/apache/rocketmq/client/impl/consumer/DefaultLitePullConsumerImpl.java +++ b/client/src/main/java/org/apache/rocketmq/client/impl/consumer/DefaultLitePullConsumerImpl.java @@ -631,7 +631,7 @@ public class DefaultLitePullConsumerImpl implements MQConsumerInner { public long committed(MessageQueue messageQueue) throws MQClientException { checkServiceState(); - long offset = this.offsetStore.readOffset(messageQueue, ReadOffsetType.READ_FROM_STORE); + long offset = this.offsetStore.readOffset(messageQueue, ReadOffsetType.MEMORY_FIRST_THEN_STORE); if (offset == -2) throw new MQClientException("Fetch consume offset from broker exception", null); return offset; diff --git a/client/src/main/java/org/apache/rocketmq/client/impl/consumer/ProcessQueue.java b/client/src/main/java/org/apache/rocketmq/client/impl/consumer/ProcessQueue.java index 092da9aa33..f659bd3f83 100644 --- a/client/src/main/java/org/apache/rocketmq/client/impl/consumer/ProcessQueue.java +++ b/client/src/main/java/org/apache/rocketmq/client/impl/consumer/ProcessQueue.java @@ -296,7 +296,7 @@ public class ProcessQueue { } } - public List takeMessags(final int batchSize) { + public List takeMessages(final int batchSize) { List result = new ArrayList(batchSize); final long now = System.currentTimeMillis(); try { diff --git a/client/src/main/java/org/apache/rocketmq/client/impl/factory/MQClientInstance.java b/client/src/main/java/org/apache/rocketmq/client/impl/factory/MQClientInstance.java index bbd2eecb1c..886203fe7d 100644 --- a/client/src/main/java/org/apache/rocketmq/client/impl/factory/MQClientInstance.java +++ b/client/src/main/java/org/apache/rocketmq/client/impl/factory/MQClientInstance.java @@ -17,7 +17,6 @@ package org.apache.rocketmq.client.impl.factory; import java.io.UnsupportedEncodingException; -import java.net.DatagramSocket; import java.util.Collections; import java.util.HashMap; import java.util.HashSet; @@ -118,7 +117,6 @@ public class MQClientInstance { private final ConsumerStatsManager consumerStatsManager; private final AtomicLong sendHeartbeatTimesTotal = new AtomicLong(0); private ServiceState serviceState = ServiceState.CREATE_JUST; - private DatagramSocket datagramSocket; private Random random = new Random(); public MQClientInstance(ClientConfig clientConfig, int instanceIndex, String clientId) { @@ -854,10 +852,6 @@ public class MQClientInstance { this.mQClientAPIImpl.shutdown(); this.rebalanceService.shutdown(); - if (this.datagramSocket != null) { - this.datagramSocket.close(); - this.datagramSocket = null; - } MQClientManager.getInstance().removeClientFactory(this.clientId); log.info("the client factory [{}] shutdown OK", this.clientId); break; diff --git a/client/src/main/java/org/apache/rocketmq/client/impl/producer/DefaultMQProducerImpl.java b/client/src/main/java/org/apache/rocketmq/client/impl/producer/DefaultMQProducerImpl.java index fca50cc565..bff1457835 100644 --- a/client/src/main/java/org/apache/rocketmq/client/impl/producer/DefaultMQProducerImpl.java +++ b/client/src/main/java/org/apache/rocketmq/client/impl/producer/DefaultMQProducerImpl.java @@ -258,7 +258,7 @@ public class DefaultMQProducerImpl implements MQProducerInner { if (shutdownFactory) { this.mQClientFactory.shutdown(); } - + this.timer.cancel(); log.info("the producer [{}] shutdown OK", this.defaultMQProducer.getProducerGroup()); this.serviceState = ServiceState.SHUTDOWN_ALREADY; break; @@ -1356,27 +1356,18 @@ public class DefaultMQProducerImpl implements MQProducerInner { this.sendDefaultImpl(msg, CommunicationMode.ASYNC, new SendCallback() { @Override public void onSuccess(SendResult sendResult) { - requestResponseFuture.setSendReqeustOk(true); + requestResponseFuture.setSendRequestOk(true); } @Override public void onException(Throwable e) { - requestResponseFuture.setSendReqeustOk(false); + requestResponseFuture.setSendRequestOk(false); requestResponseFuture.putResponseMessage(null); requestResponseFuture.setCause(e); } }, timeout - cost); - Message responseMessage = requestResponseFuture.waitResponseMessage(timeout - cost); - if (responseMessage == null) { - if (requestResponseFuture.isSendRequestOk()) { - throw new RequestTimeoutException(ClientErrorCode.REQUEST_TIMEOUT_EXCEPTION, - "send request message to <" + msg.getTopic() + "> OK, but wait reply message timeout, " + timeout + " ms."); - } else { - throw new MQClientException("send request message to <" + msg.getTopic() + "> fail", requestResponseFuture.getCause()); - } - } - return responseMessage; + return waitResponse(msg, timeout, requestResponseFuture, cost); } finally { RequestFutureTable.getRequestFutureTable().remove(correlationId); } @@ -1395,7 +1386,7 @@ public class DefaultMQProducerImpl implements MQProducerInner { this.sendDefaultImpl(msg, CommunicationMode.ASYNC, new SendCallback() { @Override public void onSuccess(SendResult sendResult) { - requestResponseFuture.setSendReqeustOk(true); + requestResponseFuture.setSendRequestOk(true); } @Override @@ -1421,27 +1412,18 @@ public class DefaultMQProducerImpl implements MQProducerInner { this.sendSelectImpl(msg, selector, arg, CommunicationMode.ASYNC, new SendCallback() { @Override public void onSuccess(SendResult sendResult) { - requestResponseFuture.setSendReqeustOk(true); + requestResponseFuture.setSendRequestOk(true); } @Override public void onException(Throwable e) { - requestResponseFuture.setSendReqeustOk(false); + requestResponseFuture.setSendRequestOk(false); requestResponseFuture.putResponseMessage(null); requestResponseFuture.setCause(e); } }, timeout - cost); - Message responseMessage = requestResponseFuture.waitResponseMessage(timeout - cost); - if (responseMessage == null) { - if (requestResponseFuture.isSendRequestOk()) { - throw new RequestTimeoutException(ClientErrorCode.REQUEST_TIMEOUT_EXCEPTION, - "send request message to <" + msg.getTopic() + "> OK, but wait reply message timeout, " + timeout + " ms."); - } else { - throw new MQClientException("send request message to <" + msg.getTopic() + "> fail", requestResponseFuture.getCause()); - } - } - return responseMessage; + return waitResponse(msg, timeout, requestResponseFuture, cost); } finally { RequestFutureTable.getRequestFutureTable().remove(correlationId); } @@ -1461,7 +1443,7 @@ public class DefaultMQProducerImpl implements MQProducerInner { this.sendSelectImpl(msg, selector, arg, CommunicationMode.ASYNC, new SendCallback() { @Override public void onSuccess(SendResult sendResult) { - requestResponseFuture.setSendReqeustOk(true); + requestResponseFuture.setSendRequestOk(true); } @Override @@ -1487,32 +1469,36 @@ public class DefaultMQProducerImpl implements MQProducerInner { this.sendKernelImpl(msg, mq, CommunicationMode.ASYNC, new SendCallback() { @Override public void onSuccess(SendResult sendResult) { - requestResponseFuture.setSendReqeustOk(true); + requestResponseFuture.setSendRequestOk(true); } @Override public void onException(Throwable e) { - requestResponseFuture.setSendReqeustOk(false); + requestResponseFuture.setSendRequestOk(false); requestResponseFuture.putResponseMessage(null); requestResponseFuture.setCause(e); } }, null, timeout - cost); - Message responseMessage = requestResponseFuture.waitResponseMessage(timeout - cost); - if (responseMessage == null) { - if (requestResponseFuture.isSendRequestOk()) { - throw new RequestTimeoutException(ClientErrorCode.REQUEST_TIMEOUT_EXCEPTION, - "send request message to <" + msg.getTopic() + "> OK, but wait reply message timeout, " + timeout + " ms."); - } else { - throw new MQClientException("send request message to <" + msg.getTopic() + "> fail", requestResponseFuture.getCause()); - } - } - return responseMessage; + return waitResponse(msg, timeout, requestResponseFuture, cost); } finally { RequestFutureTable.getRequestFutureTable().remove(correlationId); } } + private Message waitResponse(Message msg, long timeout, RequestResponseFuture requestResponseFuture, long cost) throws InterruptedException, RequestTimeoutException, MQClientException { + Message responseMessage = requestResponseFuture.waitResponseMessage(timeout - cost); + if (responseMessage == null) { + if (requestResponseFuture.isSendRequestOk()) { + throw new RequestTimeoutException(ClientErrorCode.REQUEST_TIMEOUT_EXCEPTION, + "send request message to <" + msg.getTopic() + "> OK, but wait reply message timeout, " + timeout + " ms."); + } else { + throw new MQClientException("send request message to <" + msg.getTopic() + "> fail", requestResponseFuture.getCause()); + } + } + return responseMessage; + } + public void request(final Message msg, final MessageQueue mq, final RequestCallback requestCallback, long timeout) throws RemotingException, InterruptedException, MQClientException, MQBrokerException { long beginTimestamp = System.currentTimeMillis(); @@ -1526,7 +1512,7 @@ public class DefaultMQProducerImpl implements MQProducerInner { this.sendKernelImpl(msg, mq, CommunicationMode.ASYNC, new SendCallback() { @Override public void onSuccess(SendResult sendResult) { - requestResponseFuture.setSendReqeustOk(true); + requestResponseFuture.setSendRequestOk(true); } @Override @@ -1540,7 +1526,7 @@ public class DefaultMQProducerImpl implements MQProducerInner { private void requestFail(final String correlationId) { RequestResponseFuture responseFuture = RequestFutureTable.getRequestFutureTable().remove(correlationId); if (responseFuture != null) { - responseFuture.setSendReqeustOk(false); + responseFuture.setSendRequestOk(false); responseFuture.putResponseMessage(null); try { responseFuture.executeRequestCallback(); diff --git a/client/src/main/java/org/apache/rocketmq/client/producer/RequestResponseFuture.java b/client/src/main/java/org/apache/rocketmq/client/producer/RequestResponseFuture.java index c54b236b16..e66c08fdc5 100644 --- a/client/src/main/java/org/apache/rocketmq/client/producer/RequestResponseFuture.java +++ b/client/src/main/java/org/apache/rocketmq/client/producer/RequestResponseFuture.java @@ -103,8 +103,8 @@ public class RequestResponseFuture { return sendRequestOk; } - public void setSendReqeustOk(boolean sendReqeustOk) { - this.sendRequestOk = sendReqeustOk; + public void setSendRequestOk(boolean sendRequestOk) { + this.sendRequestOk = sendRequestOk; } public Message getRequestMsg() { diff --git a/client/src/test/java/org/apache/rocketmq/client/impl/consumer/ProcessQueueTest.java b/client/src/test/java/org/apache/rocketmq/client/impl/consumer/ProcessQueueTest.java index d6a6dcf72f..16e4a0d901 100644 --- a/client/src/test/java/org/apache/rocketmq/client/impl/consumer/ProcessQueueTest.java +++ b/client/src/test/java/org/apache/rocketmq/client/impl/consumer/ProcessQueueTest.java @@ -38,7 +38,7 @@ public class ProcessQueueTest { assertThat(pq.getMsgCount().get()).isEqualTo(100); - pq.takeMessags(10); + pq.takeMessages(10); pq.commit(); assertThat(pq.getMsgCount().get()).isEqualTo(90); @@ -55,7 +55,7 @@ public class ProcessQueueTest { assertThat(pq.getMsgSize().get()).isEqualTo(100 * 123); - pq.takeMessags(10); + pq.takeMessages(10); pq.commit(); assertThat(pq.getMsgSize().get()).isEqualTo(90 * 123); @@ -74,17 +74,17 @@ public class ProcessQueueTest { assertThat(processQueueInfo.getCachedMsgSizeInMiB()).isEqualTo(12); - pq.takeMessags(10000); + pq.takeMessages(10000); pq.commit(); pq.fillProcessQueueInfo(processQueueInfo); assertThat(processQueueInfo.getCachedMsgSizeInMiB()).isEqualTo(10); - pq.takeMessags(10000); + pq.takeMessages(10000); pq.commit(); pq.fillProcessQueueInfo(processQueueInfo); assertThat(processQueueInfo.getCachedMsgSizeInMiB()).isEqualTo(9); - pq.takeMessags(80000); + pq.takeMessages(80000); pq.commit(); pq.fillProcessQueueInfo(processQueueInfo); assertThat(processQueueInfo.getCachedMsgSizeInMiB()).isEqualTo(0); diff --git a/client/src/test/java/org/apache/rocketmq/client/producer/DefaultMQProducerTest.java b/client/src/test/java/org/apache/rocketmq/client/producer/DefaultMQProducerTest.java index d572a2320e..41046fcdf5 100644 --- a/client/src/test/java/org/apache/rocketmq/client/producer/DefaultMQProducerTest.java +++ b/client/src/test/java/org/apache/rocketmq/client/producer/DefaultMQProducerTest.java @@ -373,7 +373,7 @@ public class DefaultMQProducerTest { assertThat(responseMap).isNotNull(); for (Map.Entry entry : responseMap.entrySet()) { RequestResponseFuture future = entry.getValue(); - future.setSendReqeustOk(true); + future.setSendRequestOk(true); message.setFlag(1); future.getRequestCallback().onSuccess(message); } diff --git a/client/src/test/java/org/apache/rocketmq/client/producer/RequestResponseFutureTest.java b/client/src/test/java/org/apache/rocketmq/client/producer/RequestResponseFutureTest.java index 90e4623e9b..bc20873c11 100644 --- a/client/src/test/java/org/apache/rocketmq/client/producer/RequestResponseFutureTest.java +++ b/client/src/test/java/org/apache/rocketmq/client/producer/RequestResponseFutureTest.java @@ -37,7 +37,7 @@ public class RequestResponseFutureTest { @Override public void onException(Throwable e) { } }); - future.setSendReqeustOk(true); + future.setSendRequestOk(true); future.executeRequestCallback(); assertThat(cc.get()).isEqualTo(1); } diff --git a/common/pom.xml b/common/pom.xml index 64904fc2eb..56c150f877 100644 --- a/common/pom.xml +++ b/common/pom.xml @@ -19,7 +19,7 @@ org.apache.rocketmq rocketmq-all - 4.7.0 + 4.7.1-SNAPSHOT 4.0.0 diff --git a/common/src/main/java/org/apache/rocketmq/common/BrokerConfig.java b/common/src/main/java/org/apache/rocketmq/common/BrokerConfig.java index a7568f0a20..bfe8a2104d 100644 --- a/common/src/main/java/org/apache/rocketmq/common/BrokerConfig.java +++ b/common/src/main/java/org/apache/rocketmq/common/BrokerConfig.java @@ -184,6 +184,8 @@ public class BrokerConfig { private boolean storeReplyMessageEnable = true; + private boolean autoDeleteUnusedStats = false; + public static String localHostName() { try { return InetAddress.getLocalHost().getHostName(); @@ -793,4 +795,12 @@ public class BrokerConfig { public void setStoreReplyMessageEnable(boolean storeReplyMessageEnable) { this.storeReplyMessageEnable = storeReplyMessageEnable; } + + public boolean isAutoDeleteUnusedStats() { + return autoDeleteUnusedStats; + } + + public void setAutoDeleteUnusedStats(boolean autoDeleteUnusedStats) { + this.autoDeleteUnusedStats = autoDeleteUnusedStats; + } } diff --git a/common/src/main/java/org/apache/rocketmq/common/message/MessageDecoder.java b/common/src/main/java/org/apache/rocketmq/common/message/MessageDecoder.java index 54d5e6b60f..d048dde2b9 100644 --- a/common/src/main/java/org/apache/rocketmq/common/message/MessageDecoder.java +++ b/common/src/main/java/org/apache/rocketmq/common/message/MessageDecoder.java @@ -414,6 +414,9 @@ public class MessageDecoder { final String name = entry.getKey(); final String value = entry.getValue(); + if (value == null) { + continue; + } sb.append(name); sb.append(NAME_VALUE_SEPARATOR); sb.append(value); diff --git a/common/src/main/java/org/apache/rocketmq/common/message/MessageExt.java b/common/src/main/java/org/apache/rocketmq/common/message/MessageExt.java index 7f35172534..577c4f4bc1 100644 --- a/common/src/main/java/org/apache/rocketmq/common/message/MessageExt.java +++ b/common/src/main/java/org/apache/rocketmq/common/message/MessageExt.java @@ -142,18 +142,20 @@ public class MessageExt extends Message { } public String getBornHostString() { - if (this.bornHost != null) { - InetSocketAddress inetSocketAddress = (InetSocketAddress) this.bornHost; - return inetSocketAddress.getAddress().getHostAddress(); + if (null != this.bornHost) { + InetAddress inetAddress = ((InetSocketAddress) this.bornHost).getAddress(); + + return null != inetAddress ? inetAddress.getHostAddress() : null; } return null; } public String getBornHostNameString() { - if (this.bornHost != null) { - InetSocketAddress inetSocketAddress = (InetSocketAddress) this.bornHost; - return inetSocketAddress.getAddress().getHostName(); + if (null != this.bornHost) { + InetAddress inetAddress = ((InetSocketAddress) this.bornHost).getAddress(); + + return null != inetAddress ? inetAddress.getHostName() : null; } return null; diff --git a/common/src/main/java/org/apache/rocketmq/common/protocol/RequestCode.java b/common/src/main/java/org/apache/rocketmq/common/protocol/RequestCode.java index b3009d738f..75ceff38cf 100644 --- a/common/src/main/java/org/apache/rocketmq/common/protocol/RequestCode.java +++ b/common/src/main/java/org/apache/rocketmq/common/protocol/RequestCode.java @@ -89,7 +89,7 @@ public class RequestCode { public static final int REGISTER_BROKER = 103; public static final int UNREGISTER_BROKER = 104; - public static final int GET_ROUTEINTO_BY_TOPIC = 105; + public static final int GET_ROUTEINFO_BY_TOPIC = 105; public static final int GET_BROKER_CLUSTER_INFO = 106; public static final int UPDATE_AND_CREATE_SUBSCRIPTIONGROUP = 200; diff --git a/common/src/main/java/org/apache/rocketmq/common/stats/MomentStatsItemSet.java b/common/src/main/java/org/apache/rocketmq/common/stats/MomentStatsItemSet.java index 7ff26cd3b8..4d2ce0cfcd 100644 --- a/common/src/main/java/org/apache/rocketmq/common/stats/MomentStatsItemSet.java +++ b/common/src/main/java/org/apache/rocketmq/common/stats/MomentStatsItemSet.java @@ -74,6 +74,26 @@ public class MomentStatsItemSet { statsItem.getValue().set(value); } + public void delValueByInfixKey(final String statsKey, String separator) { + Iterator> it = this.statsItemTable.entrySet().iterator(); + while (it.hasNext()) { + Entry next = it.next(); + if (next.getKey().contains(separator + statsKey + separator)) { + it.remove(); + } + } + } + + public void delValueBySuffixKey(final String statsKey, String separator) { + Iterator> it = this.statsItemTable.entrySet().iterator(); + while (it.hasNext()) { + Entry next = it.next(); + if (next.getKey().endsWith(separator + statsKey)) { + it.remove(); + } + } + } + public MomentStatsItem getAndCreateStatsItem(final String statsKey) { MomentStatsItem statsItem = this.statsItemTable.get(statsKey); if (null == statsItem) { diff --git a/common/src/main/java/org/apache/rocketmq/common/stats/StatsItemSet.java b/common/src/main/java/org/apache/rocketmq/common/stats/StatsItemSet.java index 1c2de330bc..bcf9665959 100644 --- a/common/src/main/java/org/apache/rocketmq/common/stats/StatsItemSet.java +++ b/common/src/main/java/org/apache/rocketmq/common/stats/StatsItemSet.java @@ -158,6 +158,43 @@ public class StatsItemSet { statsItem.getTimes().addAndGet(incTimes); } + public void delValue(final String statsKey) { + StatsItem statsItem = this.statsItemTable.get(statsKey); + if (null != statsItem) { + this.statsItemTable.remove(statsKey); + } + } + + public void delValueByPrefixKey(final String statsKey, String separator) { + Iterator> it = this.statsItemTable.entrySet().iterator(); + while (it.hasNext()) { + Entry next = it.next(); + if (next.getKey().startsWith(statsKey + separator)) { + it.remove(); + } + } + } + + public void delValueByInfixKey(final String statsKey, String separator) { + Iterator> it = this.statsItemTable.entrySet().iterator(); + while (it.hasNext()) { + Entry next = it.next(); + if (next.getKey().contains(separator + statsKey + separator)) { + it.remove(); + } + } + } + + public void delValueBySuffixKey(final String statsKey, String separator) { + Iterator> it = this.statsItemTable.entrySet().iterator(); + while (it.hasNext()) { + Entry next = it.next(); + if (next.getKey().endsWith(separator + statsKey)) { + it.remove(); + } + } + } + public StatsItem getAndCreateStatsItem(final String statsKey) { StatsItem statsItem = this.statsItemTable.get(statsKey); if (null == statsItem) { diff --git a/common/src/test/java/org/apache/rocketmq/common/BrokerConfigTest.java b/common/src/test/java/org/apache/rocketmq/common/BrokerConfigTest.java index aa8bcfa725..07e132ff81 100644 --- a/common/src/test/java/org/apache/rocketmq/common/BrokerConfigTest.java +++ b/common/src/test/java/org/apache/rocketmq/common/BrokerConfigTest.java @@ -37,11 +37,13 @@ public class BrokerConfigTest { brokerConfig.setBrokerId(0); brokerConfig.setBrokerClusterName("DefaultCluster"); brokerConfig.setMsgTraceTopicName("RMQ_SYS_TRACE_TOPIC4"); + brokerConfig.setAutoDeleteUnusedStats(true); assertThat(brokerConfig.getBrokerClusterName()).isEqualTo("DefaultCluster"); assertThat(brokerConfig.getNamesrvAddr()).isEqualTo("127.0.0.1:9876"); assertThat(brokerConfig.getMsgTraceTopicName()).isEqualTo("RMQ_SYS_TRACE_TOPIC4"); assertThat(brokerConfig.getBrokerId()).isEqualTo(0); assertThat(brokerConfig.getBrokerName()).isEqualTo("broker-a"); assertThat(brokerConfig.isAutoCreateTopicEnable()).isEqualTo(false); + assertThat(brokerConfig.isAutoDeleteUnusedStats()).isEqualTo(true); } } \ No newline at end of file diff --git a/common/src/test/java/org/apache/rocketmq/common/message/MessageDecoderTest.java b/common/src/test/java/org/apache/rocketmq/common/message/MessageDecoderTest.java index 82ad2714dd..fde523dba9 100644 --- a/common/src/test/java/org/apache/rocketmq/common/message/MessageDecoderTest.java +++ b/common/src/test/java/org/apache/rocketmq/common/message/MessageDecoderTest.java @@ -246,4 +246,23 @@ public class MessageDecoderTest { assertThat("abc").isEqualTo(decodedMsg.getTopic()); } -} + + public void testNullValueProperty() throws Exception { + MessageExt msg = new MessageExt(); + msg.setBody("x".getBytes()); + msg.setTopic("x"); + msg.setBornHost(new InetSocketAddress("127.0.0.1", 9000)); + msg.setStoreHost(new InetSocketAddress("127.0.0.1", 9000)); + String key = "NullValueKey"; + msg.putProperty(key, null); + try { + byte[] encode = MessageDecoder.encode(msg, false); + MessageExt decode = MessageDecoder.decode(ByteBuffer.wrap(encode)); + assertThat(decode.getProperty(key)).isNull(); + } catch (Exception e) { + e.printStackTrace(); + assertThat(Boolean.FALSE).isTrue(); + } + } + +} \ No newline at end of file diff --git a/distribution/bin/mqadmin b/distribution/bin/mqadmin index cd0253bc87..980d6e3b6b 100644 --- a/distribution/bin/mqadmin +++ b/distribution/bin/mqadmin @@ -42,4 +42,4 @@ fi export ROCKETMQ_HOME -sh ${ROCKETMQ_HOME}/bin/tools.sh org.apache.rocketmq.tools.command.MQAdminStartup $@ +sh ${ROCKETMQ_HOME}/bin/tools.sh org.apache.rocketmq.tools.command.MQAdminStartup "$@" diff --git a/distribution/bin/tools.sh b/distribution/bin/tools.sh index 071794c630..a13dc7e620 100644 --- a/distribution/bin/tools.sh +++ b/distribution/bin/tools.sh @@ -40,4 +40,4 @@ JAVA_OPT="${JAVA_OPT} -server -Xms1g -Xmx1g -Xmn256m -XX:MetaspaceSize=128m -XX: JAVA_OPT="${JAVA_OPT} -Djava.ext.dirs=${BASE_DIR}/lib:${JAVA_HOME}/jre/lib/ext:${JAVA_HOME}/lib/ext" JAVA_OPT="${JAVA_OPT} -cp ${CLASSPATH}" -$JAVA ${JAVA_OPT} $@ +$JAVA ${JAVA_OPT} "$@" diff --git a/distribution/pom.xml b/distribution/pom.xml index 887847b759..09e7ad0306 100644 --- a/distribution/pom.xml +++ b/distribution/pom.xml @@ -20,7 +20,7 @@ org.apache.rocketmq rocketmq-all - 4.7.0 + 4.7.1-SNAPSHOT rocketmq-distribution rocketmq-distribution ${project.version} diff --git a/docs/cn/RocketMQ_Example.md b/docs/cn/RocketMQ_Example.md index 09acb0de97..0cf977ed7d 100644 --- a/docs/cn/RocketMQ_Example.md +++ b/docs/cn/RocketMQ_Example.md @@ -500,48 +500,52 @@ try { 复杂度只有当你发送大批量时才会增长,你可能不确定它是否超过了大小限制(4MB)。这时候你最好把你的消息列表分割一下: ```java - -public class ListSplitter implements Iterator> { - private final int SIZE_LIMIT = 1024 * 1024 * 4; - private final List messages; - private int currIndex; - public ListSplitter(List messages) { - this.messages = messages; - } - @Override public boolean hasNext() { - return currIndex < messages.size(); - } - @Override public List next() { - int nextIndex = currIndex; - int totalSize = 0; - for (; nextIndex < messages.size(); nextIndex++) { - Message message = messages.get(nextIndex); - int tmpSize = message.getTopic().length() + message.getBody().length; - Map properties = message.getProperties(); - for (Map.Entry entry : properties.entrySet()) { - tmpSize += entry.getKey().length() + entry.getValue().length(); - } - tmpSize = tmpSize + 20; // 增加日志的开销20字节 - if (tmpSize > SIZE_LIMIT) { - //单个消息超过了最大的限制 - //忽略,否则会阻塞分裂的进程 - if (nextIndex - currIndex == 0) { - //假如下一个子列表没有元素,则添加这个子列表然后退出循环,否则只是退出循环 - nextIndex++; - } - break; - } - if (tmpSize + totalSize > SIZE_LIMIT) { - break; - } else { - totalSize += tmpSize; - } - - } - List subList = messages.subList(currIndex, nextIndex); - currIndex = nextIndex; - return subList; - } +public class ListSplitter implements Iterator> { + private final int SIZE_LIMIT = 1024 * 1024 * 4; + private final List messages; + private int currIndex; + public ListSplitter(List messages) { + this.messages = messages; + } + @Override public boolean hasNext() { + return currIndex < messages.size(); + } + @Override public List next() { + int startIndex = getStartIndex(); + int nextIndex = startIndex; + int totalSize = 0; + for (; nextIndex < messages.size(); nextIndex++) { + Message message = messages.get(nextIndex); + int tmpSize = calcMessageSize(message); + if (tmpSize + totalSize > SIZE_LIMIT) { + break; + } else { + totalSize += tmpSize; + } + } + List subList = messages.subList(startIndex, nextIndex); + currIndex = nextIndex; + return subList; + } + private int getStartIndex() { + Message currMessage = messages.get(currIndex); + int tmpSize = calcMessageSize(currMessage); + while(tmpSize > SIZE_LIMIT) { + currIndex += 1; + Message message = messages.get(curIndex); + tmpSize = calcMessageSize(message); + } + return currIndex; + } + private int calcMessageSize(Message message) { + int tmpSize = message.getTopic().length() + message.getBody().length(); + Map properties = message.getProperties(); + for (Map.Entry entry : properties.entrySet()) { + tmpSize += entry.getKey().length() + entry.getValue().length(); + } + tmpSize = tmpSize + 20; // 增加⽇日志的开销20字节 + return tmpSize; + } } //把大的消息分裂成若干个小的消息 ListSplitter splitter = new ListSplitter(messages); @@ -702,7 +706,7 @@ public class TransactionProducer { ``` #### 2、实现事务的监听接口 -当发送半消息成功时,我们使用 `executeLocalTransaction` 方法来执行本地事务。它返回前一节中提到的三个事务状态之一。`checkLocalTranscation` 方法用于检查本地事务状态,并回应消息队列的检查请求。它也是返回前一节中提到的三个事务状态之一。 +当发送半消息成功时,我们使用 `executeLocalTransaction` 方法来执行本地事务。它返回前一节中提到的三个事务状态之一。`checkLocalTransaction` 方法用于检查本地事务状态,并回应消息队列的检查请求。它也是返回前一节中提到的三个事务状态之一。 ```java public class TransactionListenerImpl implements TransactionListener { @@ -737,8 +741,8 @@ public class TransactionListenerImpl implements TransactionListener { ### 6.2 事务消息使用上的限制 1. 事务消息不支持延时消息和批量消息。 -2. 为了避免单个消息被检查太多次而导致半队列消息累积,我们默认将单个消息的检查次数限制为 15 次,但是用户可以通过 Broker 配置文件的 `transactionCheckMax`参数来修改此限制。如果已经检查某条消息超过 N 次的话( N = `transactionCheckMax` ) 则 Broker 将丢弃此消息,并在默认情况下同时打印错误日志。用户可以通过重写 `AbstractTransactionCheckListener` 类来修改这个行为。 -3. 事务消息将在 Broker 配置文件中的参数 transactionMsgTimeout 这样的特定时间长度之后被检查。当发送事务消息时,用户还可以通过设置用户属性 CHECK_IMMUNITY_TIME_IN_SECONDS 来改变这个限制,该参数优先于 `transactionMsgTimeout` 参数。 +2. 为了避免单个消息被检查太多次而导致半队列消息累积,我们默认将单个消息的检查次数限制为 15 次,但是用户可以通过 Broker 配置文件的 `transactionCheckMax`参数来修改此限制。如果已经检查某条消息超过 N 次的话( N = `transactionCheckMax` ) 则 Broker 将丢弃此消息,并在默认情况下同时打印错误日志。用户可以通过重写 `AbstractTransactionalMessageCheckListener` 类来修改这个行为。 +3. 事务消息将在 Broker 配置文件中的参数 transactionTimeout 这样的特定时间长度之后被检查。当发送事务消息时,用户还可以通过设置用户属性 CHECK_IMMUNITY_TIME_IN_SECONDS 来改变这个限制,该参数优先于 `transactionTimeout` 参数。 4. 事务性消息可能不止一次被检查或消费。 5. 提交给用户的目标主题消息可能会失败,目前这依日志的记录而定。它的高可用性通过 RocketMQ 本身的高可用性机制来保证,如果希望确保事务消息不丢失、并且事务完整性得到保证,建议使用同步的双重写入机制。 6. 事务消息的生产者 ID 不能与其他类型消息的生产者 ID 共享。与其他类型的消息不同,事务消息允许反向查询、MQ服务器能通过它们的生产者 ID 查询到消费者。 diff --git a/docs/cn/image/consumer_reply.png b/docs/cn/image/consumer_reply.png new file mode 100644 index 0000000000..ae00da665d Binary files /dev/null and b/docs/cn/image/consumer_reply.png differ diff --git a/docs/cn/image/producer_send_request.png b/docs/cn/image/producer_send_request.png new file mode 100644 index 0000000000..016fedac6f Binary files /dev/null and b/docs/cn/image/producer_send_request.png differ diff --git a/docs/cn/operation.md b/docs/cn/operation.md index e35bd6812d..d8c0bb151d 100644 --- a/docs/cn/operation.md +++ b/docs/cn/operation.md @@ -564,7 +564,7 @@ $ nohup sh mqbroker -n 192.168.1.1:9876 -c $ROCKETMQ_HOME/conf/2m-2s-sync/broker 从NameServer上清除 Broker写权限 -b - Broker 地址,地址为ip:port + BrokerName -n @@ -1161,7 +1161,7 @@ $ nohup sh mqbroker -n 192.168.1.1:9876 -c $ROCKETMQ_HOME/conf/2m-2s-sync/broker consumerConnec tion + height:89.0pt;border-top:none;width:65pt'>consumerConnection 查询 Consumer 的网络连接 -g @@ -1177,7 +1177,7 @@ $ nohup sh mqbroker -n 192.168.1.1:9876 -c $ROCKETMQ_HOME/conf/2m-2s-sync/broker producerConnec tion + height:106.0pt;border-top:none;width:65pt'>producerConnection 查询 Producer 的网络连接 -g diff --git a/docs/cn/rpc_request.md b/docs/cn/rpc_request.md new file mode 100644 index 0000000000..a8fe802a15 --- /dev/null +++ b/docs/cn/rpc_request.md @@ -0,0 +1,146 @@ +# “Request-Reply”特性 +--- + +## 1 使用场景 +随着服务规模的扩大,单机服务无法满足性能和容量的要求,此时需要将服务拆分为更小粒度的服务或者部署多个服务实例构成集群来提供服务。在分布式场景下,RPC是最常用的联机调用的方式。 + +在构建分布式应用时,有些领域,例如金融服务领域,常常使用消息队列来构建服务总线,实现联机调用的目的。消息队列的主要场景是解耦、削峰填谷,在联机调用的场景下,需要将服务的调用抽象成基于消息的交互,并增强同步调用的这种交互逻辑。为了更好地支持消息队列在联机调用场景下的应用,rocketmq-4.7.0推出了“Request-Reply”特性来支持RPC调用。 + +## 2 设计思路 +在rocketmq中,整个同步调用主要包括两个过程: + +(1)请求方生成消息,发送给响应方,并等待响应方回包; + +(2)响应方收到请求消息后,消费这条消息,并发出一条响应消息给请求方。 + +整个过程实质上是两个消息收发过程的组合。所以这里最关键的问题是如何将异步的消息收发过程构建成一个同步的过程。其中主要有两个问题需要解决: + +### 2.1 请求方如何同步等待回包 + +这个问题的解决方案中,一个关键的数据结构是RequestResponseFuture。 + +``` +public class RequestResponseFuture { + private final String correlationId; + private final RequestCallback requestCallback; + private final long beginTimestamp = System.currentTimeMillis(); + private final Message requestMsg = null; + private long timeoutMillis; + private CountDownLatch countDownLatch = new CountDownLatch(1); + private volatile Message responseMsg = null; + private volatile boolean sendRequestOk = true; + private volatile Throwable cause = null; +} +``` +RequestResponseFuture中,利用correlationId来标识一个请求。如下图所示,Producer发送request时创建一个RequestResponseFuture,以correlationId为key,RequestResponseFuture为value存入map,同时请求中带上RequestResponseFuture中的correlationId,收到回包后根据correlationId拿到对应的RequestResponseFuture,并设置回包内容。 +![](image/producer_send_request.png) + +### 2.2 consumer消费消息后,如何准确回包 + +(1)producer在发送消息的时候,会给每条消息生成唯一的标识符,同时还带上了producer的clientId。当consumer收到并消费消息后,从消息中取出消息的标识符correlationId和producer的标识符clientId,放入响应消息,用来确定此响应消息是哪条请求消息的回包,以及此响应消息应该发给哪个producer。同时响应消息中设置了消息的类型以及响应消息的topic,然后consumer将消息发给broker,如下图所示。 +![](image/consumer_reply.png) + +(2)broker收到响应消息后,需要将消息发回给指定的producer。Broker如何知道发回给哪个producer?因为消息中包含了producer的标识符clientId,在ProducerManager中,维护了标识符和channel信息的对应关系,通过这个对应关系,就能把回包发给对应的producer。 + +响应消息发送和一般的消息发送流程区别在于,响应消息不需要producer拉取,而是由broker直接推给producer。同时选择broker的策略也有变化:请求消息从哪个broker发过来,响应消息也发到对应的broker上。 + +Producer收到响应消息后,根据消息中的唯一标识符,从RequestResponseFuture的map中找到对应的RequestResponseFuture结构,设置响应消息,同时计数器减一,解除等待状态,使请求方收到响应消息。 + +## 3 使用方法 + +同步调用的示例在example文件夹的rpc目录下。 + +### 3.1 Producer +``` +Message msg = new Message(topic, + "", + "Hello world".getBytes(RemotingHelper.DEFAULT_CHARSET)); + + long begin = System.currentTimeMillis(); + Message retMsg = producer.request(msg, ttl); + long cost = System.currentTimeMillis() - begin; + System.out.printf("request to <%s> cost: %d replyMessage: %s %n", topic, cost, retMsg); +``` +调用接口替换为request即可。 + +### 3.2 Consumer +需要启动一个producer,同时在覆写consumeMessage方法的时候,自定义响应消息并发送。 + +``` + @Override + public ConsumeConcurrentlyStatus consumeMessage(List msgs, ConsumeConcurrentlyContext context) { + System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs); + for (MessageExt msg : msgs) { + try { + System.out.printf("handle message: %s", msg.toString()); + String replyTo = MessageUtil.getReplyToClient(msg); + byte[] replyContent = "reply message contents.".getBytes(); + // create reply message with given util, do not create reply message by yourself + Message replyMessage = MessageUtil.createReplyMessage(msg, replyContent); + + // send reply message with producer + SendResult replyResult = replyProducer.send(replyMessage, 3000); + System.out.printf("reply to %s , %s %n", replyTo, replyResult.toString()); + } catch (MQClientException | RemotingException | MQBrokerException | InterruptedException e) { + e.printStackTrace(); + } + } + return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; + } +``` + +## 4 接口参数 + +4.1 public Message request(Message msg,long timeout) + +msg:待发送的消息 + +timeout:同步调用超时时间 + +4.2 public void request(Message msg, final RequestCallback requestCallback, long timeout) + +msg:待发送的消息 + +requestCallback:回调函数 + +timeout:同步调用超时时间 + +4.3 public Message request(final Message msg, final MessageQueueSelector selector, final Object arg,final long timeout) + +msg:待发送的消息 + +selector:消息队列选择器 + +arg:消息队列选择器需要的参数 + +timeout:同步调用超时时间 + +4.4 public void request(final Message msg, final MessageQueueSelector selector, final Object arg,final RequestCallback requestCallback, final long timeout) + +msg:待发送的消息 + +selector:消息队列选择器 + +arg:消息队列选择器需要的参数 + +requestCallback:回调函数 + +timeout:同步调用超时时间 + +4.5 public Message request(final Message msg, final MessageQueue mq, final long timeout) + +msg:待发送的消息 + +mq:目标消息队列 + +timeout:同步调用超时时间 + +4.6 public void request(final Message msg, final MessageQueue mq, final RequestCallback requestCallback, long timeout) + +msg:待发送的消息 + +mq:目标消息队列 + +requestCallback:回调函数 + +timeout:同步调用超时时间 diff --git a/docs/en/CLITools.md b/docs/en/CLITools.md index 909f002ff4..47ba69e571 100644 --- a/docs/en/CLITools.md +++ b/docs/en/CLITools.md @@ -424,7 +424,7 @@ Before introducing the mqadmin management tool, the following points need to be Clear write permissions for broker from nameServer -b - Declare the address of the broker and format as ip:port + Declare the BrokerName -n @@ -1053,7 +1053,7 @@ Before introducing the mqadmin management tool, the following points need to be consumerConnec tion + height:89.0pt;border-top:none;width:65pt'>consumerConnection Query the network connection of consumer -g @@ -1069,7 +1069,7 @@ Before introducing the mqadmin management tool, the following points need to be producerConnec tion + height:106.0pt;border-top:none;width:65pt'>producerConnection Query the network connection of producer -g diff --git a/docs/en/Example_Batch.md b/docs/en/Example_Batch.md index d7199ca7c7..11bb42ffbe 100644 --- a/docs/en/Example_Batch.md +++ b/docs/en/Example_Batch.md @@ -19,47 +19,52 @@ try { ### 2 Split into Lists The complexity only grow when you send large batch and you may not sure if it exceeds the size limit (4MiB). At this time, you’d better split the lists: ```java -public class ListSplitter implements Iterator> { - private final int SIZE_LIMIT = 1000 * 1000; +public class ListSplitter implements Iterator> { + private final int SIZE_LIMIT = 1024 * 1024 * 4; private final List messages; private int currIndex; - public ListSplitter(List messages) { - this.messages = messages; + public ListSplitter(List messages) { + this.messages = messages; } @Override public boolean hasNext() { - return currIndex < messages.size(); + return currIndex < messages.size(); } - @Override public List next() { - int nextIndex = currIndex; + @Override public List next() { + int startIndex = getStartIndex(); + int nextIndex = startIndex; int totalSize = 0; for (; nextIndex < messages.size(); nextIndex++) { - Message message = messages.get(nextIndex); - int tmpSize = message.getTopic().length() + message.getBody().length; - Map properties = message.getProperties(); - for (Map.Entry entry : properties.entrySet()) { - tmpSize += entry.getKey().length() + entry.getValue().length(); - } - tmpSize = tmpSize + 20; //for log overhead - if (tmpSize > SIZE_LIMIT) { - //it is unexpected that single message exceeds the SIZE_LIMIT - //here just let it go, otherwise it will block the splitting process - if (nextIndex - currIndex == 0) { - //if the next sublist has no element, add this one and then break, otherwise just break - nextIndex++; - } - break; - } + Message message = messages.get(nextIndex); + int tmpSize = calcMessageSize(message); if (tmpSize + totalSize > SIZE_LIMIT) { - break; + break; } else { - totalSize += tmpSize; + totalSize += tmpSize; } - } - List subList = messages.subList(currIndex, nextIndex); + List subList = messages.subList(startIndex, nextIndex); currIndex = nextIndex; return subList; } + private int getStartIndex() { + Message currMessage = messages.get(currIndex); + int tmpSize = calcMessageSize(currMessage); + while(tmpSize > SIZE_LIMIT) { + currIndex += 1; + Message message = messages.get(curIndex); + tmpSize = calcMessageSize(message); + } + return currIndex; + } + private int calcMessageSize(Message message) { + int tmpSize = message.getTopic().length() + message.getBody().length(); + Map properties = message.getProperties(); + for (Map.Entry entry : properties.entrySet()) { + tmpSize += entry.getKey().length() + entry.getValue().length(); + } + tmpSize = tmpSize + 20; // Increase the log overhead by 20 bytes + return tmpSize; + } } // then you could split the large list into small ones: diff --git a/docs/en/Example_Transaction.md b/docs/en/Example_Transaction.md index 9dc700e375..a35b040f82 100644 --- a/docs/en/Example_Transaction.md +++ b/docs/en/Example_Transaction.md @@ -88,8 +88,8 @@ public class TransactionListenerImpl implements TransactionListener { ## 3 Usage Constraint 1. Messages of the transactional have no schedule and batch support. -2. In order to avoid a single message being checked too many times and lead to half queue message accumulation, we limited the number of checks for a single message to 15 times by default, but users can change this limit by change the ```transactionCheckMax``` parameter in the configuration of the broker, if one message has been checked over ```transactionCheckMax``` times, broker will discard this message and print an error log at the same time by default. Users can change this behavior by override the ```AbstractTransactionCheckListener``` class. -3. A transactional message will be checked after a certain period of time that determined by parameter ```transactionTimeout``` in the configuration of the broker. And users also can change this limit by set user property “CHECK_IMMUNITY_TIME_IN_SECONDS” when sending transactional message, this parameter takes precedence over the “transactionMsgTimeout” parameter. +2. In order to avoid a single message being checked too many times and lead to half queue message accumulation, we limited the number of checks for a single message to 15 times by default, but users can change this limit by change the ```transactionCheckMax``` parameter in the configuration of the broker, if one message has been checked over ```transactionCheckMax``` times, broker will discard this message and print an error log at the same time by default. Users can change this behavior by override the ```AbstractTransactionalMessageCheckListener``` class. +3. A transactional message will be checked after a certain period of time that determined by parameter g```transactionTimeout``` in the configuration of the broker. And users also can change this limit by set user property ```CHECK_IMMUNITY_TIME_IN_SECONDS``` when sending transactional message, this parameter takes precedence over the ```transactionTimeout``` parameter. 4. A transactional message maybe checked or consumed more than once. 5. Committed message reput to the user’s target topic may fail. Currently, it depends on the log record. High availability is ensured by the high availability mechanism of RocketMQ itself. If you want to ensure that the transactional message isn’t lost and the transaction integrity is guaranteed, it is recommended to use synchronous double write. mechanism. 6. Producer IDs of transactional messages cannot be shared with producer IDs of other types of messages. Unlike other types of message, transactional messages allow backward queries. MQ Server query clients by their Producer IDs. diff --git a/docs/en/README.md b/docs/en/README.md new file mode 100644 index 0000000000..b24d2403fd --- /dev/null +++ b/docs/en/README.md @@ -0,0 +1,49 @@ +Apache RocketMQ Developer Guide +-------- + +##### This guide helps develpers understand and use Apache RocketMQ quickly. + +### 1. Concepts & Features + +- [Concept](Concept.md):introduce basic concepts in RocketMQ. + +- [Feature](Feature.md):introduce functional features of RocketMQ's implementations. + + +### 2. Architecture Design + +- [Architecture](architecture.md):introduce RocketMQ's deployment and technical architecture. + +- [Design](design.md):introduce design concept of RocketMQ's key mechanisms, including message storage, communication mechanisms, message filter, loadbalance, transaction message, etc. + + +### 3. Example + +- [Example](RocketMQ_Example.md) :introduce RocketMQ's common usage, including basic example, sequence message example, delay message example, batch message example, filter message example, transaction message example, etc. + + +### 4. Best Practice +- [Best Practice](best_practice.md):introduce RocketMQ's best practice, including producer, consumer, broker, NameServer, configuration of client, and the best parameter configuration of JVM, linux. + +- [Message Trace](msg_trace/user_guide.md):introduce how to use RocketMQ's message tracing feature. + +- [Auth Management](acl/Operations_ACL.md):introduce how to deployment quickly and how to use RocketMQ cluster enabling auth management feature. + +- [Quick Start](dledger/quick_start.md):introduce how to deploy Dledger quickly. + +- [Cluster Deployment](dledger/deploy_guide.md):introduce how to deploy Dledger in cluster. + +### 5. Operation and maintenance management +- [Operation](operation.md):introduce RocketMQ's deployment modes that including single-master mode, multi-master mode, multi-master multi-slave mode and so on, as well as the usage of operation tool mqadmin. + + +### 6. API Reference(TODO) + +- [DefaultMQProducer API Reference](client/java/API_Reference_DefaultMQProducer.md) + + + + + + + diff --git a/docs/en/RocketMQ_Example.md b/docs/en/RocketMQ_Example.md new file mode 100644 index 0000000000..48011aff9a --- /dev/null +++ b/docs/en/RocketMQ_Example.md @@ -0,0 +1,15 @@ +### Examples List + +- [basic example](Example_Simple.md) + +- [sequence message example](Example_Orderly.md) + +- [delay message example](Example_Delay.md) + +- [batch message example](Example_Batch.md) + +- [filter message example](Example_Filter.md) + +- [transaction message example](Example_Transaction.md) + +- [openmessaging example](Example_OpenMessaging.md) diff --git a/docs/en/dledger/deploy_guide.md b/docs/en/dledger/deploy_guide.md new file mode 100644 index 0000000000..55509498dd --- /dev/null +++ b/docs/en/dledger/deploy_guide.md @@ -0,0 +1,78 @@ +# Dledger cluster deployment +--- +## preface +This document introduces how to deploy auto failover RocketMQ-on-DLedger Group。 + +RocketMQ-on-DLedger Group is a broker group with **same name**, needs at least 3 nodes, elect a Leader by Raft algorithm automatically, the others as Follower, replicating data between Leader and Follower for system high available. +RocketMQ-on-DLedger Group can failover automatically, and maintains consistent. +RocketMQ-on-DLedger Group can scale up horizontal, that is, can deploy any RocketMQ-on-DLedger Groups providing services external. + +## 1. New cluster deployment + +#### 1.1 Write the configuration +each RocketMQ-on-DLedger Group needs at least 3 machines.(assuming 3 in this document) +write 3 configuration files, advising refer to the directory of conf/dledger 's example configuration file. +key configuration items: + +| name | meaning | example | +| --- | --- | --- | +| enableDLegerCommitLog | whether enable DLedger  | true | +| dLegerGroup | DLedger Raft Group's name, advising maintain consistent to brokerName | RaftNode00 | +| dLegerPeers | DLedger Group's nodes port infos, each node's configuration stay consistent in the same group. | n0-127.0.0.1:40911;n1-127.0.0.1:40912;n2-127.0.0.1:40913 | +| dLegerSelfId | node id, must belongs to dLegerPeers; each node is unique in the same group. | n0 | +| sendMessageThreadPoolNums | the count of sending thread, advising set equal to the cpu cores. | 16 | + +the following presents an example configuration conf/dledger/broker-n0.conf. + +``` +brokerClusterName = RaftCluster +brokerName=RaftNode00 +listenPort=30911 +namesrvAddr=127.0.0.1:9876 +storePathRootDir=/tmp/rmqstore/node00 +storePathCommitLog=/tmp/rmqstore/node00/commitlog +enableDLegerCommitLog=true +dLegerGroup=RaftNode00 +dLegerPeers=n0-127.0.0.1:40911;n1-127.0.0.1:40912;n2-127.0.0.1:40913 +## must be unique +dLegerSelfId=n0 +sendMessageThreadPoolNums=16 +``` + +### 1.2 Start Broker + +Startup stays consistent with the old version. + +`nohup sh bin/mqbroker -c conf/dledger/xxx-n0.conf & ` +`nohup sh bin/mqbroker -c conf/dledger/xxx-n1.conf & ` +`nohup sh bin/mqbroker -c conf/dledger/xxx-n2.conf & ` + + +## 2. Upgrade old cluster + +If old cluster deployed in Master mode, then each Master needs to be transformed into a RocketMQ-on-DLedger Group. +If old cluster deployed in Master-Slave mode, then each Master-Slave group needs to be transformed into a RocketMQ-on-DLedger Group. + +### 2.1 Kill old Broker + +execute kill command, or call `bin/mqshutdown broker`. + +### 2.2 Check old Commitlog + +Each node in RocketMQ-on-DLedger group is compatible with old Commitlog, but Raft replicating process works on the adding message only. So, to avoid occurring exceptions, old Commitlog must be consistent. +If old cluster deployed in Master-Slave mode, it maybe inconsistent after shutdown. Advising use md5sum to check at least 2 recently Commitlog file, if occur inconsistent, maintain consistent by copy. + +Although RocketMQ-on-DLedger Group can deployed with 2 nodes, it lacks failover ability(at least 3 nodes can tolerate one node fail). +Make sure that both Master and Slave's Commitlog is consistent, then prepare 3 machines, copy old Commitlog from Master to this 3 machines(BTW, copy the config directory). + +Then, go ahead to set configurations. + +### 2.3 Modify configuration + +Refer to New cluster deployment. + +### 2.4 Restart Broker + +Refer to New cluster deployment. + + diff --git a/docs/en/dledger/quick_start.md b/docs/en/dledger/quick_start.md new file mode 100644 index 0000000000..f947e166d5 --- /dev/null +++ b/docs/en/dledger/quick_start.md @@ -0,0 +1,63 @@ +# Dledger Quick Deployment +--- +### preface +This document is mainly introduced for how to build and deploy auto failover RocketMQ cluster based on DLedger. + +For detailed new cluster deployment and old cluster upgrade document, please refer to [Deployment Guide](deploy_guide.md)。 + +### 1. Build from source code +Build phase contains two parts, first, build DLedger, then build RocketMQ. + +#### 1.1 Build DLedger + +`git clone https://github.com/openmessaging/openmessaging-storage-dledger.git` + +`cd openmessaging-storage-dledger` + +`mvn clean install -DskipTests` + +#### 1.2 Build RocketMQ + +`git clone https://github.com/apache/rocketmq.git` + +`cd rocketmq` + +`git checkout -b store_with_dledger origin/store_with_dledger` + +`mvn -Prelease-all -DskipTests clean install -U` + +### 2. Quick Deployment + +after build successful + +`cd distribution/target/apache-rocketmq` + +`sh bin/dledger/fast-try.sh start` + +if the above commands executed successfully, then check cluster status by using mqadmin operation commands. + +`sh bin/mqadmin clusterList -n 127.0.0.1:9876` + +If everything goes well, the following content will appear: + +![ClusterList](https://img.alicdn.com/5476e8b07b923/TB11Z.ZyCzqK1RjSZFLXXcn2XXa) + +(BID is 0 indicate Master,the others are Follower) + +After startup successful, producer can produce message, and then test failover scenario. + +Stop cluster fastly, execute the following command: + +`sh bin/dledger/fast-try.sh stop` + +Quick deployment, default configuration is in directory conf/dledger,default storage path is /tmp/rmqstore. + + +### 3. Failover + +After successful deployment, kill Leader process(as the above example, kill process that binds port 30931), about 10 seconds elapses, use clusterList command check cluster's status, Leader switch to another node. + + + + + diff --git a/docs/en/msg_trace/user_guide.md b/docs/en/msg_trace/user_guide.md new file mode 100644 index 0000000000..eb540d02be --- /dev/null +++ b/docs/en/msg_trace/user_guide.md @@ -0,0 +1,105 @@ +# message trace +---- + +## 1. Message trace data's key properties +| Producer End| Consumer End| Broker End| +| --- | --- | --- | +| produce message | consume message | message's topic | +| send message time | delivery time, delivery rounds  | message store location | +| whether the message was sent successfully | whether message was consumed successfully | message's key | +| send cost-time | consume cost-time | message's tag value | + +## 2. Enable message trace in cluster deployment + +### 2.1 Broker's configuration file +following by Broker's properties file configuration that enable message trace: +``` +brokerClusterName=DefaultCluster +brokerName=broker-a +brokerId=0 +deleteWhen=04 +fileReservedTime=48 +brokerRole=ASYNC_MASTER +flushDiskType=ASYNC_FLUSH +storePathRootDir=/data/rocketmq/rootdir-a-m +storePathCommitLog=/data/rocketmq/commitlog-a-m +autoCreateSubscriptionGroup=true +## if msg tracing is open,the flag will be true +traceTopicEnable=true +listenPort=10911 +brokerIP1=XX.XX.XX.XX1 +namesrvAddr=XX.XX.XX.XX:9876 +``` + +### 2.2 Common mode +Each Broker node in RocketMQ cluster used for storing message trace data that client collected and sent. So, there is no requirements and limitations to the size of Broker node in RocketMQ cluster. + +### 2.3 IO physical isolation mode +For huge amounts of message trace data scenario, we can select any one Broker node in RocketMQ cluster used for storing message trace data special, thus, common message data's IO are isolated from message trace data's IO in physical, not impact each other. In this mode, RocketMQ cluster must have at least two Broker nodes, the one that defined as storing message trace data. + +### 2.4 Start Broker that enable message trace +`nohup sh mqbroker -c ../conf/2m-noslave/broker-a.properties &` + +## 3. Save the definition of topic that with support message trace +RocketMQ's message trace feature supports two types of storage. + +### 3.1 System level TraceTopic +Be default, message trace data is stored in system level TraceTopic(topic name: **RMQ_SYS_TRACE_TOPIC**). That topic will be created at startup of broker(As mentioned above, set **traceTopicEnable** to **true** in Broker's configuration). + +### 3.2 User defined TraceTopic +If user don't want to store message trace data in system level TraceTopic, he can create user defined TraceTopic used for storing message trace data(that is, create common topic for storing message trace data). The following part will introduce how client SDK support user defined TraceTopic. + +## 4. Client SDK demo with message trace feature +For business system adapting to use RocketMQ's message trace feature easily, in design phase, the author add a switch parameter(**enableMsgTrace**) for enable message trace; add a custom parameter(**customizedTraceTopic**) for user defined TraceTopic. + +### 4.1 Enable message trace when sending messages +``` + DefaultMQProducer producer = new DefaultMQProducer("ProducerGroupName",true); + producer.setNamesrvAddr("XX.XX.XX.XX1"); + producer.start(); + try { + { + Message msg = new Message("TopicTest", + "TagA", + "OrderID188", + "Hello world".getBytes(RemotingHelper.DEFAULT_CHARSET)); + SendResult sendResult = producer.send(msg); + System.out.printf("%s%n", sendResult); + } + + } catch (Exception e) { + e.printStackTrace(); + } +``` + +### 4.2 Enable message trace when subscribe messages +``` + DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("CID_JODIE_1",true); + consumer.subscribe("TopicTest", "*"); + consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET); + consumer.setConsumeTimestamp("20181109221800"); + consumer.registerMessageListener(new MessageListenerConcurrently() { + @Override + public ConsumeConcurrentlyStatus consumeMessage(List msgs, ConsumeConcurrentlyContext context) { + System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs); + return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; + } + }); + consumer.start(); + System.out.printf("Consumer Started.%n"); +``` + +### 4.3 Self-defined topic support message trace +Adjusting instantiation of DefaultMQProducer and DefaultMQPushConsumer as following code to support user defined TraceTopic. +``` + ##Topic_test11111 should be created by user, used for storing message trace data. + DefaultMQProducer producer = new DefaultMQProducer("ProducerGroupName",true,"Topic_test11111"); + ...... + + DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("CID_JODIE_1",true,"Topic_test11111"); + ...... +``` + + + + diff --git a/docs/en/operation.md b/docs/en/operation.md new file mode 100644 index 0000000000..1d41e98444 --- /dev/null +++ b/docs/en/operation.md @@ -0,0 +1,1374 @@ + +# Operation Management +--- + +### 1 Deploy cluster + +#### 1.1 Single Master mode + +This mode is risky, upon broker restart or broken down, the whole service is unavailable. It's not recommended in production environment, it can be used for local test. + +##### 1)Start NameServer + +```bash +### Start Name Server +$ nohup sh mqnamesrv & + +### check whether Name Server is successfully started +$ tail -f ~/logs/rocketmqlogs/namesrv.log +The Name Server boot success... +``` + +##### 2)Start Broker + +```bash +### start Broker +$ nohup sh bin/mqbroker -n localhost:9876 & + +### check whether Name Server is successfully started, eg: Broker's IP is 192.168.1.2, Broker's name is broker-a +$ tail -f ~/logs/rocketmqlogs/Broker.log +The broker[broker-a, 192.169.1.2:10911] boot success... +``` + +#### 1.2 Multi Master mode + +Cluster contains Master node only, no Slave node, eg: 2 Master nodes, 3 Master nodes, advantages and disadvantages of this mode are shown below: + +- advantages:simple configuration, single Master node broke down or restart do not impact application. Under RAID10 disk config, even if machine broken down and cannot recover, message do not get lost because of RAID10's high reliable(async flush to disk lost little message, sync to disk do not lost message), this mode get highest performance. + +- disadvantages:during the machine's down time, messages have not be consumed on this machine can not be subscribed before recovery. That will impacts message's instantaneity. + +##### 1)Start NameServer + +NameServer should be started before broker. If under production environment, we recommend start 3 NameServer nodes for high available. Startup command is equal, as shown below: + +```bash +### start Name Server +$ nohup sh mqnamesrv & + +### check whether Name Server is successfully started +$ tail -f ~/logs/rocketmqlogs/namesrv.log +The Name Server boot success... +``` + +##### 2)start Broker cluster + +```bash +### start the first Master on machine A, eg:NameServer's IP is :192.168.1.1 +$ nohup sh mqbroker -n 192.168.1.1:9876 -c $ROCKETMQ_HOME/conf/2m-noslave/broker-a.properties & + +### start the second Master on machine B, eg:NameServer's IP is :192.168.1.1 +$ nohup sh mqbroker -n 192.168.1.1:9876 -c $ROCKETMQ_HOME/conf/2m-noslave/broker-b.properties & + +... +``` + +The above commands only used for single NameServer. In multi NameServer cluster, multi addresses concat by semicolon followed by -n in broker start command. + +#### 1.3 Multi Master Multi Slave mode - async replication + +Each Master node is equipped with one Slave node, this mode has many Master-Slave group, using async replication for HA, slaver has a lag(ms level) behind master, advantages and disadvantages of this mode are shown below: + +- advantages: message lost a little, even if disk is broken; message instantaneity do not loss; Consumer can still consume from slave when master is down, this process is transparency to user, no human intervention is required; Performance is almost equal to Multi Master mode. + +- disadvantages: message lost a little data, when Master is down and disk broken. + +##### 1)Start NameServer + +```bash +### start Name Server +$ nohup sh mqnamesrv & + +### check whether Name Server is successfully started +$ tail -f ~/logs/rocketmqlogs/namesrv.log +The Name Server boot success... +``` + +##### 2)Start Broker cluster + +```bash +### start first Master on machine A, eg: NameServer's IP is 192.168.1.1 +$ nohup sh mqbroker -n 192.168.1.1:9876 -c $ROCKETMQ_HOME/conf/2m-2s-async/broker-a.properties & + +### start second Master on machine B, eg: NameServer's IP is 192.168.1.1 +$ nohup sh mqbroker -n 192.168.1.1:9876 -c $ROCKETMQ_HOME/conf/2m-2s-async/broker-b.properties & + +### start first Slave on machine C, eg: NameServer's IP is 192.168.1.1 +$ nohup sh mqbroker -n 192.168.1.1:9876 -c $ROCKETMQ_HOME/conf/2m-2s-async/broker-a-s.properties & + +### start second Slave on machine D, eg: NameServer's IP is 192.168.1.1 +$ nohup sh mqbroker -n 192.168.1.1:9876 -c $ROCKETMQ_HOME/conf/2m-2s-async/broker-b-s.properties & +``` + +#### 1.4 Multi Master Multi Slave mode - synchronous double write + +Each Master node is equipped with one Slave node, this mode has many Master-Slave group, using synchronous double write for HA, application's write operation is successful means both master and slave write successful, advantages and disadvantages of this mode are shown below: + +- advantages:both data and service have no single point failure, message has no lantancy even if Master is down, service available and data available is very high; + +- disadvantages:this mode's performance is 10% lower than async replication mode, sending latency is a little high, in the current version, it do not have auto Master-Slave switch when Master is down. + +##### 1)Start NameServer + +```bash +### start Name Server +$ nohup sh mqnamesrv & + +### check whether Name Server is successfully started +$ tail -f ~/logs/rocketmqlogs/namesrv.log +The Name Server boot success... +``` + +##### 2)Start Broker cluster + +```bash +### start first Master on machine A, eg:NameServer's IP is 192.168.1.1 +$ nohup sh mqbroker -n 192.168.1.1:9876 -c $ROCKETMQ_HOME/conf/2m-2s-sync/broker-a.properties & + +### start second Master on machine B, eg:NameServer's IP is 192.168.1.1 +$ nohup sh mqbroker -n 192.168.1.1:9876 -c $ROCKETMQ_HOME/conf/2m-2s-sync/broker-b.properties & + +### start first Slave on machine C, eg: NameServer's IP is 192.168.1.1 +$ nohup sh mqbroker -n 192.168.1.1:9876 -c $ROCKETMQ_HOME/conf/2m-2s-sync/broker-a-s.properties & + +### start second Slave on machine D, eg: NameServer's IP is 192.168.1.1 +$ nohup sh mqbroker -n 192.168.1.1:9876 -c $ROCKETMQ_HOME/conf/2m-2s-sync/broker-b-s.properties & +``` + +The above Broker matches Slave by specifying the same BrokerName, Master's BrokerId must be 0, Slave's BrokerId must larger than 0. Besides, a Master can have multi Slaves that each has a distinct BrokerId. $ROCKETMQ_HOME indicates RocketMQ's install directory, user needs to set this environment parameter. + +### 2 mqadmin management tool + +> Attentions: +> +> 1. execute command:`./mqadmin {command} {args}` +> 2. almost all commands need -n indicates NameSerer address, format is ip:port +> 3. almost all commands can get help info by -h +> 4. if command contains both Broker address(-b) and cluster name(-c), it's prior to use broker address. If command do not contains broker address, it will executed on all hosts in this cluster. Support only one broker host. -b format is ip:port, default port is 10911 +> 5. there are many commands under tools, but not all command can be used, only commands that initialized in MQAdminStartup can be used, you can modify this class, add or self-define command. +> 6. because of version update, little command do not update timely, please refer to source code directly when occur error. +#### 2.1 Topic + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + +
namemeaningcommand itemsexplaination
updateTopiccreate or update Topic's config-bBroker address, means which Broker that topic is located, only support single Broker, address format is ip:port
-ccluster name, whic cluster that topic belongs to(query cluster info by clusterList)
-h-print help info
-nNameServer Service address, format is ip:port
-passign read write authority to new topic(W=2|R=4|WR=6)
-rthe count of queue that can be read(default is 8)
-wthe count of queue that can be wrote(default is 8)
-ttopic name(can only use characters ^[a-zA-Z0-9_-]+$ )
deleteTopicdelete Topic-ccluster name, which cluster that topic will be deleted belongs to(query cluster info by clusterList)
-hprint help info
-nNameServer Service address, format is ip:port
-ttopic name(can only use characters ^[a-zA-Z0-9_-]+$ )
topicListquery Topic list info-hprint help info
-creturn topic list only if do not contains -c, if containis -c, it will return cluster name, topic name, consumer group name
-nNameServer Service address, format is ip:port
topicRoutequery Topic's route info-ttopic name
-hprint help info
-nNameServer Service address, format is ip:port
topicStatusquery Topic's offset-ttopic name
-hprint help info
-nNameServer Service address, format is ip:port
topicClusterListquery cluster list where Topic belongs to-ttopic name
-hprint help info
-nNameServer Service address, format is ip:port
updateTopicPermupdate Topic's produce and consume authority-ttopic name
-hprint help info
-nNameServer Service address, format is ip:port
-bBroker address which topic belongs to, support single broker only, format is ip:port
-passign read and write authority to the new topic(W=2|R=4|WR=6)
-ccluster name, which topic belongs to(query cluster info by clusterList), if do not have -b, execute comman an all brokers.
updateOrderConfcreate delete get specified namespace's kv config from NameServer, have not enabled at present-hprint help info
-nNameServer Service address, format is ip:port
-ttopic, key
-vorderConf, value
-mmethod, including get, put, delete
allocateMQcalculate consumer list rebalance result by average rebalance algorithm-ttopic name
-hprint help info
-nNameServer Service address, format is ip:port
-iipList, seperate by comma, calculate which topic queue that ips will load.
statsAllprint Topic's subscribe info, TPS, size of message blocked, count of read and write at last 24h, eg.-hprint help info
-nNameServer Service address, format is ip:port
-aonly print active topic or not
-tassign topic
+ + + +#### 2.2 Cluster + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + +
名称meaningcommand itemsexplaination
clusterListquery cluster info, including cluster, BrokerName, BrokerId, TPS, eg.-mprint more infos(eg: #InTotalYest, #OutTotalYest, #InTotalToday ,#OutTotalToday)
-hprint help info
-nNameServer Service address, format is ip:port
-iprint interval, unit second
clusterRTsend message to detect each cluster's Broker RT. Message will be sent to ${BrokerName} Topic。-aamount, count of detection, RT = sum time / + amount
-ssize of message, unit B
-cwhich cluster will be detected
-pwhether print format log, splitted by |, default is not print
-hprint help info
-mwhich machine room it belongs to, just for print
-isend interval, unit second
-nNameServer Service address, format is ip:port
+ + +#### 2.3 Broker + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + +
名称meaningcommand itemsexplaination
updateBrokerConfigupdate Broker's config file, it will modify Broker.conf-bBroker address, format is ip:port
-ccluster name
-kkey
-vvalue
-hprint help info
-nNameServer Service address, format is ip:port
brokerStatusget Broker's statistics info, running status(including whatever you want).-bBroker address, fomat isip:port
-hprint help info
-nNameServer Service address, format is ip:port
brokerConsumeStatsBroker's consumer info, including Consume Offset, Broker Offset, Diff, Timestamp that ordered by essage Queue-bBroker address, fomat isip:port
-trequest timeout time
-ldiff threshold, it will print when exceed this threshold.
-owhether is sequencial topic, generally false
-hprint help info
-nNameServer Service address, format is ip:port
getBrokerConfigget Broker's config-bBroker address, fomat isip:port
-nNameServer Service address, format is ip:port
wipeWritePermrevoke broker's write authority from NameServer.-bBroker address, fomat isip:port
-nNameServer Service address, format is ip:port
-hprint help info
cleanExpiredCQclean Broker's expired Consume Queue that maybe generated by decrease queue count.-nNameServer Service address, format is ip:port
-hprint help info
-bBroker address, fomat isip:port
-ccluster name
cleanUnusedTopicclean Broker's unused Topic that deleted mannually to release memory that Topic's Consume Queue occupied.-nNameServer Service address, format is ip:port
-hprint help info
-bBroker address, fomat isip:port
-ccluster name
sendMsgStatussend message to Broker, return send status and RT-nNameServer Service address, format is ip:port
-hprint help info
-bBrokerName, is different from broker address
-smessage size, unit B
-csend count
+ + +#### 2.4 Message + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + +
名称meaningcommand itemsexplaination
queryMsgByIdquery message by offsetMsgId. If use opensource console, it should use offsetMsgId. Please refer to QueryMsgByIdSubCommand for detail.-imsgId
-hprint help info
-nNameServer Service address, format is ip:port
queryMsgByKeyquery message by Message's Key-kmsgKey
-ttopic name
-hprint help info
-nNameServer Service address, format is ip:port
queryMsgByOffsetquery message by Offset-bBroker name(it's not broker address, can query Broker name by clusterList).
-iquery queue id
-ooffset value
-ttopic name
-hprint help info
-nNameServer Service address, format is ip:port
queryMsgByUniqueKeyquery by msgId, msgId is different from offsetMsgId, please refer to Frequently asked questions about operations for details. Use -g and -d to let specified consumer return consume result.-hprint help info
-nNameServer Service address, format is ip:port
-iuniqe msg id
-gconsumerGroup
-dclientId
-ttopic name
checkMsgSendRTdetect RT of sending a message to a topic, similiar to clusterRT-hprint help info
-nNameServer Service address, format is ip:port
-ttopic name
-adetection count
-ssize of the message
sendMessagesend a message, also can send to a specified Message Queue.-hprint help info
-nNameServer Service address, format is ip:port
-ttopic name
-pbody, message entity
-kkeys
-ctags
-bBrokerName
-iqueueId
consumeMessageconsume message. Differert consume logic depends on offset, start & end timestamp, message queue, please refer to ConsumeMessageCommand for details.-hprint help info
-nNameServer Service address, format is ip:port
-ttopic name
-bBrokerName
-ooffset that consumer start consume
-iqueueId
-gconsumer gropu
-stimestamp at start, refer to -h to get format开
-dtimestamp at the end
-csize of message that consumed
printMsgconsume and print messages from broker, support a time range-hprint help info
-nNameServer Service address, format is ip:port
-ttopic name
-ccharset, eg: UTF-8
-ssubExpress, filter expression
-btimestap at start, refer to -h to get format
-etimestamp at the end
-dwhether print message entity or not
printMsgByQueuesimilar to printMsg, but it need specified Message Queue-hprint help info
-nNameServer Service address, format is ip:port
-ttopic name
-iqueueId
-aBrokerName
-ccharset, eg: UTF-8
-ssubExpress, filter expression
-btimestamp at start, refer to -h to get format
-etimestamp at the end
-pwhether print message or not
-dwhether print message entity or not
-fwhether count and print tag or not
resetOffsetByTimereset offset by timestamp, Broker and consumer will all be reseted-hprint help info
-nNameServer Service address, format is ip:port
-gconsumer group
-ttopic name
-sreset offset corresponding to this timestamp
-fwhether enforce to reset or not, if set false, only can reset offset, if set true, it omit the relationship between timestamp and consumer offset.
-cwhether reset c++ sdk's offset or not
+ + +#### 2.5 Consumer, Consumer Group + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + +
namemeaningcommand itemsexplaination
consumerProgressquery subscribe status, can get blocking counts of a concrete client ip.-gconsumer group name
-swhether print client IP or not
-hprint help info
-nNameServer Service address, format is ip:port
consumerStatusquery consumer status, including message blocking, and consumer's jstack result(please refer to ConsumerStatusSubCommand)-hprint help info
-nNameServer Service address, format is ip:port
-gconsumer group
-iclientId
-swhether execute jstack or not
updateSubGroupcreate or update subscribe info-nNameServer Service address, format is ip:port
-hprint help info
-bBroker address
-ccluster name
-gconsumer group name
-sconsumer group is allowed to consume or not
-mstart consume from minimal offset or not
-dbroadcast mode or not
-qcapacity of retry queue
-rmax retry count
-iIt works when slaveReadEnable enabled, and that not consumed from slave. Suggesting that consume from slave node by specify slave id.
-wIf broker consume from slave, whic slave node depends on this config that configed by BrokerId, eg: 1.
-awhether notify other consumers to rebalance or not when the count of consumer changes
deleteSubGroupdelete subscribe info from Broker-nNameServer Service address, format is ip:port
-hprint help info
-bBroker address
-ccluster name
-gconsumer group name
cloneGroupOffsetuse source group's offset at target group-nNameServer Service address, format is ip:port
-hprint help info
-ssource consumer group
-dtarget consumer group
-ttopic name
-onot used at present
+ + + + +#### 2.6 Connection + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + +
namemeaningcommand itemsexplaination
consumerConnec tionquery Consumer's connection-gconsumer group name
-nNameServer Service address, format is ip:port
-hprint help info
producerConnec tionquery Producer's connection-gproducer group name
-ttopic name
-nNameServer Service address, format is ip:port
-hprint help info
+ + + + +#### 2.7 NameServer + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + +
namemeaningcommand itemsexplaination
updateKvConfigupdate NameServer's kv config, not used at present-snamespace
-kkey
-vvalue
-nNameServer Service address, format is ip:port
-hprint help info
deleteKvConfigdelete NameServer's kv config-snamespace
-kkey
-nNameServer Service address, format is ip:port
-hprint help info
getNamesrvConfigget NameServer's config-nNameServer Service address, format is ip:port
-hprint help info
updateNamesrvConfigmodify NameServer's config-nNameServer Service address, format is ip:port
-hprint help info
-kkey
-vvalue
+ + + + +#### 2.8 Other + + + + + + + + + + + + + + + + + + + + + + +
namemeaningcommand itemsexplaination
startMonitoringStart the monitoring process, monitor message deletion and the number of retried messages in the queue-nNameServer Service address, format is ip:port
-hprint help info
+ + +### 3 Frequently asked questions about operations + +#### 3.1 RocketMQ's mqadmin command error + +> question description:execute mqadmin occur below exception after deploy RocketMQ cluster. +> +> ```java +> org.apache.rocketmq.remoting.exception.RemotingConnectException: connect to failed +> ``` + +Solution: execute command `export NAMESRV_ADDR=ip:9876` (ip is NameServer's ip address), then execute mqadmin commands. + +#### 3.2 RocketMQ consumer cannot consume, because of different version of producer and consumer. + +> question description: one producer produce message, consumer A can consume, consume B cannot consume, RocketMQ console print: +> +> ```java +> Not found the consumer group consume stats, because return offset table is empty, maybe the consumer not consume any message。 +> ``` + +Solution: make sure that producer and consumer has the same version of rocketmq-client. + +#### 3.3 Consumer cannot consume oldest message, when a new consumer group is added. + +> question description: when a new consumer group start, it consumes from current offset, do not fetch oldest message. + +Solution: rocketmq's default policy is consume from latest, that is skip oldest message. If you want consume oldest message, you need to set `org.apache.rocketmq.client.consumer.DefaultMQPushConsumer#setConsumeFromWhere`. The following is three common configurations: + +- default configuration, a new consumer group consume from latest position at first startup, then consume from last time's offset at next startup, that is skip oldest message; + +```java +consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET); +``` + +- a new consumer group consume from oldest postion at first startup, then consume from last time's offset at next startup, that is consume the unexpired message; + +```java +consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET); +``` + +- a new consumer group consume from specified timestamp at first startup, then consume from last time's offset at next startup, cooperate with consumer.setConsumeTimestamp(), default is half an hour before; + +```java +consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_TIMESTAMP); +``` + +#### 3.4 How to enable consume from Slave + +In some cases, consumer need reset offset to a day or two before, if Master Broker has limited memory, it's CommitLog will have a high IO load, then it will impact other message's read and write that on this broker. When `slaveReadEnable=true` is set, and consumer's offset exceeds `accessMessageInMemoryMaxRatio=40%`, Master Broker will recommend consumer consume from Slave Broker to lower Master Broker IO. + +#### 3.5 Performance tuning + +A spin lock is recommended for asynchronous disk flush, a reentrant lock is recommended for synchronous disk flush, configuration item is `useReentrantLockWhenPutMessage`, default is false; Enable `TransientStorePoolEnable` is recommended when use asynchronous disk flush; Recommend to close `transferMsgByHeap` to improve fetch efficiency; Set a little larger `sendMessageThreadPoolNums`, when use synchronous disk flush. + +#### 3.6 The meaning and difference between msgId and offsetMsgId in RocketMQ + +You will usually see the following log print message after sending message by using RocketMQ sdk. + +```java +SendResult [sendStatus=SEND_OK, msgId=0A42333A0DC818B4AAC246C290FD0000, offsetMsgId=0A42333A00002A9F000000000134F1F5, messageQueue=MessageQueue [topic=topicTest1, BrokerName=mac.local, queueId=3], queueOffset=4] +``` + +- msgId, is generated by producer sdk. In particular, call method `MessageClientIDSetter.createUniqIDBuffer()` to generate unique Id; +- offsetMsgId, offsetMsgId is generated by Broker server(format is "Ip Address + port + CommitLog offset"). offsetMsgId is messageId that is RocketMQ console's input. diff --git a/example/pom.xml b/example/pom.xml index 8fb505b6d3..dba83fe6d2 100644 --- a/example/pom.xml +++ b/example/pom.xml @@ -19,7 +19,7 @@ org.apache.rocketmq rocketmq-all - 4.7.0 + 4.7.1-SNAPSHOT 4.0.0 @@ -51,12 +51,12 @@ org.apache.rocketmq rocketmq-openmessaging - 4.7.0 + 4.7.1-SNAPSHOT org.apache.rocketmq rocketmq-acl - 4.7.0 + 4.7.1-SNAPSHOT diff --git a/filter/pom.xml b/filter/pom.xml index a02d1220a3..46debc3e38 100644 --- a/filter/pom.xml +++ b/filter/pom.xml @@ -20,7 +20,7 @@ rocketmq-all org.apache.rocketmq - 4.7.0 + 4.7.1-SNAPSHOT 4.0.0 diff --git a/filter/src/main/java/org/apache/rocketmq/filter/expression/BooleanConstantExpression.java b/filter/src/main/java/org/apache/rocketmq/filter/expression/BooleanConstantExpression.java new file mode 100644 index 0000000000..958e622f6c --- /dev/null +++ b/filter/src/main/java/org/apache/rocketmq/filter/expression/BooleanConstantExpression.java @@ -0,0 +1,37 @@ +/* + * 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.filter.expression; + +/** + * BooleanConstantExpression + */ +public class BooleanConstantExpression extends ConstantExpression implements BooleanExpression { + + public static final BooleanConstantExpression NULL = new BooleanConstantExpression(null); + public static final BooleanConstantExpression TRUE = new BooleanConstantExpression(Boolean.TRUE); + public static final BooleanConstantExpression FALSE = new BooleanConstantExpression(Boolean.FALSE); + + public BooleanConstantExpression(Object value) { + super(value); + } + + public boolean matches(EvaluationContext context) throws Exception { + Object object = evaluate(context); + return object != null && object == Boolean.TRUE; + } +} diff --git a/filter/src/main/java/org/apache/rocketmq/filter/expression/ComparisonExpression.java b/filter/src/main/java/org/apache/rocketmq/filter/expression/ComparisonExpression.java index fe898e7884..9a919c4441 100644 --- a/filter/src/main/java/org/apache/rocketmq/filter/expression/ComparisonExpression.java +++ b/filter/src/main/java/org/apache/rocketmq/filter/expression/ComparisonExpression.java @@ -90,11 +90,11 @@ public abstract class ComparisonExpression extends BinaryExpression implements B } public static BooleanExpression createIsNull(Expression left) { - return doCreateEqual(left, ConstantExpression.NULL); + return doCreateEqual(left, BooleanConstantExpression.NULL); } public static BooleanExpression createIsNotNull(Expression left) { - return UnaryExpression.createNOT(doCreateEqual(left, ConstantExpression.NULL)); + return UnaryExpression.createNOT(doCreateEqual(left, BooleanConstantExpression.NULL)); } public static BooleanExpression createNotEqual(Expression left, Expression right) { diff --git a/filter/src/main/java/org/apache/rocketmq/filter/expression/ConstantExpression.java b/filter/src/main/java/org/apache/rocketmq/filter/expression/ConstantExpression.java index e649f5a232..127bd8ba7b 100644 --- a/filter/src/main/java/org/apache/rocketmq/filter/expression/ConstantExpression.java +++ b/filter/src/main/java/org/apache/rocketmq/filter/expression/ConstantExpression.java @@ -30,21 +30,6 @@ package org.apache.rocketmq.filter.expression; */ public class ConstantExpression implements Expression { - static class BooleanConstantExpression extends ConstantExpression implements BooleanExpression { - public BooleanConstantExpression(Object value) { - super(value); - } - - public boolean matches(EvaluationContext context) throws Exception { - Object object = evaluate(context); - return object != null && object == Boolean.TRUE; - } - } - - public static final BooleanConstantExpression NULL = new BooleanConstantExpression(null); - public static final BooleanConstantExpression TRUE = new BooleanConstantExpression(Boolean.TRUE); - public static final BooleanConstantExpression FALSE = new BooleanConstantExpression(Boolean.FALSE); - private Object value; public ConstantExpression(Object value) { @@ -60,16 +45,10 @@ public class ConstantExpression implements Expression { // only support Long.MIN_VALUE ~ Long.MAX_VALUE Number value = new Long(text); -// try { -// value = new Long(text); -// } catch (NumberFormatException e) { -// // The number may be too big to fit in a long. -// value = new BigDecimal(text); -// } long l = value.longValue(); if (Integer.MIN_VALUE <= l && l <= Integer.MAX_VALUE) { - value = Integer.valueOf(value.intValue()); + value = value.intValue(); } return new ConstantExpression(value); } @@ -106,7 +85,7 @@ public class ConstantExpression implements Expression { return "NULL"; } if (value instanceof Boolean) { - return ((Boolean) value).booleanValue() ? "TRUE" : "FALSE"; + return (Boolean) value ? "TRUE" : "FALSE"; } if (value instanceof String) { return encodeString((String) value); @@ -138,17 +117,19 @@ public class ConstantExpression implements Expression { * it was provided in a selector. */ public static String encodeString(String s) { - StringBuffer b = new StringBuffer(); - b.append('\''); + + StringBuilder builder = new StringBuilder(); + + builder.append('\''); for (int i = 0; i < s.length(); i++) { char c = s.charAt(i); if (c == '\'') { - b.append(c); + builder.append(c); } - b.append(c); + builder.append(c); } - b.append('\''); - return b.toString(); + builder.append('\''); + return builder.toString(); } } diff --git a/filter/src/main/java/org/apache/rocketmq/filter/parser/SelectorParser.java b/filter/src/main/java/org/apache/rocketmq/filter/parser/SelectorParser.java index 198aacf83e..4ea7dcfd42 100644 --- a/filter/src/main/java/org/apache/rocketmq/filter/parser/SelectorParser.java +++ b/filter/src/main/java/org/apache/rocketmq/filter/parser/SelectorParser.java @@ -20,6 +20,7 @@ package org.apache.rocketmq.filter.parser; import com.google.common.cache.Cache; import com.google.common.cache.CacheBuilder; +import org.apache.rocketmq.filter.expression.BooleanConstantExpression; import org.apache.rocketmq.filter.expression.BooleanExpression; import org.apache.rocketmq.filter.expression.ComparisonExpression; import org.apache.rocketmq.filter.expression.ConstantExpression; @@ -437,15 +438,15 @@ public class SelectorParser implements SelectorParserConstants { break; case TRUE: jj_consume_token(TRUE); - left = ConstantExpression.TRUE; + left = BooleanConstantExpression.TRUE; break; case FALSE: jj_consume_token(FALSE); - left = ConstantExpression.FALSE; + left = BooleanConstantExpression.FALSE; break; case NULL: jj_consume_token(NULL); - left = ConstantExpression.NULL; + left = BooleanConstantExpression.NULL; break; default: jjLa1[12] = jjGen; diff --git a/logappender/pom.xml b/logappender/pom.xml index 884e6e9899..c4d0b217d3 100644 --- a/logappender/pom.xml +++ b/logappender/pom.xml @@ -19,7 +19,7 @@ org.apache.rocketmq rocketmq-all - 4.7.0 + 4.7.1-SNAPSHOT 4.0.0 rocketmq-logappender diff --git a/logging/pom.xml b/logging/pom.xml index 24088c40cf..b4a52a7b0a 100644 --- a/logging/pom.xml +++ b/logging/pom.xml @@ -19,7 +19,7 @@ org.apache.rocketmq rocketmq-all - 4.7.0 + 4.7.1-SNAPSHOT 4.0.0 diff --git a/namesrv/pom.xml b/namesrv/pom.xml index ff821cf5ef..d8ae216588 100644 --- a/namesrv/pom.xml +++ b/namesrv/pom.xml @@ -19,7 +19,7 @@ org.apache.rocketmq rocketmq-all - 4.7.0 + 4.7.1-SNAPSHOT 4.0.0 diff --git a/namesrv/src/main/java/org/apache/rocketmq/namesrv/processor/DefaultRequestProcessor.java b/namesrv/src/main/java/org/apache/rocketmq/namesrv/processor/DefaultRequestProcessor.java index 7210246edc..44a2b6f261 100644 --- a/namesrv/src/main/java/org/apache/rocketmq/namesrv/processor/DefaultRequestProcessor.java +++ b/namesrv/src/main/java/org/apache/rocketmq/namesrv/processor/DefaultRequestProcessor.java @@ -97,7 +97,7 @@ public class DefaultRequestProcessor extends AsyncNettyRequestProcessor implemen } case RequestCode.UNREGISTER_BROKER: return this.unregisterBroker(ctx, request); - case RequestCode.GET_ROUTEINTO_BY_TOPIC: + case RequestCode.GET_ROUTEINFO_BY_TOPIC: return this.getRouteInfoByTopic(ctx, request); case RequestCode.GET_BROKER_CLUSTER_INFO: return this.getBrokerClusterInfo(ctx, request); @@ -384,10 +384,12 @@ public class DefaultRequestProcessor extends AsyncNettyRequestProcessor implemen int wipeTopicCnt = this.namesrvController.getRouteInfoManager().wipeWritePermOfBrokerByLock(requestHeader.getBrokerName()); - log.info("wipe write perm of broker[{}], client: {}, {}", - requestHeader.getBrokerName(), - RemotingHelper.parseChannelRemoteAddr(ctx.channel()), - wipeTopicCnt); + if (ctx != null) { + log.info("wipe write perm of broker[{}], client: {}, {}", + requestHeader.getBrokerName(), + RemotingHelper.parseChannelRemoteAddr(ctx.channel()), + wipeTopicCnt); + } responseHeader.setWipeTopicCount(wipeTopicCnt); response.setCode(ResponseCode.SUCCESS); @@ -502,7 +504,9 @@ public class DefaultRequestProcessor extends AsyncNettyRequestProcessor implemen } private RemotingCommand updateConfig(ChannelHandlerContext ctx, RemotingCommand request) { - log.info("updateConfig called by {}", RemotingHelper.parseChannelRemoteAddr(ctx.channel())); + if (ctx != null) { + log.info("updateConfig called by {}", RemotingHelper.parseChannelRemoteAddr(ctx.channel())); + } final RemotingCommand response = RemotingCommand.createResponseCommand(null); @@ -518,13 +522,6 @@ public class DefaultRequestProcessor extends AsyncNettyRequestProcessor implemen return response; } - if (bodyStr == null) { - log.error("updateConfig get null body!"); - response.setCode(ResponseCode.SYSTEM_ERROR); - response.setRemark("string2Properties error"); - return response; - } - Properties properties = MixAll.string2Properties(bodyStr); if (properties == null) { log.error("updateConfig MixAll.string2Properties error {}", bodyStr); diff --git a/openmessaging/pom.xml b/openmessaging/pom.xml index 1b2fa8bcef..9fcd708ef9 100644 --- a/openmessaging/pom.xml +++ b/openmessaging/pom.xml @@ -20,7 +20,7 @@ rocketmq-all org.apache.rocketmq - 4.7.0 + 4.7.1-SNAPSHOT 4.0.0 diff --git a/pom.xml b/pom.xml index dcef6ab20b..1ade0b79fe 100644 --- a/pom.xml +++ b/pom.xml @@ -29,7 +29,7 @@ 2012 org.apache.rocketmq rocketmq-all - 4.7.0 + 4.7.1-SNAPSHOT pom Apache RocketMQ ${project.version} http://rocketmq.apache.org/ @@ -42,7 +42,7 @@ git@github.com:apache/rocketmq.git scm:git:git@github.com:apache/rocketmq.git scm:git:git@github.com:apache/rocketmq.git - rocketmq-all-4.7.0 + HEAD @@ -280,7 +280,7 @@ org.jacoco jacoco-maven-plugin - 0.7.8 + 0.8.5 default-prepare-agent diff --git a/remoting/pom.xml b/remoting/pom.xml index 64c6c11c50..d71f0364a6 100644 --- a/remoting/pom.xml +++ b/remoting/pom.xml @@ -19,7 +19,7 @@ org.apache.rocketmq rocketmq-all - 4.7.0 + 4.7.1-SNAPSHOT 4.0.0 diff --git a/remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettyRemotingClient.java b/remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettyRemotingClient.java index 415d4956b2..d836c8e0bf 100644 --- a/remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettyRemotingClient.java +++ b/remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettyRemotingClient.java @@ -354,6 +354,10 @@ public class NettyRemotingClient extends NettyRemotingAbstract implements Remoti Collections.shuffle(addrs); log.info("name server address updated. NEW : {} , OLD: {}", addrs, old); this.namesrvAddrList.set(addrs); + + if (!addrs.contains(this.namesrvAddrChoosed.get())) { + this.namesrvAddrChoosed.set(null); + } } } } diff --git a/srvutil/pom.xml b/srvutil/pom.xml index 47c3ef5610..6009097862 100644 --- a/srvutil/pom.xml +++ b/srvutil/pom.xml @@ -19,7 +19,7 @@ org.apache.rocketmq rocketmq-all - 4.7.0 + 4.7.1-SNAPSHOT 4.0.0 diff --git a/store/pom.xml b/store/pom.xml index 34c08bc87d..59a9bfa8f2 100644 --- a/store/pom.xml +++ b/store/pom.xml @@ -19,7 +19,7 @@ org.apache.rocketmq rocketmq-all - 4.7.0 + 4.7.1-SNAPSHOT 4.0.0 diff --git a/store/src/main/java/org/apache/rocketmq/store/CommitLog.java b/store/src/main/java/org/apache/rocketmq/store/CommitLog.java index 82bd6706c9..810110eeb6 100644 --- a/store/src/main/java/org/apache/rocketmq/store/CommitLog.java +++ b/store/src/main/java/org/apache/rocketmq/store/CommitLog.java @@ -1383,11 +1383,8 @@ public class CommitLog { return nextOffset; } - public void wakeupCustomer(final boolean flushOK) { - long endTimestamp = System.currentTimeMillis(); - PutMessageStatus result = (flushOK && ((endTimestamp - this.startTimestamp) <= this.timeoutMillis)) ? - PutMessageStatus.PUT_OK : PutMessageStatus.FLUSH_SLAVE_TIMEOUT; - this.flushOKFuture.complete(result); + public void wakeupCustomer(final PutMessageStatus putMessageStatus) { + this.flushOKFuture.complete(putMessageStatus); } public CompletableFuture future() { @@ -1407,9 +1404,7 @@ public class CommitLog { synchronized (this.requestsWrite) { this.requestsWrite.add(request); } - if (hasNotified.compareAndSet(false, true)) { - waitPoint.countDown(); // notify - } + this.wakeup(); } private void swapRequests() { @@ -1433,7 +1428,7 @@ public class CommitLog { } } - req.wakeupCustomer(flushOK); + req.wakeupCustomer(flushOK ? PutMessageStatus.PUT_OK : PutMessageStatus.FLUSH_DISK_TIMEOUT); } long storeTimestamp = CommitLog.this.mappedFileQueue.getStoreTimestamp(); diff --git a/store/src/main/java/org/apache/rocketmq/store/DefaultMessageStore.java b/store/src/main/java/org/apache/rocketmq/store/DefaultMessageStore.java index f59b5eb98b..900e9af31f 100644 --- a/store/src/main/java/org/apache/rocketmq/store/DefaultMessageStore.java +++ b/store/src/main/java/org/apache/rocketmq/store/DefaultMessageStore.java @@ -114,6 +114,9 @@ public class DefaultMessageStore implements MessageStore { boolean shutDownNormal = false; + private final ScheduledExecutorService diskCheckScheduledExecutorService = + Executors.newSingleThreadScheduledExecutor(new ThreadFactoryImpl("DiskCheckScheduledThread")); + public DefaultMessageStore(final MessageStoreConfig messageStoreConfig, final BrokerStatsManager brokerStatsManager, final MessageArrivingListener messageArrivingListener, final BrokerConfig brokerConfig) throws IOException { this.messageArrivingListener = messageArrivingListener; @@ -293,7 +296,7 @@ public class DefaultMessageStore implements MessageStore { this.shutdown = true; this.scheduledExecutorService.shutdown(); - + this.diskCheckScheduledExecutorService.shutdown(); try { Thread.sleep(1000); @@ -1029,6 +1032,10 @@ public class DefaultMessageStore implements MessageStore { } it.remove(); + if (this.brokerConfig.isAutoDeleteUnusedStats()) { + this.brokerStatsManager.onTopicDeleted(topic); + } + log.info("cleanUnusedTopic: {},topic destroyed", topic); } } @@ -1325,6 +1332,11 @@ public class DefaultMessageStore implements MessageStore { // DefaultMessageStore.this.cleanExpiredConsumerQueue(); // } // }, 1, 1, TimeUnit.HOURS); + this.diskCheckScheduledExecutorService.scheduleAtFixedRate(new Runnable() { + public void run() { + DefaultMessageStore.this.cleanCommitLogService.isSpaceFull(); + } + }, 1000L, 10000L, TimeUnit.MILLISECONDS); } private void cleanFilesPeriodically() { @@ -1723,6 +1735,30 @@ public class DefaultMessageStore implements MessageStore { public void setManualDeleteFileSeveralTimes(int manualDeleteFileSeveralTimes) { this.manualDeleteFileSeveralTimes = manualDeleteFileSeveralTimes; } + public boolean isSpaceFull() { + String storePathPhysic = DefaultMessageStore.this.getMessageStoreConfig().getStorePathCommitLog(); + double physicRatio = UtilAll.getDiskPartitionSpaceUsedPercent(storePathPhysic); + double ratio = DefaultMessageStore.this.getMessageStoreConfig().getDiskMaxUsedSpaceRatio() / 100.0; + if (physicRatio > ratio) { + DefaultMessageStore.log.info("physic disk of commitLog used: " + physicRatio); + } + if (physicRatio > this.diskSpaceWarningLevelRatio) { + boolean diskok = DefaultMessageStore.this.runningFlags.getAndMakeDiskFull(); + if (diskok) { + DefaultMessageStore.log.error("physic disk of commitLog maybe full soon, used " + physicRatio + ", so mark disk full"); + } + + return true; + } else { + boolean diskok = DefaultMessageStore.this.runningFlags.getAndMakeDiskOK(); + + if (!diskok) { + DefaultMessageStore.log.info("physic disk space of commitLog OK " + physicRatio + ", so mark disk ok"); + } + + return false; + } + } } class CleanConsumeQueueService { diff --git a/store/src/main/java/org/apache/rocketmq/store/ha/HAService.java b/store/src/main/java/org/apache/rocketmq/store/ha/HAService.java index 3035c575cf..a0e77fa3dc 100644 --- a/store/src/main/java/org/apache/rocketmq/store/ha/HAService.java +++ b/store/src/main/java/org/apache/rocketmq/store/ha/HAService.java @@ -39,6 +39,7 @@ import org.apache.rocketmq.logging.InternalLoggerFactory; import org.apache.rocketmq.remoting.common.RemotingUtil; import org.apache.rocketmq.store.CommitLog; import org.apache.rocketmq.store.DefaultMessageStore; +import org.apache.rocketmq.store.PutMessageStatus; public class HAService { private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.STORE_LOGGER_NAME); @@ -260,9 +261,7 @@ public class HAService { synchronized (this.requestsWrite) { this.requestsWrite.add(request); } - if (hasNotified.compareAndSet(false, true)) { - waitPoint.countDown(); // notify - } + this.wakeup(); } public void notifyTransferSome() { @@ -291,7 +290,7 @@ public class HAService { log.warn("transfer messsage to slave timeout, " + req.getNextOffset()); } - req.wakeupCustomer(transferOK); + req.wakeupCustomer(transferOK ? PutMessageStatus.PUT_OK : PutMessageStatus.FLUSH_SLAVE_TIMEOUT); } this.requestsRead.clear(); diff --git a/store/src/main/java/org/apache/rocketmq/store/ha/WaitNotifyObject.java b/store/src/main/java/org/apache/rocketmq/store/ha/WaitNotifyObject.java index 3483c99217..75b7597fea 100644 --- a/store/src/main/java/org/apache/rocketmq/store/ha/WaitNotifyObject.java +++ b/store/src/main/java/org/apache/rocketmq/store/ha/WaitNotifyObject.java @@ -21,6 +21,7 @@ import org.apache.rocketmq.logging.InternalLogger; import org.apache.rocketmq.logging.InternalLoggerFactory; import java.util.HashMap; +import java.util.Map; public class WaitNotifyObject { private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.STORE_LOGGER_NAME); @@ -65,9 +66,9 @@ public class WaitNotifyObject { synchronized (this) { boolean needNotify = false; - for (Boolean value : this.waitingThreadTable.values()) { - needNotify = needNotify || !value; - value = true; + for (Map.Entry entry : this.waitingThreadTable.entrySet()) { + needNotify = needNotify || !entry.getValue(); + entry.setValue(true); } if (needNotify) { diff --git a/store/src/main/java/org/apache/rocketmq/store/index/IndexFile.java b/store/src/main/java/org/apache/rocketmq/store/index/IndexFile.java index 3d76b73dc2..e513edf111 100644 --- a/store/src/main/java/org/apache/rocketmq/store/index/IndexFile.java +++ b/store/src/main/java/org/apache/rocketmq/store/index/IndexFile.java @@ -134,7 +134,9 @@ public class IndexFile { this.indexHeader.setBeginTimestamp(storeTimestamp); } - this.indexHeader.incHashSlotCount(); + if (invalidIndex == slotValue) { + this.indexHeader.incHashSlotCount(); + } this.indexHeader.incIndexCount(); this.indexHeader.setEndPhyOffset(phyOffset); this.indexHeader.setEndTimestamp(storeTimestamp); diff --git a/store/src/main/java/org/apache/rocketmq/store/stats/BrokerStatsManager.java b/store/src/main/java/org/apache/rocketmq/store/stats/BrokerStatsManager.java index 4adbed76ab..e151844706 100644 --- a/store/src/main/java/org/apache/rocketmq/store/stats/BrokerStatsManager.java +++ b/store/src/main/java/org/apache/rocketmq/store/stats/BrokerStatsManager.java @@ -121,6 +121,26 @@ public class BrokerStatsManager { return null; } + public void onTopicDeleted(final String topic) { + this.statsTable.get(TOPIC_PUT_NUMS).delValue(topic); + this.statsTable.get(TOPIC_PUT_SIZE).delValue(topic); + this.statsTable.get(GROUP_GET_NUMS).delValueByPrefixKey(topic, "@"); + this.statsTable.get(GROUP_GET_SIZE).delValueByPrefixKey(topic, "@"); + this.statsTable.get(SNDBCK_PUT_NUMS).delValueByPrefixKey(topic, "@"); + this.statsTable.get(GROUP_GET_LATENCY).delValueByInfixKey(topic, "@"); + this.momentStatsItemSetFallSize.delValueByInfixKey(topic, "@"); + this.momentStatsItemSetFallTime.delValueByInfixKey(topic, "@"); + } + + public void onGroupDeleted(final String group) { + this.statsTable.get(GROUP_GET_NUMS).delValueBySuffixKey(group, "@"); + this.statsTable.get(GROUP_GET_SIZE).delValueBySuffixKey(group, "@"); + this.statsTable.get(SNDBCK_PUT_NUMS).delValueBySuffixKey(group, "@"); + this.statsTable.get(GROUP_GET_LATENCY).delValueBySuffixKey(group, "@"); + this.momentStatsItemSetFallSize.delValueBySuffixKey(group, "@"); + this.momentStatsItemSetFallTime.delValueBySuffixKey(group, "@"); + } + public void incTopicPutNums(final String topic) { this.statsTable.get(TOPIC_PUT_NUMS).addValue(topic, 1, 1); } diff --git a/store/src/test/java/org/apache/rocketmq/store/DefaultMessageStoreCleanFilesTest.java b/store/src/test/java/org/apache/rocketmq/store/DefaultMessageStoreCleanFilesTest.java index fd510f11a5..7455dfea3c 100644 --- a/store/src/test/java/org/apache/rocketmq/store/DefaultMessageStoreCleanFilesTest.java +++ b/store/src/test/java/org/apache/rocketmq/store/DefaultMessageStoreCleanFilesTest.java @@ -19,8 +19,12 @@ package org.apache.rocketmq.store; import org.apache.rocketmq.common.BrokerConfig; import org.apache.rocketmq.common.UtilAll; +import org.apache.rocketmq.common.message.MessageConst; +import org.apache.rocketmq.common.message.MessageDecoder; import org.apache.rocketmq.store.config.FlushDiskType; import org.apache.rocketmq.store.config.MessageStoreConfig; +import org.apache.rocketmq.store.index.IndexFile; +import org.apache.rocketmq.store.index.IndexService; import org.apache.rocketmq.store.stats.BrokerStatsManager; import org.junit.After; import org.junit.Before; @@ -31,7 +35,9 @@ import java.lang.reflect.Field; import java.net.InetAddress; import java.net.InetSocketAddress; import java.net.SocketAddress; +import java.util.ArrayList; import java.util.Calendar; +import java.util.HashMap; import java.util.Map; import java.util.UUID; @@ -52,6 +58,7 @@ public class DefaultMessageStoreCleanFilesTest { private SocketAddress storeHost; private String topic = "test"; + private String keys = "hello"; private int queueId = 0; private int fileCountCommitLog = 55; // exactly one message per CommitLog file. @@ -65,6 +72,43 @@ public class DefaultMessageStoreCleanFilesTest { bornHost = new InetSocketAddress(InetAddress.getByName("127.0.0.1"), 0); } + @Test + public void testIsSpaceFullFunctionEmpty2Full() throws Exception { + String deleteWhen = "04"; + // the min value of diskMaxUsedSpaceRatio. + int diskMaxUsedSpaceRatio = 1; + // used to set disk-full flag + double diskSpaceCleanForciblyRatio = 0.01D; + initMessageStore(deleteWhen, diskMaxUsedSpaceRatio, diskSpaceCleanForciblyRatio); + // build and put 55 messages, exactly one message per CommitLog file. + buildAndPutMessagesToMessageStore(msgCount); + MappedFileQueue commitLogQueue = getMappedFileQueueCommitLog(); + assertEquals(fileCountCommitLog, commitLogQueue.getMappedFiles().size()); + int fileCountConsumeQueue = getFileCountConsumeQueue(); + MappedFileQueue consumeQueue = getMappedFileQueueConsumeQueue(); + assertEquals(fileCountConsumeQueue, consumeQueue.getMappedFiles().size()); + cleanCommitLogService.isSpaceFull(); + assertEquals(1 << 4, messageStore.getRunningFlags().getFlagBits() & (1 << 4)); + messageStore.shutdown(); + messageStore.destroy(); + + } + + @Test + public void testIsSpaceFullFunctionFull2Empty() throws Exception { + String deleteWhen = "04"; + // the min value of diskMaxUsedSpaceRatio. + int diskMaxUsedSpaceRatio = 1; + //use to reset disk-full flag + double diskSpaceCleanForciblyRatio = 0.999D; + initMessageStore(deleteWhen, diskMaxUsedSpaceRatio, diskSpaceCleanForciblyRatio); + //set disk full + messageStore.getRunningFlags().getAndMakeDiskFull(); + + cleanCommitLogService.isSpaceFull(); + assertEquals(0, messageStore.getRunningFlags().getFlagBits() & (1 << 4)); + } + @Test public void testDeleteExpiredFilesByTimeUp() throws Exception { String deleteWhen = Calendar.getInstance().get(Calendar.HOUR_OF_DAY) + ""; @@ -87,6 +131,9 @@ public class DefaultMessageStoreCleanFilesTest { MappedFileQueue consumeQueue = getMappedFileQueueConsumeQueue(); assertEquals(fileCountConsumeQueue, consumeQueue.getMappedFiles().size()); + int fileCountIndexFile = getFileCountIndexFile(); + assertEquals(fileCountIndexFile, getIndexFileList().size()); + int expireFileCount = 15; expireFiles(commitLogQueue, expireFileCount); @@ -101,6 +148,10 @@ public class DefaultMessageStoreCleanFilesTest { int msgCountPerFile = getMsgCountPerConsumeQueueMappedFile(); int expectDeleteCountConsumeQueue = (int) Math.floor((double) expectDeletedCount / msgCountPerFile); assertEquals(fileCountConsumeQueue - expectDeleteCountConsumeQueue, consumeQueue.getMappedFiles().size()); + + int msgCountPerIndexFile = getMsgCountPerIndexFile(); + int expectDeleteCountIndexFile = (int) Math.floor((double) expectDeletedCount / msgCountPerIndexFile); + assertEquals(fileCountIndexFile - expectDeleteCountIndexFile, getIndexFileList().size()); } } @@ -126,6 +177,9 @@ public class DefaultMessageStoreCleanFilesTest { MappedFileQueue consumeQueue = getMappedFileQueueConsumeQueue(); assertEquals(fileCountConsumeQueue, consumeQueue.getMappedFiles().size()); + int fileCountIndexFile = getFileCountIndexFile(); + assertEquals(fileCountIndexFile, getIndexFileList().size()); + int expireFileCount = 15; expireFiles(commitLogQueue, expireFileCount); @@ -140,6 +194,10 @@ public class DefaultMessageStoreCleanFilesTest { int msgCountPerFile = getMsgCountPerConsumeQueueMappedFile(); int expectDeleteCountConsumeQueue = (int) Math.floor((double) expectDeletedCount / msgCountPerFile); assertEquals(fileCountConsumeQueue - expectDeleteCountConsumeQueue, consumeQueue.getMappedFiles().size()); + + int msgCountPerIndexFile = getMsgCountPerIndexFile(); + int expectDeleteCountIndexFile = (int) Math.floor((double) expectDeletedCount / msgCountPerIndexFile); + assertEquals(fileCountIndexFile - expectDeleteCountIndexFile, getIndexFileList().size()); } } @@ -165,6 +223,9 @@ public class DefaultMessageStoreCleanFilesTest { MappedFileQueue consumeQueue = getMappedFileQueueConsumeQueue(); assertEquals(fileCountConsumeQueue, consumeQueue.getMappedFiles().size()); + int fileCountIndexFile = getFileCountIndexFile(); + assertEquals(fileCountIndexFile, getIndexFileList().size()); + // In this case, there is no need to expire the files. // int expireFileCount = 15; // expireFiles(commitLogQueue, expireFileCount); @@ -181,6 +242,10 @@ public class DefaultMessageStoreCleanFilesTest { int msgCountPerFile = getMsgCountPerConsumeQueueMappedFile(); int expectDeleteCountConsumeQueue = (int) Math.floor((double) (a * 10) / msgCountPerFile); assertEquals(fileCountConsumeQueue - expectDeleteCountConsumeQueue, consumeQueue.getMappedFiles().size()); + + int msgCountPerIndexFile = getMsgCountPerIndexFile(); + int expectDeleteCountIndexFile = (int) Math.floor((double) (a * 10) / msgCountPerIndexFile); + assertEquals(fileCountIndexFile - expectDeleteCountIndexFile, getIndexFileList().size()); } } @@ -208,6 +273,9 @@ public class DefaultMessageStoreCleanFilesTest { MappedFileQueue consumeQueue = getMappedFileQueueConsumeQueue(); assertEquals(fileCountConsumeQueue, consumeQueue.getMappedFiles().size()); + int fileCountIndexFile = getFileCountIndexFile(); + assertEquals(fileCountIndexFile, getIndexFileList().size()); + int expireFileCount = 15; expireFiles(commitLogQueue, expireFileCount); @@ -222,6 +290,10 @@ public class DefaultMessageStoreCleanFilesTest { int msgCountPerFile = getMsgCountPerConsumeQueueMappedFile(); int expectDeleteCountConsumeQueue = (int) Math.floor((double) expectDeletedCount / msgCountPerFile); assertEquals(fileCountConsumeQueue - expectDeleteCountConsumeQueue, consumeQueue.getMappedFiles().size()); + + int msgCountPerIndexFile = getMsgCountPerIndexFile(); + int expectDeleteCountIndexFile = (int) Math.floor((double) (a * 10) / msgCountPerIndexFile); + assertEquals(fileCountIndexFile - expectDeleteCountIndexFile, getIndexFileList().size()); } } @@ -274,32 +346,60 @@ public class DefaultMessageStoreCleanFilesTest { return fileQueue; } + private ArrayList getIndexFileList() throws Exception { + Field indexServiceField = messageStore.getClass().getDeclaredField("indexService"); + indexServiceField.setAccessible(true); + IndexService indexService = (IndexService) indexServiceField.get(messageStore); + + Field indexFileListField = indexService.getClass().getDeclaredField("indexFileList"); + indexFileListField.setAccessible(true); + ArrayList indexFileList = (ArrayList) indexFileListField.get(indexService); + + return indexFileList; + } + private int getFileCountConsumeQueue() { int countPerFile = getMsgCountPerConsumeQueueMappedFile(); double fileCount = (double) msgCount / countPerFile; return (int) Math.ceil(fileCount); } + private int getFileCountIndexFile() { + int countPerFile = getMsgCountPerIndexFile(); + double fileCount = (double) msgCount / countPerFile; + return (int) Math.ceil(fileCount); + } + private int getMsgCountPerConsumeQueueMappedFile() { int size = messageStore.getMessageStoreConfig().getMappedFileSizeConsumeQueue(); return size / CQ_STORE_UNIT_SIZE;// 7 in this case } + private int getMsgCountPerIndexFile() { + // 7 in this case + return messageStore.getMessageStoreConfig().getMaxIndexNum() - 1; + } + private void buildAndPutMessagesToMessageStore(int msgCount) throws Exception { int msgLen = topic.getBytes(CHARSET_UTF8).length + 91; + Map properties = new HashMap<>(4); + properties.put(MessageConst.PROPERTY_KEYS, keys); + String s = MessageDecoder.messageProperties2String(properties); + int propertiesLen = s.getBytes(CHARSET_UTF8).length; int commitLogEndFileMinBlankLength = 4 + 4; - int singleMsgBodyLen = mappedFileSize - msgLen - commitLogEndFileMinBlankLength; + int singleMsgBodyLen = mappedFileSize - msgLen - propertiesLen - commitLogEndFileMinBlankLength; for (int i = 0; i < msgCount; i++) { MessageExtBrokerInner msg = new MessageExtBrokerInner(); msg.setTopic(topic); msg.setBody(new byte[singleMsgBodyLen]); - msg.setKeys(String.valueOf(System.currentTimeMillis())); + msg.setKeys(keys); msg.setQueueId(queueId); msg.setSysFlag(0); msg.setBornTimestamp(System.currentTimeMillis()); msg.setStoreHost(storeHost); msg.setBornHost(bornHost); + msg.setPropertiesString(MessageDecoder.messageProperties2String(msg.getProperties())); PutMessageResult result = messageStore.putMessage(msg); assertTrue(result != null && result.isOk()); } @@ -324,8 +424,8 @@ public class DefaultMessageStoreCleanFilesTest { MessageStoreConfig messageStoreConfig = new MessageStoreConfigForTest(); messageStoreConfig.setMappedFileSizeCommitLog(mappedFileSize); messageStoreConfig.setMappedFileSizeConsumeQueue(mappedFileSize); - messageStoreConfig.setMaxHashSlotNum(10000); - messageStoreConfig.setMaxIndexNum(100 * 100); + messageStoreConfig.setMaxHashSlotNum(100); + messageStoreConfig.setMaxIndexNum(8); messageStoreConfig.setFlushDiskType(FlushDiskType.SYNC_FLUSH); messageStoreConfig.setFlushIntervalConsumeQueue(1); diff --git a/store/src/test/java/stats/BrokerStatsManagerTest.java b/store/src/test/java/stats/BrokerStatsManagerTest.java new file mode 100644 index 0000000000..17020729a6 --- /dev/null +++ b/store/src/test/java/stats/BrokerStatsManagerTest.java @@ -0,0 +1,141 @@ +/* + * 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 stats; + +import org.apache.rocketmq.store.stats.BrokerStatsManager; +import org.junit.After; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; + +import static org.apache.rocketmq.store.stats.BrokerStatsManager.BROKER_PUT_NUMS; +import static org.apache.rocketmq.store.stats.BrokerStatsManager.GROUP_GET_FALL_SIZE; +import static org.apache.rocketmq.store.stats.BrokerStatsManager.GROUP_GET_FALL_TIME; +import static org.apache.rocketmq.store.stats.BrokerStatsManager.GROUP_GET_LATENCY; +import static org.apache.rocketmq.store.stats.BrokerStatsManager.GROUP_GET_NUMS; +import static org.apache.rocketmq.store.stats.BrokerStatsManager.GROUP_GET_SIZE; +import static org.apache.rocketmq.store.stats.BrokerStatsManager.SNDBCK_PUT_NUMS; +import static org.apache.rocketmq.store.stats.BrokerStatsManager.TOPIC_PUT_NUMS; +import static org.apache.rocketmq.store.stats.BrokerStatsManager.TOPIC_PUT_SIZE; +import static org.assertj.core.api.Assertions.assertThat; + +public class BrokerStatsManagerTest { + private BrokerStatsManager brokerStatsManager; + + private String TOPIC = "TOPIC_TEST"; + private String GROUP_NAME = "GROUP_TEST"; + + @Before + public void init() { + brokerStatsManager = new BrokerStatsManager("DefaultCluster"); + brokerStatsManager.start(); + } + + @After + public void destory() { + brokerStatsManager.shutdown(); + } + + @Test + public void testGetStatsItem() { + assertThat(brokerStatsManager.getStatsItem("TEST", "TEST")).isNull(); + } + + @Test + public void testIncTopicPutNums() { + brokerStatsManager.incTopicPutNums(TOPIC); + assertThat(brokerStatsManager.getStatsItem(TOPIC_PUT_NUMS, TOPIC).getTimes().doubleValue()).isEqualTo(1L); + brokerStatsManager.incTopicPutNums(TOPIC, 2, 2); + assertThat(brokerStatsManager.getStatsItem(TOPIC_PUT_NUMS, TOPIC).getValue().doubleValue()).isEqualTo(3L); + } + + @Test + public void testIncTopicPutSize() { + brokerStatsManager.incTopicPutSize(TOPIC, 2); + assertThat(brokerStatsManager.getStatsItem(TOPIC_PUT_SIZE, TOPIC).getValue().doubleValue()).isEqualTo(2L); + } + + @Test + public void testIncGroupGetNums() { + brokerStatsManager.incGroupGetNums(GROUP_NAME, TOPIC, 1); + String statsKey = brokerStatsManager.buildStatsKey(TOPIC, GROUP_NAME); + assertThat(brokerStatsManager.getStatsItem(GROUP_GET_NUMS, statsKey).getValue().doubleValue()).isEqualTo(1L); + } + + @Test + public void testIncGroupGetSize() { + brokerStatsManager.incGroupGetSize(GROUP_NAME, TOPIC, 1); + String statsKey = brokerStatsManager.buildStatsKey(TOPIC, GROUP_NAME); + assertThat(brokerStatsManager.getStatsItem(GROUP_GET_SIZE, statsKey).getValue().doubleValue()).isEqualTo(1L); + } + + @Test + public void testIncGroupGetLatency() { + brokerStatsManager.incGroupGetLatency(GROUP_NAME, TOPIC, 1, 1); + String statsKey = String.format("%d@%s@%s", 1, TOPIC, GROUP_NAME); + assertThat(brokerStatsManager.getStatsItem(GROUP_GET_LATENCY, statsKey).getValue().doubleValue()).isEqualTo(1L); + } + + @Test + public void testIncBrokerPutNums() { + brokerStatsManager.incBrokerPutNums(); + assertThat(brokerStatsManager.getStatsItem(BROKER_PUT_NUMS, "DefaultCluster").getValue().doubleValue()).isEqualTo(1L); + } + + @Test + public void testOnTopicDeleted() { + brokerStatsManager.incTopicPutNums(TOPIC); + brokerStatsManager.incTopicPutSize(TOPIC, 100); + brokerStatsManager.incGroupGetNums(GROUP_NAME, TOPIC, 1); + brokerStatsManager.incGroupGetSize(GROUP_NAME, TOPIC, 100); + brokerStatsManager.incSendBackNums(GROUP_NAME, TOPIC); + brokerStatsManager.incGroupGetLatency(GROUP_NAME, TOPIC, 1, 1); + brokerStatsManager.recordDiskFallBehindTime(GROUP_NAME, TOPIC, 1, 11L); + brokerStatsManager.recordDiskFallBehindSize(GROUP_NAME, TOPIC, 1, 11L); + + brokerStatsManager.onTopicDeleted(TOPIC); + + Assert.assertNull(brokerStatsManager.getStatsItem(TOPIC_PUT_NUMS, TOPIC)); + Assert.assertNull(brokerStatsManager.getStatsItem(TOPIC_PUT_SIZE, TOPIC)); + Assert.assertNull(brokerStatsManager.getStatsItem(GROUP_GET_SIZE, TOPIC + "@" + GROUP_NAME)); + Assert.assertNull(brokerStatsManager.getStatsItem(GROUP_GET_NUMS, TOPIC + "@" + GROUP_NAME)); + Assert.assertNull(brokerStatsManager.getStatsItem(SNDBCK_PUT_NUMS, TOPIC + "@" + GROUP_NAME)); + Assert.assertNull(brokerStatsManager.getStatsItem(GROUP_GET_LATENCY, "1@" + TOPIC + "@" + GROUP_NAME)); + Assert.assertNull(brokerStatsManager.getStatsItem(GROUP_GET_FALL_SIZE, "1@" + TOPIC + "@" + GROUP_NAME)); + Assert.assertNull(brokerStatsManager.getStatsItem(GROUP_GET_FALL_TIME, "1@" + TOPIC + "@" + GROUP_NAME)); + } + + @Test + public void testOnGroupDeleted(){ + brokerStatsManager.incGroupGetNums(GROUP_NAME, TOPIC, 1); + brokerStatsManager.incGroupGetSize(GROUP_NAME, TOPIC, 100); + brokerStatsManager.incSendBackNums(GROUP_NAME, TOPIC); + brokerStatsManager.incGroupGetLatency(GROUP_NAME, TOPIC, 1, 1); + brokerStatsManager.recordDiskFallBehindTime(GROUP_NAME, TOPIC, 1, 11L); + brokerStatsManager.recordDiskFallBehindSize(GROUP_NAME, TOPIC, 1, 11L); + + brokerStatsManager.onGroupDeleted(GROUP_NAME); + + Assert.assertNull(brokerStatsManager.getStatsItem(GROUP_GET_SIZE, TOPIC + "@" + GROUP_NAME)); + Assert.assertNull(brokerStatsManager.getStatsItem(GROUP_GET_NUMS, TOPIC + "@" + GROUP_NAME)); + Assert.assertNull(brokerStatsManager.getStatsItem(SNDBCK_PUT_NUMS, TOPIC + "@" + GROUP_NAME)); + Assert.assertNull(brokerStatsManager.getStatsItem(GROUP_GET_LATENCY, "1@" + TOPIC + "@" + GROUP_NAME)); + Assert.assertNull(brokerStatsManager.getStatsItem(GROUP_GET_FALL_SIZE, "1@" + TOPIC + "@" + GROUP_NAME)); + Assert.assertNull(brokerStatsManager.getStatsItem(GROUP_GET_FALL_TIME, "1@" + TOPIC + "@" + GROUP_NAME)); + } +} diff --git a/test/pom.xml b/test/pom.xml index a6d0fdba58..ce9a8a9a7d 100644 --- a/test/pom.xml +++ b/test/pom.xml @@ -20,7 +20,7 @@ rocketmq-all org.apache.rocketmq - 4.7.0 + 4.7.1-SNAPSHOT 4.0.0 diff --git a/test/src/main/java/org/apache/rocketmq/test/listener/rmq/order/RMQOrderListener.java b/test/src/main/java/org/apache/rocketmq/test/listener/rmq/order/RMQOrderListener.java index 85c61770e0..f483866608 100644 --- a/test/src/main/java/org/apache/rocketmq/test/listener/rmq/order/RMQOrderListener.java +++ b/test/src/main/java/org/apache/rocketmq/test/listener/rmq/order/RMQOrderListener.java @@ -64,8 +64,9 @@ public class RMQOrderListener extends AbstractListener implements MessageListene return String.format("%s_%s", queueId, brokerIp); } + @Override public ConsumeOrderlyStatus consumeMessage(List msgs, - ConsumeOrderlyContext context) { + ConsumeOrderlyContext context) { for (MessageExt msg : msgs) { if (isDebug) { if (listenerName != null && listenerName != "") { diff --git a/tools/pom.xml b/tools/pom.xml index ceb58dd111..c6938afdb7 100644 --- a/tools/pom.xml +++ b/tools/pom.xml @@ -19,7 +19,7 @@ org.apache.rocketmq rocketmq-all - 4.7.0 + 4.7.1-SNAPSHOT 4.0.0