Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
小问号你是否有许多朋友
Rocketmq
提交
76464ba6
R
Rocketmq
项目概览
小问号你是否有许多朋友
/
Rocketmq
与 Fork 源项目一致
Fork自
Apache RocketMQ / Rocketmq
通知
1
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,发现更多精彩内容 >>
提交
76464ba6
编写于
8月 28, 2017
作者:
L
lindzh
提交者:
vongosling
8月 28, 2017
浏览文件
操作
浏览文件
下载
电子邮件补丁
差异文件
ROCKETMQ-264 Fix ut's time-consuming problems closes apache/incubator-rocketmq#145
上级
629c3e9e
变更
10
隐藏空白更改
内联
并排
Showing
10 changed file
with
272 addition
and
306 deletion
+272
-306
broker/src/test/java/org/apache/rocketmq/broker/BrokerControllerTest.java
...java/org/apache/rocketmq/broker/BrokerControllerTest.java
+8
-10
broker/src/test/java/org/apache/rocketmq/broker/filter/MessageStoreWithFilterTest.java
...he/rocketmq/broker/filter/MessageStoreWithFilterTest.java
+125
-131
broker/src/test/resources/logback-test.xml
broker/src/test/resources/logback-test.xml
+0
-1
client/src/main/java/org/apache/rocketmq/client/log/ClientLogger.java
...ain/java/org/apache/rocketmq/client/log/ClientLogger.java
+12
-16
client/src/test/java/org/apache/rocketmq/client/consumer/rebalance/AllocateMessageQueueConsitentHashTest.java
...umer/rebalance/AllocateMessageQueueConsitentHashTest.java
+3
-3
client/src/test/java/org/apache/rocketmq/client/log/ClientLogTest.java
...st/java/org/apache/rocketmq/client/log/ClientLogTest.java
+0
-72
client/src/test/java/org/apache/rocketmq/client/log/ClientLoggerTest.java
...java/org/apache/rocketmq/client/log/ClientLoggerTest.java
+51
-0
filter/src/test/java/org/apache/rocketmq/filter/ExpressionTest.java
.../test/java/org/apache/rocketmq/filter/ExpressionTest.java
+33
-35
namesrv/src/test/java/org/apache/rocketmq/namesrv/kvconfig/KVConfigManagerTest.java
...apache/rocketmq/namesrv/kvconfig/KVConfigManagerTest.java
+0
-1
store/src/test/java/org/apache/rocketmq/store/DefaultMessageStoreTest.java
...va/org/apache/rocketmq/store/DefaultMessageStoreTest.java
+40
-37
未找到文件。
broker/src/test/java/org/apache/rocketmq/broker/BrokerControllerTest.java
浏览文件 @
76464ba6
...
...
@@ -37,16 +37,14 @@ public class BrokerControllerTest {
*/
@Test
public
void
testBrokerRestart
()
throws
Exception
{
for
(
int
i
=
0
;
i
<
2
;
i
++)
{
BrokerController
brokerController
=
new
BrokerController
(
new
BrokerConfig
(),
new
NettyServerConfig
(),
new
NettyClientConfig
(),
new
MessageStoreConfig
());
assertThat
(
brokerController
.
initialize
());
brokerController
.
start
();
brokerController
.
shutdown
();
}
BrokerController
brokerController
=
new
BrokerController
(
new
BrokerConfig
(),
new
NettyServerConfig
(),
new
NettyClientConfig
(),
new
MessageStoreConfig
());
assertThat
(
brokerController
.
initialize
());
brokerController
.
start
();
brokerController
.
shutdown
();
}
@After
...
...
broker/src/test/java/org/apache/rocketmq/broker/filter/MessageStoreWithFilterTest.java
浏览文件 @
76464ba6
...
...
@@ -33,6 +33,8 @@ import org.apache.rocketmq.store.MessageExtBrokerInner;
import
org.apache.rocketmq.store.PutMessageResult
;
import
org.apache.rocketmq.store.config.MessageStoreConfig
;
import
org.apache.rocketmq.store.stats.BrokerStatsManager
;
import
org.junit.After
;
import
org.junit.Before
;
import
org.junit.Test
;
import
java.io.File
;
...
...
@@ -63,6 +65,14 @@ public class MessageStoreWithFilterTest {
private
static
SocketAddress
StoreHost
;
private
DefaultMessageStore
master
;
private
ConsumerFilterManager
filterManager
;
private
int
topicCount
=
3
;
private
int
msgPerTopic
=
30
;
static
{
try
{
StoreHost
=
new
InetSocketAddress
(
InetAddress
.
getLocalHost
(),
8123
);
...
...
@@ -76,6 +86,24 @@ public class MessageStoreWithFilterTest {
}
}
@Before
public
void
init
()
{
filterManager
=
ConsumerFilterManagerTest
.
gen
(
topicCount
,
msgPerTopic
);
try
{
master
=
gen
(
filterManager
);
}
catch
(
Exception
e
)
{
e
.
printStackTrace
();
assertThat
(
true
).
isFalse
();
}
}
@After
public
void
destroy
()
{
master
.
shutdown
();
master
.
destroy
();
UtilAll
.
deleteFile
(
new
File
(
storePath
));
}
public
MessageExtBrokerInner
buildMessage
()
{
MessageExtBrokerInner
msg
=
new
MessageExtBrokerInner
();
msg
.
setTopic
(
topic
);
...
...
@@ -202,177 +230,143 @@ public class MessageStoreWithFilterTest {
@Test
public
void
testGetMessage_withFilterBitMapAndConsumerChanged
()
{
int
topicCount
=
10
,
msgPerTopic
=
10
;
ConsumerFilterManager
filterManager
=
ConsumerFilterManagerTest
.
gen
(
topicCount
,
msgPerTopic
);
DefaultMessageStore
master
=
null
;
List
<
MessageExtBrokerInner
>
msgs
=
null
;
try
{
m
aster
=
gen
(
filterManager
);
m
sgs
=
putMsg
(
master
,
topicCount
,
msgPerTopic
);
}
catch
(
Exception
e
)
{
e
.
printStackTrace
();
assertThat
(
true
).
isFalse
();
}
// sleep to wait for consume queue has been constructed.
try
{
List
<
MessageExtBrokerInner
>
msgs
=
null
;
try
{
msgs
=
putMsg
(
master
,
topicCount
,
msgPerTopic
);
}
catch
(
Exception
e
)
{
e
.
printStackTrace
();
assertThat
(
true
).
isFalse
();
}
// sleep to wait for consume queue has been constructed.
try
{
Thread
.
sleep
(
1000
);
}
catch
(
InterruptedException
e
)
{
e
.
printStackTrace
();
assertThat
(
true
).
isFalse
();
}
Thread
.
sleep
(
200
);
}
catch
(
InterruptedException
e
)
{
e
.
printStackTrace
();
assertThat
(
true
).
isFalse
();
}
// reset consumer;
String
topic
=
"topic"
+
0
;
String
resetGroup
=
"CID_"
+
2
;
String
normalGroup
=
"CID_"
+
3
;
// reset consumer;
String
topic
=
"topic"
+
0
;
String
resetGroup
=
"CID_"
+
2
;
String
normalGroup
=
"CID_"
+
3
;
{
// reset CID_2@topic0 to get all messages.
SubscriptionData
resetSubData
=
new
SubscriptionData
();
resetSubData
.
setExpressionType
(
ExpressionType
.
SQL92
);
resetSubData
.
setTopic
(
topic
);
resetSubData
.
setClassFilterMode
(
false
);
resetSubData
.
setSubString
(
"a is not null OR a is null"
);
{
// reset CID_2@topic0 to get all messages.
SubscriptionData
resetSubData
=
new
SubscriptionData
();
resetSubData
.
setExpressionType
(
ExpressionType
.
SQL92
);
resetSubData
.
setTopic
(
topic
);
resetSubData
.
setClassFilterMode
(
false
);
resetSubData
.
setSubString
(
"a is not null OR a is null"
);
ConsumerFilterData
resetFilterData
=
ConsumerFilterManager
.
build
(
topic
,
resetGroup
,
resetSubData
.
getSubString
(),
resetSubData
.
getExpressionType
(),
System
.
currentTimeMillis
());
ConsumerFilterData
resetFilterData
=
ConsumerFilterManager
.
build
(
topic
,
resetGroup
,
resetSubData
.
getSubString
(),
resetSubData
.
getExpressionType
(),
System
.
currentTimeMillis
());
GetMessageResult
resetGetResult
=
master
.
getMessage
(
resetGroup
,
topic
,
queueId
,
0
,
1000
,
new
ExpressionMessageFilter
(
resetSubData
,
resetFilterData
,
filterManager
));
GetMessageResult
resetGetResult
=
master
.
getMessage
(
resetGroup
,
topic
,
queueId
,
0
,
1000
,
new
ExpressionMessageFilter
(
resetSubData
,
resetFilterData
,
filterManager
));
try
{
assertThat
(
resetGetResult
).
isNotNull
();
try
{
assertThat
(
resetGetResult
).
isNotNull
();
List
<
MessageExtBrokerInner
>
filteredMsgs
=
filtered
(
msgs
,
resetFilterData
);
List
<
MessageExtBrokerInner
>
filteredMsgs
=
filtered
(
msgs
,
resetFilterData
);
assertThat
(
resetGetResult
.
getMessageBufferList
().
size
()).
isEqualTo
(
filteredMsgs
.
size
());
}
finally
{
resetGetResult
.
release
();
}
assertThat
(
resetGetResult
.
getMessageBufferList
().
size
()).
isEqualTo
(
filteredMsgs
.
size
());
}
finally
{
resetGetResult
.
release
();
}
}
{
ConsumerFilterData
normalFilterData
=
filterManager
.
get
(
topic
,
normalGroup
);
assertThat
(
normalFilterData
).
isNotNull
();
assertThat
(
normalFilterData
.
getBornTime
()).
isLessThan
(
System
.
currentTimeMillis
());
{
ConsumerFilterData
normalFilterData
=
filterManager
.
get
(
topic
,
normalGroup
);
assertThat
(
normalFilterData
).
isNotNull
();
assertThat
(
normalFilterData
.
getBornTime
()).
isLessThan
(
System
.
currentTimeMillis
());
SubscriptionData
normalSubData
=
new
SubscriptionData
();
normalSubData
.
setExpressionType
(
normalFilterData
.
getExpressionType
());
normalSubData
.
setTopic
(
topic
);
normalSubData
.
setClassFilterMode
(
false
);
normalSubData
.
setSubString
(
normalFilterData
.
getExpression
());
SubscriptionData
normalSubData
=
new
SubscriptionData
();
normalSubData
.
setExpressionType
(
normalFilterData
.
getExpressionType
());
normalSubData
.
setTopic
(
topic
);
normalSubData
.
setClassFilterMode
(
false
);
normalSubData
.
setSubString
(
normalFilterData
.
getExpression
());
List
<
MessageExtBrokerInner
>
filteredMsgs
=
filtered
(
msgs
,
normalFilterData
);
List
<
MessageExtBrokerInner
>
filteredMsgs
=
filtered
(
msgs
,
normalFilterData
);
GetMessageResult
normalGetResult
=
master
.
getMessage
(
normalGroup
,
topic
,
queueId
,
0
,
1000
,
new
ExpressionMessageFilter
(
normalSubData
,
normalFilterData
,
filterManager
));
GetMessageResult
normalGetResult
=
master
.
getMessage
(
normalGroup
,
topic
,
queueId
,
0
,
1000
,
new
ExpressionMessageFilter
(
normalSubData
,
normalFilterData
,
filterManager
));
try
{
assertThat
(
normalGetResult
).
isNotNull
();
assertThat
(
normalGetResult
.
getMessageBufferList
().
size
()).
isEqualTo
(
filteredMsgs
.
size
());
}
finally
{
normalGetResult
.
release
();
}
try
{
assertThat
(
normalGetResult
).
isNotNull
();
assertThat
(
normalGetResult
.
getMessageBufferList
().
size
()).
isEqualTo
(
filteredMsgs
.
size
());
}
finally
{
normalGetResult
.
release
();
}
}
finally
{
master
.
shutdown
();
master
.
destroy
();
UtilAll
.
deleteFile
(
new
File
(
storePath
));
}
}
@Test
public
void
testGetMessage_withFilterBitMap
()
{
int
topicCount
=
10
,
msgPerTopic
=
500
;
ConsumerFilterManager
filterManager
=
ConsumerFilterManagerTest
.
gen
(
topicCount
,
msgPerTopic
);
DefaultMessageStore
master
=
null
;
List
<
MessageExtBrokerInner
>
msgs
=
null
;
try
{
master
=
gen
(
filterManager
);
msgs
=
putMsg
(
master
,
topicCount
,
msgPerTopic
);
// sleep to wait for consume queue has been constructed.
Thread
.
sleep
(
200
);
}
catch
(
Exception
e
)
{
e
.
printStackTrace
();
assertThat
(
true
).
isFalse
();
}
try
{
List
<
MessageExtBrokerInner
>
msgs
=
null
;
try
{
msgs
=
putMsg
(
master
,
topicCount
,
msgPerTopic
);
// sleep to wait for consume queue has been constructed.
Thread
.
sleep
(
1000
);
}
catch
(
Exception
e
)
{
e
.
printStackTrace
();
assertThat
(
true
).
isFalse
();
}
for
(
int
i
=
0
;
i
<
topicCount
;
i
++)
{
String
realTopic
=
topic
+
i
;
for
(
int
i
=
0
;
i
<
topicCount
;
i
++)
{
String
realTopic
=
topic
+
i
;
for
(
int
j
=
0
;
j
<
msgPerTopic
;
j
++)
{
String
group
=
"CID_"
+
j
;
ConsumerFilterData
filterData
=
filterManager
.
get
(
realTopic
,
group
);
assertThat
(
filterData
).
isNotNull
();
List
<
MessageExtBrokerInner
>
filteredMsgs
=
filtered
(
msgs
,
filterData
);
SubscriptionData
subscriptionData
=
new
SubscriptionData
();
subscriptionData
.
setExpressionType
(
filterData
.
getExpressionType
());
subscriptionData
.
setTopic
(
filterData
.
getTopic
());
subscriptionData
.
setClassFilterMode
(
false
);
subscriptionData
.
setSubString
(
filterData
.
getExpression
());
GetMessageResult
getMessageResult
=
master
.
getMessage
(
group
,
realTopic
,
queueId
,
0
,
10000
,
new
ExpressionMessageFilter
(
subscriptionData
,
filterData
,
filterManager
));
String
assertMsg
=
group
+
"-"
+
realTopic
;
try
{
assertThat
(
getMessageResult
).
isNotNull
();
assertThat
(
GetMessageStatus
.
FOUND
).
isEqualTo
(
getMessageResult
.
getStatus
());
assertThat
(
getMessageResult
.
getMessageBufferList
()).
isNotNull
().
isNotEmpty
();
assertThat
(
getMessageResult
.
getMessageBufferList
().
size
()).
isEqualTo
(
filteredMsgs
.
size
());
for
(
ByteBuffer
buffer
:
getMessageResult
.
getMessageBufferList
())
{
MessageExt
messageExt
=
MessageDecoder
.
decode
(
buffer
.
slice
(),
false
);
assertThat
(
messageExt
).
isNotNull
();
Object
evlRet
=
null
;
try
{
evlRet
=
filterData
.
getCompiledExpression
().
evaluate
(
new
MessageEvaluationContext
(
messageExt
.
getProperties
()));
}
catch
(
Exception
e
)
{
e
.
printStackTrace
();
assertThat
(
true
).
isFalse
();
}
for
(
int
j
=
0
;
j
<
msgPerTopic
;
j
++)
{
String
group
=
"CID_"
+
j
;
assertThat
(
evlRet
).
isNotNull
().
isEqualTo
(
Boolean
.
TRUE
);
ConsumerFilterData
filterData
=
filterManager
.
get
(
realTopic
,
group
);
assertThat
(
filterData
).
isNotNull
();
// check
boolean
find
=
false
;
for
(
MessageExtBrokerInner
messageExtBrokerInner
:
filteredMsgs
)
{
if
(
messageExtBrokerInner
.
getMsgId
().
equals
(
messageExt
.
getMsgId
()))
{
find
=
true
;
}
List
<
MessageExtBrokerInner
>
filteredMsgs
=
filtered
(
msgs
,
filterData
);
SubscriptionData
subscriptionData
=
new
SubscriptionData
();
subscriptionData
.
setExpressionType
(
filterData
.
getExpressionType
());
subscriptionData
.
setTopic
(
filterData
.
getTopic
());
subscriptionData
.
setClassFilterMode
(
false
);
subscriptionData
.
setSubString
(
filterData
.
getExpression
());
GetMessageResult
getMessageResult
=
master
.
getMessage
(
group
,
realTopic
,
queueId
,
0
,
10000
,
new
ExpressionMessageFilter
(
subscriptionData
,
filterData
,
filterManager
));
String
assertMsg
=
group
+
"-"
+
realTopic
;
try
{
assertThat
(
getMessageResult
).
isNotNull
();
assertThat
(
GetMessageStatus
.
FOUND
).
isEqualTo
(
getMessageResult
.
getStatus
());
assertThat
(
getMessageResult
.
getMessageBufferList
()).
isNotNull
().
isNotEmpty
();
assertThat
(
getMessageResult
.
getMessageBufferList
().
size
()).
isEqualTo
(
filteredMsgs
.
size
());
for
(
ByteBuffer
buffer
:
getMessageResult
.
getMessageBufferList
())
{
MessageExt
messageExt
=
MessageDecoder
.
decode
(
buffer
.
slice
(),
false
);
assertThat
(
messageExt
).
isNotNull
();
Object
evlRet
=
null
;
try
{
evlRet
=
filterData
.
getCompiledExpression
().
evaluate
(
new
MessageEvaluationContext
(
messageExt
.
getProperties
()));
}
catch
(
Exception
e
)
{
e
.
printStackTrace
();
assertThat
(
true
).
isFalse
();
}
assertThat
(
evlRet
).
isNotNull
().
isEqualTo
(
Boolean
.
TRUE
);
// check
boolean
find
=
false
;
for
(
MessageExtBrokerInner
messageExtBrokerInner
:
filteredMsgs
)
{
if
(
messageExtBrokerInner
.
getMsgId
().
equals
(
messageExt
.
getMsgId
()))
{
find
=
true
;
}
assertThat
(
find
).
isTrue
();
}
}
finally
{
getMessageResult
.
release
();
assertThat
(
find
).
isTrue
();
}
}
finally
{
getMessageResult
.
release
();
}
}
}
finally
{
master
.
shutdown
();
master
.
destroy
();
UtilAll
.
deleteFile
(
new
File
(
storePath
));
}
}
}
broker/src/test/resources/logback-test.xml
浏览文件 @
76464ba6
...
...
@@ -19,7 +19,6 @@
<configuration>
<appender
name=
"DefaultAppender"
class=
"ch.qos.logback.core.ConsoleAppender"
>
<append>
true
</append>
<encoder>
<pattern>
%d{yyy-MM-dd HH\:mm\:ss,GMT+8} %p %t - %m%n
</pattern>
<charset
class=
"java.nio.charset.Charset"
>
UTF-8
</charset>
...
...
client/src/main/java/org/apache/rocketmq/client/log/ClientLogger.java
浏览文件 @
76464ba6
...
...
@@ -16,14 +16,14 @@
*/
package
org.apache.rocketmq.client.log
;
import
java.lang.reflect.Method
;
import
java.net.URL
;
import
org.apache.rocketmq.common.constant.LoggerName
;
import
org.slf4j.ILoggerFactory
;
import
org.slf4j.Logger
;
import
org.slf4j.LoggerFactory
;
import
java.lang.reflect.Method
;
import
java.net.URL
;
public
class
ClientLogger
{
public
static
final
String
CLIENT_LOG_ROOT
=
"rocketmq.client.logRoot"
;
public
static
final
String
CLIENT_LOG_MAXINDEX
=
"rocketmq.client.logFileMaxIndex"
;
...
...
@@ -31,25 +31,21 @@ public class ClientLogger {
private
static
Logger
log
;
private
static
Class
logClass
=
null
;
private
static
Logger
createLogger
(
final
String
loggerName
)
{
String
logConfigFilePath
=
System
.
getProperty
(
"rocketmq.client.log.configFile"
,
System
.
getenv
(
"ROCKETMQ_CLIENT_LOG_CONFIGFILE"
));
String
logConfigFilePath
=
System
.
getProperty
(
"rocketmq.client.log.configFile"
,
System
.
getenv
(
"ROCKETMQ_CLIENT_LOG_CONFIGFILE"
));
Boolean
isloadconfig
=
Boolean
.
parseBoolean
(
System
.
getProperty
(
"rocketmq.client.log.loadconfig"
,
"true"
));
Boolean
.
parseBoolean
(
System
.
getProperty
(
"rocketmq.client.log.loadconfig"
,
"true"
));
final
String
log4JResourceFile
=
System
.
getProperty
(
"rocketmq.client.log4j.resource.fileName"
,
"log4j_rocketmq_client.xml"
);
System
.
getProperty
(
"rocketmq.client.log4j.resource.fileName"
,
"log4j_rocketmq_client.xml"
);
final
String
logbackResourceFile
=
System
.
getProperty
(
"rocketmq.client.logback.resource.fileName"
,
"logback_rocketmq_client.xml"
);
System
.
getProperty
(
"rocketmq.client.logback.resource.fileName"
,
"logback_rocketmq_client.xml"
);
final
String
log4J2ResourceFile
=
System
.
getProperty
(
"rocketmq.client.log4j2.resource.fileName"
,
"log4j2_rocketmq_client.xml"
);
System
.
getProperty
(
"rocketmq.client.log4j2.resource.fileName"
,
"log4j2_rocketmq_client.xml"
);
String
clientLogRoot
=
System
.
getProperty
(
CLIENT_LOG_ROOT
,
"${user.home}
/logs/rocketmqlogs"
);
String
clientLogRoot
=
System
.
getProperty
(
CLIENT_LOG_ROOT
,
System
.
getProperty
(
"user.home"
)
+
"
/logs/rocketmqlogs"
);
System
.
setProperty
(
"client.logRoot"
,
clientLogRoot
);
String
clientLogLevel
=
System
.
getProperty
(
CLIENT_LOG_LEVEL
,
"INFO"
);
System
.
setProperty
(
"client.logLevel"
,
clientLogLevel
);
...
...
@@ -85,11 +81,11 @@ public class ClientLogger {
if
(
null
==
logConfigFilePath
)
{
URL
url
=
ClientLogger
.
class
.
getClassLoader
().
getResource
(
logbackResourceFile
);
Method
doConfigure
=
joranConfiguratoroObj
.
getClass
().
getMethod
(
"doConfigure"
,
URL
.
class
);
joranConfiguratoroObj
.
getClass
().
getMethod
(
"doConfigure"
,
URL
.
class
);
doConfigure
.
invoke
(
joranConfiguratoroObj
,
url
);
}
else
{
Method
doConfigure
=
joranConfiguratoroObj
.
getClass
().
getMethod
(
"doConfigure"
,
String
.
class
);
joranConfiguratoroObj
.
getClass
().
getMethod
(
"doConfigure"
,
String
.
class
);
doConfigure
.
invoke
(
joranConfiguratoroObj
,
logConfigFilePath
);
}
...
...
@@ -102,7 +98,7 @@ public class ClientLogger {
initialize
.
invoke
(
joranConfigurator
,
"log4j2"
,
logConfigFilePath
);
}
}
logClass
=
classType
;
//
logClass = classType;
}
catch
(
Exception
e
)
{
System
.
err
.
println
(
e
);
}
...
...
client/src/test/java/org/apache/rocketmq/client/consumer/rebalance/AllocateMessageQueueConsitentHashTest.java
浏览文件 @
76464ba6
...
...
@@ -92,9 +92,9 @@ public class AllocateMessageQueueConsitentHashTest {
@Test
public
void
testRun100RandomCase
()
{
for
(
int
i
=
0
;
i
<
10
0
;
i
++)
{
int
consumerSize
=
new
Random
().
nextInt
(
20
0
)
+
1
;
//1-20
0
int
queueSize
=
new
Random
().
nextInt
(
100
)
+
1
;
//1-10
0
for
(
int
i
=
0
;
i
<
10
;
i
++)
{
int
consumerSize
=
new
Random
().
nextInt
(
20
)
+
1
;
//1-2
0
int
queueSize
=
new
Random
().
nextInt
(
20
)
+
1
;
//1-2
0
testAllocate
(
queueSize
,
consumerSize
);
try
{
Thread
.
sleep
(
1
);
...
...
client/src/test/java/org/apache/rocketmq/client/log/ClientLogTest.java
已删除
100644 → 0
浏览文件 @
629c3e9e
/*
* 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.log
;
import
org.junit.Assert
;
import
org.junit.Test
;
import
java.io.*
;
import
java.lang.reflect.Field
;
import
java.util.Date
;
public
class
ClientLogTest
{
public
static
final
String
CLIENT_LOG_ROOT
=
"rocketmq.client.logRoot"
;
public
static
final
String
LOG_DIR
;
static
{
LOG_DIR
=
System
.
getProperty
(
CLIENT_LOG_ROOT
,
"${user.home}/logs/rocketmqlogs"
);
}
// FIXME: Workarond for concret implementation for slf4j, is there any better solution for all slf4j implementations in one class ? 2017/8/1
@Test
public
void
testLog4j2
()
throws
IOException
,
NoSuchFieldException
,
IllegalAccessException
{
ClientLogger
.
getLog
();
long
seek
=
0
;
boolean
result
=
false
;
File
file
=
new
File
(
LOG_DIR
+
File
.
separator
+
"rocketmq_client.log"
);
if
(
file
.
exists
())
{
seek
=
file
.
length
();
}
Field
logClassField
=
ClientLogger
.
class
.
getDeclaredField
(
"logClass"
);
logClassField
.
setAccessible
(
true
);
Class
logClass
=
(
Class
)
logClassField
.
get
(
ClientLogger
.
class
);
Assert
.
assertEquals
(
"org.apache.logging.slf4j.Log4jLoggerFactory"
,
logClass
.
getName
());
for
(
int
i
=
0
;
i
<
10
;
i
++)
{
ClientLogger
.
getLog
().
info
(
"testcase testLog4j2 "
+
new
Date
());
}
RandomAccessFile
randomAccessFile
=
new
RandomAccessFile
(
file
,
"r"
);
randomAccessFile
.
seek
(
seek
);
String
line
=
randomAccessFile
.
readLine
();
int
idx
=
1
;
while
(
line
!=
null
)
{
if
(
line
.
contains
(
"testLog4j2"
))
{
result
=
true
;
break
;
}
line
=
randomAccessFile
.
readLine
();
idx
++;
if
(
idx
>
20
)
{
break
;
}
}
randomAccessFile
.
close
();
Assert
.
assertTrue
(
result
);
}
}
namesrv/src/test/java/org/apache/rocketmq/namesrv/NamesrvControll
erTest.java
→
client/src/test/java/org/apache/rocketmq/client/log/ClientLogg
erTest.java
浏览文件 @
76464ba6
...
...
@@ -14,33 +14,38 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package
org.apache.rocketmq.namesrv
;
import
org.apache.rocketmq.common.namesrv.NamesrvConfig
;
import
org.apache.rocketmq.remoting.netty.NettyServerConfig
;
package
org.apache.rocketmq.client.log
;
import
org.apache.rocketmq.common.UtilAll
;
import
org.junit.After
;
import
org.junit.Test
;
import
org.slf4j.Logger
;
import
java.io.File
;
import
static
org
.
junit
.
Assert
.
assertEquals
;
import
static
org
.
assertj
.
core
.
api
.
Assertions
.
assertThat
;
public
class
ClientLoggerTest
{
public
class
NamesrvControllerTest
{
private
final
static
int
RESTARTNUM
=
2
;
public
static
final
String
CLIENT_LOG_ROOT
=
"rocketmq.client.logRoot"
;
public
static
final
String
LOG_DIR
;
static
{
LOG_DIR
=
System
.
getProperty
(
CLIENT_LOG_ROOT
,
System
.
getProperty
(
"user.home"
)
+
"/logs/rocketmqlogs"
);
}
/**
* Tests if the controller can be properly stopped and started.
*
* @throws Exception If fails.
*/
@After
public
void
cleanFiles
()
{
UtilAll
.
deleteFile
(
new
File
(
LOG_DIR
));
}
// FIXME: Workaround for concrete implementation for slf4j, is there any better solution for all slf4j implementations in one class ? 2017/8/1
@Test
public
void
testRestart
()
throws
Exception
{
for
(
int
i
=
0
;
i
<
RESTARTNUM
;
i
++)
{
NamesrvController
namesrvController
=
new
NamesrvController
(
new
NamesrvConfig
(),
new
NettyServerConfig
()
);
boolean
initResult
=
namesrvController
.
initialize
();
assertThat
(
initResult
).
isEqualTo
(
true
);
namesrvController
.
start
();
namesrvController
.
shutdown
();
}
public
void
testLog4j
()
throws
Exception
{
Logger
logger
=
ClientLogger
.
getLog
();
assertEquals
(
"org.apache.logging.slf4j.Log4jLogger"
,
logger
.
getClass
().
getName
());
}
}
\ No newline at end of file
}
filter/src/test/java/org/apache/rocketmq/filter/ExpressionTest.java
浏览文件 @
76464ba6
...
...
@@ -47,7 +47,7 @@ public class ExpressionTest {
private
static
String
stringHasString
=
"TAGS is not null and TAGS='''''tag'''''"
;
@Test
public
void
testEvaluate_stringHasString
()
{
public
void
testEvaluate_stringHasString
()
throws
Exception
{
Expression
expr
=
genExp
(
stringHasString
);
EvaluationContext
context
=
genContext
(
...
...
@@ -58,7 +58,7 @@ public class ExpressionTest {
}
@Test
public
void
testEvaluate_now
()
{
public
void
testEvaluate_now
()
throws
Exception
{
EvaluationContext
context
=
genContext
(
KeyValue
.
c
(
"a"
,
System
.
currentTimeMillis
())
);
...
...
@@ -72,8 +72,8 @@ public class ExpressionTest {
eval
(
expression
,
context
,
Boolean
.
TRUE
);
}
@Test
public
void
testEvaluate_stringCompare
()
{
@Test
(
expected
=
RuntimeException
.
class
)
public
void
testEvaluate_stringCompare
()
throws
Exception
{
Expression
expression
=
genExp
(
"a between up and low"
);
EvaluationContext
context
=
genContext
(
...
...
@@ -147,7 +147,7 @@ public class ExpressionTest {
}
@Test
public
void
testEvaluate_exponent
()
{
public
void
testEvaluate_exponent
()
throws
Exception
{
Expression
expression
=
genExp
(
"a > 3.1E10"
);
EvaluationContext
context
=
genContext
(
...
...
@@ -158,7 +158,7 @@ public class ExpressionTest {
}
@Test
public
void
testEvaluate_floatNumber
()
{
public
void
testEvaluate_floatNumber
()
throws
Exception
{
Expression
expression
=
genExp
(
"a > 3.14"
);
EvaluationContext
context
=
genContext
(
...
...
@@ -169,7 +169,7 @@ public class ExpressionTest {
}
@Test
public
void
testEvaluate_twoVariable
()
{
public
void
testEvaluate_twoVariable
()
throws
Exception
{
Expression
expression
=
genExp
(
"a > b"
);
EvaluationContext
context
=
genContext
(
...
...
@@ -178,8 +178,12 @@ public class ExpressionTest {
);
eval
(
expression
,
context
,
Boolean
.
FALSE
);
}
context
=
genContext
(
@Test
public
void
testEvaluate_twoVariableGt
()
throws
Exception
{
Expression
expression
=
genExp
(
"a > b"
);
EvaluationContext
context
=
genContext
(
KeyValue
.
c
(
"b"
,
String
.
valueOf
(
10
)),
KeyValue
.
c
(
"a"
,
String
.
valueOf
(
20
))
);
...
...
@@ -188,7 +192,7 @@ public class ExpressionTest {
}
@Test
public
void
testEvaluate_nullOr
()
{
public
void
testEvaluate_nullOr
()
throws
Exception
{
Expression
expression
=
genExp
(
nullOrExpression
);
EvaluationContext
context
=
genContext
(
...
...
@@ -210,7 +214,7 @@ public class ExpressionTest {
}
@Test
public
void
testEvaluate_boolean
()
{
public
void
testEvaluate_boolean
()
throws
Exception
{
Expression
expression
=
genExp
(
booleanExpression
);
EvaluationContext
context
=
genContext
(
...
...
@@ -229,7 +233,7 @@ public class ExpressionTest {
}
@Test
public
void
testEvaluate_equal
()
{
public
void
testEvaluate_equal
()
throws
Exception
{
Expression
expression
=
genExp
(
equalExpression
);
EvaluationContext
context
=
genContext
(
...
...
@@ -245,7 +249,7 @@ public class ExpressionTest {
}
@Test
public
void
testEvaluate_andTrue
()
{
public
void
testEvaluate_andTrue
()
throws
Exception
{
Expression
expression
=
genExp
(
andExpression
);
EvaluationContext
context
=
genContext
(
...
...
@@ -278,7 +282,7 @@ public class ExpressionTest {
}
@Test
public
void
testEvaluate_andFalse
()
{
public
void
testEvaluate_andFalse
()
throws
Exception
{
Expression
expression
=
genExp
(
andExpression
);
EvaluationContext
context
=
genContext
(
...
...
@@ -302,7 +306,7 @@ public class ExpressionTest {
}
@Test
public
void
testEvaluate_orTrue
()
{
public
void
testEvaluate_orTrue
()
throws
Exception
{
Expression
expression
=
genExp
(
orExpression
);
// first
...
...
@@ -337,7 +341,7 @@ public class ExpressionTest {
}
@Test
public
void
testEvaluate_orFalse
()
{
public
void
testEvaluate_orFalse
()
throws
Exception
{
Expression
expression
=
genExp
(
orExpression
);
// forth
EvaluationContext
context
=
genContext
(
...
...
@@ -350,7 +354,7 @@ public class ExpressionTest {
}
@Test
public
void
testEvaluate_inTrue
()
{
public
void
testEvaluate_inTrue
()
throws
Exception
{
Expression
expression
=
genExp
(
inExpression
);
EvaluationContext
context
=
genContext
(
...
...
@@ -370,7 +374,7 @@ public class ExpressionTest {
}
@Test
public
void
testEvaluate_inFalse
()
{
public
void
testEvaluate_inFalse
()
throws
Exception
{
Expression
expression
=
genExp
(
inExpression
);
EvaluationContext
context
=
genContext
(
...
...
@@ -380,7 +384,7 @@ public class ExpressionTest {
}
@Test
public
void
testEvaluate_notInTrue
()
{
public
void
testEvaluate_notInTrue
()
throws
Exception
{
Expression
expression
=
genExp
(
notInExpression
);
EvaluationContext
context
=
genContext
(
...
...
@@ -390,7 +394,7 @@ public class ExpressionTest {
}
@Test
public
void
testEvaluate_notInFalse
()
{
public
void
testEvaluate_notInFalse
()
throws
Exception
{
Expression
expression
=
genExp
(
notInExpression
);
EvaluationContext
context
=
genContext
(
...
...
@@ -410,7 +414,7 @@ public class ExpressionTest {
}
@Test
public
void
testEvaluate_betweenTrue
()
{
public
void
testEvaluate_betweenTrue
()
throws
Exception
{
Expression
expression
=
genExp
(
betweenExpression
);
EvaluationContext
context
=
genContext
(
...
...
@@ -430,7 +434,7 @@ public class ExpressionTest {
}
@Test
public
void
testEvaluate_betweenFalse
()
{
public
void
testEvaluate_betweenFalse
()
throws
Exception
{
Expression
expression
=
genExp
(
betweenExpression
);
EvaluationContext
context
=
genContext
(
...
...
@@ -445,7 +449,7 @@ public class ExpressionTest {
}
@Test
public
void
testEvaluate_notBetweenTrue
()
{
public
void
testEvaluate_notBetweenTrue
()
throws
Exception
{
Expression
expression
=
genExp
(
notBetweenExpression
);
EvaluationContext
context
=
genContext
(
...
...
@@ -460,7 +464,7 @@ public class ExpressionTest {
}
@Test
public
void
testEvaluate_notBetweenFalse
()
{
public
void
testEvaluate_notBetweenFalse
()
throws
Exception
{
Expression
expression
=
genExp
(
notBetweenExpression
);
EvaluationContext
context
=
genContext
(
...
...
@@ -480,7 +484,7 @@ public class ExpressionTest {
}
@Test
public
void
testEvaluate_isNullTrue
()
{
public
void
testEvaluate_isNullTrue
()
throws
Exception
{
Expression
expression
=
genExp
(
isNullExpression
);
EvaluationContext
context
=
genContext
(
...
...
@@ -490,7 +494,7 @@ public class ExpressionTest {
}
@Test
public
void
testEvaluate_isNullFalse
()
{
public
void
testEvaluate_isNullFalse
()
throws
Exception
{
Expression
expression
=
genExp
(
isNullExpression
);
EvaluationContext
context
=
genContext
(
...
...
@@ -500,7 +504,7 @@ public class ExpressionTest {
}
@Test
public
void
testEvaluate_isNotNullTrue
()
{
public
void
testEvaluate_isNotNullTrue
()
throws
Exception
{
Expression
expression
=
genExp
(
isNotNullExpression
);
EvaluationContext
context
=
genContext
(
...
...
@@ -510,7 +514,7 @@ public class ExpressionTest {
}
@Test
public
void
testEvaluate_isNotNullFalse
()
{
public
void
testEvaluate_isNotNullFalse
()
throws
Exception
{
Expression
expression
=
genExp
(
isNotNullExpression
);
EvaluationContext
context
=
genContext
(
...
...
@@ -519,14 +523,8 @@ public class ExpressionTest {
eval
(
expression
,
context
,
Boolean
.
FALSE
);
}
protected
void
eval
(
Expression
expression
,
EvaluationContext
context
,
Boolean
result
)
{
Object
ret
=
null
;
try
{
ret
=
expression
.
evaluate
(
context
);
}
catch
(
Throwable
e
)
{
e
.
printStackTrace
();
}
protected
void
eval
(
Expression
expression
,
EvaluationContext
context
,
Boolean
result
)
throws
Exception
{
Object
ret
=
expression
.
evaluate
(
context
);
if
(
ret
==
null
||
!(
ret
instanceof
Boolean
))
{
assertThat
(
result
).
isFalse
();
}
else
{
...
...
namesrv/src/test/java/org/apache/rocketmq/namesrv/kvconfig/KVConfigManagerTest.java
浏览文件 @
76464ba6
...
...
@@ -34,7 +34,6 @@ public class KVConfigManagerTest extends NameServerInstanceTest {
@Test
public
void
testPutKVConfig
()
{
kvConfigManager
.
load
();
kvConfigManager
.
putKVConfig
(
NamesrvUtil
.
NAMESPACE_ORDER_TOPIC_CONFIG
,
"UnitTest"
,
"test"
);
byte
[]
kvConfig
=
kvConfigManager
.
getKVListByNamespace
(
NamesrvUtil
.
NAMESPACE_ORDER_TOPIC_CONFIG
);
assertThat
(
kvConfig
).
isNotNull
();
...
...
store/src/test/java/org/apache/rocketmq/store/DefaultMessageStoreTest.java
浏览文件 @
76464ba6
...
...
@@ -43,15 +43,24 @@ public class DefaultMessageStoreTest {
private
SocketAddress
BornHost
;
private
SocketAddress
StoreHost
;
private
byte
[]
MessageBody
;
private
MessageStore
messageStore
;
@Before
public
void
init
()
throws
Exception
{
StoreHost
=
new
InetSocketAddress
(
InetAddress
.
getLocalHost
(),
8123
);
BornHost
=
new
InetSocketAddress
(
InetAddress
.
getByName
(
"127.0.0.1"
),
0
);
messageStore
=
buildMessageStore
();
boolean
load
=
messageStore
.
load
();
assertTrue
(
load
);
messageStore
.
start
();
}
@After
public
void
destory
()
{
messageStore
.
shutdown
();
messageStore
.
destroy
();
MessageStoreConfig
messageStoreConfig
=
new
MessageStoreConfig
();
File
file
=
new
File
(
messageStoreConfig
.
getStorePathRootDir
());
UtilAll
.
deleteFile
(
file
);
...
...
@@ -63,7 +72,7 @@ public class DefaultMessageStoreTest {
messageStoreConfig
.
setMapedFileSizeConsumeQueue
(
1024
*
1024
*
10
);
messageStoreConfig
.
setMaxHashSlotNum
(
10000
);
messageStoreConfig
.
setMaxIndexNum
(
100
*
100
);
messageStoreConfig
.
setFlushDiskType
(
FlushDiskType
.
A
SYNC_FLUSH
);
messageStoreConfig
.
setFlushDiskType
(
FlushDiskType
.
SYNC_FLUSH
);
return
new
DefaultMessageStore
(
messageStoreConfig
,
new
BrokerStatsManager
(
"simpleTest"
),
new
MyMessageArrivingListener
(),
new
BrokerConfig
());
}
...
...
@@ -72,12 +81,16 @@ public class DefaultMessageStoreTest {
long
totalMsgs
=
100
;
QUEUE_TOTAL
=
1
;
MessageBody
=
StoreMessage
.
getBytes
();
MessageStore
master
=
buildMessageStore
();
boolean
load
=
master
.
load
(
);
assertTrue
(
load
);
for
(
long
i
=
0
;
i
<
totalMsgs
;
i
++)
{
messageStore
.
putMessage
(
buildMessage
()
);
}
master
.
start
();
verifyThatMasterIsFunctional
(
totalMsgs
,
master
);
for
(
long
i
=
0
;
i
<
totalMsgs
;
i
++)
{
GetMessageResult
result
=
messageStore
.
getMessage
(
"GROUP_A"
,
"TOPIC_A"
,
0
,
i
,
1024
*
1024
,
null
);
assertThat
(
result
).
isNotNull
();
result
.
release
();
}
verifyThatMasterIsFunctional
(
totalMsgs
,
messageStore
);
}
public
MessageExtBrokerInner
buildMessage
()
{
...
...
@@ -97,54 +110,46 @@ public class DefaultMessageStoreTest {
@Test
public
void
testGroupCommit
()
throws
Exception
{
long
totalMsgs
=
10
0
;
long
totalMsgs
=
10
;
QUEUE_TOTAL
=
1
;
MessageBody
=
StoreMessage
.
getBytes
();
MessageStoreConfig
messageStoreConfig
=
new
MessageStoreConfig
();
messageStoreConfig
.
setMapedFileSizeCommitLog
(
1024
*
8
);
messageStoreConfig
.
setFlushDiskType
(
FlushDiskType
.
SYNC_FLUSH
);
MessageStore
master
=
new
DefaultMessageStore
(
messageStoreConfig
,
null
,
new
MyMessageArrivingListener
(),
new
BrokerConfig
());
boolean
load
=
master
.
load
();
assertTrue
(
load
);
for
(
long
i
=
0
;
i
<
totalMsgs
;
i
++)
{
messageStore
.
putMessage
(
buildMessage
());
}
master
.
start
();
verifyThatMasterIsFunctional
(
totalMsgs
,
master
);
for
(
long
i
=
0
;
i
<
totalMsgs
;
i
++)
{
GetMessageResult
result
=
messageStore
.
getMessage
(
"GROUP_A"
,
"TOPIC_A"
,
0
,
i
,
1024
*
1024
,
null
);
assertThat
(
result
).
isNotNull
();
result
.
release
();
}
verifyThatMasterIsFunctional
(
totalMsgs
,
messageStore
);
}
private
void
verifyThatMasterIsFunctional
(
long
totalMsgs
,
MessageStore
master
)
{
try
{
for
(
long
i
=
0
;
i
<
totalMsgs
;
i
++)
{
master
.
putMessage
(
buildMessage
());
}
for
(
long
i
=
0
;
i
<
totalMsgs
;
i
++)
{
GetMessageResult
result
=
master
.
getMessage
(
"GROUP_A"
,
"TOPIC_A"
,
0
,
i
,
1024
*
1024
,
null
);
assertThat
(
result
).
isNotNull
();
result
.
release
();
}
}
finally
{
master
.
shutdown
();
master
.
destroy
();
for
(
long
i
=
0
;
i
<
totalMsgs
;
i
++)
{
master
.
putMessage
(
buildMessage
());
}
for
(
long
i
=
0
;
i
<
totalMsgs
;
i
++)
{
GetMessageResult
result
=
master
.
getMessage
(
"GROUP_A"
,
"TOPIC_A"
,
0
,
i
,
1024
*
1024
,
null
);
assertThat
(
result
).
isNotNull
();
result
.
release
();
}
}
@Test
public
void
testPullSize
()
throws
Exception
{
MessageStore
messageStore
=
buildMessageStore
();
boolean
load
=
messageStore
.
load
();
assertTrue
(
load
);
messageStore
.
start
();
String
topic
=
"pullSizeTopic"
;
for
(
int
i
=
0
;
i
<
32
;
i
++)
{
MessageExtBrokerInner
messageExtBrokerInner
=
buildMessage
();
messageExtBrokerInner
.
setTopic
(
topic
);
messageExtBrokerInner
.
setQueueId
(
0
);
PutMessageResult
putMessageResult
=
messageStore
.
putMessage
(
messageExtBrokerInner
);
messageStore
.
putMessage
(
messageExtBrokerInner
);
}
//wait for consume queue build
Thread
.
sleep
(
10
0
);
Thread
.
sleep
(
10
);
String
group
=
"simple"
;
GetMessageResult
getMessageResult32
=
messageStore
.
getMessage
(
group
,
topic
,
0
,
0
,
32
,
null
);
assertThat
(
getMessageResult32
.
getMessageBufferList
().
size
()).
isEqualTo
(
32
);
...
...
@@ -154,8 +159,6 @@ public class DefaultMessageStoreTest {
GetMessageResult
getMessageResult45
=
messageStore
.
getMessage
(
group
,
topic
,
0
,
0
,
10
,
null
);
assertThat
(
getMessageResult45
.
getMessageBufferList
().
size
()).
isEqualTo
(
10
);
messageStore
.
shutdown
();
}
private
class
MyMessageArrivingListener
implements
MessageArrivingListener
{
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录