You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "coderzc (via GitHub)" <gi...@apache.org> on 2023/09/15 03:07:39 UTC
[GitHub] [pulsar] coderzc opened a new pull request, #21187: [fix][broker] Avoid consumers receiving acknowledged messages from a compacted topic upon reconnection
coderzc opened a new pull request, #21187:
URL: https://github.com/apache/pulsar/pull/21187
Fixes #21074
### Motivation
// TODO
### Modifications
// TODO
### Verifying this change
- [x] Make sure that the change passes the CI checks.
*(Please pick either of the following options)*
This change is a trivial rework / code cleanup without any test coverage.
*(or)*
This change is already covered by existing tests, such as *(please describe tests)*.
*(or)*
This change added tests and can be verified as follows:
*(example:)*
- *Added integration tests for end-to-end deployment with large payloads (10MB)*
- *Extended integration test for recovery after broker failure*
### Does this pull request potentially affect one of the following parts:
<!-- DO NOT REMOVE THIS SECTION. CHECK THE PROPER BOX ONLY. -->
*If the box was checked, please highlight the changes*
- [ ] Dependencies (add or upgrade a dependency)
- [ ] The public API
- [ ] The schema
- [ ] The default values of configurations
- [ ] The threading model
- [ ] The binary protocol
- [ ] The REST endpoints
- [ ] The admin CLI options
- [ ] The metrics
- [ ] Anything that affects deployment
### Documentation
<!-- DO NOT REMOVE THIS SECTION. CHECK THE PROPER BOX ONLY. -->
- [ ] `doc` <!-- Your PR contains doc changes. -->
- [ ] `doc-required` <!-- Your PR changes impact docs and you will update later -->
- [x] `doc-not-needed` <!-- Your PR changes do not impact docs -->
- [ ] `doc-complete` <!-- Docs have been already added -->
### Matching PR in forked repository
PR in forked repository: <!-- ENTER URL HERE -->
<!--
After opening this PR, the build in apache/pulsar will fail and instructions will
be provided for opening a PR in the PR author's forked repository.
apache/pulsar pull requests should be first tested in your own fork since the
apache/pulsar CI based on GitHub Actions has constrained resources and quota.
GitHub Actions provides separate quota for pull requests that are executed in
a forked repository.
The tests will be run in the forked repository until all PR review comments have
been handled, the tests pass and the PR is approved by a reviewer.
-->
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Avoid consumers receiving acknowledged messages from compacted topic after reconnection [pulsar]
Posted by "coderzc (via GitHub)" <gi...@apache.org>.
coderzc commented on code in PR #21187:
URL: https://github.com/apache/pulsar/pull/21187#discussion_r1462670799
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java:
##########
@@ -2519,7 +2525,8 @@ public Position getPersistentMarkDeletedPosition() {
public void rewind() {
lock.writeLock().lock();
try {
- PositionImpl newReadPosition = ledger.getNextValidPosition(markDeletePosition);
+ PositionImpl newReadPosition = isCursorReadFromCompacted() ? markDeletePosition.getNext() :
+ ledger.getNextValidPosition(markDeletePosition);
Review Comment:
I don't think this affects correctness, if the newReadPosition's ledger doesn't exist in ml, it will be moved to the beginning of the next ledger.
https://github.com/apache/pulsar/blob/b14fcb47f20398bdf5b698e158346dce19f0edc6/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java#L2074-L2085
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Avoid consumers receiving acknowledged messages from compacted topic after reconnection [pulsar]
Posted by "coderzc (via GitHub)" <gi...@apache.org>.
coderzc commented on code in PR #21187:
URL: https://github.com/apache/pulsar/pull/21187#discussion_r1462658698
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorSubscriptionTest.java:
##########
@@ -868,6 +869,7 @@ public void testReplicatedSubscriptionWithCompaction() throws Exception {
.topic(topicName)
.subscriptionName("sub2")
.subscriptionType(SubscriptionType.Exclusive)
+ .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
Review Comment:
Yes, consumer always read from earliest position before applying this change.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Avoid consumers receiving acknowledged messages from compacted topic after reconnection [pulsar]
Posted by "coderzc (via GitHub)" <gi...@apache.org>.
coderzc commented on code in PR #21187:
URL: https://github.com/apache/pulsar/pull/21187#discussion_r1462691000
##########
pulsar-broker/src/main/java/org/apache/pulsar/compaction/CompactedTopicImpl.java:
##########
@@ -100,7 +100,11 @@ public void asyncReadEntriesOrWait(ManagedCursor cursor,
boolean isFirstRead,
ReadEntriesCallback callback, Consumer consumer) {
PositionImpl cursorPosition;
- if (isFirstRead && MessageId.earliest.equals(consumer.getStartMessageId())){
+ boolean readFromEarliest = isFirstRead && MessageId.earliest.equals(consumer.getStartMessageId())
+ && (!cursor.isDurable() || cursor.getName().equals(Compactor.COMPACTION_SUBSCRIPTION)
+ || cursor.getMarkDeletedPosition() == null
+ || cursor.getMarkDeletedPosition().getEntryId() == -1L);
Review Comment:
> Sorry, I mean why we need to check cursor.getMarkDeletedPosition().getEntryId() == -1L?
Oh, since if `markDeletedPosition.entryID == -1L` then the md-position is an invalid position, the initial md-position of the consumer will be set to it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Avoid consumers receiving acknowledged messages from compacted topic after reconnection [pulsar]
Posted by "coderzc (via GitHub)" <gi...@apache.org>.
coderzc commented on code in PR #21187:
URL: https://github.com/apache/pulsar/pull/21187#discussion_r1456849047
##########
pulsar-broker/src/main/java/org/apache/pulsar/compaction/CompactedTopicImpl.java:
##########
@@ -100,7 +100,13 @@ public void asyncReadEntriesOrWait(ManagedCursor cursor,
boolean isFirstRead,
ReadEntriesCallback callback, Consumer consumer) {
PositionImpl cursorPosition;
- if (isFirstRead && MessageId.earliest.equals(consumer.getStartMessageId())){
+ boolean readFromEarliest = false;
+ if (!cursor.isDurable() || ((ManagedCursorImpl) cursor).isCompactionCursor()
+ || cursor.getMarkDeletedPosition() == null
+ || cursor.getMarkDeletedPosition().getEntryId() == -1L) {
+ readFromEarliest = isFirstRead && MessageId.earliest.equals(consumer.getStartMessageId());
+ }
Review Comment:
This is for cherry-picking it to old release version.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Avoid consumers receiving acknowledged messages from compacted topic after reconnection [pulsar]
Posted by "coderzc (via GitHub)" <gi...@apache.org>.
coderzc commented on code in PR #21187:
URL: https://github.com/apache/pulsar/pull/21187#discussion_r1457382311
##########
pulsar-broker/src/main/java/org/apache/pulsar/compaction/CompactedTopicImpl.java:
##########
@@ -100,7 +100,13 @@ public void asyncReadEntriesOrWait(ManagedCursor cursor,
boolean isFirstRead,
ReadEntriesCallback callback, Consumer consumer) {
PositionImpl cursorPosition;
- if (isFirstRead && MessageId.earliest.equals(consumer.getStartMessageId())){
+ boolean readFromEarliest = false;
+ if (!cursor.isDurable() || ((ManagedCursorImpl) cursor).isCompactionCursor()
+ || cursor.getMarkDeletedPosition() == null
+ || cursor.getMarkDeletedPosition().getEntryId() == -1L) {
+ readFromEarliest = isFirstRead && MessageId.earliest.equals(consumer.getStartMessageId());
+ }
Review Comment:
Yeah, I will open a separated PR for cherry-pick, but I think we can keep this change on the master.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Avoid consumers receiving acknowledged messages from compacted topic after reconnection [pulsar]
Posted by "coderzc (via GitHub)" <gi...@apache.org>.
coderzc commented on code in PR #21187:
URL: https://github.com/apache/pulsar/pull/21187#discussion_r1456904069
##########
pulsar-broker/src/test/java/org/apache/pulsar/compaction/CompactionTest.java:
##########
@@ -1985,4 +1985,55 @@ public void testCompactionDuplicate() throws Exception {
}
}
}
+
+ @Test(timeOut = 100000)
+ public void testAcknowledge() throws Exception {
+ final String topicName = "persistent://my-property/use/my-ns/testAcknowledge" + UUID.randomUUID();
+ final String subName = "my-sub";
+ @Cleanup
+ PulsarClient client = newPulsarClient(lookupUrl.toString(), 100);
+ Producer<String> producer = pulsarClient.newProducer(Schema.STRING)
+ .enableBatching(false).topic(topicName).create();
+
+ pulsarClient.newConsumer().topic(topicName).subscriptionName(subName).readCompacted(true).subscribe().close();
+
+ Map<String, String> expected = new HashMap<>();
+ for (int i = 0; i < 10; i++) {
+ producer.newMessage().key(String.valueOf(i)).value(String.valueOf(i)).send();
+ expected.put(String.valueOf(i), String.valueOf(i));
+ }
+ producer.flush();
+
+ admin.topics().triggerCompaction(topicName);
+
+ Awaitility.await().untilAsserted(() -> {
+ assertEquals(admin.topics().compactionStatus(topicName).status,
+ LongRunningProcessStatus.Status.SUCCESS);
+ });
+
+ ConsumerImpl<String> consumer = (ConsumerImpl<String>) client.newConsumer(Schema.STRING)
+ .topic(topicName).readCompacted(true).receiverQueueSize(1).subscriptionName(subName)
+ .isAckReceiptEnabled(true)
+ .subscribe();
+
+ for (int i = 0; i < 10; i++) {
+ Message<String> message = consumer.receive(3, TimeUnit.SECONDS);
+ if (message == null) {
+ break;
+ }
+
+ consumer.acknowledge(message);
+ String remove = expected.remove(message.getKey());
+ Assert.assertEquals(remove, message.getValue());
+ }
+
+ // Make consumer reconnect to broker
Review Comment:
Fixed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Avoid consumers receiving acknowledged messages from compacted topic after reconnection [pulsar]
Posted by "coderzc (via GitHub)" <gi...@apache.org>.
coderzc commented on code in PR #21187:
URL: https://github.com/apache/pulsar/pull/21187#discussion_r1464262263
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java:
##########
@@ -2519,7 +2525,8 @@ public Position getPersistentMarkDeletedPosition() {
public void rewind() {
lock.writeLock().lock();
try {
- PositionImpl newReadPosition = ledger.getNextValidPosition(markDeletePosition);
+ PositionImpl newReadPosition = isCursorReadFromCompacted() ? markDeletePosition.getNext() :
+ ledger.getNextValidPosition(markDeletePosition);
Review Comment:
@codelipenghui
> After checking the changes again. It looks like we don't need to persist the readCompacted? If depends on the consumer settings. We only need to set the readCompacted when new consumer connected?
I don't think this will work well because when recovering the cursor, maybe the consumer is not connected yet, then our md-position will be forced to move to the first position of the ml.
please see:
https://github.com/apache/pulsar/pull/21187/files#diff-fad355f91bd15cc041161f9a46fce62b7fee87fbfb8f0ff8a8b724a1bd1f29eeR685-R697
> Oh, I found a similar PR which fixed the Reader API.
Yes, they are a issue
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Avoid consumers receiving acknowledged messages from compacted topic after reconnection [pulsar]
Posted by "codelipenghui (via GitHub)" <gi...@apache.org>.
codelipenghui commented on code in PR #21187:
URL: https://github.com/apache/pulsar/pull/21187#discussion_r1464293684
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java:
##########
@@ -2519,7 +2525,8 @@ public Position getPersistentMarkDeletedPosition() {
public void rewind() {
lock.writeLock().lock();
try {
- PositionImpl newReadPosition = ledger.getNextValidPosition(markDeletePosition);
+ PositionImpl newReadPosition = isCursorReadFromCompacted() ? markDeletePosition.getNext() :
+ ledger.getNextValidPosition(markDeletePosition);
Review Comment:
@coderzc but it just moved the read position, right? If the first consumer connected and readCompacted is enabled, we can reset the read position to the next position of the MD position.
Or can we add a new method `rewind(boolean isReadCompacted)`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Avoid consumers receiving acknowledged messages from compacted topic after reconnection [pulsar]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on code in PR #21187:
URL: https://github.com/apache/pulsar/pull/21187#discussion_r1455139453
##########
pulsar-broker/src/main/java/org/apache/pulsar/compaction/CompactedTopicImpl.java:
##########
@@ -100,7 +100,13 @@ public void asyncReadEntriesOrWait(ManagedCursor cursor,
boolean isFirstRead,
ReadEntriesCallback callback, Consumer consumer) {
PositionImpl cursorPosition;
- if (isFirstRead && MessageId.earliest.equals(consumer.getStartMessageId())){
+ boolean readFromEarliest = false;
+ if (!cursor.isDurable() || ((ManagedCursorImpl) cursor).isCompactionCursor()
+ || cursor.getMarkDeletedPosition() == null
+ || cursor.getMarkDeletedPosition().getEntryId() == -1L) {
+ readFromEarliest = isFirstRead && MessageId.earliest.equals(consumer.getStartMessageId());
+ }
Review Comment:
You don't need to modify a deprecated method.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumer.java:
##########
@@ -362,7 +363,12 @@ private void readMoreEntries(Consumer consumer) {
}
havePendingRead = true;
if (consumer.readCompacted()) {
- boolean readFromEarliest = isFirstRead && MessageId.earliest.equals(consumer.getStartMessageId());
+ boolean readFromEarliest = false;
+ if (!cursor.isDurable() || ((ManagedCursorImpl) cursor).isCompactionCursor()
+ || cursor.getMarkDeletedPosition() == null
+ || cursor.getMarkDeletedPosition().getEntryId() == -1L) {
+ readFromEarliest = isFirstRead && MessageId.earliest.equals(consumer.getStartMessageId());
+ }
Review Comment:
Could you just simplify the check with:
```java
boolean readFromEarliest = isFirstRead && cursor.isDurable()
&& cursor.getName().equals(Compactor.COMPACTION_SUBSCRIPTION);
```
You don't need to expose the `ManagedCursor#isCompactionCursor()` method.
And why did you also check the mark deleted position here?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Avoid consumers receiving acknowledged messages from compacted topic after reconnection [pulsar]
Posted by "codelipenghui (via GitHub)" <gi...@apache.org>.
codelipenghui commented on code in PR #21187:
URL: https://github.com/apache/pulsar/pull/21187#discussion_r1462680198
##########
pulsar-broker/src/main/java/org/apache/pulsar/compaction/CompactedTopicImpl.java:
##########
@@ -100,7 +100,11 @@ public void asyncReadEntriesOrWait(ManagedCursor cursor,
boolean isFirstRead,
ReadEntriesCallback callback, Consumer consumer) {
PositionImpl cursorPosition;
- if (isFirstRead && MessageId.earliest.equals(consumer.getStartMessageId())){
+ boolean readFromEarliest = isFirstRead && MessageId.earliest.equals(consumer.getStartMessageId())
+ && (!cursor.isDurable() || cursor.getName().equals(Compactor.COMPACTION_SUBSCRIPTION)
+ || cursor.getMarkDeletedPosition() == null
+ || cursor.getMarkDeletedPosition().getEntryId() == -1L);
Review Comment:
Sorry, I mean why we need to check `cursor.getMarkDeletedPosition().getEntryId() == -1L`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Avoid consumers receiving acknowledged messages from compacted topic after reconnection [pulsar]
Posted by "codelipenghui (via GitHub)" <gi...@apache.org>.
codelipenghui commented on code in PR #21187:
URL: https://github.com/apache/pulsar/pull/21187#discussion_r1462682695
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java:
##########
@@ -2519,7 +2525,8 @@ public Position getPersistentMarkDeletedPosition() {
public void rewind() {
lock.writeLock().lock();
try {
- PositionImpl newReadPosition = ledger.getNextValidPosition(markDeletePosition);
+ PositionImpl newReadPosition = isCursorReadFromCompacted() ? markDeletePosition.getNext() :
+ ledger.getNextValidPosition(markDeletePosition);
Review Comment:
If so, can we always use `markDeletePosition.getNext()`? so that we don't need to check if the subscription is enabled read compacted or not.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Avoid consumers receiving acknowledged messages from compacted topic after reconnection [pulsar]
Posted by "coderzc (via GitHub)" <gi...@apache.org>.
coderzc commented on code in PR #21187:
URL: https://github.com/apache/pulsar/pull/21187#discussion_r1462719147
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java:
##########
@@ -2519,7 +2525,8 @@ public Position getPersistentMarkDeletedPosition() {
public void rewind() {
lock.writeLock().lock();
try {
- PositionImpl newReadPosition = ledger.getNextValidPosition(markDeletePosition);
+ PositionImpl newReadPosition = isCursorReadFromCompacted() ? markDeletePosition.getNext() :
+ ledger.getNextValidPosition(markDeletePosition);
Review Comment:
But this will affect the `CursorStats` and increase some CPU cycles. I think we can apply this change only to cursors that have readCompacted enabled.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Avoid consumers receiving acknowledged messages from compacted topic after reconnection [pulsar]
Posted by "codelipenghui (via GitHub)" <gi...@apache.org>.
codelipenghui commented on code in PR #21187:
URL: https://github.com/apache/pulsar/pull/21187#discussion_r1464255843
##########
pulsar-broker/src/main/java/org/apache/pulsar/compaction/CompactedTopicImpl.java:
##########
@@ -100,7 +100,11 @@ public void asyncReadEntriesOrWait(ManagedCursor cursor,
boolean isFirstRead,
ReadEntriesCallback callback, Consumer consumer) {
PositionImpl cursorPosition;
- if (isFirstRead && MessageId.earliest.equals(consumer.getStartMessageId())){
+ boolean readFromEarliest = isFirstRead && MessageId.earliest.equals(consumer.getStartMessageId())
+ && (!cursor.isDurable() || cursor.getName().equals(Compactor.COMPACTION_SUBSCRIPTION)
+ || cursor.getMarkDeletedPosition() == null
+ || cursor.getMarkDeletedPosition().getEntryId() == -1L);
Review Comment:
Ok, I see. Could you please help add some comments to improve the code readability?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Avoid consumers receiving acknowledged messages from compacted topic after reconnection [pulsar]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #21187:
URL: https://github.com/apache/pulsar/pull/21187#issuecomment-1783973793
The pr had no activity for 30 days, mark with Stale label.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Avoid consumers receiving acknowledged messages from compacted topic after reconnection [pulsar]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on code in PR #21187:
URL: https://github.com/apache/pulsar/pull/21187#discussion_r1455325323
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java:
##########
@@ -443,6 +445,10 @@ public CompletableFuture<Void> messageAcked(CommandAck ack) {
properties = ack.getPropertiesList().stream()
.collect(Collectors.toMap(KeyLongValue::getKey, KeyLongValue::getValue));
}
+ if (readCompacted) {
+ properties = new TreeMap<>(properties);
+ properties.put(READ_COMPACTED_CURSOR_PROPERTIES, 1L);
+ }
Review Comment:
It's meaningless to use a TreeMap here.
```suggestion
if (readCompacted) {
properties = Collections.singletonMap(ManagedCursorImpl.READ_COMPACTED_CURSOR_PROPERTIES, 1L);
}
```
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java:
##########
@@ -1017,6 +1021,10 @@ private CompletableFuture<Subscription> getDurableSubscription(String subscripti
}
Map<String, Long> properties = PersistentSubscription.getBaseCursorProperties(replicated);
+ if (readCompacted) {
+ properties = new TreeMap<>(properties);
+ properties.put(READ_COMPACTED_CURSOR_PROPERTIES, 1L);
Review Comment:
Why did you create a new map instead of modifying the `properties`? (the same question as it is in `readMoreEntries`)
```java
if (readCompacted) {
properties = new HashMap<>(properties);
properties.put(READ_COMPACTED_CURSOR_PROPERTIES, 1L);
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Avoid consumers receiving acknowledged messages from compacted topic after reconnection [pulsar]
Posted by "coderzc (via GitHub)" <gi...@apache.org>.
coderzc commented on code in PR #21187:
URL: https://github.com/apache/pulsar/pull/21187#discussion_r1462658698
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorSubscriptionTest.java:
##########
@@ -868,6 +869,7 @@ public void testReplicatedSubscriptionWithCompaction() throws Exception {
.topic(topicName)
.subscriptionName("sub2")
.subscriptionType(SubscriptionType.Exclusive)
+ .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
Review Comment:
Yes, consumer always be read from earliest position before applying this change.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Avoid consumers receiving acknowledged messages from compacted topic after reconnection [pulsar]
Posted by "heesung-sn (via GitHub)" <gi...@apache.org>.
heesung-sn commented on PR #21187:
URL: https://github.com/apache/pulsar/pull/21187#issuecomment-1967402320
@coderzc could you help to cherry-pick this to branch-3.0? I see conflict.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] coderzc commented on a diff in pull request #21187: [fix][broker] Avoid consumers receiving acknowledged messages from compacted topic after reconnection
Posted by "coderzc (via GitHub)" <gi...@apache.org>.
coderzc commented on code in PR #21187:
URL: https://github.com/apache/pulsar/pull/21187#discussion_r1338063546
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumer.java:
##########
@@ -349,7 +350,11 @@ protected void readMoreEntries(Consumer consumer) {
}
havePendingRead = true;
if (consumer.readCompacted()) {
- boolean readFromEarliest = isFirstRead && MessageId.earliest.equals(consumer.getStartMessageId());
+ boolean readFromEarliest = false;
+ if (!cursor.isDurable() || ((ManagedCursorImpl) cursor).isCompactionCursor()
+ || cursor.getPersistentMarkDeletedPosition() == null) {
+ readFromEarliest = isFirstRead && MessageId.earliest.equals(consumer.getStartMessageId());
Review Comment:
For `non-durable` consumer, it seems the consumer should read from startMessageId after reconnecting.
Refer to: https://github.com/apache/pulsar/blob/7a3f304d6b6f658ef825b6464f192b42e4833a26/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java#L803-L812
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] codelipenghui commented on a diff in pull request #21187: [fix][broker] Avoid consumers receiving acknowledged messages from compacted topic after reconnection
Posted by "codelipenghui (via GitHub)" <gi...@apache.org>.
codelipenghui commented on code in PR #21187:
URL: https://github.com/apache/pulsar/pull/21187#discussion_r1338016772
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumer.java:
##########
@@ -349,7 +350,11 @@ protected void readMoreEntries(Consumer consumer) {
}
havePendingRead = true;
if (consumer.readCompacted()) {
- boolean readFromEarliest = isFirstRead && MessageId.earliest.equals(consumer.getStartMessageId());
+ boolean readFromEarliest = false;
+ if (!cursor.isDurable() || ((ManagedCursorImpl) cursor).isCompactionCursor()
+ || cursor.getPersistentMarkDeletedPosition() == null) {
+ readFromEarliest = isFirstRead && MessageId.earliest.equals(consumer.getStartMessageId());
Review Comment:
```suggestion
readFromEarliest = isFirstRead && MessageId.earliest.equals(consumer.getStartMessageId());
```
##########
pulsar-broker/src/test/java/org/apache/pulsar/compaction/CompactionTest.java:
##########
@@ -1985,4 +1985,55 @@ public void testCompactionDuplicate() throws Exception {
}
}
}
+
+ @Test(timeOut = 100000)
+ public void testAcknowledge() throws Exception {
+ final String topicName = "persistent://my-property/use/my-ns/testAcknowledge" + UUID.randomUUID();
+ final String subName = "my-sub";
+ @Cleanup
+ PulsarClient client = newPulsarClient(lookupUrl.toString(), 100);
+ Producer<String> producer = pulsarClient.newProducer(Schema.STRING)
+ .enableBatching(false).topic(topicName).create();
+
+ pulsarClient.newConsumer().topic(topicName).subscriptionName(subName).readCompacted(true).subscribe().close();
+
+ Map<String, String> expected = new HashMap<>();
+ for (int i = 0; i < 10; i++) {
+ producer.newMessage().key(String.valueOf(i)).value(String.valueOf(i)).send();
+ expected.put(String.valueOf(i), String.valueOf(i));
+ }
+ producer.flush();
+
+ admin.topics().triggerCompaction(topicName);
+
+ Awaitility.await().untilAsserted(() -> {
+ assertEquals(admin.topics().compactionStatus(topicName).status,
+ LongRunningProcessStatus.Status.SUCCESS);
+ });
+
+ ConsumerImpl<String> consumer = (ConsumerImpl<String>) client.newConsumer(Schema.STRING)
+ .topic(topicName).readCompacted(true).receiverQueueSize(1).subscriptionName(subName)
+ .isAckReceiptEnabled(true)
+ .subscribe();
+
+ for (int i = 0; i < 10; i++) {
+ Message<String> message = consumer.receive(3, TimeUnit.SECONDS);
+ if (message == null) {
+ break;
+ }
+
+ consumer.acknowledge(message);
+ String remove = expected.remove(message.getKey());
+ Assert.assertEquals(remove, message.getValue());
+ }
+
+ // Make consumer reconnect to broker
Review Comment:
It's better to check if the acknowledgments have been processed by the broker. Otherwise, it will introduce a flaky test. You can check the internal stats of the topic or check the backlog of the topic to make sure all the messages have been hacked.
For example:
```java
Awaitility.await().untilAsserted(() ->
assertEquals(admin.topics().getStats(topicName, true).getSubscriptions().get(subName).getMsgBacklog(), 0));
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] poorbarcode commented on a diff in pull request #21187: [fix][broker] Avoid consumers receiving acknowledged messages from compacted topic after reconnection
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #21187:
URL: https://github.com/apache/pulsar/pull/21187#discussion_r1338035184
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumer.java:
##########
@@ -349,7 +350,11 @@ protected void readMoreEntries(Consumer consumer) {
}
havePendingRead = true;
if (consumer.readCompacted()) {
- boolean readFromEarliest = isFirstRead && MessageId.earliest.equals(consumer.getStartMessageId());
+ boolean readFromEarliest = false;
+ if (!cursor.isDurable() || ((ManagedCursorImpl) cursor).isCompactionCursor()
+ || cursor.getPersistentMarkDeletedPosition() == null) {
+ readFromEarliest = isFirstRead && MessageId.earliest.equals(consumer.getStartMessageId());
Review Comment:
The expected behaviors from the users point:
- Create a new reader. It should be reading at the earliest entry.
- Should continue reading from `{markDeletedPosition} + 1` after reconnect (caused by socket closed or other events)<sup>[1]</sup>
- Create a new consumer(`non-durable`). It should be reading at the earliest entry.
- Should continue reading from `{markDeletedPosition} + 1` after reconnect (caused by socket closed or other events)<sup>[2]</sup>
- Create a new consumer(`durable`). It should be reading at `{markDeletedPosition} + 1.`<sup>[3]</sup>
This PR only fixes the case<sup>[3]</sup>, we should guarantee the case<sup>[1]</sup> and case<sup>[2]</sup> too.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Avoid consumers receiving acknowledged messages from compacted topic after reconnection [pulsar]
Posted by "codelipenghui (via GitHub)" <gi...@apache.org>.
codelipenghui commented on code in PR #21187:
URL: https://github.com/apache/pulsar/pull/21187#discussion_r1465816949
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java:
##########
@@ -1045,6 +1048,10 @@ private CompletableFuture<Subscription> getDurableSubscription(String subscripti
}
Map<String, Long> properties = PersistentSubscription.getBaseCursorProperties(replicated);
+ if (readCompacted) {
+ properties = new HashMap<>(properties);
+ properties.put(READ_COMPACTED_CURSOR_PROPERTIES, 1L);
Review Comment:
Actually, it will persistent noting? If a consumer enabled readCompacted but haven't ack messages for a few minutes, we will also get a chance to skip the data if the broker rewind the cursor?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Avoid consumers receiving acknowledged messages from compacted topic after reconnection [pulsar]
Posted by "coderzc (via GitHub)" <gi...@apache.org>.
coderzc commented on code in PR #21187:
URL: https://github.com/apache/pulsar/pull/21187#discussion_r1467222308
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java:
##########
@@ -2519,7 +2525,8 @@ public Position getPersistentMarkDeletedPosition() {
public void rewind() {
lock.writeLock().lock();
try {
- PositionImpl newReadPosition = ledger.getNextValidPosition(markDeletePosition);
+ PositionImpl newReadPosition = isCursorReadFromCompacted() ? markDeletePosition.getNext() :
+ ledger.getNextValidPosition(markDeletePosition);
Review Comment:
We can only skip invalid md-position when recovering cursor and add `rewind(boolean readCompacted)` method.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Avoid consumers receiving acknowledged messages from compacted topic after reconnection [pulsar]
Posted by "codelipenghui (via GitHub)" <gi...@apache.org>.
codelipenghui merged PR #21187:
URL: https://github.com/apache/pulsar/pull/21187
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Avoid consumers receiving acknowledged messages from compacted topic after reconnection [pulsar]
Posted by "coderzc (via GitHub)" <gi...@apache.org>.
coderzc commented on code in PR #21187:
URL: https://github.com/apache/pulsar/pull/21187#discussion_r1464295280
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java:
##########
@@ -2519,7 +2525,8 @@ public Position getPersistentMarkDeletedPosition() {
public void rewind() {
lock.writeLock().lock();
try {
- PositionImpl newReadPosition = ledger.getNextValidPosition(markDeletePosition);
+ PositionImpl newReadPosition = isCursorReadFromCompacted() ? markDeletePosition.getNext() :
+ ledger.getNextValidPosition(markDeletePosition);
Review Comment:
@codelipenghui I mean we will lost the previous md-position after recovering the cursor if we don't persist `readCompacted`.
<img width="755" alt="image" src="https://github.com/apache/pulsar/assets/26179648/e4e5f9c6-0e73-406b-bf04-981b6a77e433">
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Avoid consumers receiving acknowledged messages from compacted topic after reconnection [pulsar]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on code in PR #21187:
URL: https://github.com/apache/pulsar/pull/21187#discussion_r1462661498
##########
pulsar-broker/src/main/java/org/apache/pulsar/compaction/CompactedTopicImpl.java:
##########
@@ -100,7 +100,11 @@ public void asyncReadEntriesOrWait(ManagedCursor cursor,
boolean isFirstRead,
ReadEntriesCallback callback, Consumer consumer) {
PositionImpl cursorPosition;
- if (isFirstRead && MessageId.earliest.equals(consumer.getStartMessageId())){
+ boolean readFromEarliest = isFirstRead && MessageId.earliest.equals(consumer.getStartMessageId())
+ && (!cursor.isDurable() || cursor.getName().equals(Compactor.COMPACTION_SUBSCRIPTION)
+ || cursor.getMarkDeletedPosition() == null
+ || cursor.getMarkDeletedPosition().getEntryId() == -1L);
Review Comment:
Related discussion: https://github.com/apache/pulsar/pull/21187#discussion_r1455147185
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Avoid consumers receiving acknowledged messages from compacted topic after reconnection [pulsar]
Posted by "coderzc (via GitHub)" <gi...@apache.org>.
coderzc commented on code in PR #21187:
URL: https://github.com/apache/pulsar/pull/21187#discussion_r1462691766
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorSubscriptionTest.java:
##########
@@ -868,6 +869,7 @@ public void testReplicatedSubscriptionWithCompaction() throws Exception {
.topic(topicName)
.subscriptionName("sub2")
.subscriptionType(SubscriptionType.Exclusive)
+ .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
Review Comment:
Yeah, `SubscriptionInitialPosition` will work normally.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Avoid consumers receiving acknowledged messages from compacted topic after reconnection [pulsar]
Posted by "coderzc (via GitHub)" <gi...@apache.org>.
coderzc commented on code in PR #21187:
URL: https://github.com/apache/pulsar/pull/21187#discussion_r1467222308
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java:
##########
@@ -2519,7 +2525,8 @@ public Position getPersistentMarkDeletedPosition() {
public void rewind() {
lock.writeLock().lock();
try {
- PositionImpl newReadPosition = ledger.getNextValidPosition(markDeletePosition);
+ PositionImpl newReadPosition = isCursorReadFromCompacted() ? markDeletePosition.getNext() :
+ ledger.getNextValidPosition(markDeletePosition);
Review Comment:
We can only skipping invalid md-position when recovering cursor and add `rewind(boolean readCompacted)` method.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Avoid consumers receiving acknowledged messages from compacted topic after reconnection [pulsar]
Posted by "coderzc (via GitHub)" <gi...@apache.org>.
coderzc commented on code in PR #21187:
URL: https://github.com/apache/pulsar/pull/21187#discussion_r1462656597
##########
pulsar-broker/src/test/java/org/apache/pulsar/compaction/CompactionTest.java:
##########
@@ -2206,4 +2210,158 @@ public void testCompactionWithTTL() throws Exception {
Assert.assertEquals(result, List.of("V3", "V4", "V5"));
}
+
+ @Test
+ public void testAcknowledgeWithReconnection() throws Exception {
+ final String topicName = "persistent://my-property/use/my-ns/testAcknowledge" + UUID.randomUUID();
+ final String subName = "my-sub";
+ @Cleanup
+ PulsarClient client = newPulsarClient(lookupUrl.toString(), 100);
+ Producer<String> producer = pulsarClient.newProducer(Schema.STRING)
+ .enableBatching(false).topic(topicName).create();
+
+ List<String> expected = new ArrayList<>();
+ for (int i = 0; i < 10; i++) {
+ producer.newMessage().key(String.valueOf(i)).value(String.valueOf(i)).send();
+ expected.add(String.valueOf(i));
+ }
+ producer.flush();
+
+ admin.topics().triggerCompaction(topicName);
+
+ Awaitility.await().untilAsserted(() -> {
+ assertEquals(admin.topics().compactionStatus(topicName).status,
+ LongRunningProcessStatus.Status.SUCCESS);
+ });
+
+ // trim the topic
+ admin.topics().unload(topicName);
+
+ Awaitility.await().untilAsserted(() -> {
+ PersistentTopicInternalStats internalStats = admin.topics().getInternalStats(topicName, false);
+ assertEquals(internalStats.numberOfEntries, 0);
+ });
+
+ ConsumerImpl<String> consumer = (ConsumerImpl<String>) client.newConsumer(Schema.STRING)
+ .topic(topicName).readCompacted(true).receiverQueueSize(1).subscriptionName(subName)
+ .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
+ .isAckReceiptEnabled(true)
+ .subscribe();
+
+ List<String> results = new ArrayList<>();
+ for (int i = 0; i < 5; i++) {
+ Message<String> message = consumer.receive(3, TimeUnit.SECONDS);
+ if (message == null) {
+ break;
+ }
+ results.add(message.getValue());
+ consumer.acknowledge(message);
+ }
+
+ Awaitility.await().untilAsserted(() ->
+ assertEquals(admin.topics().getStats(topicName, true).getSubscriptions().get(subName).getMsgBacklog(),
+ 5));
+
+ // Make consumer reconnect to broker
+ admin.topics().unload(topicName);
+
+ // Wait for consumer to reconnect and clear incomingMessages
+ consumer.pause();
+ Awaitility.await().untilAsserted(() -> {
+ Assert.assertEquals(consumer.numMessagesInQueue(), 0);
+ });
+ consumer.resume();
+
+ for (int i = 0; i < 5; i++) {
+ Message<String> message = consumer.receive(3, TimeUnit.SECONDS);
+ if (message == null) {
+ break;
+ }
+ results.add(message.getValue());
+ 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);
+ Assert.assertNull(message);
+
+ // Make consumer reconnect to broker
+ admin.topics().unload(topicName);
+
+ producer.newMessage().key("K").value("V").send();
+ Message<String> message2 = consumer.receive(3, TimeUnit.SECONDS);
+ Assert.assertEquals(message2.getValue(), "V");
+ consumer.acknowledge(message2);
+
+ Awaitility.await().untilAsserted(() -> {
+ PersistentTopicInternalStats internalStats = admin.topics().getInternalStats(topicName);
+ Assert.assertEquals(internalStats.lastConfirmedEntry,
+ internalStats.cursors.get(subName).markDeletePosition);
+ });
+
+ consumer.close();
+ producer.close();
+ }
+
+ @Test
+ public void testEarliestSubsAfterRollover() throws Exception {
Review Comment:
Please see https://github.com/apache/pulsar/pull/21187#discussion_r1456851588
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Avoid consumers receiving acknowledged messages from compacted topic after reconnection [pulsar]
Posted by "coderzc (via GitHub)" <gi...@apache.org>.
coderzc commented on code in PR #21187:
URL: https://github.com/apache/pulsar/pull/21187#discussion_r1462655970
##########
pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ReaderTest.java:
##########
@@ -785,4 +785,32 @@ public void testReaderListenerAcknowledgement()
admin.topics().deletePartitionedTopic(partitionedTopic);
}
+ @Test
+ public void testReaderReconnectedFromNextEntry() throws Exception {
Review Comment:
This PR will not affect the behavior of the reader. This test is just to cover the reader test.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Avoid consumers receiving acknowledged messages from compacted topic after reconnection [pulsar]
Posted by "codelipenghui (via GitHub)" <gi...@apache.org>.
codelipenghui commented on code in PR #21187:
URL: https://github.com/apache/pulsar/pull/21187#discussion_r1464256888
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java:
##########
@@ -2519,7 +2525,8 @@ public Position getPersistentMarkDeletedPosition() {
public void rewind() {
lock.writeLock().lock();
try {
- PositionImpl newReadPosition = ledger.getNextValidPosition(markDeletePosition);
+ PositionImpl newReadPosition = isCursorReadFromCompacted() ? markDeletePosition.getNext() :
+ ledger.getNextValidPosition(markDeletePosition);
Review Comment:
Oh, I found a similar [PR](https://github.com/apache/pulsar/pull/13629) which fixed the Reader API.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Avoid consumers receiving acknowledged messages from compacted topic after reconnection [pulsar]
Posted by "coderzc (via GitHub)" <gi...@apache.org>.
coderzc commented on code in PR #21187:
URL: https://github.com/apache/pulsar/pull/21187#discussion_r1462691000
##########
pulsar-broker/src/main/java/org/apache/pulsar/compaction/CompactedTopicImpl.java:
##########
@@ -100,7 +100,11 @@ public void asyncReadEntriesOrWait(ManagedCursor cursor,
boolean isFirstRead,
ReadEntriesCallback callback, Consumer consumer) {
PositionImpl cursorPosition;
- if (isFirstRead && MessageId.earliest.equals(consumer.getStartMessageId())){
+ boolean readFromEarliest = isFirstRead && MessageId.earliest.equals(consumer.getStartMessageId())
+ && (!cursor.isDurable() || cursor.getName().equals(Compactor.COMPACTION_SUBSCRIPTION)
+ || cursor.getMarkDeletedPosition() == null
+ || cursor.getMarkDeletedPosition().getEntryId() == -1L);
Review Comment:
> Sorry, I mean why we need to check cursor.getMarkDeletedPosition().getEntryId() == -1L?
Oh, since if `markDeletedPosition.entryID == -1L` then the md-position is an invalid position, the initial md-position of the consumer will be set to it.
https://github.com/apache/pulsar/blob/b14fcb47f20398bdf5b698e158346dce19f0edc6/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java#L997-L998
https://github.com/apache/pulsar/blob/b14fcb47f20398bdf5b698e158346dce19f0edc6/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java#L3715-L3725
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Avoid consumers receiving acknowledged messages from compacted topic after reconnection [pulsar]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on code in PR #21187:
URL: https://github.com/apache/pulsar/pull/21187#discussion_r1456918665
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java:
##########
@@ -1017,6 +1021,10 @@ private CompletableFuture<Subscription> getDurableSubscription(String subscripti
}
Map<String, Long> properties = PersistentSubscription.getBaseCursorProperties(replicated);
+ if (readCompacted) {
+ properties = new TreeMap<>(properties);
+ properties.put(READ_COMPACTED_CURSOR_PROPERTIES, 1L);
Review Comment:
Yeah, we can create a new map and copy the elements from the immutable map.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Avoid consumers receiving acknowledged messages from compacted topic after reconnection [pulsar]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on code in PR #21187:
URL: https://github.com/apache/pulsar/pull/21187#discussion_r1456918282
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumer.java:
##########
@@ -362,7 +363,12 @@ private void readMoreEntries(Consumer consumer) {
}
havePendingRead = true;
if (consumer.readCompacted()) {
- boolean readFromEarliest = isFirstRead && MessageId.earliest.equals(consumer.getStartMessageId());
+ boolean readFromEarliest = false;
+ if (!cursor.isDurable() || ((ManagedCursorImpl) cursor).isCompactionCursor()
+ || cursor.getMarkDeletedPosition() == null
+ || cursor.getMarkDeletedPosition().getEntryId() == -1L) {
+ readFromEarliest = isFirstRead && MessageId.earliest.equals(consumer.getStartMessageId());
+ }
Review Comment:
> If the mark-delete position is null and subscriptionInitialPosition=Earliest
Could you add a test for it?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Avoid consumers receiving acknowledged messages from compacted topic after reconnection [pulsar]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on PR #21187:
URL: https://github.com/apache/pulsar/pull/21187#issuecomment-1893475424
Some tests (e.g. `testReplicatedSubscriptionWithCompaction`) failed, please check them.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Avoid consumers receiving acknowledged messages from compacted topic after reconnection [pulsar]
Posted by "coderzc (via GitHub)" <gi...@apache.org>.
coderzc commented on code in PR #21187:
URL: https://github.com/apache/pulsar/pull/21187#discussion_r1460104498
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumer.java:
##########
@@ -362,7 +363,12 @@ private void readMoreEntries(Consumer consumer) {
}
havePendingRead = true;
if (consumer.readCompacted()) {
- boolean readFromEarliest = isFirstRead && MessageId.earliest.equals(consumer.getStartMessageId());
+ boolean readFromEarliest = false;
+ if (!cursor.isDurable() || ((ManagedCursorImpl) cursor).isCompactionCursor()
+ || cursor.getMarkDeletedPosition() == null
+ || cursor.getMarkDeletedPosition().getEntryId() == -1L) {
+ readFromEarliest = isFirstRead && MessageId.earliest.equals(consumer.getStartMessageId());
+ }
Review Comment:
Yeah, after unloading the topic, also need to write a new message otherwise `ml.getFirstPosition()` returns the old ledger position, I added `testEarliestSubsAfterRollover` to cover this case.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Avoid consumers receiving acknowledged messages from compacted topic after reconnection [pulsar]
Posted by "liangyepianzhou (via GitHub)" <gi...@apache.org>.
liangyepianzhou commented on PR #21187:
URL: https://github.com/apache/pulsar/pull/21187#issuecomment-1895436206
LGTM, the system topic of Transaction is not affected by this modification.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Avoid consumers receiving acknowledged messages from compacted topic after reconnection [pulsar]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on code in PR #21187:
URL: https://github.com/apache/pulsar/pull/21187#discussion_r1458577629
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumer.java:
##########
@@ -362,7 +363,12 @@ private void readMoreEntries(Consumer consumer) {
}
havePendingRead = true;
if (consumer.readCompacted()) {
- boolean readFromEarliest = isFirstRead && MessageId.earliest.equals(consumer.getStartMessageId());
+ boolean readFromEarliest = false;
+ if (!cursor.isDurable() || ((ManagedCursorImpl) cursor).isCompactionCursor()
+ || cursor.getMarkDeletedPosition() == null
+ || cursor.getMarkDeletedPosition().getEntryId() == -1L) {
+ readFromEarliest = isFirstRead && MessageId.earliest.equals(consumer.getStartMessageId());
+ }
Review Comment:
I removed the check for mark-delete position and `testAcknowledgeWithReconnection` still passed. So I think it's not covered.
```java
boolean readFromEarliest = isFirstRead && MessageId.earliest.equals(consumer.getStartMessageId())
&& (!cursor.isDurable() || cursor.getName().equals(Compactor.COMPACTION_SUBSCRIPTION));
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Avoid consumers receiving acknowledged messages from compacted topic after reconnection [pulsar]
Posted by "codelipenghui (via GitHub)" <gi...@apache.org>.
codelipenghui commented on code in PR #21187:
URL: https://github.com/apache/pulsar/pull/21187#discussion_r1461839137
##########
pulsar-broker/src/main/java/org/apache/pulsar/compaction/CompactedTopicImpl.java:
##########
@@ -100,7 +100,11 @@ public void asyncReadEntriesOrWait(ManagedCursor cursor,
boolean isFirstRead,
ReadEntriesCallback callback, Consumer consumer) {
PositionImpl cursorPosition;
- if (isFirstRead && MessageId.earliest.equals(consumer.getStartMessageId())){
+ boolean readFromEarliest = isFirstRead && MessageId.earliest.equals(consumer.getStartMessageId())
+ && (!cursor.isDurable() || cursor.getName().equals(Compactor.COMPACTION_SUBSCRIPTION)
+ || cursor.getMarkDeletedPosition() == null
+ || cursor.getMarkDeletedPosition().getEntryId() == -1L);
Review Comment:
What does this mean?
##########
pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ReaderTest.java:
##########
@@ -785,4 +785,32 @@ public void testReaderListenerAcknowledgement()
admin.topics().deletePartitionedTopic(partitionedTopic);
}
+ @Test
+ public void testReaderReconnectedFromNextEntry() throws Exception {
Review Comment:
I tried to run this test without this fix many times.
But the test can get passed.
<img width="1649" alt="image" src="https://github.com/apache/pulsar/assets/12592133/197cdb56-49c8-44b8-af0e-9e273b145c6b">
Could you please check if the test can actually cover the issue that this PR try to fix?
##########
pulsar-broker/src/test/java/org/apache/pulsar/compaction/CompactionTest.java:
##########
@@ -2206,4 +2210,158 @@ public void testCompactionWithTTL() throws Exception {
Assert.assertEquals(result, List.of("V3", "V4", "V5"));
}
+
+ @Test
+ public void testAcknowledgeWithReconnection() throws Exception {
+ final String topicName = "persistent://my-property/use/my-ns/testAcknowledge" + UUID.randomUUID();
+ final String subName = "my-sub";
+ @Cleanup
+ PulsarClient client = newPulsarClient(lookupUrl.toString(), 100);
+ Producer<String> producer = pulsarClient.newProducer(Schema.STRING)
+ .enableBatching(false).topic(topicName).create();
+
+ List<String> expected = new ArrayList<>();
+ for (int i = 0; i < 10; i++) {
+ producer.newMessage().key(String.valueOf(i)).value(String.valueOf(i)).send();
+ expected.add(String.valueOf(i));
+ }
+ producer.flush();
+
+ admin.topics().triggerCompaction(topicName);
+
+ Awaitility.await().untilAsserted(() -> {
+ assertEquals(admin.topics().compactionStatus(topicName).status,
+ LongRunningProcessStatus.Status.SUCCESS);
+ });
+
+ // trim the topic
+ admin.topics().unload(topicName);
+
+ Awaitility.await().untilAsserted(() -> {
+ PersistentTopicInternalStats internalStats = admin.topics().getInternalStats(topicName, false);
+ assertEquals(internalStats.numberOfEntries, 0);
+ });
+
+ ConsumerImpl<String> consumer = (ConsumerImpl<String>) client.newConsumer(Schema.STRING)
+ .topic(topicName).readCompacted(true).receiverQueueSize(1).subscriptionName(subName)
+ .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
+ .isAckReceiptEnabled(true)
+ .subscribe();
+
+ List<String> results = new ArrayList<>();
+ for (int i = 0; i < 5; i++) {
+ Message<String> message = consumer.receive(3, TimeUnit.SECONDS);
+ if (message == null) {
+ break;
+ }
+ results.add(message.getValue());
+ consumer.acknowledge(message);
+ }
+
+ Awaitility.await().untilAsserted(() ->
+ assertEquals(admin.topics().getStats(topicName, true).getSubscriptions().get(subName).getMsgBacklog(),
+ 5));
+
+ // Make consumer reconnect to broker
+ admin.topics().unload(topicName);
+
+ // Wait for consumer to reconnect and clear incomingMessages
+ consumer.pause();
+ Awaitility.await().untilAsserted(() -> {
+ Assert.assertEquals(consumer.numMessagesInQueue(), 0);
+ });
+ consumer.resume();
+
+ for (int i = 0; i < 5; i++) {
+ Message<String> message = consumer.receive(3, TimeUnit.SECONDS);
+ if (message == null) {
+ break;
+ }
+ results.add(message.getValue());
+ 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);
+ Assert.assertNull(message);
+
+ // Make consumer reconnect to broker
+ admin.topics().unload(topicName);
+
+ producer.newMessage().key("K").value("V").send();
+ Message<String> message2 = consumer.receive(3, TimeUnit.SECONDS);
+ Assert.assertEquals(message2.getValue(), "V");
+ consumer.acknowledge(message2);
+
+ Awaitility.await().untilAsserted(() -> {
+ PersistentTopicInternalStats internalStats = admin.topics().getInternalStats(topicName);
+ Assert.assertEquals(internalStats.lastConfirmedEntry,
+ internalStats.cursors.get(subName).markDeletePosition);
+ });
+
+ consumer.close();
+ producer.close();
+ }
+
+ @Test
+ public void testEarliestSubsAfterRollover() throws Exception {
Review Comment:
This test can also get passed without this fix.
@coderzc Could you please check if the test can cover the issue?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Avoid consumers receiving acknowledged messages from compacted topic after reconnection [pulsar]
Posted by "coderzc (via GitHub)" <gi...@apache.org>.
coderzc commented on code in PR #21187:
URL: https://github.com/apache/pulsar/pull/21187#discussion_r1457383266
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumer.java:
##########
@@ -362,7 +363,12 @@ private void readMoreEntries(Consumer consumer) {
}
havePendingRead = true;
if (consumer.readCompacted()) {
- boolean readFromEarliest = isFirstRead && MessageId.earliest.equals(consumer.getStartMessageId());
+ boolean readFromEarliest = false;
+ if (!cursor.isDurable() || ((ManagedCursorImpl) cursor).isCompactionCursor()
+ || cursor.getMarkDeletedPosition() == null
+ || cursor.getMarkDeletedPosition().getEntryId() == -1L) {
+ readFromEarliest = isFirstRead && MessageId.earliest.equals(consumer.getStartMessageId());
+ }
Review Comment:
`testAcknowledgeWithReconnection` already covers this case.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Avoid consumers receiving acknowledged messages from compacted topic after reconnection [pulsar]
Posted by "coderzc (via GitHub)" <gi...@apache.org>.
coderzc commented on code in PR #21187:
URL: https://github.com/apache/pulsar/pull/21187#discussion_r1462691766
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorSubscriptionTest.java:
##########
@@ -868,6 +869,7 @@ public void testReplicatedSubscriptionWithCompaction() throws Exception {
.topic(topicName)
.subscriptionName("sub2")
.subscriptionType(SubscriptionType.Exclusive)
+ .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
Review Comment:
Yeah, `SubscriptionInitialPosition` will work normally
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Avoid consumers receiving acknowledged messages from compacted topic after reconnection [pulsar]
Posted by "codelipenghui (via GitHub)" <gi...@apache.org>.
codelipenghui commented on code in PR #21187:
URL: https://github.com/apache/pulsar/pull/21187#discussion_r1465816949
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java:
##########
@@ -1045,6 +1048,10 @@ private CompletableFuture<Subscription> getDurableSubscription(String subscripti
}
Map<String, Long> properties = PersistentSubscription.getBaseCursorProperties(replicated);
+ if (readCompacted) {
+ properties = new HashMap<>(properties);
+ properties.put(READ_COMPACTED_CURSOR_PROPERTIES, 1L);
Review Comment:
Actually, it will persistent noting? If a consumer enabled readCompacted but haven't ack messages for a few minutes, we will also get a chance to skip the data if the broker rewind the cursor?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Avoid consumers receiving acknowledged messages from compacted topic after reconnection [pulsar]
Posted by "coderzc (via GitHub)" <gi...@apache.org>.
coderzc commented on code in PR #21187:
URL: https://github.com/apache/pulsar/pull/21187#discussion_r1462658698
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorSubscriptionTest.java:
##########
@@ -868,6 +869,7 @@ public void testReplicatedSubscriptionWithCompaction() throws Exception {
.topic(topicName)
.subscriptionName("sub2")
.subscriptionType(SubscriptionType.Exclusive)
+ .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
Review Comment:
Yes, consumer always be read from earliest position after this PR.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] poorbarcode commented on a diff in pull request #21187: [fix][broker] Avoid consumers receiving acknowledged messages from compacted topic after reconnection
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #21187:
URL: https://github.com/apache/pulsar/pull/21187#discussion_r1338461297
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumer.java:
##########
@@ -349,7 +350,11 @@ protected void readMoreEntries(Consumer consumer) {
}
havePendingRead = true;
if (consumer.readCompacted()) {
- boolean readFromEarliest = isFirstRead && MessageId.earliest.equals(consumer.getStartMessageId());
+ boolean readFromEarliest = false;
+ if (!cursor.isDurable() || ((ManagedCursorImpl) cursor).isCompactionCursor()
+ || cursor.getPersistentMarkDeletedPosition() == null) {
+ readFromEarliest = isFirstRead && MessageId.earliest.equals(consumer.getStartMessageId());
Review Comment:
Got it. I wrote a test below(it works as expected). Could you add it into the current PR to guarantee this feature will not be broken in the future?
```java
@Test
public void testReaderReconnectedFromNextEntry() throws Exception {
final String topic = newTopicName();
Reader<String> reader = pulsarClient.newReader(Schema.STRING).topic(topic).receiverQueueSize(1)
.startMessageId(MessageId.earliest).create();
Producer<String> producer = pulsarClient.newProducer(Schema.STRING).topic(topic).create();
// Send 3 and consume 1.
producer.send("1");
producer.send("2");
producer.send("3");
Message<String> msg1 = reader.readNext(2, TimeUnit.SECONDS);
assertEquals(msg1.getValue(), "1");
// Trigger reader reconnect.
admin.topics().unload(topic);
// For non-durable we are going to restart from the next entry.
Message<String> msg2 = reader.readNext(2, TimeUnit.SECONDS);
assertEquals(msg2.getValue(), "2");
Message<String> msg3 = reader.readNext(2, TimeUnit.SECONDS);
assertEquals(msg3.getValue(), "3");
// cleanup.
reader.close();
producer.close();
admin.topics().delete(topic, false);
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] poorbarcode commented on a diff in pull request #21187: [fix][broker] Avoid consumers receiving acknowledged messages from compacted topic after reconnection
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #21187:
URL: https://github.com/apache/pulsar/pull/21187#discussion_r1338035184
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumer.java:
##########
@@ -349,7 +350,11 @@ protected void readMoreEntries(Consumer consumer) {
}
havePendingRead = true;
if (consumer.readCompacted()) {
- boolean readFromEarliest = isFirstRead && MessageId.earliest.equals(consumer.getStartMessageId());
+ boolean readFromEarliest = false;
+ if (!cursor.isDurable() || ((ManagedCursorImpl) cursor).isCompactionCursor()
+ || cursor.getPersistentMarkDeletedPosition() == null) {
+ readFromEarliest = isFirstRead && MessageId.earliest.equals(consumer.getStartMessageId());
Review Comment:
The expected behaviors from the users point:
- Create a new reader. It should be reading at the earliest entry.
- Should continue reading from `{markDeletedPosition} + 1` after reconnect(caused by socket closed or other cases)<sup>[1]</sup>
- Create a new consumer(`non-durable`). It should be reading at the earliest entry.
- Should continue reading from `{markDeletedPosition} + 1` after reconnect(caused by socket closed or other cases)<sup>[2]</sup>
- Create a new consumer(`durable`). It should be reading at `{markDeletedPosition} + 1.`<sup>[3]</sup>
This PR only fixes the case<sup>[3]</sup>, we should guarantee the case<sup>[1]</sup> and case<sup>[2]</sup> too.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] coderzc commented on a diff in pull request #21187: [fix][broker] Avoid consumers receiving acknowledged messages from compacted topic after reconnection
Posted by "coderzc (via GitHub)" <gi...@apache.org>.
coderzc commented on code in PR #21187:
URL: https://github.com/apache/pulsar/pull/21187#discussion_r1339428718
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumer.java:
##########
@@ -349,7 +350,11 @@ protected void readMoreEntries(Consumer consumer) {
}
havePendingRead = true;
if (consumer.readCompacted()) {
- boolean readFromEarliest = isFirstRead && MessageId.earliest.equals(consumer.getStartMessageId());
+ boolean readFromEarliest = false;
+ if (!cursor.isDurable() || ((ManagedCursorImpl) cursor).isCompactionCursor()
+ || cursor.getPersistentMarkDeletedPosition() == null) {
+ readFromEarliest = isFirstRead && MessageId.earliest.equals(consumer.getStartMessageId());
Review Comment:
Thanks, I already add this test into PR.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Avoid consumers receiving acknowledged messages from compacted topic after reconnection [pulsar]
Posted by "coderzc (via GitHub)" <gi...@apache.org>.
coderzc commented on code in PR #21187:
URL: https://github.com/apache/pulsar/pull/21187#discussion_r1462661695
##########
pulsar-broker/src/main/java/org/apache/pulsar/compaction/CompactedTopicImpl.java:
##########
@@ -100,7 +100,11 @@ public void asyncReadEntriesOrWait(ManagedCursor cursor,
boolean isFirstRead,
ReadEntriesCallback callback, Consumer consumer) {
PositionImpl cursorPosition;
- if (isFirstRead && MessageId.earliest.equals(consumer.getStartMessageId())){
+ boolean readFromEarliest = isFirstRead && MessageId.earliest.equals(consumer.getStartMessageId())
+ && (!cursor.isDurable() || cursor.getName().equals(Compactor.COMPACTION_SUBSCRIPTION)
+ || cursor.getMarkDeletedPosition() == null
+ || cursor.getMarkDeletedPosition().getEntryId() == -1L);
Review Comment:
Just like described in the PR's description，If a durable cursor (exclude __compaction cursor) has the mark-delete position, don't read messages from the earliest.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Avoid consumers receiving acknowledged messages from compacted topic after reconnection [pulsar]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #21187:
URL: https://github.com/apache/pulsar/pull/21187#issuecomment-1893863461
## [Codecov](https://app.codecov.io/gh/apache/pulsar/pull/21187?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
Attention: `7 lines` in your changes are missing coverage. Please review.
> Comparison is base [(`c3f1ee0`)](https://app.codecov.io/gh/apache/pulsar/commit/c3f1ee01a376541dde8a5deda7ee99c0beb65dc0?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) 36.58% compared to head [(`7b77cbe`)](https://app.codecov.io/gh/apache/pulsar/pull/21187?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) 73.58%.
> Report is 5 commits behind head on master.
<details><summary>Additional details and impacted files</summary>
[](https://app.codecov.io/gh/apache/pulsar/pull/21187?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
```diff
@@ Coverage Diff @@
## master #21187 +/- ##
=============================================
+ Coverage 36.58% 73.58% +37.00%
- Complexity 12348 32370 +20022
=============================================
Files 1723 1859 +136
Lines 131459 138397 +6938
Branches 14374 15175 +801
=============================================
+ Hits 48092 101845 +53753
+ Misses 76948 28668 -48280
- Partials 6419 7884 +1465
```
| [Flag](https://app.codecov.io/gh/apache/pulsar/pull/21187/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [inttests](https://app.codecov.io/gh/apache/pulsar/pull/21187/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `24.12% <68.96%> (-0.08%)` | :arrow_down: |
| [systests](https://app.codecov.io/gh/apache/pulsar/pull/21187/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `23.75% <65.51%> (-0.13%)` | :arrow_down: |
| [unittests](https://app.codecov.io/gh/apache/pulsar/pull/21187/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `72.86% <75.86%> (+40.91%)` | :arrow_up: |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#carryforward-flags-in-the-pull-request-comment) to find out more.
| [Files](https://app.codecov.io/gh/apache/pulsar/pull/21187?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [...che/bookkeeper/mledger/impl/ManagedCursorImpl.java](https://app.codecov.io/gh/apache/pulsar/pull/21187?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-bWFuYWdlZC1sZWRnZXIvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2Jvb2trZWVwZXIvbWxlZGdlci9pbXBsL01hbmFnZWRDdXJzb3JJbXBsLmphdmE=) | `79.55% <100.00%> (+31.84%)` | :arrow_up: |
| [...ava/org/apache/pulsar/broker/service/Consumer.java](https://app.codecov.io/gh/apache/pulsar/pull/21187?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWJyb2tlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2Jyb2tlci9zZXJ2aWNlL0NvbnN1bWVyLmphdmE=) | `87.41% <100.00%> (+24.67%)` | :arrow_up: |
| [...sar/broker/service/persistent/PersistentTopic.java](https://app.codecov.io/gh/apache/pulsar/pull/21187?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWJyb2tlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2Jyb2tlci9zZXJ2aWNlL3BlcnNpc3RlbnQvUGVyc2lzdGVudFRvcGljLmphdmE=) | `78.52% <100.00%> (+26.71%)` | :arrow_up: |
| [...tent/PersistentDispatcherSingleActiveConsumer.java](https://app.codecov.io/gh/apache/pulsar/pull/21187?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWJyb2tlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2Jyb2tlci9zZXJ2aWNlL3BlcnNpc3RlbnQvUGVyc2lzdGVudERpc3BhdGNoZXJTaW5nbGVBY3RpdmVDb25zdW1lci5qYXZh) | `70.55% <80.00%> (+10.92%)` | :arrow_up: |
| [...g/apache/pulsar/compaction/CompactedTopicImpl.java](https://app.codecov.io/gh/apache/pulsar/pull/21187?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWJyb2tlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2NvbXBhY3Rpb24vQ29tcGFjdGVkVG9waWNJbXBsLmphdmE=) | `59.45% <0.00%> (-0.69%)` | :arrow_down: |
... and [1429 files with indirect coverage changes](https://app.codecov.io/gh/apache/pulsar/pull/21187/indirect-changes?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Avoid consumers receiving acknowledged messages from compacted topic after reconnection [pulsar]
Posted by "codelipenghui (via GitHub)" <gi...@apache.org>.
codelipenghui commented on code in PR #21187:
URL: https://github.com/apache/pulsar/pull/21187#discussion_r1462681458
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorSubscriptionTest.java:
##########
@@ -868,6 +869,7 @@ public void testReplicatedSubscriptionWithCompaction() throws Exception {
.topic(topicName)
.subscriptionName("sub2")
.subscriptionType(SubscriptionType.Exclusive)
+ .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
Review Comment:
Ok, I see. So after this change, the consumer will start to read from the latest if it doesn't have `SubscriptionInitialPosition.Earliest`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Avoid consumers receiving acknowledged messages from compacted topic after reconnection [pulsar]
Posted by "coderzc (via GitHub)" <gi...@apache.org>.
coderzc commented on code in PR #21187:
URL: https://github.com/apache/pulsar/pull/21187#discussion_r1458174072
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java:
##########
@@ -443,6 +445,10 @@ public CompletableFuture<Void> messageAcked(CommandAck ack) {
properties = ack.getPropertiesList().stream()
.collect(Collectors.toMap(KeyLongValue::getKey, KeyLongValue::getValue));
}
+ if (readCompacted) {
+ properties = new TreeMap<>(properties);
+ properties.put(READ_COMPACTED_CURSOR_PROPERTIES, 1L);
+ }
Review Comment:
Using `HashMap` instead.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Avoid consumers receiving acknowledged messages from compacted topic after reconnection [pulsar]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on code in PR #21187:
URL: https://github.com/apache/pulsar/pull/21187#discussion_r1456923986
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumer.java:
##########
@@ -362,7 +363,12 @@ private void readMoreEntries(Consumer consumer) {
}
havePendingRead = true;
if (consumer.readCompacted()) {
- boolean readFromEarliest = isFirstRead && MessageId.earliest.equals(consumer.getStartMessageId());
+ boolean readFromEarliest = false;
+ if (!cursor.isDurable() || cursor.getName().equals(Compactor.COMPACTION_SUBSCRIPTION)
+ || cursor.getMarkDeletedPosition() == null
+ || cursor.getMarkDeletedPosition().getEntryId() == -1L) {
+ readFromEarliest = isFirstRead && MessageId.earliest.equals(consumer.getStartMessageId());
+ }
Review Comment:
```suggestion
boolean readFromEarliest = isFirstRead && MessageId.earliest.equals(consumer.getStartMessageId())
&& (!cursor.isDurable()
|| cursor.getName().equals(Compactor.COMPACTION_SUBSCRIPTION)
|| cursor.getMarkDeletedPosition() == null
|| cursor.getMarkDeletedPosition().getEntryId() == -1L);
```
Merge all conditions into a single expression. And you don't need to check `cursor.xxx()` when `isFirstRead` is false after that.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Avoid consumers receiving acknowledged messages from compacted topic after reconnection [pulsar]
Posted by "coderzc (via GitHub)" <gi...@apache.org>.
coderzc commented on code in PR #21187:
URL: https://github.com/apache/pulsar/pull/21187#discussion_r1462719147
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java:
##########
@@ -2519,7 +2525,8 @@ public Position getPersistentMarkDeletedPosition() {
public void rewind() {
lock.writeLock().lock();
try {
- PositionImpl newReadPosition = ledger.getNextValidPosition(markDeletePosition);
+ PositionImpl newReadPosition = isCursorReadFromCompacted() ? markDeletePosition.getNext() :
+ ledger.getNextValidPosition(markDeletePosition);
Review Comment:
But this will affect the `CursorStats` and add some CPU cycles. I think we can apply this change only to cursors that have readCompacted enabled.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Avoid consumers receiving acknowledged messages from compacted topic after reconnection [pulsar]
Posted by "coderzc (via GitHub)" <gi...@apache.org>.
coderzc commented on code in PR #21187:
URL: https://github.com/apache/pulsar/pull/21187#discussion_r1464262263
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java:
##########
@@ -2519,7 +2525,8 @@ public Position getPersistentMarkDeletedPosition() {
public void rewind() {
lock.writeLock().lock();
try {
- PositionImpl newReadPosition = ledger.getNextValidPosition(markDeletePosition);
+ PositionImpl newReadPosition = isCursorReadFromCompacted() ? markDeletePosition.getNext() :
+ ledger.getNextValidPosition(markDeletePosition);
Review Comment:
@codelipenghui
> After checking the changes again. It looks like we don't need to persist the readCompacted? If depends on the consumer settings. We only need to set the readCompacted when new consumer connected?
I don't think this will work well because when recovering the cursor, maybe the consumer is not connected yet, then our md-position will be forced to move to the first position of the ml.
please see:
https://github.com/apache/pulsar/pull/21187/files#diff-fad355f91bd15cc041161f9a46fce62b7fee87fbfb8f0ff8a8b724a1bd1f29eeR685-R695
> Oh, I found a similar PR which fixed the Reader API.
Yes, they are a issue
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java:
##########
@@ -2519,7 +2525,8 @@ public Position getPersistentMarkDeletedPosition() {
public void rewind() {
lock.writeLock().lock();
try {
- PositionImpl newReadPosition = ledger.getNextValidPosition(markDeletePosition);
+ PositionImpl newReadPosition = isCursorReadFromCompacted() ? markDeletePosition.getNext() :
+ ledger.getNextValidPosition(markDeletePosition);
Review Comment:
@codelipenghui
> After checking the changes again. It looks like we don't need to persist the readCompacted? If depends on the consumer settings. We only need to set the readCompacted when new consumer connected?
I don't think this will work well because when recovering the cursor, maybe the consumer is not connected yet, then our md-position will be forced to move to the first position of the ml.
please see:
https://github.com/apache/pulsar/pull/21187/files#diff-fad355f91bd15cc041161f9a46fce62b7fee87fbfb8f0ff8a8b724a1bd1f29eeR685-R695
> Oh, I found a similar PR which fixed the Reader API.
Yes, they are a issue
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Avoid consumers receiving acknowledged messages from compacted topic after reconnection [pulsar]
Posted by "coderzc (via GitHub)" <gi...@apache.org>.
coderzc commented on code in PR #21187:
URL: https://github.com/apache/pulsar/pull/21187#discussion_r1464262263
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java:
##########
@@ -2519,7 +2525,8 @@ public Position getPersistentMarkDeletedPosition() {
public void rewind() {
lock.writeLock().lock();
try {
- PositionImpl newReadPosition = ledger.getNextValidPosition(markDeletePosition);
+ PositionImpl newReadPosition = isCursorReadFromCompacted() ? markDeletePosition.getNext() :
+ ledger.getNextValidPosition(markDeletePosition);
Review Comment:
@codelipenghui
> After checking the changes again. It looks like we don't need to persist the readCompacted? If depends on the consumer settings. We only need to set the readCompacted when new consumer connected?
I don't think this will work well because when recovering the cursor, maybe the consumer is not connected yet, then the md-position will be forced to move to the first position of the ml.
please see:
https://github.com/apache/pulsar/pull/21187/files#diff-fad355f91bd15cc041161f9a46fce62b7fee87fbfb8f0ff8a8b724a1bd1f29eeR685-R697
> Oh, I found a similar PR which fixed the Reader API.
Yes, they are a issue
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Avoid consumers receiving acknowledged messages from compacted topic after reconnection [pulsar]
Posted by "coderzc (via GitHub)" <gi...@apache.org>.
coderzc commented on code in PR #21187:
URL: https://github.com/apache/pulsar/pull/21187#discussion_r1464295280
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java:
##########
@@ -2519,7 +2525,8 @@ public Position getPersistentMarkDeletedPosition() {
public void rewind() {
lock.writeLock().lock();
try {
- PositionImpl newReadPosition = ledger.getNextValidPosition(markDeletePosition);
+ PositionImpl newReadPosition = isCursorReadFromCompacted() ? markDeletePosition.getNext() :
+ ledger.getNextValidPosition(markDeletePosition);
Review Comment:
@codelipenghui I mean we will lost the previous md-position after recovering the cursor if we don't record `readCompacted`.
<img width="755" alt="image" src="https://github.com/apache/pulsar/assets/26179648/e4e5f9c6-0e73-406b-bf04-981b6a77e433">
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Avoid consumers receiving acknowledged messages from compacted topic after reconnection [pulsar]
Posted by "coderzc (via GitHub)" <gi...@apache.org>.
coderzc commented on code in PR #21187:
URL: https://github.com/apache/pulsar/pull/21187#discussion_r1456851588
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumer.java:
##########
@@ -362,7 +363,12 @@ private void readMoreEntries(Consumer consumer) {
}
havePendingRead = true;
if (consumer.readCompacted()) {
- boolean readFromEarliest = isFirstRead && MessageId.earliest.equals(consumer.getStartMessageId());
+ boolean readFromEarliest = false;
+ if (!cursor.isDurable() || ((ManagedCursorImpl) cursor).isCompactionCursor()
+ || cursor.getMarkDeletedPosition() == null
+ || cursor.getMarkDeletedPosition().getEntryId() == -1L) {
+ readFromEarliest = isFirstRead && MessageId.earliest.equals(consumer.getStartMessageId());
+ }
Review Comment:
> And why did you also check the mark deleted position here?
If the mark-delete position is null and `subscriptionInitialPosition=Earliest` then we still need to read messages from earliest when the consumer first reads.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Avoid consumers receiving acknowledged messages from compacted topic after reconnection [pulsar]
Posted by "coderzc (via GitHub)" <gi...@apache.org>.
coderzc commented on code in PR #21187:
URL: https://github.com/apache/pulsar/pull/21187#discussion_r1464267713
##########
pulsar-broker/src/main/java/org/apache/pulsar/compaction/CompactedTopicImpl.java:
##########
@@ -100,7 +100,11 @@ public void asyncReadEntriesOrWait(ManagedCursor cursor,
boolean isFirstRead,
ReadEntriesCallback callback, Consumer consumer) {
PositionImpl cursorPosition;
- if (isFirstRead && MessageId.earliest.equals(consumer.getStartMessageId())){
+ boolean readFromEarliest = isFirstRead && MessageId.earliest.equals(consumer.getStartMessageId())
+ && (!cursor.isDurable() || cursor.getName().equals(Compactor.COMPACTION_SUBSCRIPTION)
+ || cursor.getMarkDeletedPosition() == null
+ || cursor.getMarkDeletedPosition().getEntryId() == -1L);
Review Comment:
> Ok, I see. Could you please help add some comments to improve the code readability?
@codelipenghui I add some comments and `hasValidMarkDeletePosition` method.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Avoid consumers receiving acknowledged messages from compacted topic after reconnection [pulsar]
Posted by "coderzc (via GitHub)" <gi...@apache.org>.
coderzc commented on code in PR #21187:
URL: https://github.com/apache/pulsar/pull/21187#discussion_r1464295280
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java:
##########
@@ -2519,7 +2525,8 @@ public Position getPersistentMarkDeletedPosition() {
public void rewind() {
lock.writeLock().lock();
try {
- PositionImpl newReadPosition = ledger.getNextValidPosition(markDeletePosition);
+ PositionImpl newReadPosition = isCursorReadFromCompacted() ? markDeletePosition.getNext() :
+ ledger.getNextValidPosition(markDeletePosition);
Review Comment:
@codelipenghui I mean we will lost the previous md-position after recovering the cursor if we don't record `readCompacted`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Avoid consumers receiving acknowledged messages from compacted topic after reconnection [pulsar]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on code in PR #21187:
URL: https://github.com/apache/pulsar/pull/21187#discussion_r1456919270
##########
pulsar-broker/src/main/java/org/apache/pulsar/compaction/CompactedTopicImpl.java:
##########
@@ -100,7 +100,13 @@ public void asyncReadEntriesOrWait(ManagedCursor cursor,
boolean isFirstRead,
ReadEntriesCallback callback, Consumer consumer) {
PositionImpl cursorPosition;
- if (isFirstRead && MessageId.earliest.equals(consumer.getStartMessageId())){
+ boolean readFromEarliest = false;
+ if (!cursor.isDurable() || ((ManagedCursorImpl) cursor).isCompactionCursor()
+ || cursor.getMarkDeletedPosition() == null
+ || cursor.getMarkDeletedPosition().getEntryId() == -1L) {
+ readFromEarliest = isFirstRead && MessageId.earliest.equals(consumer.getStartMessageId());
+ }
Review Comment:
You'd better open a separated PR for cherry-pick.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Avoid consumers receiving acknowledged messages from compacted topic after reconnection [pulsar]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on code in PR #21187:
URL: https://github.com/apache/pulsar/pull/21187#discussion_r1455312306
##########
pulsar-broker/src/test/java/org/apache/pulsar/compaction/CompactionTest.java:
##########
@@ -1985,4 +1985,55 @@ public void testCompactionDuplicate() throws Exception {
}
}
}
+
+ @Test(timeOut = 100000)
+ public void testAcknowledge() throws Exception {
+ final String topicName = "persistent://my-property/use/my-ns/testAcknowledge" + UUID.randomUUID();
+ final String subName = "my-sub";
+ @Cleanup
+ PulsarClient client = newPulsarClient(lookupUrl.toString(), 100);
+ Producer<String> producer = pulsarClient.newProducer(Schema.STRING)
+ .enableBatching(false).topic(topicName).create();
+
+ pulsarClient.newConsumer().topic(topicName).subscriptionName(subName).readCompacted(true).subscribe().close();
+
+ Map<String, String> expected = new HashMap<>();
+ for (int i = 0; i < 10; i++) {
+ producer.newMessage().key(String.valueOf(i)).value(String.valueOf(i)).send();
+ expected.put(String.valueOf(i), String.valueOf(i));
+ }
+ producer.flush();
+
+ admin.topics().triggerCompaction(topicName);
+
+ Awaitility.await().untilAsserted(() -> {
+ assertEquals(admin.topics().compactionStatus(topicName).status,
+ LongRunningProcessStatus.Status.SUCCESS);
+ });
+
+ ConsumerImpl<String> consumer = (ConsumerImpl<String>) client.newConsumer(Schema.STRING)
+ .topic(topicName).readCompacted(true).receiverQueueSize(1).subscriptionName(subName)
+ .isAckReceiptEnabled(true)
+ .subscribe();
+
+ for (int i = 0; i < 10; i++) {
+ Message<String> message = consumer.receive(3, TimeUnit.SECONDS);
+ if (message == null) {
+ break;
+ }
+
+ consumer.acknowledge(message);
+ String remove = expected.remove(message.getKey());
+ Assert.assertEquals(remove, message.getValue());
+ }
+
+ // Make consumer reconnect to broker
Review Comment:
@coderzc This comment seems not addressed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Avoid consumers receiving acknowledged messages from compacted topic after reconnection [pulsar]
Posted by "codelipenghui (via GitHub)" <gi...@apache.org>.
codelipenghui commented on code in PR #21187:
URL: https://github.com/apache/pulsar/pull/21187#discussion_r1461858708
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorSubscriptionTest.java:
##########
@@ -868,6 +869,7 @@ public void testReplicatedSubscriptionWithCompaction() throws Exception {
.topic(topicName)
.subscriptionName("sub2")
.subscriptionType(SubscriptionType.Exclusive)
+ .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
Review Comment:
Is it reflect the PR's description `This change will also make subscriptionInitialPosition=Latest work normally since the mark-delete doesn't is null.` ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Avoid consumers receiving acknowledged messages from compacted topic after reconnection [pulsar]
Posted by "coderzc (via GitHub)" <gi...@apache.org>.
coderzc commented on PR #21187:
URL: https://github.com/apache/pulsar/pull/21187#issuecomment-1905219220
> To avoid skipping the remaining data in the compacted ledger after rewind, we need to record readCompacted to cursor.properties and just rewind to the next message of the mark delete position instead of resetting the read position to the next valid position of the original topic.
> Is it better to fix this one by a separate PR? It's a potential data loss issue, doesn't seem related to the PR title.
And which test can reproduce this issue?
@codelipenghui Because consumer is always read from the earliest position before applying this change, we can't reproduce `skipping the remaining data in the compacted ledger after rewind` issue, it will happen after changing the `readFromEarliest` condition,so I'm not sure if we should split it into two PRs.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Avoid consumers receiving acknowledged messages from compacted topic after reconnection [pulsar]
Posted by "codelipenghui (via GitHub)" <gi...@apache.org>.
codelipenghui commented on PR #21187:
URL: https://github.com/apache/pulsar/pull/21187#issuecomment-1905230908
> @codelipenghui Because consumer always read from the earliest position before applying this change, we can't reproduce skipping the remaining data in the compacted ledger after rewind issue, it will happen after changing the readFromEarliest condition,so I'm not sure if we should split it into two PRs.
Ok, I guess you want to say we don't have this issue without this change. But with this PR, if we don't fix this issue, the consumer might skip the compacted data when the cursor been rewinded.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Avoid consumers receiving acknowledged messages from compacted topic after reconnection [pulsar]
Posted by "codelipenghui (via GitHub)" <gi...@apache.org>.
codelipenghui commented on code in PR #21187:
URL: https://github.com/apache/pulsar/pull/21187#discussion_r1462680314
##########
pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ReaderTest.java:
##########
@@ -785,4 +785,32 @@ public void testReaderListenerAcknowledgement()
admin.topics().deletePartitionedTopic(partitionedTopic);
}
+ @Test
+ public void testReaderReconnectedFromNextEntry() throws Exception {
Review Comment:
Ok, I see.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Avoid consumers receiving acknowledged messages from compacted topic after reconnection [pulsar]
Posted by "coderzc (via GitHub)" <gi...@apache.org>.
coderzc commented on code in PR #21187:
URL: https://github.com/apache/pulsar/pull/21187#discussion_r1460104498
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumer.java:
##########
@@ -362,7 +363,12 @@ private void readMoreEntries(Consumer consumer) {
}
havePendingRead = true;
if (consumer.readCompacted()) {
- boolean readFromEarliest = isFirstRead && MessageId.earliest.equals(consumer.getStartMessageId());
+ boolean readFromEarliest = false;
+ if (!cursor.isDurable() || ((ManagedCursorImpl) cursor).isCompactionCursor()
+ || cursor.getMarkDeletedPosition() == null
+ || cursor.getMarkDeletedPosition().getEntryId() == -1L) {
+ readFromEarliest = isFirstRead && MessageId.earliest.equals(consumer.getStartMessageId());
+ }
Review Comment:
Yeah, after unloading the topic, also need to write a new message otherwise, `ml.getFirstPosition()` still returns the old ledger first position, I added `testEarliestSubsAfterRollover` to cover this case.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Avoid consumers receiving acknowledged messages from compacted topic after reconnection [pulsar]
Posted by "coderzc (via GitHub)" <gi...@apache.org>.
coderzc commented on code in PR #21187:
URL: https://github.com/apache/pulsar/pull/21187#discussion_r1456851588
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumer.java:
##########
@@ -362,7 +363,12 @@ private void readMoreEntries(Consumer consumer) {
}
havePendingRead = true;
if (consumer.readCompacted()) {
- boolean readFromEarliest = isFirstRead && MessageId.earliest.equals(consumer.getStartMessageId());
+ boolean readFromEarliest = false;
+ if (!cursor.isDurable() || ((ManagedCursorImpl) cursor).isCompactionCursor()
+ || cursor.getMarkDeletedPosition() == null
+ || cursor.getMarkDeletedPosition().getEntryId() == -1L) {
+ readFromEarliest = isFirstRead && MessageId.earliest.equals(consumer.getStartMessageId());
+ }
Review Comment:
> And why did you also check the mark deleted position here?
If the mark-delete position is null and `subscriptionInitialPosition=Earliest` then we still need to read messages from earliest.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Avoid consumers receiving acknowledged messages from compacted topic after reconnection [pulsar]
Posted by "coderzc (via GitHub)" <gi...@apache.org>.
coderzc commented on code in PR #21187:
URL: https://github.com/apache/pulsar/pull/21187#discussion_r1456854058
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java:
##########
@@ -1017,6 +1021,10 @@ private CompletableFuture<Subscription> getDurableSubscription(String subscripti
}
Map<String, Long> properties = PersistentSubscription.getBaseCursorProperties(replicated);
+ if (readCompacted) {
+ properties = new TreeMap<>(properties);
+ properties.put(READ_COMPACTED_CURSOR_PROPERTIES, 1L);
Review Comment:
We need to merge customer properties but original map may be an immutable map.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Avoid consumers receiving acknowledged messages from compacted topic after reconnection [pulsar]
Posted by "coderzc (via GitHub)" <gi...@apache.org>.
coderzc commented on code in PR #21187:
URL: https://github.com/apache/pulsar/pull/21187#discussion_r1460104498
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumer.java:
##########
@@ -362,7 +363,12 @@ private void readMoreEntries(Consumer consumer) {
}
havePendingRead = true;
if (consumer.readCompacted()) {
- boolean readFromEarliest = isFirstRead && MessageId.earliest.equals(consumer.getStartMessageId());
+ boolean readFromEarliest = false;
+ if (!cursor.isDurable() || ((ManagedCursorImpl) cursor).isCompactionCursor()
+ || cursor.getMarkDeletedPosition() == null
+ || cursor.getMarkDeletedPosition().getEntryId() == -1L) {
+ readFromEarliest = isFirstRead && MessageId.earliest.equals(consumer.getStartMessageId());
+ }
Review Comment:
Yeah, after unloading the topic, also need to write a new message otherwise, `ml.getFirstPosition()` still returns the old ledger position, I added `testEarliestSubsAfterRollover` to cover this case.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Avoid consumers receiving acknowledged messages from compacted topic after reconnection [pulsar]
Posted by "codelipenghui (via GitHub)" <gi...@apache.org>.
codelipenghui commented on code in PR #21187:
URL: https://github.com/apache/pulsar/pull/21187#discussion_r1461879932
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java:
##########
@@ -2519,7 +2525,8 @@ public Position getPersistentMarkDeletedPosition() {
public void rewind() {
lock.writeLock().lock();
try {
- PositionImpl newReadPosition = ledger.getNextValidPosition(markDeletePosition);
+ PositionImpl newReadPosition = isCursorReadFromCompacted() ? markDeletePosition.getNext() :
+ ledger.getNextValidPosition(markDeletePosition);
Review Comment:
If we don't know the compaction horizon here, will it introduce a new issue if the read compacted is enabled but the topic don't have any compacted data (All the messages are
not compacted)?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Avoid consumers receiving acknowledged messages from compacted topic after reconnection [pulsar]
Posted by "codelipenghui (via GitHub)" <gi...@apache.org>.
codelipenghui commented on PR #21187:
URL: https://github.com/apache/pulsar/pull/21187#issuecomment-1904024978
> To avoid skipping the remaining data in the compacted ledger after rewind, we need to record readCompacted to cursor.properties and just rewind to the next message of the mark delete position instead of resetting the read position to the next valid position of the original topic.
Is it better to fix this one by a separate PR? It's a potential data loss issue, doesn't seem related to the PR title.
And which test can reproduce this issue?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Avoid consumers receiving acknowledged messages from compacted topic after reconnection [pulsar]
Posted by "coderzc (via GitHub)" <gi...@apache.org>.
coderzc commented on code in PR #21187:
URL: https://github.com/apache/pulsar/pull/21187#discussion_r1462691766
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorSubscriptionTest.java:
##########
@@ -868,6 +869,7 @@ public void testReplicatedSubscriptionWithCompaction() throws Exception {
.topic(topicName)
.subscriptionName("sub2")
.subscriptionType(SubscriptionType.Exclusive)
+ .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
Review Comment:
Yesh, `SubscriptionInitialPosition` will work normally
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Avoid consumers receiving acknowledged messages from compacted topic after reconnection [pulsar]
Posted by "coderzc (via GitHub)" <gi...@apache.org>.
coderzc commented on PR #21187:
URL: https://github.com/apache/pulsar/pull/21187#issuecomment-1905285907
> Ok, I guess you want to say we don't have this issue without this change. But with this PR, if we don't fix this issue, the consumer might skip the compacted data when the cursor been rewinded. That makes sense to me.
Yes.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Avoid consumers receiving acknowledged messages from compacted topic after reconnection [pulsar]
Posted by "codelipenghui (via GitHub)" <gi...@apache.org>.
codelipenghui commented on code in PR #21187:
URL: https://github.com/apache/pulsar/pull/21187#discussion_r1464255148
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java:
##########
@@ -2519,7 +2525,8 @@ public Position getPersistentMarkDeletedPosition() {
public void rewind() {
lock.writeLock().lock();
try {
- PositionImpl newReadPosition = ledger.getNextValidPosition(markDeletePosition);
+ PositionImpl newReadPosition = isCursorReadFromCompacted() ? markDeletePosition.getNext() :
+ ledger.getNextValidPosition(markDeletePosition);
Review Comment:
@coderzc I see.
After checking the changes again. It looks like we don't need to persist the `readCompacted`? If depends on the consumer settings. We only need to set the `readCompacted` when new consumer connected?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Avoid consumers receiving acknowledged messages from compacted topic after reconnection [pulsar]
Posted by "coderzc (via GitHub)" <gi...@apache.org>.
coderzc commented on code in PR #21187:
URL: https://github.com/apache/pulsar/pull/21187#discussion_r1462691000
##########
pulsar-broker/src/main/java/org/apache/pulsar/compaction/CompactedTopicImpl.java:
##########
@@ -100,7 +100,11 @@ public void asyncReadEntriesOrWait(ManagedCursor cursor,
boolean isFirstRead,
ReadEntriesCallback callback, Consumer consumer) {
PositionImpl cursorPosition;
- if (isFirstRead && MessageId.earliest.equals(consumer.getStartMessageId())){
+ boolean readFromEarliest = isFirstRead && MessageId.earliest.equals(consumer.getStartMessageId())
+ && (!cursor.isDurable() || cursor.getName().equals(Compactor.COMPACTION_SUBSCRIPTION)
+ || cursor.getMarkDeletedPosition() == null
+ || cursor.getMarkDeletedPosition().getEntryId() == -1L);
Review Comment:
> Sorry, I mean why we need to check cursor.getMarkDeletedPosition().getEntryId() == -1L?
Oh, since if `markDeletedPosition.entryID == -1L` then the md-position is an invalid position, the initial md-position of the consumer will be set to it.
https://github.com/apache/pulsar/blob/b14fcb47f20398bdf5b698e158346dce19f0edc6/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java#L997-L998
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Avoid consumers receiving acknowledged messages from compacted topic after reconnection [pulsar]
Posted by "coderzc (via GitHub)" <gi...@apache.org>.
coderzc commented on code in PR #21187:
URL: https://github.com/apache/pulsar/pull/21187#discussion_r1462670799
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java:
##########
@@ -2519,7 +2525,8 @@ public Position getPersistentMarkDeletedPosition() {
public void rewind() {
lock.writeLock().lock();
try {
- PositionImpl newReadPosition = ledger.getNextValidPosition(markDeletePosition);
+ PositionImpl newReadPosition = isCursorReadFromCompacted() ? markDeletePosition.getNext() :
+ ledger.getNextValidPosition(markDeletePosition);
Review Comment:
I don't think this affects correctness, if the newReadPosition's ledger doesn't exist in ml, it will be moved to the beginning of the next ledger.
https://github.com/apache/pulsar/blob/b14fcb47f20398bdf5b698e158346dce19f0edc6/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java#L1887-L1894
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] poorbarcode commented on a diff in pull request #21187: [fix][broker] Avoid consumers receiving acknowledged messages from compacted topic after reconnection
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #21187:
URL: https://github.com/apache/pulsar/pull/21187#discussion_r1338461297
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumer.java:
##########
@@ -349,7 +350,11 @@ protected void readMoreEntries(Consumer consumer) {
}
havePendingRead = true;
if (consumer.readCompacted()) {
- boolean readFromEarliest = isFirstRead && MessageId.earliest.equals(consumer.getStartMessageId());
+ boolean readFromEarliest = false;
+ if (!cursor.isDurable() || ((ManagedCursorImpl) cursor).isCompactionCursor()
+ || cursor.getPersistentMarkDeletedPosition() == null) {
+ readFromEarliest = isFirstRead && MessageId.earliest.equals(consumer.getStartMessageId());
Review Comment:
Got it. I wrote a test below(it works as expected). Could you add it into the current PR to guarantee this feature will not be broken?
```java
@Test
public void testReaderReconnectedFromNextEntry() throws Exception {
final String topic = newTopicName();
Reader<String> reader = pulsarClient.newReader(Schema.STRING).topic(topic).receiverQueueSize(1)
.startMessageId(MessageId.earliest).create();
Producer<String> producer = pulsarClient.newProducer(Schema.STRING).topic(topic).create();
// Send 3 and consume 1.
producer.send("1");
producer.send("2");
producer.send("3");
Message<String> msg1 = reader.readNext(2, TimeUnit.SECONDS);
assertEquals(msg1.getValue(), "1");
// Trigger reader reconnect.
admin.topics().unload(topic);
// For non-durable we are going to restart from the next entry.
Message<String> msg2 = reader.readNext(2, TimeUnit.SECONDS);
assertEquals(msg2.getValue(), "2");
Message<String> msg3 = reader.readNext(2, TimeUnit.SECONDS);
assertEquals(msg3.getValue(), "3");
// cleanup.
reader.close();
producer.close();
admin.topics().delete(topic, false);
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org