opensource/rocketmq
3
0
Fork 0
mirror of https://github.com/apache/rocketmq.git synced 2026-09-24 16:04:00 +08:00
Code Issues Packages Projects Releases Wiki Activity

[ISSUE 3613] bug fix, solve message hash conflict in index file #3616

Browse Source
This commit is contained in:
xijiu
2021-12-21 14:12:31 +08:00
committed by GitHub
parent f620a50bc7
commit 0bc011cef3
3 changed files with 66 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files
client/src/main/java/org/apache/rocketmq/client/impl/MQAdminImpl.java
+4 -1
View File
@@ -22,6 +22,7 @@ import java.util.Collections;
import java.util.LinkedList;
import java.util.List;
import java.util.Set;
import java.util.Objects;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReadWriteLock;
@@ -406,12 +407,14 @@ public class MQAdminImpl {
}
} else {
String keys = msgExt.getKeys();
String msgTopic = msgExt.getTopic();
if (keys != null) {
boolean matched = false;
String[] keyArray = keys.split(MessageConst.KEY_SEPARATOR);
if (keyArray != null) {
for (String k : keyArray) {
if (key.equals(k)) {
// both topic and key must be equal at the same time
if (Objects.equals(key, k) && Objects.equals(topic, msgTopic)) {
matched = true;
break;
}
test/src/test/java/org/apache/rocketmq/test/base/BaseConf.java
+5 -2
View File
@@ -70,9 +70,12 @@ public class BaseConf {
public static String initTopic() {
String topic = MQRandomUtils.getRandomTopic();
IntegrationTestBase.initTopic(topic, nsAddr, clusterName);
return initTopicWithName(topic);
}
return topic;
public static String initTopicWithName(String topicName) {
IntegrationTestBase.initTopic(topicName, nsAddr, clusterName);
return topicName;
}
public static String initConsumerGroup() {
test/src/test/java/org/apache/rocketmq/test/client/producer/querymsg/QueryMsgByKeyIT.java
+57
View File
@@ -18,7 +18,9 @@
package org.apache.rocketmq.test.client.producer.querymsg;
import java.util.List;
import org.apache.log4j.Logger;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.test.base.BaseConf;
import org.apache.rocketmq.test.client.rmq.RMQNormalProducer;
@@ -101,4 +103,59 @@ public class QueryMsgByKeyIT extends BaseConf {
assertThat(queryMsgs).isNotNull();
assertThat(queryMsgs.size()).isEqualTo(max);
}
@Test(expected = MQClientException.class)
public void testQueryMsgWithSameHash1() throws Exception {
int msgSize = 1;
String topicA = "AaTopic";
String keyA = "Aa";
String topicB = "BBTopic";
String keyB = "BB";
initTopicWithName(topicA);
initTopicWithName(topicB);
RMQNormalProducer producerA = getProducer(nsAddr, topicA);
RMQNormalProducer producerB = getProducer(nsAddr, topicB);
List<Object> msgA = MQMessageFactory.getKeyMsg(topicA, keyA, msgSize);
List<Object> msgB = MQMessageFactory.getKeyMsg(topicB, keyB, msgSize);
producerA.send(msgA);
producerB.send(msgB);
long begin = System.currentTimeMillis() - 500000;
long end = System.currentTimeMillis() + 500000;
producerA.getProducer().queryMessage(topicA, keyB, msgSize * 10, begin, end).getMessageList();
}
@Test
public void testQueryMsgWithSameHash2() throws Exception {
int msgSize = 1;
String topicA = "AaAaTopic";
String keyA = "Aa";
String topicB = "BBBBTopic";
String keyB = "Aa";
initTopicWithName(topicA);
initTopicWithName(topicB);
RMQNormalProducer producerA = getProducer(nsAddr, topicA);
RMQNormalProducer producerB = getProducer(nsAddr, topicB);
List<Object> msgA = MQMessageFactory.getKeyMsg(topicA, keyA, msgSize);
List<Object> msgB = MQMessageFactory.getKeyMsg(topicB, keyB, msgSize);
producerA.send(msgA);
producerB.send(msgB);
long begin = System.currentTimeMillis() - 500000;
long end = System.currentTimeMillis() + 500000;
List<MessageExt> list = producerA.getProducer().queryMessage(topicA, keyA, msgSize * 10, begin, end).getMessageList();
assertThat(list).isNotNull();
assertThat(list.size()).isEqualTo(1);
}
}