getTopicPublishInfoTable() {
return topicPublishInfoTable;
}
diff --git a/client/src/main/java/org/apache/rocketmq/client/producer/DefaultMQProducer.java b/client/src/main/java/org/apache/rocketmq/client/producer/DefaultMQProducer.java
index b4acf8f1c3..faa79f54c9 100644
--- a/client/src/main/java/org/apache/rocketmq/client/producer/DefaultMQProducer.java
+++ b/client/src/main/java/org/apache/rocketmq/client/producer/DefaultMQProducer.java
@@ -24,6 +24,7 @@ import org.apache.rocketmq.client.QueryResult;
import org.apache.rocketmq.client.Validators;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
+import org.apache.rocketmq.client.exception.RequestTimeoutException;
import org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl;
import org.apache.rocketmq.client.log.ClientLogger;
import org.apache.rocketmq.client.trace.AsyncTraceDispatcher;
@@ -42,38 +43,29 @@ import org.apache.rocketmq.remoting.RPCHook;
import org.apache.rocketmq.remoting.exception.RemotingException;
/**
- * This class is the entry point for applications intending to send messages.
- *
+ * This class is the entry point for applications intending to send messages.
*
* It's fine to tune fields which exposes getter/setter methods, but keep in mind, all of them should work well out of
- * box for most scenarios.
- *
+ * box for most scenarios.
*
* This class aggregates various send methods to deliver messages to brokers. Each of them has pros and
- * cons; you'd better understand strengths and weakness of them before actually coding.
- *
+ * cons; you'd better understand strengths and weakness of them before actually coding.
*
- *
- * Thread Safety: After configuring and starting process, this class can be regarded as thread-safe
- * and used among multiple threads context.
- *
+ * Thread Safety: After configuring and starting process, this class can be regarded as thread-safe
+ * and used among multiple threads context.
*/
public class DefaultMQProducer extends ClientConfig implements MQProducer {
- private final InternalLogger log = ClientLogger.getLog();
-
/**
* Wrapping internal implementations for virtually all methods presented in this class.
*/
protected final transient DefaultMQProducerImpl defaultMQProducerImpl;
-
+ private final InternalLogger log = ClientLogger.getLog();
/**
* Producer group conceptually aggregates all producer instances of exactly same role, which is particularly
- * important when transactional messages are involved.
- *
+ * important when transactional messages are involved.
*
- * For non-transactional messages, it does not matter as long as it's unique per process.
- *
+ * For non-transactional messages, it does not matter as long as it's unique per process.
*
* See {@linktourl http://rocketmq.apache.org/docs/core-concept/} for more discussion.
*/
@@ -100,16 +92,14 @@ public class DefaultMQProducer extends ClientConfig implements MQProducer {
private int compressMsgBodyOverHowmuch = 1024 * 4;
/**
- * Maximum number of retry to perform internally before claiming sending failure in synchronous mode.
- *
+ * Maximum number of retry to perform internally before claiming sending failure in synchronous mode.
*
* This may potentially cause message duplication which is up to application developers to resolve.
*/
private int retryTimesWhenSendFailed = 2;
/**
- * Maximum number of retry to perform internally before claiming sending failure in asynchronous mode.
- *
+ * Maximum number of retry to perform internally before claiming sending failure in asynchronous mode.
*
* This may potentially cause message duplication which is up to application developers to resolve.
*/
@@ -268,14 +258,10 @@ public class DefaultMQProducer extends ClientConfig implements MQProducer {
}
/**
- * Start this producer instance.
- *
+ * Start this producer instance.
*
- *
- * Much internal initializing procedures are carried out to make this instance prepared, thus, it's a must to invoke
- * this method before sending or querying messages.
- *
- *
+ * Much internal initializing procedures are carried out to make this instance prepared, thus, it's a must
+ * to invoke this method before sending or querying messages.
*
* @throws MQClientException if there is any unexpected error.
*/
@@ -316,8 +302,7 @@ public class DefaultMQProducer extends ClientConfig implements MQProducer {
}
/**
- * Send message in synchronous mode. This method returns only when the sending procedure totally completes.
- *
+ * Send message in synchronous mode. This method returns only when the sending procedure totally completes.
*
* Warn: this method has internal retry-mechanism, that is, internal implementation will retry
* {@link #retryTimesWhenSendFailed} times before claiming failure. As a result, multiple messages may potentially
@@ -359,11 +344,9 @@ public class DefaultMQProducer extends ClientConfig implements MQProducer {
}
/**
- * Send message to broker asynchronously.
- *
+ * Send message to broker asynchronously.
*
- * This method returns immediately. On sending completion, sendCallback will be executed.
- *
+ * This method returns immediately. On sending completion, sendCallback will be executed.
*
* Similar to {@link #send(Message)}, internal implementation would potentially retry up to {@link
* #retryTimesWhenSendAsyncFailed} times before claiming sending failure, which may yield message duplication and
@@ -582,6 +565,133 @@ public class DefaultMQProducer extends ClientConfig implements MQProducer {
this.defaultMQProducerImpl.send(msg, selector, arg, sendCallback, timeout);
}
+ /**
+ * Send request message in synchronous mode. This method returns only when the consumer consume the request message and reply a message.
+ *
+ * Warn: this method has internal retry-mechanism, that is, internal implementation will retry
+ * {@link #retryTimesWhenSendFailed} times before claiming failure. As a result, multiple messages may potentially
+ * delivered to broker(s). It's up to the application developers to resolve potential duplication issue.
+ *
+ * @param msg request message to send
+ * @param timeout request timeout
+ * @return reply message
+ * @throws MQClientException if there is any client error.
+ * @throws RemotingException if there is any network-tier error.
+ * @throws MQBrokerException if there is any broker error.
+ * @throws InterruptedException if the thread is interrupted.
+ * @throws RequestTimeoutException if request timeout.
+ */
+ @Override
+ public Message request(final Message msg, final long timeout) throws RequestTimeoutException, MQClientException,
+ RemotingException, MQBrokerException, InterruptedException {
+ msg.setTopic(withNamespace(msg.getTopic()));
+ return this.defaultMQProducerImpl.request(msg, timeout);
+ }
+
+ /**
+ * Request asynchronously.
+ * This method returns immediately. On receiving reply message, requestCallback will be executed.
+ *
+ * Similar to {@link #request(Message, long)}, internal implementation would potentially retry up to {@link
+ * #retryTimesWhenSendAsyncFailed} times before claiming sending failure, which may yield message duplication and
+ * application developers are the one to resolve this potential issue.
+ *
+ * @param msg request message to send
+ * @param requestCallback callback to execute on request completion.
+ * @param timeout request timeout
+ * @throws MQClientException if there is any client error.
+ * @throws RemotingException if there is any network-tier error.
+ * @throws InterruptedException if the thread is interrupted.
+ * @throws MQBrokerException if there is any broker error.
+ */
+ @Override
+ public void request(final Message msg, final RequestCallback requestCallback, final long timeout)
+ throws MQClientException, RemotingException, InterruptedException, MQBrokerException {
+ msg.setTopic(withNamespace(msg.getTopic()));
+ this.defaultMQProducerImpl.request(msg, requestCallback, timeout);
+ }
+
+ /**
+ * Same to {@link #request(Message, long)} with message queue selector specified.
+ *
+ * @param msg request message to send
+ * @param selector message queue selector, through which we get target message queue to deliver message to.
+ * @param arg argument to work along with message queue selector.
+ * @param timeout timeout of request.
+ * @return reply message
+ * @throws MQClientException if there is any client error.
+ * @throws RemotingException if there is any network-tier error.
+ * @throws MQBrokerException if there is any broker error.
+ * @throws InterruptedException if the thread is interrupted.
+ * @throws RequestTimeoutException if request timeout.
+ */
+ @Override
+ public Message request(final Message msg, final MessageQueueSelector selector, final Object arg,
+ final long timeout) throws MQClientException, RemotingException, MQBrokerException,
+ InterruptedException, RequestTimeoutException {
+ msg.setTopic(withNamespace(msg.getTopic()));
+ return this.defaultMQProducerImpl.request(msg, selector, arg, timeout);
+ }
+
+ /**
+ * Same to {@link #request(Message, RequestCallback, long)} with target message selector specified.
+ *
+ * @param msg requst message to send
+ * @param selector message queue selector, through which we get target message queue to deliver message to.
+ * @param arg argument to work along with message queue selector.
+ * @param requestCallback callback to execute on request completion.
+ * @param timeout timeout of request.
+ * @throws MQClientException if there is any client error.
+ * @throws RemotingException if there is any network-tier error.
+ * @throws InterruptedException if the thread is interrupted.
+ * @throws MQBrokerException if there is any broker error.
+ */
+ @Override
+ public void request(final Message msg, final MessageQueueSelector selector, final Object arg,
+ final RequestCallback requestCallback, final long timeout) throws MQClientException, RemotingException,
+ InterruptedException, MQBrokerException {
+ msg.setTopic(withNamespace(msg.getTopic()));
+ this.defaultMQProducerImpl.request(msg, selector, arg, requestCallback, timeout);
+ }
+
+ /**
+ * Same to {@link #request(Message, long)} with target message queue specified in addition.
+ *
+ * @param msg request message to send
+ * @param mq target message queue.
+ * @param timeout request timeout
+ * @throws MQClientException if there is any client error.
+ * @throws RemotingException if there is any network-tier error.
+ * @throws MQBrokerException if there is any broker error.
+ * @throws InterruptedException if the thread is interrupted.
+ * @throws RequestTimeoutException if request timeout.
+ */
+ @Override
+ public Message request(final Message msg, final MessageQueue mq, final long timeout)
+ throws MQClientException, RemotingException, MQBrokerException, InterruptedException, RequestTimeoutException {
+ msg.setTopic(withNamespace(msg.getTopic()));
+ return this.defaultMQProducerImpl.request(msg, mq, timeout);
+ }
+
+ /**
+ * Same to {@link #request(Message, RequestCallback, long)} with target message queue specified.
+ *
+ * @param msg request message to send
+ * @param mq target message queue.
+ * @param requestCallback callback to execute on request completion.
+ * @param timeout timeout of request.
+ * @throws MQClientException if there is any client error.
+ * @throws RemotingException if there is any network-tier error.
+ * @throws InterruptedException if the thread is interrupted.
+ * @throws MQBrokerException if there is any broker error.
+ */
+ @Override
+ public void request(final Message msg, final MessageQueue mq, final RequestCallback requestCallback, long timeout)
+ throws MQClientException, RemotingException, InterruptedException, MQBrokerException {
+ msg.setTopic(withNamespace(msg.getTopic()));
+ this.defaultMQProducerImpl.request(msg, mq, requestCallback, timeout);
+ }
+
/**
* Same to {@link #sendOneway(Message)} with message queue selector specified.
*
diff --git a/client/src/main/java/org/apache/rocketmq/client/producer/MQProducer.java b/client/src/main/java/org/apache/rocketmq/client/producer/MQProducer.java
index 1af6005748..c6cf4c9359 100644
--- a/client/src/main/java/org/apache/rocketmq/client/producer/MQProducer.java
+++ b/client/src/main/java/org/apache/rocketmq/client/producer/MQProducer.java
@@ -21,6 +21,7 @@ import java.util.List;
import org.apache.rocketmq.client.MQAdmin;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
+import org.apache.rocketmq.client.exception.RequestTimeoutException;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.exception.RemotingException;
@@ -98,4 +99,26 @@ public interface MQProducer extends MQAdmin {
SendResult send(final Collection msgs, final MessageQueue mq, final long timeout)
throws MQClientException, RemotingException, MQBrokerException, InterruptedException;
+
+ //for rpc
+ Message request(final Message msg, final long timeout) throws RequestTimeoutException, MQClientException,
+ RemotingException, MQBrokerException, InterruptedException;
+
+ void request(final Message msg, final RequestCallback requestCallback, final long timeout)
+ throws MQClientException, RemotingException, InterruptedException, MQBrokerException;
+
+ Message request(final Message msg, final MessageQueueSelector selector, final Object arg,
+ final long timeout) throws RequestTimeoutException, MQClientException, RemotingException, MQBrokerException,
+ InterruptedException;
+
+ void request(final Message msg, final MessageQueueSelector selector, final Object arg,
+ final RequestCallback requestCallback,
+ final long timeout) throws MQClientException, RemotingException,
+ InterruptedException, MQBrokerException;
+
+ Message request(final Message msg, final MessageQueue mq, final long timeout)
+ throws RequestTimeoutException, MQClientException, RemotingException, MQBrokerException, InterruptedException;
+
+ void request(final Message msg, final MessageQueue mq, final RequestCallback requestCallback, long timeout)
+ throws MQClientException, RemotingException, MQBrokerException, InterruptedException;
}
diff --git a/client/src/main/java/org/apache/rocketmq/client/producer/RequestCallback.java b/client/src/main/java/org/apache/rocketmq/client/producer/RequestCallback.java
new file mode 100644
index 0000000000..3107ba57d6
--- /dev/null
+++ b/client/src/main/java/org/apache/rocketmq/client/producer/RequestCallback.java
@@ -0,0 +1,26 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.client.producer;
+
+import org.apache.rocketmq.common.message.Message;
+
+public interface RequestCallback {
+ void onSuccess(final Message message);
+
+ void onException(final Throwable e);
+}
diff --git a/client/src/main/java/org/apache/rocketmq/client/producer/RequestFutureTable.java b/client/src/main/java/org/apache/rocketmq/client/producer/RequestFutureTable.java
new file mode 100644
index 0000000000..3d4caa208b
--- /dev/null
+++ b/client/src/main/java/org/apache/rocketmq/client/producer/RequestFutureTable.java
@@ -0,0 +1,62 @@
+/*
+ * 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.client.producer;
+
+import java.util.Iterator;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import org.apache.rocketmq.client.common.ClientErrorCode;
+import org.apache.rocketmq.client.exception.RequestTimeoutException;
+import org.apache.rocketmq.client.log.ClientLogger;
+import org.apache.rocketmq.logging.InternalLogger;
+
+public class RequestFutureTable {
+ private static InternalLogger log = ClientLogger.getLog();
+ private static ConcurrentHashMap requestFutureTable = new ConcurrentHashMap();
+
+ public static ConcurrentHashMap getRequestFutureTable() {
+ return requestFutureTable;
+ }
+
+ public static void scanExpiredRequest() {
+ final List rfList = new LinkedList();
+ Iterator> it = requestFutureTable.entrySet().iterator();
+ while (it.hasNext()) {
+ Map.Entry next = it.next();
+ RequestResponseFuture rep = next.getValue();
+
+ if (rep.isTimeout()) {
+ it.remove();
+ rfList.add(rep);
+ log.warn("remove timeout request, CorrelationId={}" + rep.getCorrelationId());
+ }
+ }
+
+ for (RequestResponseFuture rf : rfList) {
+ try {
+ Throwable cause = new RequestTimeoutException(ClientErrorCode.REQUEST_TIMEOUT_EXCEPTION, "request timeout, no reply message.");
+ rf.setCause(cause);
+ rf.executeRequestCallback();
+ } catch (Throwable e) {
+ log.warn("scanResponseTable, operationComplete Exception", e);
+ }
+ }
+ }
+}
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
new file mode 100644
index 0000000000..c54b236b16
--- /dev/null
+++ b/client/src/main/java/org/apache/rocketmq/client/producer/RequestResponseFuture.java
@@ -0,0 +1,121 @@
+/*
+ * 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.client.producer;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import org.apache.rocketmq.common.message.Message;
+
+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;
+
+ public RequestResponseFuture(String correlationId, long timeoutMillis, RequestCallback requestCallback) {
+ this.correlationId = correlationId;
+ this.timeoutMillis = timeoutMillis;
+ this.requestCallback = requestCallback;
+ }
+
+ public void executeRequestCallback() {
+ if (requestCallback != null) {
+ if (sendRequestOk && cause == null) {
+ requestCallback.onSuccess(responseMsg);
+ } else {
+ requestCallback.onException(cause);
+ }
+ }
+ }
+
+ public boolean isTimeout() {
+ long diff = System.currentTimeMillis() - this.beginTimestamp;
+ return diff > this.timeoutMillis;
+ }
+
+ public Message waitResponseMessage(final long timeout) throws InterruptedException {
+ this.countDownLatch.await(timeout, TimeUnit.MILLISECONDS);
+ return this.responseMsg;
+ }
+
+ public void putResponseMessage(final Message responseMsg) {
+ this.responseMsg = responseMsg;
+ this.countDownLatch.countDown();
+ }
+
+ public String getCorrelationId() {
+ return correlationId;
+ }
+
+ public long getTimeoutMillis() {
+ return timeoutMillis;
+ }
+
+ public void setTimeoutMillis(long timeoutMillis) {
+ this.timeoutMillis = timeoutMillis;
+ }
+
+ public RequestCallback getRequestCallback() {
+ return requestCallback;
+ }
+
+ public long getBeginTimestamp() {
+ return beginTimestamp;
+ }
+
+ public CountDownLatch getCountDownLatch() {
+ return countDownLatch;
+ }
+
+ public void setCountDownLatch(CountDownLatch countDownLatch) {
+ this.countDownLatch = countDownLatch;
+ }
+
+ public Message getResponseMsg() {
+ return responseMsg;
+ }
+
+ public void setResponseMsg(Message responseMsg) {
+ this.responseMsg = responseMsg;
+ }
+
+ public boolean isSendRequestOk() {
+ return sendRequestOk;
+ }
+
+ public void setSendReqeustOk(boolean sendReqeustOk) {
+ this.sendRequestOk = sendReqeustOk;
+ }
+
+ public Message getRequestMsg() {
+ return requestMsg;
+ }
+
+ public Throwable getCause() {
+ return cause;
+ }
+
+ public void setCause(Throwable cause) {
+ this.cause = cause;
+ }
+}
diff --git a/client/src/main/java/org/apache/rocketmq/client/utils/MessageUtil.java b/client/src/main/java/org/apache/rocketmq/client/utils/MessageUtil.java
new file mode 100644
index 0000000000..416ba44da3
--- /dev/null
+++ b/client/src/main/java/org/apache/rocketmq/client/utils/MessageUtil.java
@@ -0,0 +1,55 @@
+/*
+ * 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.client.utils;
+
+import org.apache.rocketmq.client.common.ClientErrorCode;
+import org.apache.rocketmq.client.exception.MQClientException;
+import org.apache.rocketmq.common.MixAll;
+import org.apache.rocketmq.common.message.Message;
+import org.apache.rocketmq.common.message.MessageAccessor;
+import org.apache.rocketmq.common.message.MessageConst;
+
+public class MessageUtil {
+ public static Message createReplyMessage(final Message requestMessage, final byte[] body) throws MQClientException {
+ if (requestMessage != null) {
+ Message replyMessage = new Message();
+ String cluster = requestMessage.getProperty(MessageConst.PROPERTY_CLUSTER);
+ String replyTo = requestMessage.getProperty(MessageConst.PROPERTY_MESSAGE_REPLY_TO_CLIENT);
+ String correlationId = requestMessage.getProperty(MessageConst.PROPERTY_CORRELATION_ID);
+ String ttl = requestMessage.getProperty(MessageConst.PROPERTY_MESSAGE_TTL);
+ replyMessage.setBody(body);
+ if (cluster != null) {
+ String replyTopic = MixAll.getReplyTopic(cluster);
+ replyMessage.setTopic(replyTopic);
+ MessageAccessor.putProperty(replyMessage, MessageConst.PROPERTY_MESSAGE_TYPE, MixAll.REPLY_MESSAGE_FLAG);
+ MessageAccessor.putProperty(replyMessage, MessageConst.PROPERTY_CORRELATION_ID, correlationId);
+ MessageAccessor.putProperty(replyMessage, MessageConst.PROPERTY_MESSAGE_REPLY_TO_CLIENT, replyTo);
+ MessageAccessor.putProperty(replyMessage, MessageConst.PROPERTY_MESSAGE_TTL, ttl);
+
+ return replyMessage;
+ } else {
+ throw new MQClientException(ClientErrorCode.CREATE_REPLY_MESSAGE_EXCEPTION, "create reply message fail, requestMessage error, property[" + MessageConst.PROPERTY_CLUSTER + "] is null.");
+ }
+ }
+ throw new MQClientException(ClientErrorCode.CREATE_REPLY_MESSAGE_EXCEPTION, "create reply message fail, requestMessage cannot be null.");
+ }
+
+ public static String getReplyToClient(final Message msg) {
+ return msg.getProperty(MessageConst.PROPERTY_MESSAGE_REPLY_TO_CLIENT);
+ }
+}
diff --git a/client/src/test/java/org/apache/rocketmq/client/consumer/DefaultLitePullConsumerTest.java b/client/src/test/java/org/apache/rocketmq/client/consumer/DefaultLitePullConsumerTest.java
index b55f8a0a1c..d2cb0571ed 100644
--- a/client/src/test/java/org/apache/rocketmq/client/consumer/DefaultLitePullConsumerTest.java
+++ b/client/src/test/java/org/apache/rocketmq/client/consumer/DefaultLitePullConsumerTest.java
@@ -189,6 +189,36 @@ public class DefaultLitePullConsumerTest {
litePullConsumer.shutdown();
}
+ @Test
+ public void testSeek_SeekToBegin() throws Exception {
+ DefaultLitePullConsumer litePullConsumer = createStartLitePullConsumer();
+ when(mQAdminImpl.minOffset(any(MessageQueue.class))).thenReturn(0L);
+ when(mQAdminImpl.maxOffset(any(MessageQueue.class))).thenReturn(500L);
+ MessageQueue messageQueue = createMessageQueue();
+ litePullConsumer.assign(Collections.singletonList(messageQueue));
+ litePullConsumer.seekToBegin(messageQueue);
+ Field field = DefaultLitePullConsumerImpl.class.getDeclaredField("assignedMessageQueue");
+ field.setAccessible(true);
+ AssignedMessageQueue assignedMessageQueue = (AssignedMessageQueue) field.get(litePullConsumerImpl);
+ assertEquals(assignedMessageQueue.getSeekOffset(messageQueue), 0L);
+ litePullConsumer.shutdown();
+ }
+
+ @Test
+ public void testSeek_SeekToEnd() throws Exception {
+ DefaultLitePullConsumer litePullConsumer = createStartLitePullConsumer();
+ when(mQAdminImpl.minOffset(any(MessageQueue.class))).thenReturn(0L);
+ when(mQAdminImpl.maxOffset(any(MessageQueue.class))).thenReturn(500L);
+ MessageQueue messageQueue = createMessageQueue();
+ litePullConsumer.assign(Collections.singletonList(messageQueue));
+ litePullConsumer.seekToEnd(messageQueue);
+ Field field = DefaultLitePullConsumerImpl.class.getDeclaredField("assignedMessageQueue");
+ field.setAccessible(true);
+ AssignedMessageQueue assignedMessageQueue = (AssignedMessageQueue) field.get(litePullConsumerImpl);
+ assertEquals(assignedMessageQueue.getSeekOffset(messageQueue), 500L);
+ litePullConsumer.shutdown();
+ }
+
@Test
public void testSeek_SeekOffsetIllegal() throws Exception {
DefaultLitePullConsumer litePullConsumer = createStartLitePullConsumer();
diff --git a/client/src/test/java/org/apache/rocketmq/client/impl/MQClientAPIImplTest.java b/client/src/test/java/org/apache/rocketmq/client/impl/MQClientAPIImplTest.java
index 84af632355..3f00d9e403 100644
--- a/client/src/test/java/org/apache/rocketmq/client/impl/MQClientAPIImplTest.java
+++ b/client/src/test/java/org/apache/rocketmq/client/impl/MQClientAPIImplTest.java
@@ -42,6 +42,7 @@ import org.apache.rocketmq.remoting.protocol.RemotingCommand;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
+import org.mockito.Matchers;
import org.mockito.Mock;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.junit.MockitoJUnitRunner;
@@ -164,7 +165,7 @@ public class MQClientAPIImplTest {
public Object answer(InvocationOnMock mock) throws Throwable {
InvokeCallback callback = mock.getArgument(3);
RemotingCommand request = mock.getArgument(1);
- ResponseFuture responseFuture = new ResponseFuture(null,request.getOpaque(), 3 * 1000, null, null);
+ ResponseFuture responseFuture = new ResponseFuture(null, request.getOpaque(), 3 * 1000, null, null);
responseFuture.setResponseCommand(createSuccessResponse(request));
callback.operationComplete(responseFuture);
return null;
@@ -289,6 +290,7 @@ public class MQClientAPIImplTest {
assertThat(ex.getErrorMessage()).isEqualTo("corresponding to accessConfig has been deleted failed");
}
}
+
@Test
public void testResumeCheckHalfMessage_WithException() throws RemotingException, InterruptedException, MQBrokerException, MQClientException {
doAnswer(new Answer() {
@@ -322,6 +324,38 @@ public class MQClientAPIImplTest {
assertThat(result).isEqualTo(true);
}
+ @Test
+ public void testSendMessageTypeofReply() throws Exception {
+ doAnswer(new Answer() {
+ @Override
+ public Object answer(InvocationOnMock mock) throws Throwable {
+ InvokeCallback callback = mock.getArgument(3);
+ RemotingCommand request = mock.getArgument(1);
+ ResponseFuture responseFuture = new ResponseFuture(null, request.getOpaque(), 3 * 1000, null, null);
+ responseFuture.setResponseCommand(createSuccessResponse(request));
+ callback.operationComplete(responseFuture);
+ return null;
+ }
+ }).when(remotingClient).invokeAsync(Matchers.anyString(), Matchers.any(RemotingCommand.class), Matchers.anyLong(), Matchers.any(InvokeCallback.class));
+ SendMessageContext sendMessageContext = new SendMessageContext();
+ sendMessageContext.setProducer(new DefaultMQProducerImpl(new DefaultMQProducer()));
+ msg.getProperties().put("MSG_TYPE", "reply");
+ mqClientAPI.sendMessage(brokerAddr, brokerName, msg, new SendMessageRequestHeader(), 3 * 1000, CommunicationMode.ASYNC,
+ new SendCallback() {
+ @Override
+ public void onSuccess(SendResult sendResult) {
+ assertThat(sendResult.getSendStatus()).isEqualTo(SendStatus.SEND_OK);
+ assertThat(sendResult.getOffsetMsgId()).isEqualTo("123");
+ assertThat(sendResult.getQueueOffset()).isEqualTo(123L);
+ assertThat(sendResult.getMessageQueue().getQueueId()).isEqualTo(1);
+ }
+
+ @Override
+ public void onException(Throwable e) {
+ }
+ }, null, null, 0, sendMessageContext, defaultMQProducerImpl);
+ }
+
private RemotingCommand createResumeSuccessResponse(RemotingCommand request) {
RemotingCommand response = RemotingCommand.createResponseCommand(null);
response.setCode(ResponseCode.SUCCESS);
diff --git a/client/src/test/java/org/apache/rocketmq/client/impl/consumer/ConsumeMessageOrderlyServiceTest.java b/client/src/test/java/org/apache/rocketmq/client/impl/consumer/ConsumeMessageOrderlyServiceTest.java
new file mode 100644
index 0000000000..4cfa011434
--- /dev/null
+++ b/client/src/test/java/org/apache/rocketmq/client/impl/consumer/ConsumeMessageOrderlyServiceTest.java
@@ -0,0 +1,86 @@
+/*
+ * 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.client.impl.consumer;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
+import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
+import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
+import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
+import org.apache.rocketmq.common.message.MessageExt;
+import org.apache.rocketmq.common.protocol.body.CMResult;
+import org.junit.Before;
+import org.junit.Test;
+
+import static org.junit.Assert.assertTrue;
+
+public class ConsumeMessageOrderlyServiceTest {
+ private String consumerGroup;
+ private String topic = "FooBar";
+ private String brokerName = "BrokerA";
+ private DefaultMQPushConsumer pushConsumer;
+
+ @Before
+ public void init() throws Exception {
+ consumerGroup = "FooBarGroup" + System.currentTimeMillis();
+ pushConsumer = new DefaultMQPushConsumer(consumerGroup);
+ }
+
+ @Test
+ public void testConsumeMessageDirectly_WithNoException() {
+ Map map = new HashMap();
+ map.put(ConsumeOrderlyStatus.SUCCESS, CMResult.CR_SUCCESS);
+ map.put(ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT, CMResult.CR_LATER);
+ map.put(ConsumeOrderlyStatus.COMMIT, CMResult.CR_COMMIT);
+ map.put(ConsumeOrderlyStatus.ROLLBACK, CMResult.CR_ROLLBACK);
+ map.put(null, CMResult.CR_RETURN_NULL);
+
+ for (ConsumeOrderlyStatus consumeOrderlyStatus : map.keySet()) {
+ final ConsumeOrderlyStatus status = consumeOrderlyStatus;
+ MessageListenerOrderly listenerOrderly = new MessageListenerOrderly() {
+ @Override
+ public ConsumeOrderlyStatus consumeMessage(List msgs, ConsumeOrderlyContext context) {
+ return status;
+ }
+ };
+
+ ConsumeMessageOrderlyService consumeMessageOrderlyService = new ConsumeMessageOrderlyService(pushConsumer.getDefaultMQPushConsumerImpl(), listenerOrderly);
+ MessageExt msg = new MessageExt();
+ msg.setTopic(topic);
+ assertTrue(consumeMessageOrderlyService.consumeMessageDirectly(msg, brokerName).getConsumeResult().equals(map.get(consumeOrderlyStatus)));
+ }
+
+ }
+
+ @Test
+ public void testConsumeMessageDirectly_WithException() {
+ MessageListenerOrderly listenerOrderly = new MessageListenerOrderly() {
+ @Override
+ public ConsumeOrderlyStatus consumeMessage(List msgs, ConsumeOrderlyContext context) {
+ throw new RuntimeException();
+ }
+ };
+
+ ConsumeMessageOrderlyService consumeMessageOrderlyService = new ConsumeMessageOrderlyService(pushConsumer.getDefaultMQPushConsumerImpl(), listenerOrderly);
+ MessageExt msg = new MessageExt();
+ msg.setTopic(topic);
+ assertTrue(consumeMessageOrderlyService.consumeMessageDirectly(msg, brokerName).getConsumeResult().equals(CMResult.CR_THROW_EXCEPTION));
+ }
+
+}
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 b9be49c033..d572a2320e 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
@@ -21,15 +21,18 @@ import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
-
import org.apache.rocketmq.client.ClientConfig;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
+import org.apache.rocketmq.client.exception.RequestTimeoutException;
import org.apache.rocketmq.client.hook.SendMessageContext;
import org.apache.rocketmq.client.hook.SendMessageHook;
import org.apache.rocketmq.client.impl.CommunicationMode;
@@ -45,6 +48,7 @@ import org.apache.rocketmq.common.protocol.route.BrokerData;
import org.apache.rocketmq.common.protocol.route.QueueData;
import org.apache.rocketmq.common.protocol.route.TopicRouteData;
import org.apache.rocketmq.remoting.exception.RemotingException;
+import org.apache.rocketmq.remoting.exception.RemotingSendRequestException;
import org.apache.rocketmq.remoting.netty.NettyRemotingClient;
import org.junit.After;
import org.junit.Before;
@@ -52,7 +56,9 @@ import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.Mock;
import org.mockito.Spy;
+import org.mockito.invocation.InvocationOnMock;
import org.mockito.junit.MockitoJUnitRunner;
+import org.mockito.stubbing.Answer;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Fail.failBecauseExceptionWasNotThrown;
@@ -168,6 +174,7 @@ public class DefaultMQProducerTest {
@Test
public void testSendMessageAsync_Success() throws RemotingException, InterruptedException, MQBrokerException, MQClientException {
final CountDownLatch countDownLatch = new CountDownLatch(1);
+ when(mQClientAPIImpl.getTopicRouteInfoFromNameServer(anyString(), anyLong())).thenReturn(createTopicRoute());
producer.send(message, new SendCallback() {
@Override
public void onSuccess(SendResult sendResult) {
@@ -190,9 +197,11 @@ public class DefaultMQProducerTest {
final AtomicInteger cc = new AtomicInteger(0);
final CountDownLatch countDownLatch = new CountDownLatch(6);
+ when(mQClientAPIImpl.getTopicRouteInfoFromNameServer(anyString(), anyLong())).thenReturn(createTopicRoute());
SendCallback sendCallback = new SendCallback() {
@Override
public void onSuccess(SendResult sendResult) {
+ countDownLatch.countDown();
}
@Override
@@ -213,20 +222,21 @@ public class DefaultMQProducerTest {
message.setTopic("test");
message.setBody("hello world".getBytes());
producer.send(new Message(), sendCallback);
- producer.send(message, sendCallback, 1000);
producer.send(message, new MessageQueue(), sendCallback);
producer.send(new Message(), new MessageQueue(), sendCallback, 1000);
producer.send(new Message(), messageQueueSelector, null, sendCallback);
producer.send(message, messageQueueSelector, null, sendCallback, 1000);
+ //this message is send success
+ producer.send(message, sendCallback, 1000);
countDownLatch.await(3000L, TimeUnit.MILLISECONDS);
- assertThat(cc.get()).isEqualTo(6);
+ assertThat(cc.get()).isEqualTo(5);
}
@Test
public void testSendMessageAsync_BodyCompressed() throws RemotingException, InterruptedException, MQBrokerException, MQClientException {
-
final CountDownLatch countDownLatch = new CountDownLatch(1);
+ when(mQClientAPIImpl.getTopicRouteInfoFromNameServer(anyString(), anyLong())).thenReturn(createTopicRoute());
producer.send(bigMessage, new SendCallback() {
@Override
public void onSuccess(SendResult sendResult) {
@@ -238,7 +248,6 @@ public class DefaultMQProducerTest {
@Override
public void onException(Throwable e) {
- countDownLatch.countDown();
}
});
countDownLatch.await(3000L, TimeUnit.MILLISECONDS);
@@ -312,6 +321,101 @@ public class DefaultMQProducerTest {
assertThat(remotingClient.getCallbackExecutor()).isEqualTo(customized);
}
+ @Test
+ public void testRequestMessage() throws RemotingException, RequestTimeoutException, MQClientException, InterruptedException, MQBrokerException {
+ when(mQClientAPIImpl.getTopicRouteInfoFromNameServer(anyString(), anyLong())).thenReturn(createTopicRoute());
+ final AtomicBoolean finish = new AtomicBoolean(false);
+ new Thread(new Runnable() {
+ @Override public void run() {
+ ConcurrentHashMap responseMap = RequestFutureTable.getRequestFutureTable();
+ assertThat(responseMap).isNotNull();
+ while (!finish.get()) {
+ try {
+ Thread.sleep(10);
+ } catch (InterruptedException e) {
+ }
+ for (Map.Entry entry : responseMap.entrySet()) {
+ RequestResponseFuture future = entry.getValue();
+ future.putResponseMessage(message);
+ }
+ }
+ }
+ }).start();
+ Message result = producer.request(message, 3 * 1000L);
+ finish.getAndSet(true);
+ assertThat(result.getTopic()).isEqualTo("FooBar");
+ assertThat(result.getBody()).isEqualTo(new byte[] {'a'});
+ }
+
+ @Test(expected = RequestTimeoutException.class)
+ public void testRequestMessage_RequestTimeoutException() throws RemotingException, RequestTimeoutException, MQClientException, InterruptedException, MQBrokerException {
+ when(mQClientAPIImpl.getTopicRouteInfoFromNameServer(anyString(), anyLong())).thenReturn(createTopicRoute());
+ Message result = producer.request(message, 3 * 1000L);
+ }
+
+ @Test
+ public void testAsyncRequest_OnSuccess() throws Exception {
+ when(mQClientAPIImpl.getTopicRouteInfoFromNameServer(anyString(), anyLong())).thenReturn(createTopicRoute());
+ final CountDownLatch countDownLatch = new CountDownLatch(1);
+ RequestCallback requestCallback = new RequestCallback() {
+ @Override public void onSuccess(Message message) {
+ assertThat(message.getTopic()).isEqualTo("FooBar");
+ assertThat(message.getBody()).isEqualTo(new byte[] {'a'});
+ assertThat(message.getFlag()).isEqualTo(1);
+ countDownLatch.countDown();
+ }
+
+ @Override public void onException(Throwable e) {
+ }
+ };
+ producer.request(message, requestCallback, 3 * 1000L);
+ ConcurrentHashMap responseMap = RequestFutureTable.getRequestFutureTable();
+ assertThat(responseMap).isNotNull();
+ for (Map.Entry entry : responseMap.entrySet()) {
+ RequestResponseFuture future = entry.getValue();
+ future.setSendReqeustOk(true);
+ message.setFlag(1);
+ future.getRequestCallback().onSuccess(message);
+ }
+ countDownLatch.await(3000L, TimeUnit.MILLISECONDS);
+ }
+
+ @Test
+ public void testAsyncRequest_OnException() throws Exception {
+ final AtomicInteger cc = new AtomicInteger(0);
+ final CountDownLatch countDownLatch = new CountDownLatch(1);
+ RequestCallback requestCallback = new RequestCallback() {
+ @Override public void onSuccess(Message message) {
+
+ }
+
+ @Override public void onException(Throwable e) {
+ cc.incrementAndGet();
+ countDownLatch.countDown();
+ }
+ };
+ MessageQueueSelector messageQueueSelector = new MessageQueueSelector() {
+ @Override
+ public MessageQueue select(List mqs, Message msg, Object arg) {
+ return null;
+ }
+ };
+
+ try {
+ producer.request(message, requestCallback, 3 * 1000L);
+ failBecauseExceptionWasNotThrown(Exception.class);
+ } catch (Exception e) {
+ ConcurrentHashMap responseMap = RequestFutureTable.getRequestFutureTable();
+ assertThat(responseMap).isNotNull();
+ for (Map.Entry entry : responseMap.entrySet()) {
+ RequestResponseFuture future = entry.getValue();
+ future.getRequestCallback().onException(e);
+ }
+ }
+ countDownLatch.await(3000L, TimeUnit.MILLISECONDS);
+ assertThat(cc.get()).isEqualTo(1);
+ }
+
public static TopicRouteData createTopicRoute() {
TopicRouteData topicRouteData = new TopicRouteData();
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
new file mode 100644
index 0000000000..90e4623e9b
--- /dev/null
+++ b/client/src/test/java/org/apache/rocketmq/client/producer/RequestResponseFutureTest.java
@@ -0,0 +1,45 @@
+/*
+ * 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.client.producer;
+
+import java.util.UUID;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.rocketmq.common.message.Message;
+import org.junit.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+public class RequestResponseFutureTest {
+
+ @Test
+ public void testExecuteRequestCallback() throws Exception {
+ final AtomicInteger cc = new AtomicInteger(0);
+ RequestResponseFuture future = new RequestResponseFuture(UUID.randomUUID().toString(), 3 * 1000L, new RequestCallback() {
+ @Override public void onSuccess(Message message) {
+ cc.incrementAndGet();
+ }
+
+ @Override public void onException(Throwable e) {
+ }
+ });
+ future.setSendReqeustOk(true);
+ future.executeRequestCallback();
+ assertThat(cc.get()).isEqualTo(1);
+ }
+
+}
diff --git a/client/src/test/java/org/apache/rocketmq/client/utils/MessageUtilsTest.java b/client/src/test/java/org/apache/rocketmq/client/utils/MessageUtilsTest.java
new file mode 100644
index 0000000000..803e596fc8
--- /dev/null
+++ b/client/src/test/java/org/apache/rocketmq/client/utils/MessageUtilsTest.java
@@ -0,0 +1,80 @@
+/*
+ * 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.client.utils;
+
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.rocketmq.client.exception.MQClientException;
+import org.apache.rocketmq.common.MixAll;
+import org.apache.rocketmq.common.message.Message;
+import org.apache.rocketmq.common.message.MessageAccessor;
+import org.apache.rocketmq.common.message.MessageConst;
+import org.junit.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Fail.failBecauseExceptionWasNotThrown;
+
+public class MessageUtilsTest {
+
+ @Test
+ public void testCreateReplyMessage() throws MQClientException {
+ Message msg = MessageUtil.createReplyMessage(createReplyMessage("clusterName"), new byte[] {'a'});
+ assertThat(msg.getTopic()).isEqualTo("clusterName" + "_" + MixAll.REPLY_TOPIC_POSTFIX);
+ assertThat(msg.getProperty(MessageConst.PROPERTY_MESSAGE_REPLY_TO_CLIENT)).isEqualTo("127.0.0.1");
+ assertThat(msg.getProperty(MessageConst.PROPERTY_MESSAGE_TTL)).isEqualTo("3000");
+ }
+
+ @Test
+ public void testCreateReplyMessage_Exception() throws MQClientException {
+ try {
+ Message msg = MessageUtil.createReplyMessage(createReplyMessage(null), new byte[] {'a'});
+ failBecauseExceptionWasNotThrown(MQClientException.class);
+ } catch (MQClientException e) {
+ assertThat(e).hasMessageContaining("create reply message fail, requestMessage error, property[" + MessageConst.PROPERTY_CLUSTER + "] is null.");
+ }
+ }
+
+ @Test
+ public void testCreateReplyMessage_reqMsgIsNull() throws MQClientException {
+ try {
+ Message msg = MessageUtil.createReplyMessage(null, new byte[] {'a'});
+ failBecauseExceptionWasNotThrown(MQClientException.class);
+ } catch (MQClientException e) {
+ assertThat(e).hasMessageContaining("create reply message fail, requestMessage cannot be null.");
+ }
+ }
+
+ @Test
+ public void testGetReplyToClient() throws MQClientException {
+ Message msg = createReplyMessage("clusterName");
+ String replyToClient = MessageUtil.getReplyToClient(msg);
+ assertThat(replyToClient).isNotNull();
+ assertThat(replyToClient).isEqualTo("127.0.0.1");
+ }
+
+ private Message createReplyMessage(String clusterName) {
+ Message requestMessage = new Message();
+ Map map = new HashMap();
+ map.put(MessageConst.PROPERTY_MESSAGE_REPLY_TO_CLIENT, "127.0.0.1");
+ map.put(MessageConst.PROPERTY_CLUSTER, clusterName);
+ map.put(MessageConst.PROPERTY_MESSAGE_TTL, "3000");
+ MessageAccessor.setProperties(requestMessage, map);
+ return requestMessage;
+ }
+
+}
diff --git a/common/pom.xml b/common/pom.xml
index 11df833bff..b5a566dc18 100644
--- a/common/pom.xml
+++ b/common/pom.xml
@@ -19,7 +19,7 @@
org.apache.rocketmq
rocketmq-all
- 4.6.0-SNAPSHOT
+ 4.6.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 1c3f37d00d..a7568f0a20 100644
--- a/common/src/main/java/org/apache/rocketmq/common/BrokerConfig.java
+++ b/common/src/main/java/org/apache/rocketmq/common/BrokerConfig.java
@@ -61,6 +61,7 @@ public class BrokerConfig {
*/
private int sendMessageThreadPoolNums = 1; //16 + Runtime.getRuntime().availableProcessors() * 4;
private int pullMessageThreadPoolNums = 16 + Runtime.getRuntime().availableProcessors() * 2;
+ private int processReplyMessageThreadPoolNums = 16 + Runtime.getRuntime().availableProcessors() * 2;
private int queryMessageThreadPoolNums = 8 + Runtime.getRuntime().availableProcessors();
private int adminBrokerThreadPoolNums = 16;
@@ -83,6 +84,7 @@ public class BrokerConfig {
private boolean fetchNamesrvAddrByAddressServer = false;
private int sendThreadPoolQueueCapacity = 10000;
private int pullThreadPoolQueueCapacity = 100000;
+ private int replyThreadPoolQueueCapacity = 10000;
private int queryThreadPoolQueueCapacity = 20000;
private int clientManagerThreadPoolQueueCapacity = 1000000;
private int consumerManagerThreadPoolQueueCapacity = 1000000;
@@ -180,6 +182,8 @@ public class BrokerConfig {
@ImportantField
private boolean aclEnable = false;
+ private boolean storeReplyMessageEnable = true;
+
public static String localHostName() {
try {
return InetAddress.getLocalHost().getHostName();
@@ -374,6 +378,14 @@ public class BrokerConfig {
this.pullMessageThreadPoolNums = pullMessageThreadPoolNums;
}
+ public int getProcessReplyMessageThreadPoolNums() {
+ return processReplyMessageThreadPoolNums;
+ }
+
+ public void setProcessReplyMessageThreadPoolNums(int processReplyMessageThreadPoolNums) {
+ this.processReplyMessageThreadPoolNums = processReplyMessageThreadPoolNums;
+ }
+
public int getQueryMessageThreadPoolNums() {
return queryMessageThreadPoolNums;
}
@@ -470,6 +482,14 @@ public class BrokerConfig {
this.pullThreadPoolQueueCapacity = pullThreadPoolQueueCapacity;
}
+ public int getReplyThreadPoolQueueCapacity() {
+ return replyThreadPoolQueueCapacity;
+ }
+
+ public void setReplyThreadPoolQueueCapacity(int replyThreadPoolQueueCapacity) {
+ this.replyThreadPoolQueueCapacity = replyThreadPoolQueueCapacity;
+ }
+
public int getQueryThreadPoolQueueCapacity() {
return queryThreadPoolQueueCapacity;
}
@@ -749,7 +769,7 @@ public class BrokerConfig {
public void setMsgTraceTopicName(String msgTraceTopicName) {
this.msgTraceTopicName = msgTraceTopicName;
}
-
+
public boolean isTraceTopicEnable() {
return traceTopicEnable;
}
@@ -765,4 +785,12 @@ public class BrokerConfig {
public void setAclEnable(boolean aclEnable) {
this.aclEnable = aclEnable;
}
+
+ public boolean isStoreReplyMessageEnable() {
+ return storeReplyMessageEnable;
+ }
+
+ public void setStoreReplyMessageEnable(boolean storeReplyMessageEnable) {
+ this.storeReplyMessageEnable = storeReplyMessageEnable;
+ }
}
diff --git a/common/src/main/java/org/apache/rocketmq/common/MQVersion.java b/common/src/main/java/org/apache/rocketmq/common/MQVersion.java
index 47cfb6ebd0..019afd8c7f 100644
--- a/common/src/main/java/org/apache/rocketmq/common/MQVersion.java
+++ b/common/src/main/java/org/apache/rocketmq/common/MQVersion.java
@@ -18,7 +18,7 @@ package org.apache.rocketmq.common;
public class MQVersion {
- public static final int CURRENT_VERSION = Version.V4_5_2.ordinal();
+ public static final int CURRENT_VERSION = Version.V4_6_0.ordinal();
public static String getVersionDesc(int value) {
int length = Version.values().length;
diff --git a/common/src/main/java/org/apache/rocketmq/common/MixAll.java b/common/src/main/java/org/apache/rocketmq/common/MixAll.java
index bba8b36ff5..e9a67bb5c0 100644
--- a/common/src/main/java/org/apache/rocketmq/common/MixAll.java
+++ b/common/src/main/java/org/apache/rocketmq/common/MixAll.java
@@ -45,8 +45,6 @@ import org.apache.rocketmq.logging.InternalLogger;
import org.apache.rocketmq.logging.InternalLoggerFactory;
public class MixAll {
- private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.COMMON_LOGGER_NAME);
-
public static final String ROCKETMQ_HOME_ENV = "ROCKETMQ_HOME";
public static final String ROCKETMQ_HOME_PROPERTY = "rocketmq.home.dir";
public static final String NAMESRV_ADDR_ENV = "NAMESRV_ADDR";
@@ -74,27 +72,26 @@ public class MixAll {
public static final String CID_ONSAPI_OWNER_GROUP = "CID_ONSAPI_OWNER";
public static final String CID_ONSAPI_PULL_GROUP = "CID_ONSAPI_PULL";
public static final String CID_RMQ_SYS_PREFIX = "CID_RMQ_SYS_";
-
public static final List LOCAL_INET_ADDRESS = getLocalInetAddress();
public static final String LOCALHOST = localhost();
public static final String DEFAULT_CHARSET = "UTF-8";
public static final long MASTER_ID = 0L;
public static final long CURRENT_JVM_PID = getPID();
-
public static final String RETRY_GROUP_TOPIC_PREFIX = "%RETRY%";
-
public static final String DLQ_GROUP_TOPIC_PREFIX = "%DLQ%";
+ public static final String REPLY_TOPIC_POSTFIX = "REPLY_TOPIC";
public static final String SYSTEM_TOPIC_PREFIX = "rmq_sys_";
public static final String UNIQUE_MSG_QUERY_FLAG = "_UNIQUE_KEY_QUERY";
public static final String DEFAULT_TRACE_REGION_ID = "DefaultRegion";
public static final String CONSUME_CONTEXT_TYPE = "ConsumeContextType";
-
public static final String RMQ_SYS_TRANS_HALF_TOPIC = "RMQ_SYS_TRANS_HALF_TOPIC";
public static final String RMQ_SYS_TRACE_TOPIC = "RMQ_SYS_TRACE_TOPIC";
public static final String RMQ_SYS_TRANS_OP_HALF_TOPIC = "RMQ_SYS_TRANS_OP_HALF_TOPIC";
public static final String TRANS_CHECK_MAX_TIME_TOPIC = "TRANS_CHECK_MAX_TIME_TOPIC";
public static final String CID_SYS_RMQ_TRANS = "CID_RMQ_SYS_TRANS";
public static final String ACL_CONF_TOOLS_FILE = "/conf/tools.yml";
+ public static final String REPLY_MESSAGE_FLAG = "reply";
+ private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.COMMON_LOGGER_NAME);
public static String getWSAddr() {
String wsDomainName = System.getProperty("rocketmq.namesrv.domain", DEFAULT_NAMESRV_ADDR_LOOKUP);
@@ -110,6 +107,10 @@ public class MixAll {
return RETRY_GROUP_TOPIC_PREFIX + consumerGroup;
}
+ public static String getReplyTopic(final String clusterName) {
+ return clusterName + "_" + REPLY_TOPIC_POSTFIX;
+ }
+
public static boolean isSysConsumerGroup(final String consumerGroup) {
return consumerGroup.startsWith(CID_RMQ_SYS_PREFIX);
}
diff --git a/common/src/main/java/org/apache/rocketmq/common/UtilAll.java b/common/src/main/java/org/apache/rocketmq/common/UtilAll.java
index 85a4cde121..222f697eb5 100644
--- a/common/src/main/java/org/apache/rocketmq/common/UtilAll.java
+++ b/common/src/main/java/org/apache/rocketmq/common/UtilAll.java
@@ -440,13 +440,11 @@ public class UtilAll {
return false;
}
- public static boolean isInternalV6IP(byte[] ip) {
- if (ip.length != 16) {
- throw new RuntimeException("illegal ipv6 bytes");
- }
-
- //FEC0:0000:0000:0000:0000:0000:0000:0000/10
- if (ip[0] == (byte) 254 && ip[1] >= (byte) 192) {
+ public static boolean isInternalV6IP(InetAddress inetAddr) {
+ if (inetAddr.isAnyLocalAddress() // Wild card ipv6
+ || inetAddr.isLinkLocalAddress() // Single broadcast ipv6 address: fe80:xx:xx...
+ || inetAddr.isLoopbackAddress() //Loopback ipv6 address
+ || inetAddr.isSiteLocalAddress()) { // Site local ipv6 address: fec0:xx:xx...
return true;
}
return false;
@@ -457,9 +455,6 @@ public class UtilAll {
throw new RuntimeException("illegal ipv4 bytes");
}
-// if (ip[0] == (byte)30 && ip[1] == (byte)10 && ip[2] == (byte)163 && ip[3] == (byte)120) {
-// }
-
if (ip[0] >= (byte) 1 && ip[0] <= (byte) 126) {
if (ip[1] == (byte) 1 && ip[2] == (byte) 1 && ip[3] == (byte) 1) {
return false;
@@ -550,7 +545,7 @@ public class UtilAll {
byte[] ipByte = ip.getAddress();
if (ipByte.length == 16) {
if (ipV6Check(ipByte)) {
- if (!isInternalV6IP(ipByte)) {
+ if (!isInternalV6IP(ip)) {
return ipByte;
} else if (internalIP == null) {
internalIP = ipByte;
diff --git a/common/src/main/java/org/apache/rocketmq/common/message/MessageConst.java b/common/src/main/java/org/apache/rocketmq/common/message/MessageConst.java
index aa8481643a..5bdc846562 100644
--- a/common/src/main/java/org/apache/rocketmq/common/message/MessageConst.java
+++ b/common/src/main/java/org/apache/rocketmq/common/message/MessageConst.java
@@ -45,6 +45,13 @@ public class MessageConst {
public static final String PROPERTY_TRANSACTION_CHECK_TIMES = "TRANSACTION_CHECK_TIMES";
public static final String PROPERTY_CHECK_IMMUNITY_TIME_IN_SECONDS = "CHECK_IMMUNITY_TIME_IN_SECONDS";
public static final String PROPERTY_INSTANCE_ID = "INSTANCE_ID";
+ public static final String PROPERTY_CORRELATION_ID = "CORRELATION_ID";
+ public static final String PROPERTY_MESSAGE_REPLY_TO_CLIENT = "REPLY_TO_CLIENT";
+ public static final String PROPERTY_MESSAGE_TTL = "TTL";
+ public static final String PROPERTY_REPLY_MESSAGE_ARRIVE_TIME = "ARRIVE_TIME";
+ public static final String PROPERTY_PUSH_REPLY_TIME = "PUSH_REPLY_TIME";
+ public static final String PROPERTY_CLUSTER = "CLUSTER";
+ public static final String PROPERTY_MESSAGE_TYPE = "MSG_TYPE";
public static final String KEY_SEPARATOR = " ";
@@ -74,5 +81,12 @@ public class MessageConst {
STRING_HASH_SET.add(PROPERTY_MAX_RECONSUME_TIMES);
STRING_HASH_SET.add(PROPERTY_CONSUME_START_TIMESTAMP);
STRING_HASH_SET.add(PROPERTY_INSTANCE_ID);
+ STRING_HASH_SET.add(PROPERTY_CORRELATION_ID);
+ STRING_HASH_SET.add(PROPERTY_MESSAGE_REPLY_TO_CLIENT);
+ STRING_HASH_SET.add(PROPERTY_MESSAGE_TTL);
+ STRING_HASH_SET.add(PROPERTY_REPLY_MESSAGE_ARRIVE_TIME);
+ STRING_HASH_SET.add(PROPERTY_PUSH_REPLY_TIME);
+ STRING_HASH_SET.add(PROPERTY_CLUSTER);
+ STRING_HASH_SET.add(PROPERTY_MESSAGE_TYPE);
}
}
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 dbdabbcde9..b3009d738f 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
@@ -182,4 +182,10 @@ public class RequestCode {
* resume logic of checking half messages that have been put in TRANS_CHECK_MAXTIME_TOPIC before
*/
public static final int RESUME_CHECK_HALF_MESSAGE = 323;
+
+ public static final int SEND_REPLY_MESSAGE = 324;
+
+ public static final int SEND_REPLY_MESSAGE_V2 = 325;
+
+ public static final int PUSH_REPLY_MESSAGE_TO_CLIENT = 326;
}
diff --git a/common/src/main/java/org/apache/rocketmq/common/protocol/header/ReplyMessageRequestHeader.java b/common/src/main/java/org/apache/rocketmq/common/protocol/header/ReplyMessageRequestHeader.java
new file mode 100644
index 0000000000..3bb09073f7
--- /dev/null
+++ b/common/src/main/java/org/apache/rocketmq/common/protocol/header/ReplyMessageRequestHeader.java
@@ -0,0 +1,170 @@
+/*
+ * 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.common.protocol.header;
+
+import org.apache.rocketmq.remoting.CommandCustomHeader;
+import org.apache.rocketmq.remoting.annotation.CFNotNull;
+import org.apache.rocketmq.remoting.annotation.CFNullable;
+import org.apache.rocketmq.remoting.exception.RemotingCommandException;
+
+public class ReplyMessageRequestHeader implements CommandCustomHeader {
+ @CFNotNull
+ private String producerGroup;
+ @CFNotNull
+ private String topic;
+ @CFNotNull
+ private String defaultTopic;
+ @CFNotNull
+ private Integer defaultTopicQueueNums;
+ @CFNotNull
+ private Integer queueId;
+ @CFNotNull
+ private Integer sysFlag;
+ @CFNotNull
+ private Long bornTimestamp;
+ @CFNotNull
+ private Integer flag;
+ @CFNullable
+ private String properties;
+ @CFNullable
+ private Integer reconsumeTimes;
+ @CFNullable
+ private boolean unitMode = false;
+
+ @CFNotNull
+ private String bornHost;
+ @CFNotNull
+ private String storeHost;
+ @CFNotNull
+ private long storeTimestamp;
+
+ public void checkFields() throws RemotingCommandException {
+ }
+
+ public String getProducerGroup() {
+ return producerGroup;
+ }
+
+ public void setProducerGroup(String producerGroup) {
+ this.producerGroup = producerGroup;
+ }
+
+ public String getTopic() {
+ return topic;
+ }
+
+ public void setTopic(String topic) {
+ this.topic = topic;
+ }
+
+ public String getDefaultTopic() {
+ return defaultTopic;
+ }
+
+ public void setDefaultTopic(String defaultTopic) {
+ this.defaultTopic = defaultTopic;
+ }
+
+ public Integer getDefaultTopicQueueNums() {
+ return defaultTopicQueueNums;
+ }
+
+ public void setDefaultTopicQueueNums(Integer defaultTopicQueueNums) {
+ this.defaultTopicQueueNums = defaultTopicQueueNums;
+ }
+
+ public Integer getQueueId() {
+ return queueId;
+ }
+
+ public void setQueueId(Integer queueId) {
+ this.queueId = queueId;
+ }
+
+ public Integer getSysFlag() {
+ return sysFlag;
+ }
+
+ public void setSysFlag(Integer sysFlag) {
+ this.sysFlag = sysFlag;
+ }
+
+ public Long getBornTimestamp() {
+ return bornTimestamp;
+ }
+
+ public void setBornTimestamp(Long bornTimestamp) {
+ this.bornTimestamp = bornTimestamp;
+ }
+
+ public Integer getFlag() {
+ return flag;
+ }
+
+ public void setFlag(Integer flag) {
+ this.flag = flag;
+ }
+
+ public String getProperties() {
+ return properties;
+ }
+
+ public void setProperties(String properties) {
+ this.properties = properties;
+ }
+
+ public Integer getReconsumeTimes() {
+ return reconsumeTimes;
+ }
+
+ public void setReconsumeTimes(Integer reconsumeTimes) {
+ this.reconsumeTimes = reconsumeTimes;
+ }
+
+ public boolean isUnitMode() {
+ return unitMode;
+ }
+
+ public void setUnitMode(boolean unitMode) {
+ this.unitMode = unitMode;
+ }
+
+ public String getBornHost() {
+ return bornHost;
+ }
+
+ public void setBornHost(String bornHost) {
+ this.bornHost = bornHost;
+ }
+
+ public String getStoreHost() {
+ return storeHost;
+ }
+
+ public void setStoreHost(String storeHost) {
+ this.storeHost = storeHost;
+ }
+
+ public long getStoreTimestamp() {
+ return storeTimestamp;
+ }
+
+ public void setStoreTimestamp(long storeTimestamp) {
+ this.storeTimestamp = storeTimestamp;
+ }
+}
diff --git a/common/src/main/java/org/apache/rocketmq/common/utils/CorrelationIdUtil.java b/common/src/main/java/org/apache/rocketmq/common/utils/CorrelationIdUtil.java
new file mode 100644
index 0000000000..86d1fd4d42
--- /dev/null
+++ b/common/src/main/java/org/apache/rocketmq/common/utils/CorrelationIdUtil.java
@@ -0,0 +1,26 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.common.utils;
+
+import java.util.UUID;
+
+public class CorrelationIdUtil {
+ public static String createCorrelationId() {
+ return UUID.randomUUID().toString();
+ }
+}
diff --git a/common/src/test/java/org/apache/rocketmq/common/UtilAllTest.java b/common/src/test/java/org/apache/rocketmq/common/UtilAllTest.java
index 3991367330..b854099a5b 100644
--- a/common/src/test/java/org/apache/rocketmq/common/UtilAllTest.java
+++ b/common/src/test/java/org/apache/rocketmq/common/UtilAllTest.java
@@ -17,6 +17,8 @@
package org.apache.rocketmq.common;
+import java.net.InetAddress;
+import java.net.UnknownHostException;
import java.util.Properties;
import org.junit.Test;
@@ -99,12 +101,12 @@ public class UtilAllTest {
}
@Test
- public void testIPv6Check() {
- byte[] nonInternalIp = UtilAll.string2bytes("24084004018081003FAA1DDE2B3F898A");
- byte[] internalIp = UtilAll.string2bytes("FEC0000000000000000000000000FFFF");
- assertThat(UtilAll.isInternalV6IP(nonInternalIp)).isFalse();
- assertThat(UtilAll.isInternalV6IP(internalIp)).isTrue();
- assertThat(UtilAll.ipToIPv6Str(nonInternalIp).toUpperCase()).isEqualTo("2408:4004:0180:8100:3FAA:1DDE:2B3F:898A");
+ public void testIPv6Check() throws UnknownHostException {
+ InetAddress nonInternal = InetAddress.getByName("2408:4004:0180:8100:3FAA:1DDE:2B3F:898A");
+ InetAddress internal = InetAddress.getByName("FE80:0000:0000:0000:0000:0000:0000:FFFF");
+ assertThat(UtilAll.isInternalV6IP(nonInternal)).isFalse();
+ assertThat(UtilAll.isInternalV6IP(internal)).isTrue();
+ assertThat(UtilAll.ipToIPv6Str(nonInternal.getAddress()).toUpperCase()).isEqualTo("2408:4004:0180:8100:3FAA:1DDE:2B3F:898A");
}
static class DemoConfig {
diff --git a/common/src/test/java/org/apache/rocketmq/common/message/MessageClientIDSetterTest.java b/common/src/test/java/org/apache/rocketmq/common/message/MessageClientIDSetterTest.java
index 0df75eb30a..1ec6d93cff 100644
--- a/common/src/test/java/org/apache/rocketmq/common/message/MessageClientIDSetterTest.java
+++ b/common/src/test/java/org/apache/rocketmq/common/message/MessageClientIDSetterTest.java
@@ -17,9 +17,8 @@
package org.apache.rocketmq.common.message;
-import java.util.Calendar;
-import java.util.Date;
import org.junit.Test;
+
import static org.assertj.core.api.Assertions.assertThat;
public class MessageClientIDSetterTest {
diff --git a/distribution/pom.xml b/distribution/pom.xml
index 4e3e78fcd9..c023c9847e 100644
--- a/distribution/pom.xml
+++ b/distribution/pom.xml
@@ -20,7 +20,7 @@
org.apache.rocketmq
rocketmq-all
- 4.6.0-SNAPSHOT
+ 4.6.1-SNAPSHOT
rocketmq-distribution
rocketmq-distribution ${project.version}
diff --git a/docs/cn/README b/docs/cn/README
new file mode 100644
index 0000000000..e69de29bb2
diff --git a/docs/cn/README.md b/docs/cn/README.md
index 3247985e53..2dbd854913 100644
--- a/docs/cn/README.md
+++ b/docs/cn/README.md
@@ -1,7 +1,7 @@
Apache RocketMQ开发者指南
--------
-##### 这个开发者指南是帮忙您快速了解,并使用 Apache RocketMQ
+##### 这个开发者指南是帮助您快速了解,并使用 Apache RocketMQ
### 1. 概念和特性
diff --git a/docs/cn/RocketMQ_Example.md b/docs/cn/RocketMQ_Example.md
index d298db8420..09acb0de97 100644
--- a/docs/cn/RocketMQ_Example.md
+++ b/docs/cn/RocketMQ_Example.md
@@ -66,7 +66,11 @@ public class AsyncProducer {
// 启动Producer实例
producer.start();
producer.setRetryTimesWhenSendAsyncFailed(0);
- for (int i = 0; i < 100; i++) {
+
+ int messageCount = 100;
+ // 根据消息数量实例化倒计时计算器
+ final CountDownLatch2 countDownLatch = new CountDownLatch2(messageCount);
+ for (int i = 0; i < messageCount; i++) {
final int index = i;
// 创建消息，并指定Topic，Tag和消息体
Message msg = new Message("TopicTest",
@@ -87,6 +91,8 @@ public class AsyncProducer {
}
});
}
+ // 等待5s
+ countDownLatch.await(5, TimeUnit.SECONDS);
// 如果不再发送消息，关闭Producer实例。
producer.shutdown();
}
diff --git a/docs/cn/features.md b/docs/cn/features.md
index 859e0f84a3..e7b63864e9 100644
--- a/docs/cn/features.md
+++ b/docs/cn/features.md
@@ -16,7 +16,7 @@
RocketMQ的消费者可以根据Tag进行消息过滤，也支持自定义属性过滤。消息过滤目前是在Broker端实现的，优点是减少了对于Consumer无用消息的网络传输，缺点是增加了Broker的负担、而且实现相对复杂。
## 4 消息可靠性
RocketMQ支持消息的高可靠，影响消息可靠性的几种情况：
-1) Broker正常关闭
+1) Broker非正常关闭
2) Broker异常Crash
3) OS Crash
4) 机器掉电，但是能立即恢复供电情况
diff --git a/example/pom.xml b/example/pom.xml
index 5e160276e5..8340d82642 100644
--- a/example/pom.xml
+++ b/example/pom.xml
@@ -19,7 +19,7 @@
org.apache.rocketmq
rocketmq-all
- 4.6.0-SNAPSHOT
+ 4.6.1-SNAPSHOT
4.0.0
@@ -51,12 +51,12 @@
org.apache.rocketmq
rocketmq-openmessaging
- 4.6.0-SNAPSHOT
+ 4.6.1-SNAPSHOT
org.apache.rocketmq
rocketmq-acl
- 4.6.0-SNAPSHOT
+ 4.6.1-SNAPSHOT
diff --git a/example/src/main/java/org/apache/rocketmq/example/benchmark/Consumer.java b/example/src/main/java/org/apache/rocketmq/example/benchmark/Consumer.java
index d431d3ecc6..4724a1d7ab 100644
--- a/example/src/main/java/org/apache/rocketmq/example/benchmark/Consumer.java
+++ b/example/src/main/java/org/apache/rocketmq/example/benchmark/Consumer.java
@@ -103,6 +103,10 @@ public class Consumer {
}, 10000, 10000);
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(group);
+ if (commandLine.hasOption('n')) {
+ String ns = commandLine.getOptionValue('n');
+ consumer.setNamesrvAddr(ns);
+ }
consumer.setInstanceName(Long.toString(System.currentTimeMillis()));
if (filterType == null || expression == null) {
diff --git a/example/src/main/java/org/apache/rocketmq/example/benchmark/TransactionProducer.java b/example/src/main/java/org/apache/rocketmq/example/benchmark/TransactionProducer.java
index d9fafdd08e..04707e1a28 100644
--- a/example/src/main/java/org/apache/rocketmq/example/benchmark/TransactionProducer.java
+++ b/example/src/main/java/org/apache/rocketmq/example/benchmark/TransactionProducer.java
@@ -24,6 +24,11 @@ import java.util.TimerTask;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicLong;
+
+import org.apache.commons.cli.CommandLine;
+import org.apache.commons.cli.Option;
+import org.apache.commons.cli.Options;
+import org.apache.commons.cli.PosixParser;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.LocalTransactionExecuter;
import org.apache.rocketmq.client.producer.LocalTransactionState;
@@ -33,20 +38,19 @@ import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.common.RemotingHelper;
+import org.apache.rocketmq.srvutil.ServerUtil;
public class TransactionProducer {
- private static int threadCount;
- private static int messageSize;
- private static boolean ischeck;
- private static boolean ischeckffalse;
public static void main(String[] args) throws MQClientException, UnsupportedEncodingException {
- threadCount = args.length >= 1 ? Integer.parseInt(args[0]) : 32;
- messageSize = args.length >= 2 ? Integer.parseInt(args[1]) : 1024 * 2;
- ischeck = args.length >= 3 && Boolean.parseBoolean(args[2]);
- ischeckffalse = args.length >= 4 && Boolean.parseBoolean(args[3]);
+ Options options = ServerUtil.buildCommandlineOptions(new Options());
+ CommandLine commandLine = ServerUtil.parseCmdLine("TransactionProducer", args, buildCommandlineOptions(options), new PosixParser());
- final Message msg = buildMessage(messageSize);
+ final String topic = commandLine.hasOption('t') ? commandLine.getOptionValue('t').trim() : "BenchmarkTest";
+ final int threadCount = commandLine.hasOption('w') ? Integer.parseInt(commandLine.getOptionValue('w')) : 32;
+ final int messageSize = commandLine.hasOption('s') ? Integer.parseInt(commandLine.getOptionValue('s')) : 2048;
+ final boolean ischeck = commandLine.hasOption('c') ? Boolean.parseBoolean(commandLine.getOptionValue('c')) : false;
+ final boolean ischeckffalse = commandLine.hasOption('r') ? Boolean.parseBoolean(commandLine.getOptionValue('r')) : true;
final ExecutorService sendThreadPool = Executors.newFixedThreadPool(threadCount);
@@ -98,6 +102,10 @@ public class TransactionProducer {
producer.setInstanceName(Long.toString(System.currentTimeMillis()));
producer.setTransactionCheckListener(transactionCheckListener);
producer.setDefaultTopicQueueNums(1000);
+ if (commandLine.hasOption('n')) {
+ String ns = commandLine.getOptionValue('n');
+ producer.setNamesrvAddr(ns);
+ }
producer.start();
final TransactionExecuterBImpl tranExecuter = new TransactionExecuterBImpl(ischeck);
@@ -111,7 +119,7 @@ public class TransactionProducer {
// Thread.sleep(1000);
final long beginTimestamp = System.currentTimeMillis();
SendResult sendResult =
- producer.sendMessageInTransaction(msg, tranExecuter, null);
+ producer.sendMessageInTransaction(buildMessage(messageSize, topic), tranExecuter, null);
if (sendResult != null) {
statsBenchmark.getSendRequestSuccessCount().incrementAndGet();
statsBenchmark.getReceiveResponseSuccessCount().incrementAndGet();
@@ -138,18 +146,45 @@ public class TransactionProducer {
}
}
- private static Message buildMessage(final int messageSize) throws UnsupportedEncodingException {
- Message msg = new Message();
- msg.setTopic("BenchmarkTest");
+ private static Message buildMessage(final int messageSize, String topic) {
+ try {
+ Message msg = new Message();
+ msg.setTopic(topic);
- StringBuilder sb = new StringBuilder();
- for (int i = 0; i < messageSize; i += 10) {
- sb.append("hello baby");
+ StringBuilder sb = new StringBuilder();
+ for (int i = 0; i < messageSize; i += 10) {
+ sb.append("hello baby");
+ }
+ msg.setBody(sb.toString().getBytes(RemotingHelper.DEFAULT_CHARSET));
+ return msg;
+ } catch (UnsupportedEncodingException e) {
+ throw new RuntimeException(e);
}
+ }
- msg.setBody(sb.toString().getBytes(RemotingHelper.DEFAULT_CHARSET));
+ public static Options buildCommandlineOptions(final Options options) {
+ Option opt = new Option("w", "threadCount", true, "Thread count, Default: 32");
+ opt.setRequired(false);
+ options.addOption(opt);
- return msg;
+ opt = new Option("s", "messageSize", true, "Message Size, Default: 2048");
+ opt.setRequired(false);
+ options.addOption(opt);
+
+ opt = new Option("t", "topic", true, "Topic name, Default: BenchmarkTest");
+ opt.setRequired(false);
+ options.addOption(opt);
+
+ opt = new Option("c", "check", true, "Check the message, Default: false");
+ opt.setRequired(false);
+ options.addOption(opt);
+
+ opt = new Option("r", "checkResult", true, "Message check result, Default: true");
+ opt.setRequired(false);
+ options.addOption(opt);
+
+
+ return options;
}
}
diff --git a/example/src/main/java/org/apache/rocketmq/example/rpc/AsyncRequestProducer.java b/example/src/main/java/org/apache/rocketmq/example/rpc/AsyncRequestProducer.java
new file mode 100644
index 0000000000..072291d5c2
--- /dev/null
+++ b/example/src/main/java/org/apache/rocketmq/example/rpc/AsyncRequestProducer.java
@@ -0,0 +1,63 @@
+/*
+ * 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.example.rpc;
+
+import org.apache.rocketmq.client.exception.MQClientException;
+import org.apache.rocketmq.client.log.ClientLogger;
+import org.apache.rocketmq.client.producer.DefaultMQProducer;
+import org.apache.rocketmq.client.producer.RequestCallback;
+import org.apache.rocketmq.common.message.Message;
+import org.apache.rocketmq.logging.InternalLogger;
+import org.apache.rocketmq.remoting.common.RemotingHelper;
+
+public class AsyncRequestProducer {
+ private static final InternalLogger log = ClientLogger.getLog();
+
+ public static void main(String[] args) throws MQClientException, InterruptedException {
+ String producerGroup = "please_rename_unique_group_name";
+ String topic = "RequestTopic";
+ long ttl = 3000;
+
+ DefaultMQProducer producer = new DefaultMQProducer(producerGroup);
+ producer.start();
+
+ try {
+ Message msg = new Message(topic,
+ "",
+ "Hello world".getBytes(RemotingHelper.DEFAULT_CHARSET));
+
+ long begin = System.currentTimeMillis();
+ producer.request(msg, new RequestCallback() {
+ @Override
+ public void onSuccess(Message message) {
+ long cost = System.currentTimeMillis() - begin;
+ System.out.printf("request to <%s> cost: %d replyMessage: %s %n", topic, cost, message);
+ }
+
+ @Override
+ public void onException(Throwable e) {
+ System.err.printf("request to <%s> fail.", topic);
+ }
+ }, ttl);
+ } catch (Exception e) {
+ log.warn("", e);
+ }
+ /* shutdown after your request callback is finished */
+// producer.shutdown();
+ }
+}
diff --git a/example/src/main/java/org/apache/rocketmq/example/rpc/RequestProducer.java b/example/src/main/java/org/apache/rocketmq/example/rpc/RequestProducer.java
new file mode 100644
index 0000000000..b34908b84f
--- /dev/null
+++ b/example/src/main/java/org/apache/rocketmq/example/rpc/RequestProducer.java
@@ -0,0 +1,48 @@
+/*
+ * 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.example.rpc;
+
+import org.apache.rocketmq.client.exception.MQClientException;
+import org.apache.rocketmq.client.producer.DefaultMQProducer;
+import org.apache.rocketmq.common.message.Message;
+import org.apache.rocketmq.remoting.common.RemotingHelper;
+
+public class RequestProducer {
+ public static void main(String[] args) throws MQClientException, InterruptedException {
+ String producerGroup = "please_rename_unique_group_name";
+ String topic = "RequestTopic";
+ long ttl = 3000;
+
+ DefaultMQProducer producer = new DefaultMQProducer(producerGroup);
+ producer.start();
+
+ try {
+ 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);
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
+ producer.shutdown();
+ }
+}
diff --git a/example/src/main/java/org/apache/rocketmq/example/rpc/ResponseConsumer.java b/example/src/main/java/org/apache/rocketmq/example/rpc/ResponseConsumer.java
new file mode 100644
index 0000000000..c62c7d4eb6
--- /dev/null
+++ b/example/src/main/java/org/apache/rocketmq/example/rpc/ResponseConsumer.java
@@ -0,0 +1,79 @@
+/*
+ * 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.example.rpc;
+
+import java.util.List;
+import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
+import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
+import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
+import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
+import org.apache.rocketmq.client.exception.MQBrokerException;
+import org.apache.rocketmq.client.exception.MQClientException;
+import org.apache.rocketmq.client.producer.DefaultMQProducer;
+import org.apache.rocketmq.client.producer.SendResult;
+import org.apache.rocketmq.client.utils.MessageUtil;
+import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
+import org.apache.rocketmq.common.message.Message;
+import org.apache.rocketmq.common.message.MessageExt;
+import org.apache.rocketmq.remoting.exception.RemotingException;
+
+public class ResponseConsumer {
+ public static void main(String[] args) throws InterruptedException, MQClientException {
+ String producerGroup = "please_rename_unique_group_name";
+ String consumerGroup = "please_rename_unique_group_name";
+ String topic = "RequestTopic";
+
+ // create a producer to send reply message
+ DefaultMQProducer replyProducer = new DefaultMQProducer(producerGroup);
+ replyProducer.start();
+
+ // create consumer
+ DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(consumerGroup);
+ consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
+
+ // recommend client configs
+ consumer.setPullTimeDelayMillsWhenException(0L);
+
+ 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);
+ 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;
+ }
+ });
+
+ consumer.subscribe(topic, "*");
+ consumer.start();
+ System.out.printf("Consumer Started.%n");
+ }
+}
diff --git a/filter/pom.xml b/filter/pom.xml
index 9b2128d957..e170f26884 100644
--- a/filter/pom.xml
+++ b/filter/pom.xml
@@ -20,7 +20,7 @@
rocketmq-all
org.apache.rocketmq
- 4.6.0-SNAPSHOT
+ 4.6.1-SNAPSHOT
4.0.0
diff --git a/logappender/pom.xml b/logappender/pom.xml
index 667f213587..83cd3957f7 100644
--- a/logappender/pom.xml
+++ b/logappender/pom.xml
@@ -19,7 +19,7 @@
org.apache.rocketmq
rocketmq-all
- 4.6.0-SNAPSHOT
+ 4.6.1-SNAPSHOT
4.0.0
rocketmq-logappender
diff --git a/logging/pom.xml b/logging/pom.xml
index 95c70b504f..4b92a6ac7b 100644
--- a/logging/pom.xml
+++ b/logging/pom.xml
@@ -19,7 +19,7 @@
org.apache.rocketmq
rocketmq-all
- 4.6.0-SNAPSHOT
+ 4.6.1-SNAPSHOT
4.0.0
diff --git a/namesrv/pom.xml b/namesrv/pom.xml
index bb858b91e1..3b5fc5635e 100644
--- a/namesrv/pom.xml
+++ b/namesrv/pom.xml
@@ -19,7 +19,7 @@
org.apache.rocketmq
rocketmq-all
- 4.6.0-SNAPSHOT
+ 4.6.1-SNAPSHOT
4.0.0
diff --git a/openmessaging/pom.xml b/openmessaging/pom.xml
index 8215a3f628..4a9354f1a3 100644
--- a/openmessaging/pom.xml
+++ b/openmessaging/pom.xml
@@ -20,7 +20,7 @@
rocketmq-all
org.apache.rocketmq
- 4.6.0-SNAPSHOT
+ 4.6.1-SNAPSHOT
4.0.0
diff --git a/pom.xml b/pom.xml
index 0161c7b924..f8a391b20e 100644
--- a/pom.xml
+++ b/pom.xml
@@ -16,8 +16,7 @@
limitations under the License.
-->
-
+
org.apache
@@ -30,7 +29,7 @@
2012
org.apache.rocketmq
rocketmq-all
- 4.6.0-SNAPSHOT
+ 4.6.1-SNAPSHOT
pom
Apache RocketMQ ${project.version}
http://rocketmq.apache.org/
@@ -160,7 +159,7 @@
-
+
true
@@ -569,7 +568,7 @@
com.alibaba
fastjson
- 1.2.51
+ 1.2.61
org.javassist
diff --git a/remoting/pom.xml b/remoting/pom.xml
index 68aebd9484..3443302f08 100644
--- a/remoting/pom.xml
+++ b/remoting/pom.xml
@@ -19,7 +19,7 @@
org.apache.rocketmq
rocketmq-all
- 4.6.0-SNAPSHOT
+ 4.6.1-SNAPSHOT
4.0.0
diff --git a/srvutil/pom.xml b/srvutil/pom.xml
index 6aa9fa31a2..5f4c4616ef 100644
--- a/srvutil/pom.xml
+++ b/srvutil/pom.xml
@@ -19,7 +19,7 @@
org.apache.rocketmq
rocketmq-all
- 4.6.0-SNAPSHOT
+ 4.6.1-SNAPSHOT
4.0.0
diff --git a/store/pom.xml b/store/pom.xml
index 79e337b0b8..8aa9a84f59 100644
--- a/store/pom.xml
+++ b/store/pom.xml
@@ -19,7 +19,7 @@
org.apache.rocketmq
rocketmq-all
- 4.6.0-SNAPSHOT
+ 4.6.1-SNAPSHOT
4.0.0
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 84fb4217ea..3035c575cf 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
@@ -280,7 +280,9 @@ public class HAService {
if (!this.requestsRead.isEmpty()) {
for (CommitLog.GroupCommitRequest req : this.requestsRead) {
boolean transferOK = HAService.this.push2SlaveMaxOffset.get() >= req.getNextOffset();
- for (int i = 0; !transferOK && i < 5; i++) {
+ long waitUntilWhen = HAService.this.defaultMessageStore.getSystemClock().now()
+ + HAService.this.defaultMessageStore.getMessageStoreConfig().getSyncFlushTimeout();
+ while (!transferOK && HAService.this.defaultMessageStore.getSystemClock().now() < waitUntilWhen) {
this.notifyTransferObject.waitForRunning(1000);
transferOK = HAService.this.push2SlaveMaxOffset.get() >= req.getNextOffset();
}
diff --git a/test/pom.xml b/test/pom.xml
index c3d1318884..b7f8933e32 100644
--- a/test/pom.xml
+++ b/test/pom.xml
@@ -20,7 +20,7 @@
rocketmq-all
org.apache.rocketmq
- 4.6.0-SNAPSHOT
+ 4.6.1-SNAPSHOT
4.0.0
diff --git a/tools/pom.xml b/tools/pom.xml
index 061f3c76bd..1d099eea5a 100644
--- a/tools/pom.xml
+++ b/tools/pom.xml
@@ -19,7 +19,7 @@
org.apache.rocketmq
rocketmq-all
- 4.6.0-SNAPSHOT
+ 4.6.1-SNAPSHOT
4.0.0