Skip to content

Commit ede33c1

Browse files
hdchenShannonDing
hdchen
authored and
ShannonDing
committed
[RIP-10] optimization test case of DefaultMessageStore (alibaba#995)
1 parent 690a406 commit ede33c1

File tree

1 file changed

+39
-12
lines changed

1 file changed

+39
-12
lines changed

‎store/src/test/java/org/apache/rocketmq/store/DefaultMessageStoreTest.java

+39-12
Original file line numberDiff line numberDiff line change
@@ -40,6 +40,7 @@
4040
import org.apache.rocketmq.store.stats.BrokerStatsManager;
4141
import org.junit.After;
4242
import org.junit.Before;
43+
import org.junit.Ignore;
4344
import org.junit.Test;
4445
import org.junit.runner.RunWith;
4546
import org.mockito.junit.MockitoJUnitRunner;
@@ -123,6 +124,8 @@ public void testWriteAndRead() {
123124
messageStore.putMessage(buildMessage());
124125
}
125126

127+
StoreTestUtil.waitCommitLogReput((DefaultMessageStore) messageStore);
128+
126129
for (long i = 0; i < totalMsgs; i++) {
127130
GetMessageResult result = messageStore.getMessage("GROUP_A", "TOPIC_A", 0, i, 1024 * 1024, null);
128131
assertThat(result).isNotNull();
@@ -180,7 +183,8 @@ public void should_get_consume_queue_offset_successfully_when_incomming_by_times
180183
int queueId = 0;
181184
String topic = "FooBar";
182185
AppendMessageResult[] appendMessageResults = putMessages(totalCount, topic, queueId, true);
183-
Thread.sleep(10);
186+
//Thread.sleep(10);
187+
StoreTestUtil.waitCommitLogReput((DefaultMessageStore) messageStore);
184188

185189
ConsumeQueue consumeQueue = getDefaultMessageStore().findConsumeQueue(topic, queueId);
186190
for (AppendMessageResult appendMessageResult : appendMessageResults) {
@@ -198,7 +202,8 @@ public void should_get_consume_queue_offset_successfully_when_timestamp_is_skewi
198202
int queueId = 0;
199203
String topic = "FooBar";
200204
AppendMessageResult[] appendMessageResults = putMessages(totalCount, topic, queueId, true);
201-
Thread.sleep(10);
205+
//Thread.sleep(10);
206+
StoreTestUtil.waitCommitLogReput((DefaultMessageStore) messageStore);
202207
int skewing = 2;
203208

204209
ConsumeQueue consumeQueue = getDefaultMessageStore().findConsumeQueue(topic, queueId);
@@ -222,7 +227,8 @@ public void should_get_min_of_max_consume_queue_offset_when_timestamp_s_skewing_
222227
int queueId = 0;
223228
String topic = "FooBar";
224229
AppendMessageResult[] appendMessageResults = putMessages(totalCount, topic, queueId, true);
225-
Thread.sleep(10);
230+
//Thread.sleep(10);
231+
StoreTestUtil.waitCommitLogReput((DefaultMessageStore) messageStore);
226232
int skewing = 20000;
227233

228234
ConsumeQueue consumeQueue = getDefaultMessageStore().findConsumeQueue(topic, queueId);
@@ -235,6 +241,9 @@ public void should_get_min_of_max_consume_queue_offset_when_timestamp_s_skewing_
235241
assertThat(indexBuffer.getByteBuffer().getInt()).isEqualTo(appendMessageResults[totalCount - 1].getWroteBytes());
236242
assertThat(indexBuffer2.getByteBuffer().getLong()).isEqualTo(appendMessageResults[0].getWroteOffset());
237243
assertThat(indexBuffer2.getByteBuffer().getInt()).isEqualTo(appendMessageResults[0].getWroteBytes());
244+
245+
indexBuffer.release();
246+
indexBuffer2.release();
238247
}
239248
}
240249

@@ -245,7 +254,9 @@ public void should_return_zero_when_consume_queue_not_found() throws Interrupted
245254
int wrongQueueId = 1;
246255
String topic = "FooBar";
247256
AppendMessageResult[] appendMessageResults = putMessages(totalCount, topic, queueId, false);
248-
Thread.sleep(10);
257+
//Thread.sleep(10);
258+
259+
StoreTestUtil.waitCommitLogReput((DefaultMessageStore) messageStore);
249260

250261
long offset = messageStore.getOffsetInQueueByTime(topic, wrongQueueId, appendMessageResults[0].getStoreTimestamp());
251262

@@ -259,7 +270,8 @@ public void should_return_negative_one_when_invoke_getMessageStoreTimeStamp_if_c
259270
int wrongQueueId = 1;
260271
String topic = "FooBar";
261272
putMessages(totalCount, topic, queueId, false);
262-
Thread.sleep(10);
273+
//Thread.sleep(10);
274+
StoreTestUtil.waitCommitLogReput((DefaultMessageStore) messageStore);
263275

264276
long messageStoreTimeStamp = messageStore.getMessageStoreTimeStamp(topic, wrongQueueId, 0);
265277

@@ -273,7 +285,9 @@ public void should_return_negative_one_when_invoke_getMessageStoreTimeStamp_if_c
273285
int wrongQueueId = 1;
274286
String topic = "FooBar";
275287
putMessages(totalCount, topic, queueId, true);
276-
Thread.sleep(10);
288+
//Thread.sleep(10);
289+
290+
StoreTestUtil.waitCommitLogReput((DefaultMessageStore) messageStore);
277291

278292
long messageStoreTimeStamp = messageStore.getMessageStoreTimeStamp(topic, wrongQueueId, -1);
279293

@@ -287,7 +301,8 @@ public void should_get_message_store_timestamp_successfully_when_incomming_by_to
287301
int queueId = 0;
288302
String topic = "FooBar";
289303
AppendMessageResult[] appendMessageResults = putMessages(totalCount, topic, queueId, false);
290-
Thread.sleep(10);
304+
//Thread.sleep(10);
305+
StoreTestUtil.waitCommitLogReput((DefaultMessageStore) messageStore);
291306

292307
ConsumeQueue consumeQueue = getDefaultMessageStore().findConsumeQueue(topic, queueId);
293308
int minOffsetInQueue = (int)consumeQueue.getMinOffsetInQueue();
@@ -310,7 +325,8 @@ public void should_get_store_time_successfully_when_invoke_getStoreTime_if_every
310325
int queueId = 0;
311326
String topic = "FooBar";
312327
AppendMessageResult[] appendMessageResults = putMessages(totalCount, topic, queueId, false);
313-
Thread.sleep(10);
328+
//Thread.sleep(10);
329+
StoreTestUtil.waitCommitLogReput((DefaultMessageStore) messageStore);
314330
ConsumeQueue consumeQueue = messageStore.getConsumeQueue(topic, queueId);
315331

316332
for (int i = 0; i < totalCount; i++) {
@@ -412,6 +428,8 @@ private void verifyThatMasterIsFunctional(long totalMsgs, MessageStore master) {
412428
master.putMessage(buildMessage());
413429
}
414430

431+
StoreTestUtil.waitCommitLogReput((DefaultMessageStore) messageStore);
432+
415433
for (long i = 0; i < totalMsgs; i++) {
416434
GetMessageResult result = master.getMessage("GROUP_A", "TOPIC_A", 0, i, 1024 * 1024, null);
417435
assertThat(result).isNotNull();
@@ -432,16 +450,21 @@ public void testPullSize() throws Exception {
432450
}
433451
// wait for consume queue build
434452
// the sleep time should be great than consume queue flush interval
435-
Thread.sleep(100);
453+
//Thread.sleep(100);
454+
StoreTestUtil.waitCommitLogReput((DefaultMessageStore) messageStore);
436455
String group = "simple";
437456
GetMessageResult getMessageResult32 = messageStore.getMessage(group, topic, 0, 0, 32, null);
438457
assertThat(getMessageResult32.getMessageBufferList().size()).isEqualTo(32);
458+
getMessageResult32.release();
439459

440460
GetMessageResult getMessageResult20 = messageStore.getMessage(group, topic, 0, 0, 20, null);
441461
assertThat(getMessageResult20.getMessageBufferList().size()).isEqualTo(20);
442462

463+
getMessageResult20.release();
443464
GetMessageResult getMessageResult45 = messageStore.getMessage(group, topic, 0, 0, 10, null);
444465
assertThat(getMessageResult45.getMessageBufferList().size()).isEqualTo(10);
466+
getMessageResult45.release();
467+
445468
}
446469

447470
@Test
@@ -455,7 +478,9 @@ public void testRecover() throws Exception {
455478
messageStore.putMessage(messageExtBrokerInner);
456479
}
457480

458-
Thread.sleep(100);//wait for build consumer queue
481+
// Thread.sleep(100);//wait for build consumer queue
482+
StoreTestUtil.waitCommitLogReput((DefaultMessageStore) messageStore);
483+
459484
long maxPhyOffset = messageStore.getMaxPhyOffset();
460485
long maxCqOffset = messageStore.getMaxOffsetInQueue(topic, 0);
461486

@@ -475,7 +500,8 @@ public void testRecover() throws Exception {
475500
messageExtBrokerInner.setQueueId(0);
476501
messageStore.putMessage(messageExtBrokerInner);
477502
}
478-
Thread.sleep(100);
503+
//Thread.sleep(100);
504+
StoreTestUtil.waitCommitLogReput((DefaultMessageStore) messageStore);
479505
long secondLastPhyOffset = messageStore.getMaxPhyOffset();
480506
long secondLastCqOffset = messageStore.getMaxOffsetInQueue(topic, 0);
481507

@@ -504,7 +530,8 @@ public void testRecover() throws Exception {
504530
messageExtBrokerInner.setQueueId(0);
505531
messageStore.putMessage(messageExtBrokerInner);
506532
}
507-
Thread.sleep(100);
533+
//Thread.sleep(100);
534+
StoreTestUtil.waitCommitLogReput((DefaultMessageStore) messageStore);
508535
secondLastPhyOffset = messageStore.getMaxPhyOffset();
509536
secondLastCqOffset = messageStore.getMaxOffsetInQueue(topic, 0);
510537

0 commit comments

Comments
 (0)