@Override
    public void acknowledgeMessage(List<Position> positions, AckType ackType, Map<String, Long> properties) {
        cursor.updateLastActive();
        Position previousMarkDeletePosition = cursor.getMarkDeletedPosition();

        if (ackType == AckType.Cumulative) {

            if (positions.size() != 1) {
                log.warn("[{}][{}] Invalid cumulative ack received with multiple message ids.", topicName, subName);
                return;
            }

            Position position = positions.get(0);
            if (log.isDebugEnabled()) {
                log.debug("[{}][{}] Cumulative ack on {}", topicName, subName, position);
            }
            cursor.asyncMarkDelete(position, mergeCursorProperties(properties),
                    markDeleteCallback, previousMarkDeletePosition);

        } else {
            if (log.isDebugEnabled()) {
                log.debug("[{}][{}] Individual acks on {}", topicName, subName, positions);
            }
            cursor.asyncDelete(positions, deleteCallback, previousMarkDeletePosition);
            if (topic.getBrokerService().getPulsar().getConfig().isTransactionCoordinatorEnabled()) {
                positions.forEach(position -> {
                    if (((ManagedCursorImpl) cursor).isMessageDeleted(position)) {
                        pendingAckHandle.clearIndividualPosition(position);
                    }
                });
            }

            if (dispatcher != null) {
                dispatcher.getRedeliveryTracker().removeBatch(positions);
            }
        }

        if (!cursor.getMarkDeletedPosition().equals(previousMarkDeletePosition)) {
            this.updateLastMarkDeleteAdvancedTimestamp();

            // Mark delete position advance
            ReplicatedSubscriptionSnapshotCache snapshotCache = this.replicatedSubscriptionSnapshotCache;
            if (snapshotCache != null) {
                ReplicatedSubscriptionsSnapshot snapshot = snapshotCache
                        .advancedMarkDeletePosition(cursor.getMarkDeletedPosition());
                if (snapshot != null) {
                    topic.getReplicatedSubscriptionController()
                            .ifPresent(c -> c.localSubscriptionUpdated(subName, snapshot));
                }
            }
        }

        if (topic.getManagedLedger().isTerminated() && cursor.getNumberOfEntriesInBacklog(false) == 0) {
            // Notify all consumer that the end of topic was reached
            if (dispatcher != null) {
                checkAndApplyReachedEndOfTopicOrTopicMigration(topic, dispatcher.getConsumers());
            }
        }
    }
@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(PositionFactory.create(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);
    }