[fix][broker] Fix unexpected subscription deletion caused by the cursor last active time not updated in time

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java

@@ -412,7 +412,9 @@ public void operationComplete(ManagedCursorInfo info, Stat stat) {
 
                 cursorLedgerStat = stat;
                 lastActive = info.getLastActive() != 0 ? info.getLastActive() : lastActive;
-
+                if (log.isDebugEnabled()) {
+                    log.debug("[{}] [{}] Recover cursor last active to [{}]", ledger.getName(), name, lastActive);
+                }
 
                 Map<String, String> recoveredCursorProperties = Collections.emptyMap();
                 if (info.getCursorPropertiesCount() > 0) {

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/MetaStoreImpl.java

@@ -215,8 +215,9 @@ public void asyncGetCursorInfo(String ledgerName, String cursorName,
     public void asyncUpdateCursorInfo(String ledgerName, String cursorName, ManagedCursorInfo info, Stat stat,
             MetaStoreCallback<Void> callback) {
         if (log.isDebugEnabled()) {
-            log.debug("[{}] [{}] Updating cursor info ledgerId={} mark-delete={}:{}", ledgerName, cursorName,
-                    info.getCursorsLedgerId(), info.getMarkDeleteLedgerId(), info.getMarkDeleteEntryId());
+            log.debug("[{}] [{}] Updating cursor info ledgerId={} mark-delete={}:{} lastActive={}",
+                    ledgerName, cursorName, info.getCursorsLedgerId(), info.getMarkDeleteLedgerId(),
+                    info.getMarkDeleteEntryId(), info.getLastActive());
         }
 
         String path = PREFIX + ledgerName + "/" + cursorName;

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentTopic.java

@@ -1001,8 +1001,8 @@ public void checkInactiveSubscriptions() {
                         return;
                     }
                     if (System.currentTimeMillis() - sub.getLastActive() > expirationTimeMillis) {
-                        sub.delete().thenAccept(v -> log.info("[{}][{}] The subscription was deleted due to expiration",
-                                topic, subName));
+                        sub.delete().thenAccept(v -> log.info("[{}][{}] The subscription was deleted due to expiration "
+                                + "with last active [{}]", topic, subName, sub.getLastActive()));
                     }
                 });
             }

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java

@@ -373,6 +373,7 @@ public void consumerFlow(Consumer consumer, int additionalNumberOfMessages) {
 
     @Override
     public void acknowledgeMessage(List<Position> positions, AckType ackType, Map<String, Long> properties) {
+        cursor.updateLastActive();
         Position previousMarkDeletePosition = cursor.getMarkDeletedPosition();
 
         if (ackType == AckType.Cumulative) {

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java

@@ -2352,8 +2352,8 @@ public void checkInactiveSubscriptions() {
                         return;
                     }
                     if (System.currentTimeMillis() - sub.cursor.getLastActive() > expirationTimeMillis) {
-                        sub.delete().thenAccept(v -> log.info("[{}][{}] The subscription was deleted due to expiration",
-                                topic, subName));
+                        sub.delete().thenAccept(v -> log.info("[{}][{}] The subscription was deleted due to expiration "
+                                + "with last active [{}]", topic, subName, sub.cursor.getLastActive()));
                     }
                 });
             }

pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentSubscriptionTest.java

@@ -23,6 +23,7 @@
 import static org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest.createMockZooKeeper;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.doCallRealMethod;
 import static org.mockito.Mockito.doNothing;
 import static org.mockito.Mockito.doReturn;
 import static org.mockito.Mockito.mock;
@@ -348,4 +349,25 @@ public void testCanAcknowledgeAndAbortForTransaction() throws Exception {
 
         persistentSubscription.transactionIndividualAcknowledge(txnID2, positionsPair);
     }
+
+    @Test
+    public void testAcknowledgeUpdateCursorLastActive() throws Exception {
+        doAnswer((invocationOnMock) -> {
+            ((AsyncCallbacks.DeleteCallback) invocationOnMock.getArguments()[1])
+                    .deleteComplete(invocationOnMock.getArguments()[2]);
+            return null;
+        }).when(cursorMock).asyncDelete(any(List.class), any(AsyncCallbacks.DeleteCallback.class), any());
+
+        doCallRealMethod().when(cursorMock).updateLastActive();
+        doCallRealMethod().when(cursorMock).getLastActive();
+
+        List<Position> positionList = new ArrayList<>();
+        positionList.add(new PositionImpl(1, 1));
+        long beforeAcknowledgeTimestamp = System.currentTimeMillis();
+        Thread.sleep(1);
+        persistentSubscription.acknowledgeMessage(positionList, AckType.Individual, Collections.emptyMap());
+
+        // `acknowledgeMessage` should update cursor last active
+        assertTrue(persistentSubscription.cursor.getLastActive() > beforeAcknowledgeTimestamp);
+    }
 }

pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentTopicTest.java

@@ -347,4 +347,59 @@ public void testDelayedDeliveryTrackerMemoryUsageMetric(String topic, boolean ex
             Assert.assertEquals(topicLevelNum, 0);
         }
     }
+
+    @Test
+    public void testUpdateCursorLastActive() throws Exception {
+        final String topicName = "persistent://prop/ns-abc/aTopic";
+        final String sharedSubName = "shared";
+        final String failoverSubName = "failOver";
+
+        long beforeAddConsumerTimestamp = System.currentTimeMillis();
+        Thread.sleep(1);
+        Consumer<String> consumer = pulsarClient.newConsumer(Schema.STRING).topic(topicName)
+                .subscriptionType(SubscriptionType.Shared).subscriptionName(sharedSubName)
+                .acknowledgmentGroupTime(0, TimeUnit.MILLISECONDS).subscribe();
+        Consumer<String> consumer2 = pulsarClient.newConsumer(Schema.STRING).topic(topicName)
+                .subscriptionType(SubscriptionType.Failover).subscriptionName(failoverSubName)
+                .acknowledgmentGroupTime(0, TimeUnit.MILLISECONDS).subscribe();
+        Producer<String> producer = pulsarClient.newProducer(Schema.STRING).topic(topicName).create();
+
+        PersistentTopic topic = (PersistentTopic) pulsar.getBrokerService().getTopicReference(topicName).get();
+        PersistentSubscription persistentSubscription =  topic.getSubscription(sharedSubName);
+        PersistentSubscription persistentSubscription2 =  topic.getSubscription(failoverSubName);
+
+        // `addConsumer` should update last active
+        assertTrue(persistentSubscription.getCursor().getLastActive() > beforeAddConsumerTimestamp);
+        assertTrue(persistentSubscription2.getCursor().getLastActive() > beforeAddConsumerTimestamp);
+
+        long beforeAckTimestamp = System.currentTimeMillis();
+        Thread.sleep(1);
+        producer.newMessage().value("test").send();
+        Message<String> msg = consumer.receive();
+        assertNotNull(msg);
+        consumer.acknowledge(msg);
+        msg = consumer2.receive();
+        assertNotNull(msg);
+        consumer2.acknowledge(msg);
+
+        // Make sure ack commands have been sent to broker
+        Awaitility.waitAtMost(5, TimeUnit.SECONDS)
+                .until(() -> persistentSubscription.getCursor().getLastActive() > beforeAckTimestamp);
+        Awaitility.waitAtMost(5, TimeUnit.SECONDS)
+                .until(() -> persistentSubscription2.getCursor().getLastActive() > beforeAckTimestamp);
+
+        // `acknowledgeMessage` should update last active
+        assertTrue(persistentSubscription.getCursor().getLastActive() > beforeAckTimestamp);
+        assertTrue(persistentSubscription2.getCursor().getLastActive() > beforeAckTimestamp);
+
+        long beforeRemoveConsumerTimestamp = System.currentTimeMillis();
+        Thread.sleep(1);
+        consumer.unsubscribe();
+        consumer2.unsubscribe();
+        producer.close();
+
+        // `removeConsumer` should update last active
+        assertTrue(persistentSubscription.getCursor().getLastActive() > beforeRemoveConsumerTimestamp);
+        assertTrue(persistentSubscription2.getCursor().getLastActive() > beforeRemoveConsumerTimestamp);
+    }
 }