Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
raphaelmichael
Rocketmq
提交
6a9628b3
R
Rocketmq
项目概览
raphaelmichael
/
Rocketmq
与 Fork 源项目一致
Fork自
Apache RocketMQ / Rocketmq
通知
2
Star
0
Fork
0
代码
文件
提交
分支
Tags
贡献者
分支图
Diff
Issue
0
列表
看板
标记
里程碑
合并请求
0
DevOps
流水线
流水线任务
计划
Wiki
0
Wiki
分析
仓库
DevOps
项目成员
Pages
R
Rocketmq
项目概览
项目概览
详情
发布
仓库
仓库
文件
提交
分支
标签
贡献者
分支图
比较
Issue
0
Issue
0
列表
看板
标记
里程碑
合并请求
0
合并请求
0
Pages
DevOps
DevOps
流水线
流水线任务
计划
分析
分析
仓库分析
DevOps
Wiki
0
Wiki
成员
成员
收起侧边栏
关闭侧边栏
动态
分支图
创建新Issue
流水线任务
提交
Issue看板
体验新版 GitCode,发现更多精彩内容 >>
提交
6a9628b3
编写于
4月 21, 2017
作者:
V
vsair
提交者:
dongeforever
4月 21, 2017
浏览文件
操作
浏览文件
下载
电子邮件补丁
差异文件
[ROCKETMQ-179] Fix errors of IT test cases closes apache/incubator-rocketmq#94
上级
58f1574b
变更
9
隐藏空白更改
内联
并排
Showing
9 changed file
with
53 addition
and
34 deletion
+53
-34
pom.xml
pom.xml
+0
-4
test/src/main/java/org/apache/rocketmq/test/clientinterface/MQCollector.java
...org/apache/rocketmq/test/clientinterface/MQCollector.java
+15
-5
test/src/main/java/org/apache/rocketmq/test/listener/rmq/concurrent/RMQNormalListner.java
...cketmq/test/listener/rmq/concurrent/RMQNormalListner.java
+3
-1
test/src/main/java/org/apache/rocketmq/test/util/MQAdmin.java
.../src/main/java/org/apache/rocketmq/test/util/MQAdmin.java
+0
-1
test/src/test/java/org/apache/rocketmq/test/base/IntegrationTestBase.java
...va/org/apache/rocketmq/test/base/IntegrationTestBase.java
+28
-11
test/src/test/java/org/apache/rocketmq/test/client/consumer/tag/TagMessageWithSameGroupConsumerIT.java
...lient/consumer/tag/TagMessageWithSameGroupConsumerIT.java
+4
-6
test/src/test/java/org/apache/rocketmq/test/client/producer/async/AsyncSendWithMessageQueueIT.java
...st/client/producer/async/AsyncSendWithMessageQueueIT.java
+2
-3
test/src/test/java/org/apache/rocketmq/test/client/producer/async/AsyncSendWithMessageQueueSelectorIT.java
...t/producer/async/AsyncSendWithMessageQueueSelectorIT.java
+0
-1
test/src/test/java/org/apache/rocketmq/test/delay/NormalMsgDelayIT.java
...java/org/apache/rocketmq/test/delay/NormalMsgDelayIT.java
+1
-2
未找到文件。
pom.xml
浏览文件 @
6a9628b3
...
...
@@ -461,10 +461,6 @@
<argLine>
@{failsafeArgLine}
</argLine>
<excludes>
<exclude>
**/NormalMsgDelayIT.java
</exclude>
<exclude>
**/BroadCastNormalMsgNotRecvIT.java
</exclude>
<exclude>
**/TagMessageWithSameGroupConsumerIT.java
</exclude>
<exclude>
**/AsyncSendWithMessageQueueSelectorIT.java
</exclude>
<exclude>
**/AsyncSendWithMessageQueueIT.java
</exclude>
</excludes>
</configuration>
<executions>
...
...
test/src/main/java/org/apache/rocketmq/test/clientinterface/MQCollector.java
浏览文件 @
6a9628b3
...
...
@@ -91,11 +91,21 @@ public abstract class MQCollector {
}
public
void
clearMsg
()
{
msgBodys
.
resetData
();
originMsgs
.
resetData
();
errorMsgs
.
resetData
();
originMsgIndex
.
clear
();
msgRTs
.
resetData
();
if
(
msgBodys
!=
null
)
{
msgBodys
.
resetData
();
}
if
(
originMsgs
!=
null
)
{
originMsgs
.
resetData
();
}
if
(
originMsgs
!=
null
)
{
errorMsgs
.
resetData
();
}
if
(
originMsgIndex
!=
null
)
{
originMsgIndex
.
clear
();
}
if
(
msgRTs
!=
null
)
{
msgRTs
.
resetData
();
}
}
public
void
lockCollectors
()
{
...
...
test/src/main/java/org/apache/rocketmq/test/listener/rmq/concurrent/RMQNormalListner.java
浏览文件 @
6a9628b3
...
...
@@ -63,7 +63,9 @@ public class RMQNormalListner extends AbstractListener implements MessageListene
msgBodys
.
addData
(
new
String
(
msg
.
getBody
()));
originMsgs
.
addData
(
msg
);
originMsgIndex
.
put
(
new
String
(
msg
.
getBody
()),
msg
);
if
(
originMsgIndex
!=
null
)
{
originMsgIndex
.
put
(
new
String
(
msg
.
getBody
()),
msg
);
}
}
return
consumeStatus
;
}
...
...
test/src/main/java/org/apache/rocketmq/test/util/MQAdmin.java
浏览文件 @
6a9628b3
...
...
@@ -45,7 +45,6 @@ public class MQAdmin {
mqAdminExt
.
start
();
mqAdminExt
.
createTopic
(
clusterName
,
topic
,
queueNum
);
}
catch
(
Exception
e
)
{
e
.
printStackTrace
();
}
long
startTime
=
System
.
currentTimeMillis
();
...
...
test/src/test/java/org/apache/rocketmq/test/base/IntegrationTestBase.java
浏览文件 @
6a9628b3
...
...
@@ -46,6 +46,8 @@ public class IntegrationTestBase {
protected
static
final
List
<
BrokerController
>
BROKER_CONTROLLERS
=
new
ArrayList
<>();
protected
static
final
List
<
NamesrvController
>
NAMESRV_CONTROLLERS
=
new
ArrayList
<>();
protected
static
int
topicCreateTime
=
30
*
1000
;
protected
static
final
int
COMMIT_LOG_SIZE
=
1024
*
1024
*
256
;
protected
static
final
int
INDEX_NUM
=
1000
;
protected
static
Random
random
=
new
Random
();
...
...
@@ -53,18 +55,30 @@ public class IntegrationTestBase {
Runtime
.
getRuntime
().
addShutdownHook
(
new
Thread
()
{
@Override
public
void
run
()
{
for
(
NamesrvController
namesrvController
:
NAMESRV_CONTROLLERS
)
{
if
(
namesrvController
!=
null
)
{
namesrvController
.
shutdown
();
try
{
for
(
BrokerController
brokerController
:
BROKER_CONTROLLERS
)
{
if
(
brokerController
!=
null
)
{
brokerController
.
shutdown
();
}
}
}
for
(
BrokerController
brokerController
:
BROKER_CONTROLLERS
)
{
if
(
brokerController
!=
null
)
{
brokerController
.
shutdown
();
// should destroy message store, otherwise could not delete the temp files.
for
(
BrokerController
brokerController
:
BROKER_CONTROLLERS
)
{
if
(
brokerController
!=
null
)
{
brokerController
.
getMessageStore
().
destroy
();
}
}
}
for
(
File
file
:
TMPE_FILES
)
{
deleteFile
(
file
);
for
(
NamesrvController
namesrvController
:
NAMESRV_CONTROLLERS
)
{
if
(
namesrvController
!=
null
)
{
namesrvController
.
shutdown
();
}
}
for
(
File
file
:
TMPE_FILES
)
{
deleteFile
(
file
);
}
}
catch
(
Exception
e
){
logger
.
error
(
"Shutdown error"
,
e
);
}
}
});
...
...
@@ -75,7 +89,7 @@ public class IntegrationTestBase {
String
baseDir
=
System
.
getProperty
(
"user.home"
)
+
SEP
+
"unitteststore-"
+
UUID
.
randomUUID
();
final
File
file
=
new
File
(
baseDir
);
if
(
file
.
exists
())
{
logger
.
info
(
String
.
format
(
"[%s] has already existed, please ba
ke
up and remove it for integration tests"
,
baseDir
));
logger
.
info
(
String
.
format
(
"[%s] has already existed, please ba
ck
up and remove it for integration tests"
,
baseDir
));
System
.
exit
(
1
);
}
TMPE_FILES
.
add
(
file
);
...
...
@@ -116,6 +130,9 @@ public class IntegrationTestBase {
storeConfig
.
setStorePathRootDir
(
baseDir
);
storeConfig
.
setStorePathCommitLog
(
baseDir
+
SEP
+
"commitlog"
);
storeConfig
.
setHaListenPort
(
8000
+
random
.
nextInt
(
1000
));
storeConfig
.
setMapedFileSizeCommitLog
(
COMMIT_LOG_SIZE
);
storeConfig
.
setMaxIndexNum
(
INDEX_NUM
);
storeConfig
.
setMaxHashSlotNum
(
INDEX_NUM
*
4
);
nettyServerConfig
.
setListenPort
(
10000
+
random
.
nextInt
(
1000
));
BrokerController
brokerController
=
new
BrokerController
(
brokerConfig
,
nettyServerConfig
,
nettyClientConfig
,
storeConfig
);
try
{
...
...
test/src/test/java/org/apache/rocketmq/test/client/consumer/tag/TagMessageWithSameGroupConsumerIT.java
浏览文件 @
6a9628b3
...
...
@@ -36,6 +36,7 @@ public class TagMessageWithSameGroupConsumerIT extends BaseConf {
private
static
Logger
logger
=
Logger
.
getLogger
(
TagMessageWith1ConsumerIT
.
class
);
private
RMQNormalProducer
producer
=
null
;
private
String
topic
=
null
;
private
String
tag
=
"tag"
;
@Before
public
void
setUp
()
{
...
...
@@ -51,13 +52,12 @@ public class TagMessageWithSameGroupConsumerIT extends BaseConf {
@Test
public
void
testTwoConsumerWithSameGroup
()
{
String
tag
=
"jueyin"
;
int
msgSize
=
20
;
String
originMsgDCName
=
RandomUtils
.
getStringByUUID
();
String
msgBodyDCName
=
RandomUtils
.
getStringByUUID
();
RMQNormalConsumer
consumer1
=
getConsumer
(
nsAddr
,
topic
,
tag
,
new
RMQNormalListner
(
originMsgDCName
,
msgBodyDCName
));
RMQNormalConsumer
consumer2
=
getConsumer
(
nsAddr
,
consumer1
.
getConsumerGroup
(),
tag
,
getConsumer
(
nsAddr
,
consumer1
.
getConsumerGroup
(),
tag
,
new
RMQNormalListner
(
originMsgDCName
,
msgBodyDCName
));
producer
.
send
(
tag
,
msgSize
);
Assert
.
assertEquals
(
"Not all are sent"
,
msgSize
,
producer
.
getAllUndupMsgBody
().
size
());
...
...
@@ -70,7 +70,6 @@ public class TagMessageWithSameGroupConsumerIT extends BaseConf {
@Test
public
void
testConsumerStartWithInterval
()
{
String
tag
=
"jueyin"
;
int
msgSize
=
100
;
String
originMsgDCName
=
RandomUtils
.
getStringByUUID
();
String
msgBodyDCName
=
RandomUtils
.
getStringByUUID
();
...
...
@@ -79,7 +78,7 @@ public class TagMessageWithSameGroupConsumerIT extends BaseConf {
new
RMQNormalListner
(
originMsgDCName
,
msgBodyDCName
));
producer
.
send
(
tag
,
msgSize
,
100
);
TestUtils
.
waitForMoment
(
5
);
RMQNormalConsumer
consumer2
=
getConsumer
(
nsAddr
,
consumer1
.
getConsumerGroup
(),
tag
,
getConsumer
(
nsAddr
,
consumer1
.
getConsumerGroup
(),
tag
,
new
RMQNormalListner
(
originMsgDCName
,
msgBodyDCName
));
TestUtils
.
waitForMoment
(
5
);
...
...
@@ -90,8 +89,7 @@ public class TagMessageWithSameGroupConsumerIT extends BaseConf {
}
@Test
public
void
testConsumerStartTwoAndCrashOnsAfterWhile
()
{
String
tag
=
"jueyin"
;
public
void
testConsumerStartTwoAndCrashOneAfterWhile
()
{
int
msgSize
=
100
;
String
originMsgDCName
=
RandomUtils
.
getStringByUUID
();
String
msgBodyDCName
=
RandomUtils
.
getStringByUUID
();
...
...
test/src/test/java/org/apache/rocketmq/test/client/producer/async/AsyncSendWithMessageQueueIT.java
浏览文件 @
6a9628b3
...
...
@@ -33,7 +33,6 @@ import static com.google.common.truth.Truth.assertThat;
public
class
AsyncSendWithMessageQueueIT
extends
BaseConf
{
private
static
Logger
logger
=
Logger
.
getLogger
(
TagMessageWith1ConsumerIT
.
class
);
private
static
boolean
sendFail
=
false
;
private
RMQAsyncSendProducer
producer
=
null
;
private
String
topic
=
null
;
...
...
@@ -57,7 +56,7 @@ public class AsyncSendWithMessageQueueIT extends BaseConf {
MessageQueue
mq
=
new
MessageQueue
(
topic
,
broker1Name
,
queueId
);
producer
.
asyncSend
(
msgSize
,
mq
);
producer
.
waitForResponse
(
5
*
1000
);
producer
.
waitForResponse
(
10
*
1000
);
assertThat
(
producer
.
getSuccessMsgCount
()).
isEqualTo
(
msgSize
);
consumer
.
getListner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
...
...
@@ -72,7 +71,7 @@ public class AsyncSendWithMessageQueueIT extends BaseConf {
mq
=
new
MessageQueue
(
topic
,
broker2Name
,
queueId
);
producer
.
asyncSend
(
msgSize
,
mq
);
producer
.
waitForResponse
(
5
*
1000
);
producer
.
waitForResponse
(
10
*
1000
);
assertThat
(
producer
.
getSuccessMsgCount
()).
isEqualTo
(
msgSize
);
consumer
.
getListner
().
waitForMessageConsume
(
producer
.
getAllMsgBody
(),
consumeTime
);
...
...
test/src/test/java/org/apache/rocketmq/test/client/producer/async/AsyncSendWithMessageQueueSelectorIT.java
浏览文件 @
6a9628b3
...
...
@@ -36,7 +36,6 @@ import static com.google.common.truth.Truth.assertThat;
public
class
AsyncSendWithMessageQueueSelectorIT
extends
BaseConf
{
private
static
Logger
logger
=
Logger
.
getLogger
(
TagMessageWith1ConsumerIT
.
class
);
private
static
boolean
sendFail
=
false
;
private
RMQAsyncSendProducer
producer
=
null
;
private
String
topic
=
null
;
...
...
test/src/test/java/org/apache/rocketmq/test/delay/NormalMsgDelayIT.java
浏览文件 @
6a9628b3
...
...
@@ -24,7 +24,6 @@ import org.apache.rocketmq.test.client.rmq.RMQNormalConsumer;
import
org.apache.rocketmq.test.client.rmq.RMQNormalProducer
;
import
org.apache.rocketmq.test.factory.MQMessageFactory
;
import
org.apache.rocketmq.test.listener.rmq.concurrent.RMQDelayListner
;
import
org.apache.rocketmq.test.listener.rmq.order.RMQOrderListener
;
import
org.apache.rocketmq.test.util.VerifyUtils
;
import
org.junit.After
;
import
org.junit.Assert
;
...
...
@@ -43,7 +42,7 @@ public class NormalMsgDelayIT extends DelayConf {
topic
=
initTopic
();
logger
.
info
(
String
.
format
(
"use topic: %s;"
,
topic
));
producer
=
getProducer
(
nsAddr
,
topic
);
consumer
=
getConsumer
(
nsAddr
,
topic
,
"*"
,
new
RMQ
OrderListe
ner
());
consumer
=
getConsumer
(
nsAddr
,
topic
,
"*"
,
new
RMQ
DelayList
ner
());
}
@After
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录