Skip to content

[fix][test] Fix flaky test TopicReaderTest.testMultiReaderIsAbleToSeekWithTimeOnMiddleOfTopic #17147

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
nicoloboschi wants to merge 3 commits into from
Closed

[fix][test] Fix flaky test TopicReaderTest.testMultiReaderIsAbleToSeekWithTimeOnMiddleOfTopic #17147

Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
7007d2b
[fix][test] Fix flaky test TopicReaderTest.testMultiReaderIsAbleToSee…
nicoloboschi Aug 17, 2022
6b5ef38
also fix testReaderIsAbleToSeekWithTimeOnMiddleOfTopic
nicoloboschi Aug 18, 2022
777bebf
Merge branch 'master' into fix-test-testMultiReaderIsAbleToSeekWithTi…
mattisonchao Aug 24, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
38 changes: 24 additions & 14 deletions pulsar-broker/src/test/java/org/apache/pulsar/client/api/TopicReaderTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1350,30 +1350,36 @@ public void testReaderIsAbleToSeekWithMessageIdOnMiddleOfTopic() throws Exceptio
public void testReaderIsAbleToSeekWithTimeOnMiddleOfTopic() throws Exception {
final String topicName = "persistent://my-property/my-ns/ReaderIsAbleToSeekWithTimeOnMiddleOfTopic";
final int numOfMessage = 10;
final int halfMessages = numOfMessage / 2;

Producer<byte[]> producer = pulsarClient.newProducer()
.topic(topicName).create();

long l = System.currentTimeMillis();
long halfTime = 0;
for (int i = 0; i < numOfMessage; i++) {
if (i == 6) {
halfTime = System.currentTimeMillis();
}
producer.send(String.format("msg num %d", i).getBytes());
Thread.sleep(100);
}

Reader<byte[]> reader = pulsarClient.newReader().topic(topicName)
.startMessageId(MessageId.earliest).create();
.startMessageId(MessageId.latest).create();

int plusTime = (halfMessages + 1) * 100;
reader.seek(l + plusTime);
reader.seek(halfTime);

Set<String> messageSet = Sets.newHashSet();
for (int i = halfMessages + 1; i < numOfMessage; i++) {
Message<byte[]> message = reader.readNext();
int i = 6;
while (true) {
Message<byte[]> message = reader.readNext(1, TimeUnit.SECONDS);
// make sure we are headed to the end of the topic
if (message == null) {
break;
}
String receivedMessage = new String(message.getData());
String expectedMessage = String.format("msg num %d", i);
String expectedMessage = String.format("msg num %d", i++);
testMessageOrderAndDuplicates(messageSet, receivedMessage, expectedMessage);
}
assertEquals(messageSet.size(), 4);

reader.close();
producer.close();
Expand All @@ -1383,26 +1389,30 @@ public void testReaderIsAbleToSeekWithTimeOnMiddleOfTopic() throws Exception {
public void testMultiReaderIsAbleToSeekWithTimeOnMiddleOfTopic() throws Exception {
final String topicName = "persistent://my-property/my-ns/testMultiReaderIsAbleToSeekWithTimeOnMiddleOfTopic" + System.currentTimeMillis();
final int numOfMessage = 10;
final int halfMessages = numOfMessage / 2;
admin.topics().createPartitionedTopic(topicName, 3);
Producer<byte[]> producer = pulsarClient.newProducer().topic(topicName).create();
long halfTime = 0;
for (int i = 0; i < numOfMessage; i++) {
if (i == numOfMessage / 2) {
if (i == 6) {
halfTime = System.currentTimeMillis();
}
producer.send(String.format("msg num %d", i).getBytes());
}
Assert.assertTrue(halfTime != 0);
Reader<byte[]> reader = pulsarClient.newReader().topic(topicName).startMessageId(MessageId.earliest).create();
Reader<byte[]> reader = pulsarClient.newReader().topic(topicName).startMessageId(MessageId.latest).create();

reader.seek(halfTime);
Set<String> messageSet = Sets.newHashSet();
for (int i = halfMessages + 1; i < numOfMessage; i++) {
Message<byte[]> message = reader.readNext(10, TimeUnit.SECONDS);
while (true) {
Message<byte[]> message = reader.readNext(1, TimeUnit.SECONDS);
// make sure we are headed to the end of the topic
if (message == null) {
break;
}
String receivedMessage = new String(message.getData());
Assert.assertTrue(messageSet.add(receivedMessage), "Received duplicate message " + receivedMessage);
}
assertEquals(messageSet.size(), 4);
reader.close();
producer.close();
}
Expand Down