Skip to content
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

[fix][broker] Avoid consumers receiving acknowledged messages from compacted topic after reconnection #21187

Merged
merged 15 commits into from
Jan 26, 2024
Merged

[fix][broker] Avoid consumers receiving acknowledged messages from compacted topic after reconnection #21187

Changes from 1 commit
Commits
Show all changes
15 commits
Select commit Hold shift + click to select a range
142d2d0
Avoid consumers receiving acknowledged messages from a compacted topi…
coderzc Sep 8, 2023
f67d814
Address comment
coderzc Sep 27, 2023
861ec67
Add testReaderReconnectedFromNextEntry
coderzc Sep 28, 2023
b605dd3
Avoid skip the remaining data in the compacted ledger after rewind
coderzc Jan 16, 2024
781f9dd
Fix test
coderzc Jan 16, 2024
44ac16e
Improve code
coderzc Jan 16, 2024
f399fae
Fix test
coderzc Jan 16, 2024
2ef2520
Rebase master
coderzc Jan 18, 2024
03bb922
Address comment
coderzc Jan 18, 2024
e329208
Address comment
coderzc Jan 19, 2024
786d9e0
Add `testEarliestSubsAfterRollover`
coderzc Jan 20, 2024
2634f38
Add some comments
coderzc Jan 24, 2024
6bd50de
Remove move to the next existing ledger when recovering cursor
coderzc Jan 25, 2024
ca6902a
Add `rewind(boolean readCompacted)` method in ManagedCursor
coderzc Jan 25, 2024
bad719f
Only skipping invalid md-position when recovering cursor
coderzc Jan 26, 2024
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
Prev Previous commit
Next Next commit
Address comment
  • Loading branch information
@coderzc
coderzc committed Jan 26, 2024
commit 03bb92291db20c5900f89ebe84ebebba178a53f8
15 changes: 13 additions & 2 deletions pulsar-broker/src/test/java/org/apache/pulsar/compaction/CompactionTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2259,6 +2259,10 @@ public void testAcknowledgeWithReconnection() throws Exception {
consumer.acknowledge(message);
}

Awaitility.await().untilAsserted(() ->
assertEquals(admin.topics().getStats(topicName, true).getSubscriptions().get(subName).getMsgBacklog(),
5));

// Make consumer reconnect to broker
BewareMyPower marked this conversation as resolved.
Show resolved Hide resolved
admin.topics().unload(topicName);

Expand All @@ -2272,6 +2276,10 @@ public void testAcknowledgeWithReconnection() throws Exception {
consumer.acknowledge(message);
}

Awaitility.await().untilAsserted(() ->
assertEquals(admin.topics().getStats(topicName, true).getSubscriptions().get(subName).getMsgBacklog(),
0));

Assert.assertEquals(results, expected);

Message<String> message = consumer.receive(3, TimeUnit.SECONDS);
Expand All @@ -2285,8 +2293,11 @@ public void testAcknowledgeWithReconnection() throws Exception {
Assert.assertEquals(message2.getValue(), "V");
consumer.acknowledge(message2);

PersistentTopicInternalStats internalStats = admin.topics().getInternalStats(topicName);
Assert.assertEquals(internalStats.lastConfirmedEntry, internalStats.cursors.get(subName).markDeletePosition);
Awaitility.await().untilAsserted(() -> {
PersistentTopicInternalStats internalStats = admin.topics().getInternalStats(topicName);
Assert.assertEquals(internalStats.lastConfirmedEntry,
internalStats.cursors.get(subName).markDeletePosition);
});

consumer.close();
producer.close();
Expand Down