[GitHub] incubator-rocketmq pull request #126: [ROCKETMQ-231] fix pull consumer pull ...

[lindzh]

Github user lindzh closed the pull request at:

https://github.com/apache/incubator-rocketmq/pull/126


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[GitHub] incubator-rocketmq pull request #126: [ROCKETMQ-231] fix pull consumer pull ...

[lindzh]

Github user lindzh commented on a diff in the pull request:

https://github.com/apache/incubator-rocketmq/pull/126#discussion_r132622123
  
--- Diff: 
store/src/test/java/org/apache/rocketmq/store/DefaultMessageStoreTest.java ---
@@ -45,19 +47,22 @@ public void init() throws Exception {
 BornHost = new 
InetSocketAddress(InetAddress.getByName("127.0.0.1"), 0);
 }
 
+public MessageStore buildMessageStore() throws Exception {
+MessageStoreConfig messageStoreConfig = new MessageStoreConfig();
+messageStoreConfig.setMapedFileSizeCommitLog(1024 * 1024 * 10);
+messageStoreConfig.setMapedFileSizeConsumeQueue(1024 * 1024 * 10);
+messageStoreConfig.setMaxHashSlotNum(1);
+messageStoreConfig.setMaxIndexNum(100 * 100);
+messageStoreConfig.setFlushDiskType(FlushDiskType.ASYNC_FLUSH);
+return new DefaultMessageStore(messageStoreConfig, new 
BrokerStatsManager("simpleTest"), new MyMessageArrivingListener(), new 
BrokerConfig());
--- End diff --

Indeed this is a must after testing,and this bug has been fix in PR 
https://github.com/apache/incubator-rocketmq/pull/141 


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[GitHub] incubator-rocketmq pull request #126: [ROCKETMQ-231] fix pull consumer pull ...

[dongeforever]

Github user dongeforever commented on a diff in the pull request:

https://github.com/apache/incubator-rocketmq/pull/126#discussion_r132427103
  
--- Diff: 
store/src/test/java/org/apache/rocketmq/store/DefaultMessageStoreTest.java ---
@@ -45,19 +47,22 @@ public void init() throws Exception {
 BornHost = new 
InetSocketAddress(InetAddress.getByName("127.0.0.1"), 0);
 }
 
+public MessageStore buildMessageStore() throws Exception {
+MessageStoreConfig messageStoreConfig = new MessageStoreConfig();
+messageStoreConfig.setMapedFileSizeCommitLog(1024 * 1024 * 10);
+messageStoreConfig.setMapedFileSizeConsumeQueue(1024 * 1024 * 10);
+messageStoreConfig.setMaxHashSlotNum(1);
+messageStoreConfig.setMaxIndexNum(100 * 100);
+messageStoreConfig.setFlushDiskType(FlushDiskType.ASYNC_FLUSH);
+return new DefaultMessageStore(messageStoreConfig, new 
BrokerStatsManager("simpleTest"), new MyMessageArrivingListener(), new 
BrokerConfig());
--- End diff --

The MessageStore uses the default store path(~/store).
And the test forgets to delete files in that path.
Two suggestions:
1. create temp path, and set it as storePath
2. delete all the files after the testing



---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[GitHub] incubator-rocketmq pull request #126: [ROCKETMQ-231] fix pull consumer pull ...

[lindzh]

Github user lindzh commented on a diff in the pull request:

https://github.com/apache/incubator-rocketmq/pull/126#discussion_r130558354
  
--- Diff: 
store/src/main/java/org/apache/rocketmq/store/DefaultMessageStore.java ---
@@ -1110,7 +1110,7 @@ private boolean isTheBatchFull(int sizePy, int 
maxMsgNums, int bufferTotal, int
 return false;
 }
 
-if ((messageTotal + 1) >= maxMsgNums) {
--- End diff --

Yes,this problem has been fixed to maxMsgNums <= messageTotal


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[GitHub] incubator-rocketmq pull request #126: [ROCKETMQ-231] fix pull consumer pull ...

[Jaskey]

Github user Jaskey commented on a diff in the pull request:

https://github.com/apache/incubator-rocketmq/pull/126#discussion_r125448602
  
--- Diff: 
store/src/main/java/org/apache/rocketmq/store/DefaultMessageStore.java ---
@@ -1110,7 +1110,7 @@ private boolean isTheBatchFull(int sizePy, int 
maxMsgNums, int bufferTotal, int
 return false;
 }
 
-if ((messageTotal + 1) >= maxMsgNums) {
--- End diff --

I will still suggest using `>=`


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[GitHub] incubator-rocketmq pull request #126: [ROCKETMQ-231] fix pull consumer pull ...

[vongosling]

Github user vongosling commented on a diff in the pull request:

https://github.com/apache/incubator-rocketmq/pull/126#discussion_r125432988
  
--- Diff: 
store/src/main/java/org/apache/rocketmq/store/DefaultMessageStore.java ---
@@ -1110,7 +1110,7 @@ private boolean isTheBatchFull(int sizePy, int 
maxMsgNums, int bufferTotal, int
 return false;
 }
 
-if ((messageTotal + 1) >= maxMsgNums) {
--- End diff --

why change to -1 in your new change:-)


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[GitHub] incubator-rocketmq pull request #126: [ROCKETMQ-231] fix pull consumer pull ...

[lindzh]

Github user lindzh commented on a diff in the pull request:

https://github.com/apache/incubator-rocketmq/pull/126#discussion_r124758258
  
--- Diff: 
test/src/test/java/org/apache/rocketmq/test/client/consumer/pull/PullSizeTest.java
 ---
@@ -0,0 +1,135 @@
+/*
+ * 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.test.client.consumer.pull;
+
+import org.apache.log4j.Logger;
+import org.apache.rocketmq.client.consumer.DefaultMQPullConsumer;
+import org.apache.rocketmq.client.consumer.PullResult;
+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.producer.SendStatus;
+import org.apache.rocketmq.common.message.Message;
+import org.apache.rocketmq.common.message.MessageExt;
+import org.apache.rocketmq.common.message.MessageQueue;
+import org.apache.rocketmq.remoting.common.RemotingHelper;
+import org.apache.rocketmq.test.base.BaseConf;
+import org.apache.rocketmq.test.base.IntegrationTestBase;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.nio.channels.Pipe;
+import java.util.*;
+
+public class PullSizeTest extends BaseConf {
+
+private static Logger logger = Logger.getLogger(PullSizeTest.class);
+
+private static final Map OFFSE_TABLE = new 
HashMap();
+
+public static final String PULL_SIZE_TOPIC = "TopicPullTest";
+
+public static final String PULL_SIZE_GROUP = "pullSizeTest";
+
+public boolean send() throws MQClientException, InterruptedException {
+DefaultMQProducer producer = new 
DefaultMQProducer(PULL_SIZE_GROUP);
+producer.setNamesrvAddr(nsAddr);
+producer.start();
+int successCount = 0;
+for (int i = 0; i < 1000; i++) {
+try {
+Message msg = new Message(PULL_SIZE_TOPIC, "TagA", 
("RocketMQ pull size test index " + 
i).getBytes(RemotingHelper.DEFAULT_CHARSET));
+SendResult sendResult = producer.send(msg);
+if (sendResult.getSendStatus().equals(SendStatus.SEND_OK)) 
{
+successCount++;
+}
+} catch (Exception e) {
+e.printStackTrace();
+Thread.sleep(500);
+}
+}
+producer.shutdown();
+if (successCount > 800) {
--- End diff --

This test case has removed and add new unit test in store.


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[GitHub] incubator-rocketmq pull request #126: [ROCKETMQ-231] fix pull consumer pull ...

[vongosling]

Github user vongosling commented on a diff in the pull request:

https://github.com/apache/incubator-rocketmq/pull/126#discussion_r123730992
  
--- Diff: 
test/src/test/java/org/apache/rocketmq/test/client/consumer/pull/PullSizeTest.java
 ---
@@ -0,0 +1,135 @@
+/*
+ * 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.test.client.consumer.pull;
+
+import org.apache.log4j.Logger;
+import org.apache.rocketmq.client.consumer.DefaultMQPullConsumer;
+import org.apache.rocketmq.client.consumer.PullResult;
+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.producer.SendStatus;
+import org.apache.rocketmq.common.message.Message;
+import org.apache.rocketmq.common.message.MessageExt;
+import org.apache.rocketmq.common.message.MessageQueue;
+import org.apache.rocketmq.remoting.common.RemotingHelper;
+import org.apache.rocketmq.test.base.BaseConf;
+import org.apache.rocketmq.test.base.IntegrationTestBase;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.nio.channels.Pipe;
+import java.util.*;
+
+public class PullSizeTest extends BaseConf {
+
+private static Logger logger = Logger.getLogger(PullSizeTest.class);
+
+private static final Map OFFSE_TABLE = new 
HashMap();
+
+public static final String PULL_SIZE_TOPIC = "TopicPullTest";
+
+public static final String PULL_SIZE_GROUP = "pullSizeTest";
+
+public boolean send() throws MQClientException, InterruptedException {
+DefaultMQProducer producer = new 
DefaultMQProducer(PULL_SIZE_GROUP);
+producer.setNamesrvAddr(nsAddr);
+producer.start();
+int successCount = 0;
+for (int i = 0; i < 1000; i++) {
+try {
+Message msg = new Message(PULL_SIZE_TOPIC, "TagA", 
("RocketMQ pull size test index " + 
i).getBytes(RemotingHelper.DEFAULT_CHARSET));
+SendResult sendResult = producer.send(msg);
+if (sendResult.getSendStatus().equals(SendStatus.SEND_OK)) 
{
+successCount++;
+}
+} catch (Exception e) {
+e.printStackTrace();
+Thread.sleep(500);
+}
+}
+producer.shutdown();
+if (successCount > 800) {
--- End diff --

why use this magic number 800 in your if statement


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[GitHub] incubator-rocketmq pull request #126: [ROCKETMQ-231] fix pull consumer pull ...

[vongosling]

Github user vongosling commented on a diff in the pull request:

https://github.com/apache/incubator-rocketmq/pull/126#discussion_r123730667
  
--- Diff: 
test/src/test/java/org/apache/rocketmq/test/client/consumer/pull/PullSizeTest.java
 ---
@@ -0,0 +1,135 @@
+/*
+ * 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.test.client.consumer.pull;
+
+import org.apache.log4j.Logger;
+import org.apache.rocketmq.client.consumer.DefaultMQPullConsumer;
+import org.apache.rocketmq.client.consumer.PullResult;
+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.producer.SendStatus;
+import org.apache.rocketmq.common.message.Message;
+import org.apache.rocketmq.common.message.MessageExt;
+import org.apache.rocketmq.common.message.MessageQueue;
+import org.apache.rocketmq.remoting.common.RemotingHelper;
+import org.apache.rocketmq.test.base.BaseConf;
+import org.apache.rocketmq.test.base.IntegrationTestBase;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.nio.channels.Pipe;
+import java.util.*;
+
+public class PullSizeTest extends BaseConf {
+
+private static Logger logger = Logger.getLogger(PullSizeTest.class);
--- End diff --

Could we remove logger in unit test ?


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[GitHub] incubator-rocketmq pull request #126: [ROCKETMQ-231] fix pull consumer pull ...

[vongosling]

Github user vongosling commented on a diff in the pull request:

https://github.com/apache/incubator-rocketmq/pull/126#discussion_r123731300
  
--- Diff: 
test/src/test/java/org/apache/rocketmq/test/client/consumer/pull/PullSizeTest.java
 ---
@@ -0,0 +1,135 @@
+/*
+ * 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.test.client.consumer.pull;
+
+import org.apache.log4j.Logger;
+import org.apache.rocketmq.client.consumer.DefaultMQPullConsumer;
+import org.apache.rocketmq.client.consumer.PullResult;
+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.producer.SendStatus;
+import org.apache.rocketmq.common.message.Message;
+import org.apache.rocketmq.common.message.MessageExt;
+import org.apache.rocketmq.common.message.MessageQueue;
+import org.apache.rocketmq.remoting.common.RemotingHelper;
+import org.apache.rocketmq.test.base.BaseConf;
+import org.apache.rocketmq.test.base.IntegrationTestBase;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.nio.channels.Pipe;
+import java.util.*;
+
+public class PullSizeTest extends BaseConf {
+
+private static Logger logger = Logger.getLogger(PullSizeTest.class);
+
+private static final Map OFFSE_TABLE = new 
HashMap();
+
+public static final String PULL_SIZE_TOPIC = "TopicPullTest";
+
+public static final String PULL_SIZE_GROUP = "pullSizeTest";
+
+public boolean send() throws MQClientException, InterruptedException {
+DefaultMQProducer producer = new 
DefaultMQProducer(PULL_SIZE_GROUP);
+producer.setNamesrvAddr(nsAddr);
+producer.start();
+int successCount = 0;
+for (int i = 0; i < 1000; i++) {
+try {
+Message msg = new Message(PULL_SIZE_TOPIC, "TagA", 
("RocketMQ pull size test index " + 
i).getBytes(RemotingHelper.DEFAULT_CHARSET));
+SendResult sendResult = producer.send(msg);
+if (sendResult.getSendStatus().equals(SendStatus.SEND_OK)) 
{
+successCount++;
+}
+} catch (Exception e) {
+e.printStackTrace();
+Thread.sleep(500);
+}
+}
+producer.shutdown();
+if (successCount > 800) {
+return true;
+} else {
+return false;
+}
+}
+
+
+public void pullMsg() throws MQClientException {
+boolean result = false;
+DefaultMQPullConsumer consumer = new 
DefaultMQPullConsumer(PULL_SIZE_GROUP);
+consumer.setNamesrvAddr(nsAddr);
+consumer.start();
+Set mqs = 
consumer.fetchSubscribeMessageQueues(PULL_SIZE_TOPIC);
+for (MessageQueue mq : mqs) {
+if (result) {
+break;
+}
+try {
+PullResult pullResult = consumer.pull(mq, null, 
getMessageQueueOffset(mq), 32);
+putMessageQueueOffset(mq, pullResult.getNextBeginOffset());
+switch (pullResult.getPullStatus()) {
+case FOUND:
+List msgFoundList = 
pullResult.getMsgFoundList();
+if (msgFoundList != null) {
+logger.info("[RECV] received msg queue:" + 
mq.getBrokerName() + "-" + mq.getQueueId());
+result |= msgFoundList.size() >= 32;
+}
+break;
+case NO_MATCHED_MSG:
+break;
+case NO_NEW_MSG:
+break;
+case OFFSET_ILLEGAL:
+   

[GitHub] incubator-rocketmq pull request #126: [ROCKETMQ-231] fix pull consumer pull ...

[lindzh]

GitHub user lindzh opened a pull request:

https://github.com/apache/incubator-rocketmq/pull/126

[ROCKETMQ-231] fix pull consumer pull result size

When using PullConsumer pull message by default result size is 32,and 
messages is more than 32 in a queue,but broker always returns 31.


You can merge this pull request into a Git repository by running:

$ git pull https://github.com/lindzh/incubator-rocketmq 
fix_consumer_pull_msg_size

Alternatively you can review and apply these changes as the patch at:

https://github.com/apache/incubator-rocketmq/pull/126.patch

To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:

This closes #126


commit 1d27251df0265e19397cffc1ee5098e12bc4cd0d
Author: lindzh 
Date:   2017-06-23T10:56:49Z

fix pull consumer pull result size




---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---