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 2023/12/08 04:18:01 UTC
[PR] [Fix] [Broker] fix incorrect delete sub when lastActive do not update. [pulsar]
thetumbled opened a new pull request, #21692:
URL: https://github.com/apache/pulsar/pull/21692
### Motivation
This 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.
### Modifications
Update the `lastActive` field when check the inactive sub.
### Verifying this change
- [ ] 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.
### 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] fix incorrect delete sub when lastActive do not update. [pulsar]
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #21692:
URL: https://github.com/apache/pulsar/pull/21692#discussion_r1617309066
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceTest.java:
##########
@@ -1313,6 +1314,66 @@ 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
+ // disable the updateLastActive method to simulate the ungraceful shutdown
Review Comment:
You want to mock an ungraceful shutdown, but the exact behavior you mocked is below:
- the result of `cursor.updateLastActive()` will not be persisted.
- the result of `sub.cursor.updateLastActive()` you added at [this line](https://github.com/apache/pulsar/pull/21692/files#diff-5edf14cc6f25857d0cfdd26b2d3b3141230ecfb0dfa95aebf7583fd76ede4c4bR3233) will be persisted.
This mock is wrong.
--
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 incorrect delete sub when lastActive do not update. [pulsar]
Posted by "thetumbled (via GitHub)" <gi...@apache.org>.
thetumbled commented on code in PR #21692:
URL: https://github.com/apache/pulsar/pull/21692#discussion_r1618069635
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceTest.java:
##########
@@ -1313,6 +1314,66 @@ 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);
Review Comment:
It is, but the minimum expiration time is 1min, we have to wait for 1min to trigger the bug.
--
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 incorrect delete sub when lastActive do not update. [pulsar]
Posted by "thetumbled (via GitHub)" <gi...@apache.org>.
thetumbled commented on code in PR #21692:
URL: https://github.com/apache/pulsar/pull/21692#discussion_r1618136680
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceTest.java:
##########
@@ -1313,6 +1314,66 @@ 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
+ // disable the updateLastActive method to simulate the ungraceful shutdown
Review Comment:
We can't ensure that the updated timestamp is persisted in crash time. With this PR, we can fix most of the cases, as we update it 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 deletions are better than incorrect deletion, this is PR https://github.com/apache/pulsar/pull/22794 try to do.
--
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 incorrect delete sub when lastActive do not update. [pulsar]
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #21692:
URL: https://github.com/apache/pulsar/pull/21692#discussion_r1617309066
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceTest.java:
##########
@@ -1313,6 +1314,66 @@ 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
+ // disable the updateLastActive method to simulate the ungraceful shutdown
Review Comment:
You want to mock an ungraceful shutdown, but the exact behavior you mocked is below:
- the result of `cursor.updateLastActive()` will not be persisted.
- the result of `sub.cursor.updateLastActive()` you added [this line](https://github.com/apache/pulsar/pull/21692/files#diff-5edf14cc6f25857d0cfdd26b2d3b3141230ecfb0dfa95aebf7583fd76ede4c4bR3233) will be persisted.
This mock is wrong.
--
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 incorrect delete sub when lastActive do not update. [pulsar]
Posted by "Technoboy- (via GitHub)" <gi...@apache.org>.
Technoboy- commented on PR #21692:
URL: https://github.com/apache/pulsar/pull/21692#issuecomment-1866028268
> > why do we update the cursor when every check ?
>
> If the consumer do not commit the offset for any reason, the `lastActive` will never be updated, which will result into incorrect subscription deletion. As we know that the consumer is connected, we should update `lastActive` when doing check.
in this case, the consumer is connected, how could the code run into `sub.delete`
https://github.com/apache/pulsar/blob/8beac8b12ef7c0ef54529fbb7e4e76c54dea6283/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java#L2982-L2985
--
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 incorrect delete sub when lastActive do not update. [pulsar]
Posted by "thetumbled (via GitHub)" <gi...@apache.org>.
thetumbled commented on PR #21692:
URL: https://github.com/apache/pulsar/pull/21692#issuecomment-2134776551
> @thetumbled Could you add a test for this case? Thanks
I push a new commit to add the test code, PTAL, 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] fix incorrect delete sub when lastActive do not update. [pulsar]
Posted by "thetumbled (via GitHub)" <gi...@apache.org>.
thetumbled commented on code in PR #21692:
URL: https://github.com/apache/pulsar/pull/21692#discussion_r1618136680
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceTest.java:
##########
@@ -1313,6 +1314,66 @@ 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
+ // disable the updateLastActive method to simulate the ungraceful shutdown
Review Comment:
We can't ensure that the updated timestamp is persisted in crash time. With this PR, we can fix most of the cases, as we update it every time broker do expiration check.
But strictly speaking, we can't fix all the cases. If we always 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 deletions are better than incorrect deletion, this is PR https://github.com/apache/pulsar/pull/22794 try to do.
--
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 incorrect delete sub when lastActive do not update. [pulsar]
Posted by "thetumbled (via GitHub)" <gi...@apache.org>.
thetumbled commented on PR #21692:
URL: https://github.com/apache/pulsar/pull/21692#issuecomment-1854998088
> why do we update the cursor when every check ?
If the consumer do not commit the offset for any reason, the `lastActive` will never be updated, which will result into incorrect subscription deletion.
--
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 incorrect delete sub when lastActive do not update. [pulsar]
Posted by "thetumbled (via GitHub)" <gi...@apache.org>.
thetumbled closed pull request #21692: [fix] [broker] fix incorrect delete sub when lastActive do not update.
URL: https://github.com/apache/pulsar/pull/21692
--
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 incorrect delete sub when lastActive do not update. [pulsar]
Posted by "thetumbled (via GitHub)" <gi...@apache.org>.
thetumbled commented on PR #21692:
URL: https://github.com/apache/pulsar/pull/21692#issuecomment-1867283912
> > > why do we update the cursor when every check ?
> >
> >
> > If the consumer do not commit the offset for any reason, the `lastActive` will never be updated, which will result into incorrect subscription deletion. As we know that the consumer is connected, we should update `lastActive` when doing check.
>
> in this case, the consumer is connected, how could the code run into `sub.delete`
>
> https://github.com/apache/pulsar/blob/8beac8b12ef7c0ef54529fbb7e4e76c54dea6283/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java#L2982-L2985
If the broker1 shutdown ungracefully, the topic will be loaded on broker2 without update the `lastActive` field to zk, and the broker2 load the topic with old `lastActive` field from zk. If broker2 check the inactive sub before any consumer connect to this sub, the sub will be deleted.
This problem is simillar to https://github.com/apache/pulsar/pull/17573.
--
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 incorrect delete sub when lastActive do not update. [pulsar]
Posted by "thetumbled (via GitHub)" <gi...@apache.org>.
thetumbled commented on code in PR #21692:
URL: https://github.com/apache/pulsar/pull/21692#discussion_r1618136680
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceTest.java:
##########
@@ -1313,6 +1314,66 @@ 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
+ // disable the updateLastActive method to simulate the ungraceful shutdown
Review Comment:
We can't ensure that the updated timestamp is persisted in crash time. With this PR, we can fix most of the cases, as we update it every time broker do expiration check.
But strictly speaking, we can't fix all the cases. If we always 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 deletions are better than incorrect deletion, this is what PR https://github.com/apache/pulsar/pull/22794 try to do.
--
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 incorrect delete sub when lastActive do not update. [pulsar]
Posted by "thetumbled (via GitHub)" <gi...@apache.org>.
thetumbled commented on PR #21692:
URL: https://github.com/apache/pulsar/pull/21692#issuecomment-2134788032
PTAL, thanks. @dao-jun @lhotari @Technoboy- @poorbarcode @coderzc
--
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 incorrect delete sub when lastActive do not update. [pulsar]
Posted by "dao-jun (via GitHub)" <gi...@apache.org>.
dao-jun commented on code in PR #21692:
URL: https://github.com/apache/pulsar/pull/21692#discussion_r1617387556
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceTest.java:
##########
@@ -1313,6 +1314,66 @@ 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);
Review Comment:
sleep too long
--
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 incorrect delete sub when lastActive do not update. [pulsar]
Posted by "thetumbled (via GitHub)" <gi...@apache.org>.
thetumbled commented on PR #21692:
URL: https://github.com/apache/pulsar/pull/21692#issuecomment-2144172176
closed as https://github.com/apache/pulsar/pull/22794 has been merged.
--
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