You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "Technoboy- (via GitHub)" <gi...@apache.org> on 2024/04/07 12:54:41 UTC
[PR] [fix][broker] Fix consumer stops receiving messages when with large backlogs processing [pulsar]
Technoboy- opened a new pull request, #22454:
URL: https://github.com/apache/pulsar/pull/22454
Fixes #22435
### Motivation
When the backlog size > `managedLedgerCursorBackloggedThreshold` (default 1000), the cursor will be set inactive first. so this will cause the non-durable cursor not to add to the `waiting cursor.` This leads to the consumer stopping to receive messages.
https://github.com/apache/pulsar/blob/2469b97b7e4de10fec64cc7ff1f4f46a410ad125/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java#L3082-L3090
https://github.com/apache/pulsar/blob/2469b97b7e4de10fec64cc7ff1f4f46a410ad125/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java#L3816-L3824
### 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 -->
--
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] Fix consumer stops receiving messages when with large backlogs processing [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #22454:
URL: https://github.com/apache/pulsar/pull/22454#discussion_r1555206870
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java:
##########
@@ -338,11 +337,14 @@ public synchronized void removeConsumer(Consumer consumer, boolean isResetCursor
if (!isResetCursor) {
try {
topic.getManagedLedger().deleteCursor(cursor.getName());
+ topic.getManagedLedger().removeWaitingCursor(cursor);
} catch (InterruptedException | ManagedLedgerException e) {
log.warn("[{}] [{}] Failed to remove non durable cursor", topic.getName(), subName, e);
}
}
});
+ } else {
+ topic.getManagedLedger().removeWaitingCursor(cursor);
Review Comment:
sure, no need for explicit revert of #22191 with "git revert". I simply meant that changes to addWaitingCursor method can be removed.
--
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] Fix consumer stops receiving messages when with large backlogs processing [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #22454:
URL: https://github.com/apache/pulsar/pull/22454#discussion_r1555192761
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java:
##########
@@ -338,11 +337,14 @@ public synchronized void removeConsumer(Consumer consumer, boolean isResetCursor
if (!isResetCursor) {
try {
topic.getManagedLedger().deleteCursor(cursor.getName());
+ topic.getManagedLedger().removeWaitingCursor(cursor);
} catch (InterruptedException | ManagedLedgerException e) {
log.warn("[{}] [{}] Failed to remove non durable cursor", topic.getName(), subName, e);
}
}
});
+ } else {
+ topic.getManagedLedger().removeWaitingCursor(cursor);
Review Comment:
Great! It looks like this fixes #22157. I tested that all changes to addWaitingCursor can be removed that were made to fix #22191.
--
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] Fix consumer stops receiving messages when with large backlogs processing [pulsar]
Posted by "Technoboy- (via GitHub)" <gi...@apache.org>.
Technoboy- commented on code in PR #22454:
URL: https://github.com/apache/pulsar/pull/22454#discussion_r1555200148
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java:
##########
@@ -338,11 +337,14 @@ public synchronized void removeConsumer(Consumer consumer, boolean isResetCursor
if (!isResetCursor) {
try {
topic.getManagedLedger().deleteCursor(cursor.getName());
+ topic.getManagedLedger().removeWaitingCursor(cursor);
} catch (InterruptedException | ManagedLedgerException e) {
log.warn("[{}] [{}] Failed to remove non durable cursor", topic.getName(), subName, e);
}
}
});
+ } else {
+ topic.getManagedLedger().removeWaitingCursor(cursor);
Review Comment:
yes, since this fix can fix #22157 and override #22191, we can keep #22191.
--
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] Fix consumer stops receiving messages when with large backlogs processing [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #22454:
URL: https://github.com/apache/pulsar/pull/22454#discussion_r1555220150
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java:
##########
@@ -338,11 +337,14 @@ public synchronized void removeConsumer(Consumer consumer, boolean isResetCursor
if (!isResetCursor) {
try {
topic.getManagedLedger().deleteCursor(cursor.getName());
+ topic.getManagedLedger().removeWaitingCursor(cursor);
} catch (InterruptedException | ManagedLedgerException e) {
log.warn("[{}] [{}] Failed to remove non durable cursor", topic.getName(), subName, e);
}
}
});
+ } else {
+ topic.getManagedLedger().removeWaitingCursor(cursor);
Review Comment:
That proves that the method could be removed, however since there's also `removeWaitingCursor`, it makes sense to have `addWaitingCursor` (which doesn't have any extra conditions).
--
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] Fix consumer stops receiving messages when with large backlogs processing [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #22454:
URL: https://github.com/apache/pulsar/pull/22454#discussion_r1555263091
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java:
##########
@@ -338,11 +337,14 @@ public synchronized void removeConsumer(Consumer consumer, boolean isResetCursor
if (!isResetCursor) {
try {
topic.getManagedLedger().deleteCursor(cursor.getName());
+ topic.getManagedLedger().removeWaitingCursor(cursor);
} catch (InterruptedException | ManagedLedgerException e) {
log.warn("[{}] [{}] Failed to remove non durable cursor", topic.getName(), subName, e);
}
}
});
+ } else {
+ topic.getManagedLedger().removeWaitingCursor(cursor);
Review Comment:
Thanks @Technoboy- , I got it reproduced with https://github.com/lhotari/pulsar/commit/e2efe24.
Regardless of this, I think that the problem is deeper.
I was expecting that the race condition would be fixed by adding this to ManagedCursorImpl.checkForNewEntries method:
```
if (isClosed()) {
callback.readEntriesFailed(new CursorAlreadyClosedException("Cursor was already closed"), ctx);
return;
}
```
However, it doesn't work since when a cursor is deleted, it's status isn't switched to closed. This isn't very logical in the ManagedCursorImpl code and I think it should be fixed.
I made this commit in my experiment https://github.com/lhotari/pulsar/commit/43da5894 . It resolves the race condition that exists in `checkForNewEntries`.
--
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] Fix consumer stops receiving messages when with large backlogs processing [pulsar]
Posted by "Technoboy- (via GitHub)" <gi...@apache.org>.
Technoboy- commented on code in PR #22454:
URL: https://github.com/apache/pulsar/pull/22454#discussion_r1555210553
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java:
##########
@@ -338,11 +337,14 @@ public synchronized void removeConsumer(Consumer consumer, boolean isResetCursor
if (!isResetCursor) {
try {
topic.getManagedLedger().deleteCursor(cursor.getName());
+ topic.getManagedLedger().removeWaitingCursor(cursor);
} catch (InterruptedException | ManagedLedgerException e) {
log.warn("[{}] [{}] Failed to remove non durable cursor", topic.getName(), subName, e);
}
}
});
+ } else {
+ topic.getManagedLedger().removeWaitingCursor(cursor);
Review Comment:
Oh, we can't remove this method. because we have removed the cursor from the waitingCursor when consumer unsubscribe. and there are still race conditions with `consumer.close` and `checkForNewEntries`.which will lead to OOM
You can use testAddWaitingCursorsForNonDurable to reproduce
--
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] Fix consumer stops receiving messages when with large backlogs processing [pulsar]
Posted by "Technoboy- (via GitHub)" <gi...@apache.org>.
Technoboy- merged PR #22454:
URL: https://github.com/apache/pulsar/pull/22454
--
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] Fix consumer stops receiving messages when with large backlogs processing [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on PR #22454:
URL: https://github.com/apache/pulsar/pull/22454#issuecomment-2041815550
Great work on this fix @Technoboy- ! Just one minor comment in https://github.com/apache/pulsar/pull/22454#discussion_r1555193276 .
--
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] Fix consumer stops receiving messages when with large backlogs processing [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #22454:
URL: https://github.com/apache/pulsar/pull/22454#discussion_r1555222724
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentTopicTest.java:
##########
@@ -708,4 +709,52 @@ public void testAddWaitingCursorsForNonDurable() throws Exception {
assertEquals(ledger.getWaitingCursorsCount(), 0);
});
}
+
+ @Test
+ public void testAddWaitingCursorsForNonDurable2() throws Exception {
+ final String ns = "prop/ns-test";
+ admin.namespaces().createNamespace(ns, 2);
+ final String topicName = "persistent://prop/ns-test/testAddWaitingCursors2";
+ admin.topics().createNonPartitionedTopic(topicName);
+ pulsarClient.newConsumer(Schema.STRING).topic(topicName)
+ .subscriptionMode(SubscriptionMode.Durable)
+ .subscriptionType(SubscriptionType.Shared)
+ .subscriptionName("sub-1").subscribe().close();
+ @Cleanup
+ final Producer<String> producer = pulsarClient.newProducer(Schema.STRING).topic(topicName).create();
+ for (int i = 0; i < 2_000; i ++) {
+ producer.send("test-" + i);
Review Comment:
+1
I guess disabling batching makes sense since backlog is counted in entries.
to further reduce the test time, it would be possible to also adjust the config of `managedLedgerCursorBackloggedThreshold` since less messages would be required to reproduce the issue.
```
@Override
protected void doInitConf() throws Exception {
super.doInitConf();
this.conf.setManagedLedgerCursorBackloggedThreshold(10);
}
```
--
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] Fix consumer stops receiving messages when with large backlogs processing [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #22454:
URL: https://github.com/apache/pulsar/pull/22454#discussion_r1555882121
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java:
##########
@@ -308,7 +308,6 @@ public synchronized void removeConsumer(Consumer consumer, boolean isResetCursor
if (dispatcher != null && dispatcher.getConsumers().isEmpty()) {
deactivateCursor();
- topic.getManagedLedger().removeWaitingCursor(cursor);
Review Comment:
just wondering about this removal now after the PR has been merged. It seems that whatever `isResetCursor` means that it would be skipped in that case. @Technoboy- is that a problem?
--
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] Fix consumer stops receiving messages when with large backlogs processing [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #22454:
URL: https://github.com/apache/pulsar/pull/22454#discussion_r1555180329
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java:
##########
@@ -3815,7 +3815,7 @@ public void removeWaitingCursor(ManagedCursor cursor) {
public void addWaitingCursor(ManagedCursorImpl cursor) {
if (cursor instanceof NonDurableCursorImpl) {
- if (cursor.isActive()) {
+ if (cursors.get(cursor.getName()) != null) {
Review Comment:
Is it possible to add a comment to explain this condition?
It seems that the original reason to add this was to fix #22157 in #22191 . It would be great to have the rational explained in a comment.
When looking at the code, it seems that this would always evaluate to true since non durable cursors are always "cached" and part of the `cursors` map. Only when the non durable cursor is deleted, would it be removed from `cursors`.
Is the logic in `addWaitingCursor` useful? Do we have a separate test case for #22191 that proves that it fixes the memory leak?
--
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] Fix consumer stops receiving messages when with large backlogs processing [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #22454:
URL: https://github.com/apache/pulsar/pull/22454#discussion_r1555193276
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java:
##########
@@ -3815,7 +3815,7 @@ public void removeWaitingCursor(ManagedCursor cursor) {
public void addWaitingCursor(ManagedCursorImpl cursor) {
if (cursor instanceof NonDurableCursorImpl) {
- if (cursor.isActive()) {
+ if (cursors.get(cursor.getName()) != null) {
Review Comment:
Please check my other comment https://github.com/apache/pulsar/pull/22454/files#r1555192761 . This method "addWaitingCursor" can be removed completely since the other change fixes #22157.
--
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] Fix consumer stops receiving messages when with large backlogs processing [pulsar]
Posted by "Technoboy- (via GitHub)" <gi...@apache.org>.
Technoboy- commented on code in PR #22454:
URL: https://github.com/apache/pulsar/pull/22454#discussion_r1555201939
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java:
##########
@@ -3815,7 +3815,7 @@ public void removeWaitingCursor(ManagedCursor cursor) {
public void addWaitingCursor(ManagedCursorImpl cursor) {
if (cursor instanceof NonDurableCursorImpl) {
- if (cursor.isActive()) {
+ if (cursors.get(cursor.getName()) != null) {
Review Comment:
> Please remove addWaitingCursor method or make it simply this.waitingCursors.add(cursor); without any conditions.
Oh, we can't remove this method. because we have removed the cursor from the `waitingCursor ` when consumer unsubscribe.
You can use `testAddWaitingCursorsForNonDurable` to reproduce
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java:
##########
@@ -3815,7 +3815,7 @@ public void removeWaitingCursor(ManagedCursor cursor) {
public void addWaitingCursor(ManagedCursorImpl cursor) {
if (cursor instanceof NonDurableCursorImpl) {
- if (cursor.isActive()) {
+ if (cursors.get(cursor.getName()) != null) {
Review Comment:
> Please remove addWaitingCursor method or make it simply this.waitingCursors.add(cursor); without any conditions.
Oh, we can't remove this method. because we have removed the cursor from the `waitingCursor ` when consumer unsubscribe.
You can use `testAddWaitingCursorsForNonDurable` to reproduce
--
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] Fix consumer stops receiving messages when with large backlogs processing [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #22454:
URL: https://github.com/apache/pulsar/pull/22454#discussion_r1555269716
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java:
##########
@@ -338,11 +337,14 @@ public synchronized void removeConsumer(Consumer consumer, boolean isResetCursor
if (!isResetCursor) {
try {
topic.getManagedLedger().deleteCursor(cursor.getName());
+ topic.getManagedLedger().removeWaitingCursor(cursor);
} catch (InterruptedException | ManagedLedgerException e) {
log.warn("[{}] [{}] Failed to remove non durable cursor", topic.getName(), subName, e);
}
}
});
+ } else {
+ topic.getManagedLedger().removeWaitingCursor(cursor);
Review Comment:
here's the diff of my experiment compared to the changes of this PR: https://github.com/Technoboy-/pulsar/compare/fix-22435...lhotari:pulsar:lh-remove-addWaitingCursor-conditions
I think that the proper way to fix the problem is to add the closed check to `checkForNewEntries` method and set the state to closed when the cursor is deleted.
--
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] Fix consumer stops receiving messages when with large backlogs processing [pulsar]
Posted by "Technoboy- (via GitHub)" <gi...@apache.org>.
Technoboy- commented on code in PR #22454:
URL: https://github.com/apache/pulsar/pull/22454#discussion_r1555202537
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java:
##########
@@ -3815,7 +3815,7 @@ public void removeWaitingCursor(ManagedCursor cursor) {
public void addWaitingCursor(ManagedCursorImpl cursor) {
if (cursor instanceof NonDurableCursorImpl) {
- if (cursor.isActive()) {
+ if (cursors.get(cursor.getName()) != null) {
Review Comment:
> Is the logic in addWaitingCursor useful? Do we have a separate test case for #22191 that proves that it fixes the memory leak?
`testAddWaitingCursorsForNonDurable` could help reproduce 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] Fix consumer stops receiving messages when with large backlogs processing [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #22454:
URL: https://github.com/apache/pulsar/pull/22454#discussion_r1555193564
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java:
##########
@@ -3815,7 +3815,7 @@ public void removeWaitingCursor(ManagedCursor cursor) {
public void addWaitingCursor(ManagedCursorImpl cursor) {
if (cursor instanceof NonDurableCursorImpl) {
- if (cursor.isActive()) {
+ if (cursors.get(cursor.getName()) != null) {
Review Comment:
Please remove addWaitingCursor method or make it simply `this.waitingCursors.add(cursor);` without any conditions.
--
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] Fix consumer stops receiving messages when with large backlogs processing [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #22454:
URL: https://github.com/apache/pulsar/pull/22454#discussion_r1555219351
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java:
##########
@@ -338,11 +337,14 @@ public synchronized void removeConsumer(Consumer consumer, boolean isResetCursor
if (!isResetCursor) {
try {
topic.getManagedLedger().deleteCursor(cursor.getName());
+ topic.getManagedLedger().removeWaitingCursor(cursor);
} catch (InterruptedException | ManagedLedgerException e) {
log.warn("[{}] [{}] Failed to remove non durable cursor", topic.getName(), subName, e);
}
}
});
+ } else {
+ topic.getManagedLedger().removeWaitingCursor(cursor);
Review Comment:
I did test that. The test passes with this change: https://github.com/lhotari/pulsar/commit/d46ebc41
--
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] Fix consumer stops receiving messages when with large backlogs processing [pulsar]
Posted by "codelipenghui (via GitHub)" <gi...@apache.org>.
codelipenghui commented on code in PR #22454:
URL: https://github.com/apache/pulsar/pull/22454#discussion_r1555207870
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentTopicTest.java:
##########
@@ -708,4 +709,52 @@ public void testAddWaitingCursorsForNonDurable() throws Exception {
assertEquals(ledger.getWaitingCursorsCount(), 0);
});
}
+
+ @Test
+ public void testAddWaitingCursorsForNonDurable2() throws Exception {
+ final String ns = "prop/ns-test";
+ admin.namespaces().createNamespace(ns, 2);
+ final String topicName = "persistent://prop/ns-test/testAddWaitingCursors2";
+ admin.topics().createNonPartitionedTopic(topicName);
+ pulsarClient.newConsumer(Schema.STRING).topic(topicName)
+ .subscriptionMode(SubscriptionMode.Durable)
+ .subscriptionType(SubscriptionType.Shared)
+ .subscriptionName("sub-1").subscribe().close();
+ @Cleanup
+ final Producer<String> producer = pulsarClient.newProducer(Schema.STRING).topic(topicName).create();
+ for (int i = 0; i < 2_000; i ++) {
+ producer.send("test-" + i);
Review Comment:
sendAsync is better to reduce the test time. If you don't want the message batching, you can just disable the batch.
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentTopicTest.java:
##########
@@ -708,4 +709,52 @@ public void testAddWaitingCursorsForNonDurable() throws Exception {
assertEquals(ledger.getWaitingCursorsCount(), 0);
});
}
+
+ @Test
+ public void testAddWaitingCursorsForNonDurable2() throws Exception {
+ final String ns = "prop/ns-test";
+ admin.namespaces().createNamespace(ns, 2);
+ final String topicName = "persistent://prop/ns-test/testAddWaitingCursors2";
+ admin.topics().createNonPartitionedTopic(topicName);
+ pulsarClient.newConsumer(Schema.STRING).topic(topicName)
+ .subscriptionMode(SubscriptionMode.Durable)
+ .subscriptionType(SubscriptionType.Shared)
+ .subscriptionName("sub-1").subscribe().close();
+ @Cleanup
+ final Producer<String> producer = pulsarClient.newProducer(Schema.STRING).topic(topicName).create();
+ for (int i = 0; i < 2_000; i ++) {
+ producer.send("test-" + i);
+ }
+ @Cleanup
+ final Consumer<String> consumer = pulsarClient.newConsumer(Schema.STRING).topic(topicName)
+ .subscriptionMode(SubscriptionMode.NonDurable)
+ .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
+ .subscriptionType(SubscriptionType.Exclusive)
+ .subscriptionName("sub-2").subscribe();
+ int count = 0;
+ while(true) {
+ final Message<String> msg = consumer.receive(3, TimeUnit.SECONDS);
+ if (msg != null) {
+ consumer.acknowledge(msg);
+ count++;
+ } else {
+ break;
+ }
+ }
+ Assert.assertEquals(count, 2000);
+ Thread.sleep(5_000);
+ for (int i = 0; i < 2_000; i ++) {
+ producer.send("test-" + i);
Review Comment:
Same as the above comment.
--
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] Fix consumer stops receiving messages when with large backlogs processing [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #22454:
URL: https://github.com/apache/pulsar/pull/22454#discussion_r1555386021
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentTopicTest.java:
##########
@@ -704,8 +710,57 @@ public void testAddWaitingCursorsForNonDurable() throws Exception {
Awaitility.await()
.pollDelay(5, TimeUnit.SECONDS)
.pollInterval(1, TimeUnit.SECONDS)
+ .atMost(30, TimeUnit.SECONDS)
Review Comment:
I guess this isn't needed since the delay is 5000 ms in the spy.
--
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] Fix consumer stops receiving messages when with large backlogs processing [pulsar]
Posted by "Technoboy- (via GitHub)" <gi...@apache.org>.
Technoboy- commented on code in PR #22454:
URL: https://github.com/apache/pulsar/pull/22454#discussion_r1555359865
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java:
##########
@@ -338,11 +337,14 @@ public synchronized void removeConsumer(Consumer consumer, boolean isResetCursor
if (!isResetCursor) {
try {
topic.getManagedLedger().deleteCursor(cursor.getName());
+ topic.getManagedLedger().removeWaitingCursor(cursor);
} catch (InterruptedException | ManagedLedgerException e) {
log.warn("[{}] [{}] Failed to remove non durable cursor", topic.getName(), subName, e);
}
}
});
+ } else {
+ topic.getManagedLedger().removeWaitingCursor(cursor);
Review Comment:
More better , addressed. Thanks @lhotari .
--
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] Fix consumer stops receiving messages when with large backlogs processing [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #22454:
URL: https://github.com/apache/pulsar/pull/22454#discussion_r1555263373
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java:
##########
@@ -3815,7 +3815,7 @@ public void removeWaitingCursor(ManagedCursor cursor) {
public void addWaitingCursor(ManagedCursorImpl cursor) {
if (cursor instanceof NonDurableCursorImpl) {
- if (cursor.isActive()) {
+ if (cursors.get(cursor.getName()) != null) {
Review Comment:
thanks, continued in the other thread, https://github.com/apache/pulsar/pull/22454#discussion_r1555263091
--
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] Fix consumer stops receiving messages when with large backlogs processing [pulsar]
Posted by "Technoboy- (via GitHub)" <gi...@apache.org>.
Technoboy- commented on code in PR #22454:
URL: https://github.com/apache/pulsar/pull/22454#discussion_r1555356601
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentTopicTest.java:
##########
@@ -708,4 +709,52 @@ public void testAddWaitingCursorsForNonDurable() throws Exception {
assertEquals(ledger.getWaitingCursorsCount(), 0);
});
}
+
+ @Test
+ public void testAddWaitingCursorsForNonDurable2() throws Exception {
+ final String ns = "prop/ns-test";
+ admin.namespaces().createNamespace(ns, 2);
+ final String topicName = "persistent://prop/ns-test/testAddWaitingCursors2";
+ admin.topics().createNonPartitionedTopic(topicName);
+ pulsarClient.newConsumer(Schema.STRING).topic(topicName)
+ .subscriptionMode(SubscriptionMode.Durable)
+ .subscriptionType(SubscriptionType.Shared)
+ .subscriptionName("sub-1").subscribe().close();
+ @Cleanup
+ final Producer<String> producer = pulsarClient.newProducer(Schema.STRING).topic(topicName).create();
+ for (int i = 0; i < 2_000; i ++) {
+ producer.send("test-" + i);
Review Comment:
Thanks @lhotari @codelipenghui
addressed
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentTopicTest.java:
##########
@@ -708,4 +709,52 @@ public void testAddWaitingCursorsForNonDurable() throws Exception {
assertEquals(ledger.getWaitingCursorsCount(), 0);
});
}
+
+ @Test
+ public void testAddWaitingCursorsForNonDurable2() throws Exception {
+ final String ns = "prop/ns-test";
+ admin.namespaces().createNamespace(ns, 2);
+ final String topicName = "persistent://prop/ns-test/testAddWaitingCursors2";
+ admin.topics().createNonPartitionedTopic(topicName);
+ pulsarClient.newConsumer(Schema.STRING).topic(topicName)
+ .subscriptionMode(SubscriptionMode.Durable)
+ .subscriptionType(SubscriptionType.Shared)
+ .subscriptionName("sub-1").subscribe().close();
+ @Cleanup
+ final Producer<String> producer = pulsarClient.newProducer(Schema.STRING).topic(topicName).create();
+ for (int i = 0; i < 2_000; i ++) {
+ producer.send("test-" + i);
+ }
+ @Cleanup
+ final Consumer<String> consumer = pulsarClient.newConsumer(Schema.STRING).topic(topicName)
+ .subscriptionMode(SubscriptionMode.NonDurable)
+ .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
+ .subscriptionType(SubscriptionType.Exclusive)
+ .subscriptionName("sub-2").subscribe();
+ int count = 0;
+ while(true) {
+ final Message<String> msg = consumer.receive(3, TimeUnit.SECONDS);
+ if (msg != null) {
+ consumer.acknowledge(msg);
+ count++;
+ } else {
+ break;
+ }
+ }
+ Assert.assertEquals(count, 2000);
+ Thread.sleep(5_000);
+ for (int i = 0; i < 2_000; i ++) {
+ producer.send("test-" + i);
Review Comment:
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] Fix consumer stops receiving messages when with large backlogs processing [pulsar]
Posted by "Technoboy- (via GitHub)" <gi...@apache.org>.
Technoboy- commented on code in PR #22454:
URL: https://github.com/apache/pulsar/pull/22454#discussion_r1555391292
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentTopicTest.java:
##########
@@ -704,8 +710,57 @@ public void testAddWaitingCursorsForNonDurable() throws Exception {
Awaitility.await()
.pollDelay(5, TimeUnit.SECONDS)
.pollInterval(1, TimeUnit.SECONDS)
+ .atMost(30, TimeUnit.SECONDS)
Review Comment:
removed
--
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