You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "thetumbled (via GitHub)" <gi...@apache.org> on 2024/05/29 03:37:24 UTC
[PR] [fix] [broker] maintain last active info in memory only. [pulsar]
thetumbled opened a new pull request, #22794:
URL: https://github.com/apache/pulsar/pull/22794
### Motivation
PR https://github.com/apache/pulsar/pull/17573 has fixed cases that consumer commit offset periodically. It update `lastActive` field of cursor when consumer commit the offset.
But there is still corner case that the consumer do not commit the offset for any reason. For example, there is no content in the topic could be read, so the consumer do not commit the offset reasonably. In such case we has no reason delete the subscription.
Anyway, **we should be cautious with the delete operation, which could result into data duplication or loss.**
This is the second solution trying to solve problem described in https://github.com/apache/pulsar/pull/21692.
PR https://github.com/apache/pulsar/pull/21692 try to update the last active info every time broker do expiration check.
But strictly speaking, we can't fix all the cases. If we fail to persist the cursor info, the incorrect sub deletion occurs too.
Incorrect deletion of subscription causes serious consequence, such as duplication or data lost.
In my opinion, missing deletion are better than incorrect deletion, so this pr try to avoid all the incorrect deletion even there is posibility of missing deletion.
### Modifications
Do not recover last active info from cursor info in zookeeper.
### Verifying this change
- [ ] Make sure that the change passes the CI checks.
*(Please pick either of the following options)*
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 -->
- [ ] `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 -->
--
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] maintain last active info in memory only. [pulsar]
Posted by "dao-jun (via GitHub)" <gi...@apache.org>.
dao-jun closed pull request #22794: [fix] [broker] maintain last active info in memory only.
URL: https://github.com/apache/pulsar/pull/22794
--
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] maintain last active info in memory only. [pulsar]
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode merged PR #22794:
URL: https://github.com/apache/pulsar/pull/22794
--
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] maintain last active info in memory only. [pulsar]
Posted by "dao-jun (via GitHub)" <gi...@apache.org>.
dao-jun commented on code in PR #22794:
URL: https://github.com/apache/pulsar/pull/22794#discussion_r1622384766
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java:
##########
@@ -3227,26 +3227,30 @@ public void checkInactiveSubscriptions() {
final Integer nsExpirationTime = policies.subscription_expiration_time_minutes;
final long expirationTimeMillis = TimeUnit.MINUTES
.toMillis(nsExpirationTime == null ? defaultExpirationTime : nsExpirationTime);
- if (expirationTimeMillis > 0) {
- subscriptions.forEach((subName, sub) -> {
- if (sub.dispatcher != null && sub.dispatcher.isConsumerConnected()
- || sub.isReplicated()
- || isCompactionSubscription(subName)) {
- return;
- }
- if (System.currentTimeMillis() - sub.cursor.getLastActive() > expirationTimeMillis) {
- sub.delete().thenAccept(v -> log.info("[{}][{}] The subscription was deleted due to expiration "
- + "with last active [{}]", topic, subName, sub.cursor.getLastActive()));
- }
- });
- }
+ checkInactiveSubscriptions(expirationTimeMillis);
} catch (Exception e) {
if (log.isDebugEnabled()) {
log.debug("[{}] Error getting policies", topic);
}
}
}
+ public void checkInactiveSubscriptions(long expirationTimeMillis) {
Review Comment:
```java
@VasableForTesting
void checkInactiveSubscriptions(long expirationTimeMillis)
```
--
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] maintain last active info in memory only. [pulsar]
Posted by "nodece (via GitHub)" <gi...@apache.org>.
nodece commented on code in PR #22794:
URL: https://github.com/apache/pulsar/pull/22794#discussion_r1621699092
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceTest.java:
##########
@@ -1313,6 +1314,67 @@ public void testCheckInactiveSubscriptionsShouldNotDeleteCompactionCursor() thro
}
+ @Test
+ public void testCheckInactiveSubscriptionWhenNoMessageToAck() throws Exception {
+ String namespace = "prop/testInactiveSubscriptionWhenNoMessageToAck";
+
+ try {
+ admin.namespaces().createNamespace(namespace);
+ } catch (PulsarAdminException.ConflictException e) {
+ // Ok.. (if test fails intermittently and namespace is already created)
+ }
+ // set enable subscription expiration.
+ admin.namespaces().setSubscriptionExpirationTime(namespace, 1);
+
+ String topic = "persistent://" + namespace + "/my-topic";
+ Producer<byte[]> producer = pulsarClient.newProducer().topic(topic).create();
+ producer.send("test".getBytes());
+ producer.close();
+
+ // create consumer to consume all messages
+ Consumer<byte[]> consumer = pulsarClient.newConsumer().topic(topic).subscriptionName("sub1")
+ .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest).subscribe();
+ consumer.acknowledge(consumer.receive());
+
+ Optional<Topic> topicOptional = pulsar.getBrokerService().getTopic(topic, true).get();
+ assertTrue(topicOptional.isPresent());
+ PersistentTopic persistentTopic = (PersistentTopic) topicOptional.get();
+
+ // wait for 1min, but consumer is still connected all the time.
+ // so subscription should not be deleted.
+ Thread.sleep(60000);
+ persistentTopic.checkInactiveSubscriptions();
Review Comment:
The 60s are too long, please add `org.apache.pulsar.broker.service.persistent.PersistentTopic#checkInactiveSubscriptions(long)`, so like:
```
@VisibleForTesting
public void checkInactiveSubscriptions(long expirationTimeMillis) {
if (expirationTimeMillis > 0) {
subscriptions.forEach((subName, sub) -> {
if (sub.dispatcher != null && sub.dispatcher.isConsumerConnected()
|| sub.isReplicated()
|| isCompactionSubscription(subName)) {
return;
}
if (System.currentTimeMillis() - sub.cursor.getLastActive() > expirationTimeMillis) {
sub.delete().thenAccept(v -> log.info("[{}][{}] The subscription was deleted due to expiration "
+ "with last active [{}]", topic, subName, sub.cursor.getLastActive()));
}
});
}
}
```
And then you call `persistentTopic.checkInactiveSubscriptions(1);`
--
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] maintain last active info in memory only. [pulsar]
Posted by "thetumbled (via GitHub)" <gi...@apache.org>.
thetumbled commented on code in PR #22794:
URL: https://github.com/apache/pulsar/pull/22794#discussion_r1622390487
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java:
##########
@@ -3227,26 +3227,30 @@ public void checkInactiveSubscriptions() {
final Integer nsExpirationTime = policies.subscription_expiration_time_minutes;
final long expirationTimeMillis = TimeUnit.MINUTES
.toMillis(nsExpirationTime == null ? defaultExpirationTime : nsExpirationTime);
- if (expirationTimeMillis > 0) {
- subscriptions.forEach((subName, sub) -> {
- if (sub.dispatcher != null && sub.dispatcher.isConsumerConnected()
- || sub.isReplicated()
- || isCompactionSubscription(subName)) {
- return;
- }
- if (System.currentTimeMillis() - sub.cursor.getLastActive() > expirationTimeMillis) {
- sub.delete().thenAccept(v -> log.info("[{}][{}] The subscription was deleted due to expiration "
- + "with last active [{}]", topic, subName, sub.cursor.getLastActive()));
- }
- });
- }
+ checkInactiveSubscriptions(expirationTimeMillis);
} catch (Exception e) {
if (log.isDebugEnabled()) {
log.debug("[{}] Error getting policies", topic);
}
}
}
+ public void checkInactiveSubscriptions(long expirationTimeMillis) {
Review Comment:
Done, thanks.
--
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] maintain last active info in memory only. [pulsar]
Posted by "thetumbled (via GitHub)" <gi...@apache.org>.
thetumbled commented on PR #22794:
URL: https://github.com/apache/pulsar/pull/22794#issuecomment-2136457856
PTAL, thanks. @dao-jun @lhotari @Technoboy- @poorbarcode @coderzc @codelipenghui @BewareMyPower
--
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] maintain last active info in memory only. [pulsar]
Posted by "nodece (via GitHub)" <gi...@apache.org>.
nodece commented on code in PR #22794:
URL: https://github.com/apache/pulsar/pull/22794#discussion_r1621705155
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceTest.java:
##########
@@ -1313,6 +1314,67 @@ public void testCheckInactiveSubscriptionsShouldNotDeleteCompactionCursor() thro
}
+ @Test
+ public void testCheckInactiveSubscriptionWhenNoMessageToAck() throws Exception {
+ String namespace = "prop/testInactiveSubscriptionWhenNoMessageToAck";
+
+ try {
+ admin.namespaces().createNamespace(namespace);
+ } catch (PulsarAdminException.ConflictException e) {
+ // Ok.. (if test fails intermittently and namespace is already created)
+ }
+ // set enable subscription expiration.
+ admin.namespaces().setSubscriptionExpirationTime(namespace, 1);
+
+ String topic = "persistent://" + namespace + "/my-topic";
+ Producer<byte[]> producer = pulsarClient.newProducer().topic(topic).create();
+ producer.send("test".getBytes());
+ producer.close();
+
+ // create consumer to consume all messages
+ Consumer<byte[]> consumer = pulsarClient.newConsumer().topic(topic).subscriptionName("sub1")
+ .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest).subscribe();
+ consumer.acknowledge(consumer.receive());
+
+ Optional<Topic> topicOptional = pulsar.getBrokerService().getTopic(topic, true).get();
+ assertTrue(topicOptional.isPresent());
+ PersistentTopic persistentTopic = (PersistentTopic) topicOptional.get();
+
+ // wait for 1min, but consumer is still connected all the time.
+ // so subscription should not be deleted.
+ Thread.sleep(60000);
+ persistentTopic.checkInactiveSubscriptions();
+ assertTrue(persistentTopic.getSubscriptions().containsKey("sub1"));
+ PersistentSubscription sub = persistentTopic.getSubscription("sub1");
+
+ // shutdown pulsar ungracefully
Review Comment:
A simpler way:
1. Close the topic
2. Remove the topic from the broker cache
3. Start the consumer, and then close the consumer to load the `sub1`.
4. Check inactive subscriptions
--
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] maintain last active info in memory only. [pulsar]
Posted by "nodece (via GitHub)" <gi...@apache.org>.
nodece commented on code in PR #22794:
URL: https://github.com/apache/pulsar/pull/22794#discussion_r1621705155
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceTest.java:
##########
@@ -1313,6 +1314,67 @@ public void testCheckInactiveSubscriptionsShouldNotDeleteCompactionCursor() thro
}
+ @Test
+ public void testCheckInactiveSubscriptionWhenNoMessageToAck() throws Exception {
+ String namespace = "prop/testInactiveSubscriptionWhenNoMessageToAck";
+
+ try {
+ admin.namespaces().createNamespace(namespace);
+ } catch (PulsarAdminException.ConflictException e) {
+ // Ok.. (if test fails intermittently and namespace is already created)
+ }
+ // set enable subscription expiration.
+ admin.namespaces().setSubscriptionExpirationTime(namespace, 1);
+
+ String topic = "persistent://" + namespace + "/my-topic";
+ Producer<byte[]> producer = pulsarClient.newProducer().topic(topic).create();
+ producer.send("test".getBytes());
+ producer.close();
+
+ // create consumer to consume all messages
+ Consumer<byte[]> consumer = pulsarClient.newConsumer().topic(topic).subscriptionName("sub1")
+ .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest).subscribe();
+ consumer.acknowledge(consumer.receive());
+
+ Optional<Topic> topicOptional = pulsar.getBrokerService().getTopic(topic, true).get();
+ assertTrue(topicOptional.isPresent());
+ PersistentTopic persistentTopic = (PersistentTopic) topicOptional.get();
+
+ // wait for 1min, but consumer is still connected all the time.
+ // so subscription should not be deleted.
+ Thread.sleep(60000);
+ persistentTopic.checkInactiveSubscriptions();
+ assertTrue(persistentTopic.getSubscriptions().containsKey("sub1"));
+ PersistentSubscription sub = persistentTopic.getSubscription("sub1");
+
+ // shutdown pulsar ungracefully
Review Comment:
A simpler way:
1. Close the topic
2. Remove the topic from the broker cache
3. Start the consumer, and then close the consumer to load the `sub1`.
4. Check inactive subscriptions
--
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] maintain last active info in memory only. [pulsar]
Posted by "nodece (via GitHub)" <gi...@apache.org>.
nodece commented on code in PR #22794:
URL: https://github.com/apache/pulsar/pull/22794#discussion_r1622157785
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceTest.java:
##########
@@ -1313,6 +1314,67 @@ public void testCheckInactiveSubscriptionsShouldNotDeleteCompactionCursor() thro
}
+ @Test
+ public void testCheckInactiveSubscriptionWhenNoMessageToAck() throws Exception {
+ String namespace = "prop/testInactiveSubscriptionWhenNoMessageToAck";
+
+ try {
+ admin.namespaces().createNamespace(namespace);
+ } catch (PulsarAdminException.ConflictException e) {
+ // Ok.. (if test fails intermittently and namespace is already created)
+ }
+
+ String topic = "persistent://" + namespace + "/my-topic";
+ Producer<byte[]> producer = pulsarClient.newProducer().topic(topic).create();
+ producer.send("test".getBytes());
+ producer.close();
+
+ // create consumer to consume all messages
+ Consumer<byte[]> consumer = pulsarClient.newConsumer().topic(topic).subscriptionName("sub1")
+ .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest).subscribe();
+ consumer.acknowledge(consumer.receive());
+
+ Optional<Topic> topicOptional = pulsar.getBrokerService().getTopic(topic, true).get();
+ assertTrue(topicOptional.isPresent());
+ PersistentTopic persistentTopic = (PersistentTopic) topicOptional.get();
+
+ // wait for 1s, but consumer is still connected all the time.
+ // so subscription should not be deleted.
+ Thread.sleep(1000);
Review Comment:
This consumer is connected, the subscription doesn't be 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] maintain last active info in memory only. [pulsar]
Posted by "thetumbled (via GitHub)" <gi...@apache.org>.
thetumbled commented on code in PR #22794:
URL: https://github.com/apache/pulsar/pull/22794#discussion_r1621814036
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java:
##########
@@ -2678,8 +2676,7 @@ private void persistPositionMetaStore(long cursorsLedgerId, PositionImpl positio
ManagedCursorInfo.Builder info = ManagedCursorInfo.newBuilder() //
.setCursorsLedgerId(cursorsLedgerId) //
.setMarkDeleteLedgerId(position.getLedgerId()) //
- .setMarkDeleteEntryId(position.getEntryId()) //
- .setLastActive(lastActive); //
+ .setMarkDeleteEntryId(position.getEntryId()); //
Review Comment:
Done, thanks for review.
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceTest.java:
##########
@@ -1313,6 +1314,67 @@ public void testCheckInactiveSubscriptionsShouldNotDeleteCompactionCursor() thro
}
+ @Test
+ public void testCheckInactiveSubscriptionWhenNoMessageToAck() throws Exception {
+ String namespace = "prop/testInactiveSubscriptionWhenNoMessageToAck";
+
+ try {
+ admin.namespaces().createNamespace(namespace);
+ } catch (PulsarAdminException.ConflictException e) {
+ // Ok.. (if test fails intermittently and namespace is already created)
+ }
+ // set enable subscription expiration.
+ admin.namespaces().setSubscriptionExpirationTime(namespace, 1);
+
+ String topic = "persistent://" + namespace + "/my-topic";
+ Producer<byte[]> producer = pulsarClient.newProducer().topic(topic).create();
+ producer.send("test".getBytes());
+ producer.close();
+
+ // create consumer to consume all messages
+ Consumer<byte[]> consumer = pulsarClient.newConsumer().topic(topic).subscriptionName("sub1")
+ .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest).subscribe();
+ consumer.acknowledge(consumer.receive());
+
+ Optional<Topic> topicOptional = pulsar.getBrokerService().getTopic(topic, true).get();
+ assertTrue(topicOptional.isPresent());
+ PersistentTopic persistentTopic = (PersistentTopic) topicOptional.get();
+
+ // wait for 1min, but consumer is still connected all the time.
+ // so subscription should not be deleted.
+ Thread.sleep(60000);
+ persistentTopic.checkInactiveSubscriptions();
Review Comment:
Good idea, updated.
--
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] maintain last active info in memory only. [pulsar]
Posted by "nodece (via GitHub)" <gi...@apache.org>.
nodece commented on code in PR #22794:
URL: https://github.com/apache/pulsar/pull/22794#discussion_r1621682764
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java:
##########
@@ -2678,8 +2676,7 @@ private void persistPositionMetaStore(long cursorsLedgerId, PositionImpl positio
ManagedCursorInfo.Builder info = ManagedCursorInfo.newBuilder() //
.setCursorsLedgerId(cursorsLedgerId) //
.setMarkDeleteLedgerId(position.getLedgerId()) //
- .setMarkDeleteEntryId(position.getEntryId()) //
- .setLastActive(lastActive); //
+ .setMarkDeleteEntryId(position.getEntryId()); //
Review Comment:
Please keep `setLastActive` for compatibility.
--
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] maintain last active info in memory only. [pulsar]
Posted by "dao-jun (via GitHub)" <gi...@apache.org>.
dao-jun commented on code in PR #22794:
URL: https://github.com/apache/pulsar/pull/22794#discussion_r1622379651
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java:
##########
@@ -3227,26 +3227,30 @@ public void checkInactiveSubscriptions() {
final Integer nsExpirationTime = policies.subscription_expiration_time_minutes;
final long expirationTimeMillis = TimeUnit.MINUTES
.toMillis(nsExpirationTime == null ? defaultExpirationTime : nsExpirationTime);
- if (expirationTimeMillis > 0) {
- subscriptions.forEach((subName, sub) -> {
- if (sub.dispatcher != null && sub.dispatcher.isConsumerConnected()
- || sub.isReplicated()
- || isCompactionSubscription(subName)) {
- return;
- }
- if (System.currentTimeMillis() - sub.cursor.getLastActive() > expirationTimeMillis) {
- sub.delete().thenAccept(v -> log.info("[{}][{}] The subscription was deleted due to expiration "
- + "with last active [{}]", topic, subName, sub.cursor.getLastActive()));
- }
- });
- }
+ checkInactiveSubscriptions(expirationTimeMillis);
} catch (Exception e) {
if (log.isDebugEnabled()) {
log.debug("[{}] Error getting policies", topic);
}
}
}
+ public void checkInactiveSubscriptions(long expirationTimeMillis) {
Review Comment:
Make the method private or `VisableTesting`
--
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] maintain last active info in memory only. [pulsar]
Posted by "thetumbled (via GitHub)" <gi...@apache.org>.
thetumbled commented on PR #22794:
URL: https://github.com/apache/pulsar/pull/22794#issuecomment-2142514864
/pulsarbot rerun-failure-checks
--
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] maintain last active info in memory only. [pulsar]
Posted by "nodece (via GitHub)" <gi...@apache.org>.
nodece commented on code in PR #22794:
URL: https://github.com/apache/pulsar/pull/22794#discussion_r1621700246
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceTest.java:
##########
@@ -1313,6 +1314,67 @@ public void testCheckInactiveSubscriptionsShouldNotDeleteCompactionCursor() thro
}
+ @Test
+ public void testCheckInactiveSubscriptionWhenNoMessageToAck() throws Exception {
+ String namespace = "prop/testInactiveSubscriptionWhenNoMessageToAck";
+
+ try {
+ admin.namespaces().createNamespace(namespace);
+ } catch (PulsarAdminException.ConflictException e) {
+ // Ok.. (if test fails intermittently and namespace is already created)
+ }
+ // set enable subscription expiration.
+ admin.namespaces().setSubscriptionExpirationTime(namespace, 1);
+
+ String topic = "persistent://" + namespace + "/my-topic";
+ Producer<byte[]> producer = pulsarClient.newProducer().topic(topic).create();
+ producer.send("test".getBytes());
+ producer.close();
+
+ // create consumer to consume all messages
+ Consumer<byte[]> consumer = pulsarClient.newConsumer().topic(topic).subscriptionName("sub1")
+ .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest).subscribe();
+ consumer.acknowledge(consumer.receive());
+
+ Optional<Topic> topicOptional = pulsar.getBrokerService().getTopic(topic, true).get();
+ assertTrue(topicOptional.isPresent());
+ PersistentTopic persistentTopic = (PersistentTopic) topicOptional.get();
+
+ // wait for 1min, but consumer is still connected all the time.
+ // so subscription should not be deleted.
+ Thread.sleep(60000);
+ persistentTopic.checkInactiveSubscriptions();
+ assertTrue(persistentTopic.getSubscriptions().containsKey("sub1"));
Review Comment:
`checkInactiveSubscriptions` is aync task, please use `Awaitility.await().pollDelay(3,TimeUnit.SECONDS).until(....)` to check if the `sub1` exists.
--
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] maintain last active info in memory only. [pulsar]
Posted by "thetumbled (via GitHub)" <gi...@apache.org>.
thetumbled commented on code in PR #22794:
URL: https://github.com/apache/pulsar/pull/22794#discussion_r1621814470
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceTest.java:
##########
@@ -1313,6 +1314,67 @@ public void testCheckInactiveSubscriptionsShouldNotDeleteCompactionCursor() thro
}
+ @Test
+ public void testCheckInactiveSubscriptionWhenNoMessageToAck() throws Exception {
+ String namespace = "prop/testInactiveSubscriptionWhenNoMessageToAck";
+
+ try {
+ admin.namespaces().createNamespace(namespace);
+ } catch (PulsarAdminException.ConflictException e) {
+ // Ok.. (if test fails intermittently and namespace is already created)
+ }
+ // set enable subscription expiration.
+ admin.namespaces().setSubscriptionExpirationTime(namespace, 1);
+
+ String topic = "persistent://" + namespace + "/my-topic";
+ Producer<byte[]> producer = pulsarClient.newProducer().topic(topic).create();
+ producer.send("test".getBytes());
+ producer.close();
+
+ // create consumer to consume all messages
+ Consumer<byte[]> consumer = pulsarClient.newConsumer().topic(topic).subscriptionName("sub1")
+ .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest).subscribe();
+ consumer.acknowledge(consumer.receive());
+
+ Optional<Topic> topicOptional = pulsar.getBrokerService().getTopic(topic, true).get();
+ assertTrue(topicOptional.isPresent());
+ PersistentTopic persistentTopic = (PersistentTopic) topicOptional.get();
+
+ // wait for 1min, but consumer is still connected all the time.
+ // so subscription should not be deleted.
+ Thread.sleep(60000);
+ persistentTopic.checkInactiveSubscriptions();
+ assertTrue(persistentTopic.getSubscriptions().containsKey("sub1"));
Review Comment:
Done.
--
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] maintain last active info in memory only. [pulsar]
Posted by "nodece (via GitHub)" <gi...@apache.org>.
nodece commented on code in PR #22794:
URL: https://github.com/apache/pulsar/pull/22794#discussion_r1621866758
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceTest.java:
##########
@@ -1313,6 +1314,67 @@ public void testCheckInactiveSubscriptionsShouldNotDeleteCompactionCursor() thro
}
+ @Test
+ public void testCheckInactiveSubscriptionWhenNoMessageToAck() throws Exception {
+ String namespace = "prop/testInactiveSubscriptionWhenNoMessageToAck";
+
+ try {
+ admin.namespaces().createNamespace(namespace);
+ } catch (PulsarAdminException.ConflictException e) {
+ // Ok.. (if test fails intermittently and namespace is already created)
+ }
+
+ String topic = "persistent://" + namespace + "/my-topic";
+ Producer<byte[]> producer = pulsarClient.newProducer().topic(topic).create();
+ producer.send("test".getBytes());
+ producer.close();
+
+ // create consumer to consume all messages
+ Consumer<byte[]> consumer = pulsarClient.newConsumer().topic(topic).subscriptionName("sub1")
+ .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest).subscribe();
+ consumer.acknowledge(consumer.receive());
+
+ Optional<Topic> topicOptional = pulsar.getBrokerService().getTopic(topic, true).get();
+ assertTrue(topicOptional.isPresent());
+ PersistentTopic persistentTopic = (PersistentTopic) topicOptional.get();
+
+ // wait for 1s, but consumer is still connected all the time.
+ // so subscription should not be deleted.
+ Thread.sleep(1000);
+ persistentTopic.checkInactiveSubscriptions(1000);
Review Comment:
```suggestion
persistentTopic.checkInactiveSubscriptions(1);
```
--
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] maintain last active info in memory only. [pulsar]
Posted by "nodece (via GitHub)" <gi...@apache.org>.
nodece commented on code in PR #22794:
URL: https://github.com/apache/pulsar/pull/22794#discussion_r1621866358
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceTest.java:
##########
@@ -1313,6 +1314,67 @@ public void testCheckInactiveSubscriptionsShouldNotDeleteCompactionCursor() thro
}
+ @Test
+ public void testCheckInactiveSubscriptionWhenNoMessageToAck() throws Exception {
+ String namespace = "prop/testInactiveSubscriptionWhenNoMessageToAck";
+
+ try {
+ admin.namespaces().createNamespace(namespace);
+ } catch (PulsarAdminException.ConflictException e) {
+ // Ok.. (if test fails intermittently and namespace is already created)
+ }
+
+ String topic = "persistent://" + namespace + "/my-topic";
+ Producer<byte[]> producer = pulsarClient.newProducer().topic(topic).create();
+ producer.send("test".getBytes());
+ producer.close();
+
+ // create consumer to consume all messages
+ Consumer<byte[]> consumer = pulsarClient.newConsumer().topic(topic).subscriptionName("sub1")
+ .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest).subscribe();
+ consumer.acknowledge(consumer.receive());
+
+ Optional<Topic> topicOptional = pulsar.getBrokerService().getTopic(topic, true).get();
+ assertTrue(topicOptional.isPresent());
+ PersistentTopic persistentTopic = (PersistentTopic) topicOptional.get();
+
+ // wait for 1s, but consumer is still connected all the time.
+ // so subscription should not be deleted.
+ Thread.sleep(1000);
Review Comment:
```suggestion
```
--
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