[fix] [broker] fix incorrect delete sub when lastActive do not update.

[thetumbled]

Motivation

PR #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:

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

• doc
• doc-required
• doc-not-needed
• doc-complete

Matching PR in forked repository

PR in forked repository: thetumbled#31

[poorbarcode]

@thetumbled Could you add a test for this case? Thanks

[Technoboy-]

why do we update the cursor when every check ?

[thetumbled]

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.

[Technoboy-]

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

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

Lines 2982 to 2985 in 8beac8b

	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()));
	}

[thetumbled]

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

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

Lines 2982 to 2985 in 8beac8b

	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()));
	}

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 #17573.

[thetumbled]

@thetumbled Could you add a test for this case? Thanks

I push a new commit to add the test code, PTAL, thanks.

[thetumbled]

PTAL, thanks. @dao-jun @lhotari @Technoboy- @poorbarcode @coderzc @codelipenghui

[poorbarcode]

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 will be persisted.

This mock is wrong.

[thetumbled]

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 #22794 try to do.

[dao-jun]

sleep too long

[thetumbled]

It is, but the minimum expiration time is 1min, we have to wait for 1min to trigger the bug.

[liangyepianzhou]

Update last active when a consumer disconnect instead of every check.

[thetumbled]

closed as #22794 has been merged.