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

dragonls · 2022-09-09T11:12:07Z

Motivation

The lastActive in ManagedCursorImpl only be updated in 3 places:

org.apache.pulsar.broker.service.persistent.PersistentSubscription#addConsumer
org.apache.pulsar.broker.service.persistent.PersistentSubscription#removeConsumer
org.apache.bookkeeper.mledger.impl.ManagedCursorImpl#internalResetCursor

If there are brokers shutdown with OOM or other situation, which make the broker not shutdown gracefully, the new broker load the topics and then do subscription expiry check soon before consumer reconnected, then the subscription may be deleted unexpectedly.

Need to update lastActive in ManagedCursorImpl while consuming stably, such as consumer acknowledged messages, which makes lastActive closer to the last active meaning.

Note

This PR mainly fix the bug that subscription may be deleted unexpectedly caused by lastActive not being updated in time, which depends on the lastActive persistence in zk.

lastActive is saved in zk in org.apache.bookkeeper.mledger.impl.ManagedCursorImpl#persistPositionMetaStore:

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

Lines 2500 to 2514 in d7c09be

    
           private void persistPositionMetaStore(long cursorsLedgerId, PositionImpl position, Map<String, Long> properties, 
        
                   MetaStoreCallback<Void> callback, boolean persistIndividualDeletedMessageRanges) { 
        
               if (state == State.Closed) { 
        
                   ledger.getExecutor().execute(safeRun(() -> callback.operationFailed(new MetaStoreException( 
        
                           new CursorAlreadyClosedException(name + " cursor already closed"))))); 
        
                   return; 
        
               } 
        
               // When closing we store the last mark-delete position in the z-node itself, so we won't need the cursor ledger, 
        
               // hence we write it as -1. The cursor ledger is deleted once the z-node write is confirmed. 
        
               ManagedCursorInfo.Builder info = ManagedCursorInfo.newBuilder() // 
        
                       .setCursorsLedgerId(cursorsLedgerId) // 
        
                       .setMarkDeleteLedgerId(position.getLedgerId()) // 
        
                       .setMarkDeleteEntryId(position.getEntryId()) // 
        
                       .setLastActive(lastActive); //

However, the persistence is triggered only in 4 places:

All of them are not high frequency triggered. The new broker may recover the cursor with lastActive from zk which is not the real last active time. The unexpectedly subscription deletion still can appear during two persistence having broker shutdown not gracefully.

The better solution is to save the lastActive into cursor ledger instead of only save to zk, but this is not the problem this PR solving, and need further discussion.

Modifications

Add cursor.updateLastActive() in org.apache.pulsar.broker.service.persistent.PersistentSubscription#acknowledgeMessage.
Add some logs about cursor last active timestamp.

Verifying this change

Make sure that the change passes the CI checks.

This change added tests and can be verified as follows:

org.apache.pulsar.broker.service.persistent.PersistentTopicTest#testUpdateCursorLastActive
org.apache.pulsar.broker.service.persistent.PersistentSubscriptionTest#testAcknowledgeUpdateCursorLastActive

Does this pull request potentially affect one of the following parts:

If the box was checked, please highlight the changes

Documentation

doc-required
(Your PR needs to update docs and you will update later)
doc-not-needed
(Please explain why)
doc
(Your PR contains doc changes)
doc-complete
(Docs have been already added)

…e log. (apache#17572)

liangyepianzhou

If I understand correctly, this issue does not relate to the broker restart, right?
It is that the subscriptionExpirationTimeMinutes should be the last ack time of the consumers in the subscription.
Instead of the last time that the consumer was modified (consumer add or remove) in the subscription.

.../src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentSubscription.java

dragonls · 2022-09-13T02:50:04Z

If I understand correctly, this issue does not relate to the broker restart, right? It is that the subscriptionExpirationTimeMinutes should be the last ack time of the consumers in the subscription. Instead of the last time that the consumer was modified (consumer add or remove) in the subscription.

If the broker shutdown gracefully, everything is fine. Because it will go through the RemoveConsumer, the lastActive will be updated.

As for the check of subscriptionExpirationTimeMinutes, see

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

Lines 2351 to 2358 in 200f433

    
               if (sub.dispatcher != null && sub.dispatcher.isConsumerConnected() || sub.isReplicated()) { 
        
                   return; 
        
               } 
        
               if (System.currentTimeMillis() - sub.cursor.getLastActive() > expirationTimeMillis) { 
        
                   sub.delete().thenAccept(v -> log.info("[{}][{}] The subscription was deleted due to expiration", 
        
                           topic, subName)); 
        
               } 
        
           });

If the consumer is connected, it will skip the check.

…pache#17572)

codelipenghui

If there are brokers shutdown with OOM or other situation, which make the broker not shutdown gracefully, the new broker load the topics and then do subscription expiry check soon before consumer reconnected, then the subscription may be deleted unexpectedly.

If the topic is loaded by another broker, the lastActive will be reset to clock.millis()

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

Line 305 in a6fe5bb

this.lastActive = this.clock.millis();

How the above situation can happen?

dragonls · 2022-09-15T14:11:46Z

If there are brokers shutdown with OOM or other situation, which make the broker not shutdown gracefully, the new broker load the topics and then do subscription expiry check soon before consumer reconnected, then the subscription may be deleted unexpectedly.

If the topic is loaded by another broker, the lastActive will be reset to clock.millis()

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

Line 305 in a6fe5bb

this.lastActive = this.clock.millis();

How the above situation can happen?

lastActive is set to clock.millis() while initilized, but will be covered by recover() soon.

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

Line 414 in a6fe5bb

lastActive = info.getLastActive() != 0 ? info.getLastActive() : lastActive;

codelipenghui · 2022-09-15T14:52:48Z

lastActive is set to clock.millis() while initilized, but will be covered by recover() soon.

I see.

Actually, only ledger rollover and updating the cursor Ledger failed will cause the lastActive update to zookeeper right?
It also has a chance to delete the subscription unexpectedly if no ledger rollover and cursor ledger update failure happens.

codelipenghui · 2022-09-15T15:02:24Z

If we can persistent the lastActive to cursor Ledger, we will get better for this situation.

dragonls · 2022-09-16T02:58:22Z

If we can persistent the lastActive to cursor Ledger, we will get better for this situation.

I will take a look at this and try to persistent the lastActive to cursor ledger.

dragonls · 2022-09-16T10:30:24Z

After discussed with @codelipenghui , this PR mainly fix the bug that subscription may be deleted unexpectedly caused by lastActive not being updated in time, which depends on the lastActive persistence in zk.

lastActive is saved in zk in org.apache.bookkeeper.mledger.impl.ManagedCursorImpl#persistPositionMetaStore:

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

Lines 2500 to 2514 in d7c09be

    
           private void persistPositionMetaStore(long cursorsLedgerId, PositionImpl position, Map<String, Long> properties, 
        
                   MetaStoreCallback<Void> callback, boolean persistIndividualDeletedMessageRanges) { 
        
               if (state == State.Closed) { 
        
                   ledger.getExecutor().execute(safeRun(() -> callback.operationFailed(new MetaStoreException( 
        
                           new CursorAlreadyClosedException(name + " cursor already closed"))))); 
        
                   return; 
        
               } 
        
               // When closing we store the last mark-delete position in the z-node itself, so we won't need the cursor ledger, 
        
               // hence we write it as -1. The cursor ledger is deleted once the z-node write is confirmed. 
        
               ManagedCursorInfo.Builder info = ManagedCursorInfo.newBuilder() // 
        
                       .setCursorsLedgerId(cursorsLedgerId) // 
        
                       .setMarkDeleteLedgerId(position.getLedgerId()) // 
        
                       .setMarkDeleteEntryId(position.getEntryId()) // 
        
                       .setLastActive(lastActive); //

However, the persistence is triggered only in 4 places:

All of them are not high frequency triggered. The new broker may recover the cursor with lastActive from zk which is not the real last active time. The unexpectedly subscription deletion still can appear during two persistence having broker shutdown not gracefully.

The better solution is to save the lastActive into cursor ledger instead of only save to zk, but this is not the problem this PR solving, and need further discussion.

codelipenghui · 2022-09-16T10:56:56Z

@dragonls I have approved the PR. Please also update your last comment to the PR description. So that we don't to go through all the comment to understand what is the scope of this PR and what is not the scope of this PR

…or last active time not updated in time (apache#17573) (cherry picked from commit 8a9d70a) Signed-off-by: Zixuan Liu <[email protected]>

[fix][broker] Fix unexpected subscription deletion and add last activ…

b3a415b

…e log. (apache#17572)

github-actions bot added the doc-not-needed Your PR changes do not impact docs label Sep 9, 2022

liangyepianzhou reviewed Sep 10, 2022

View reviewed changes

HQebupt reviewed Sep 12, 2022

View reviewed changes

.../src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentSubscription.java Outdated Show resolved Hide resolved

[fix][broker] Add test cases for updating cursor last active time. (a…

cae504f

…pache#17572)

dragonls requested review from HQebupt and liangyepianzhou and removed request for HQebupt and liangyepianzhou September 15, 2022 03:26

mattisonchao assigned dragonls Sep 15, 2022

mattisonchao added this to the 2.12.0 milestone Sep 15, 2022

mattisonchao added area/broker type/bug The PR fixed a bug or issue reported a bug labels Sep 15, 2022

mattisonchao requested review from codelipenghui, eolivelli and nicoloboschi and removed request for HQebupt September 15, 2022 11:32

gaozhangmin approved these changes Sep 15, 2022

View reviewed changes

codelipenghui reviewed Sep 15, 2022

View reviewed changes

dragonls changed the title ~~[fix][broker] Fix unexpected subscription deletion~~ [fix][broker] Fix unexpected subscription deletion caused by the cursor last active time not updated in time Sep 16, 2022

codelipenghui approved these changes Sep 16, 2022

View reviewed changes

HQebupt approved these changes Sep 16, 2022

View reviewed changes

codelipenghui merged commit 8a9d70a into apache:master Sep 16, 2022

lhotari mentioned this pull request Sep 19, 2022

lh reduce memory usage stub only mocks spies lhotari/pulsar#85

Closed

dragonls deleted the fix-subscription-last-active branch September 19, 2022 10:05

thetumbled mentioned this pull request Dec 8, 2023

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

Closed

15 tasks

thetumbled mentioned this pull request May 29, 2024

[fix] [broker] maintain last active info in memory only. #22794

Merged

15 tasks

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

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

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

dragonls commented Sep 9, 2022 •

edited

Loading

liangyepianzhou left a comment

dragonls commented Sep 13, 2022 •

edited

Loading

codelipenghui left a comment

dragonls commented Sep 15, 2022

codelipenghui commented Sep 15, 2022

codelipenghui commented Sep 15, 2022

dragonls commented Sep 16, 2022

dragonls commented Sep 16, 2022 •

edited

Loading

codelipenghui commented Sep 16, 2022

	private void persistPositionMetaStore(long cursorsLedgerId, PositionImpl position, Map<String, Long> properties,
	MetaStoreCallback<Void> callback, boolean persistIndividualDeletedMessageRanges) {
	if (state == State.Closed) {
	ledger.getExecutor().execute(safeRun(() -> callback.operationFailed(new MetaStoreException(
	new CursorAlreadyClosedException(name + " cursor already closed")))));
	return;
	}

	// When closing we store the last mark-delete position in the z-node itself, so we won't need the cursor ledger,
	// hence we write it as -1. The cursor ledger is deleted once the z-node write is confirmed.
	ManagedCursorInfo.Builder info = ManagedCursorInfo.newBuilder() //
	.setCursorsLedgerId(cursorsLedgerId) //
	.setMarkDeleteLedgerId(position.getLedgerId()) //
	.setMarkDeleteEntryId(position.getEntryId()) //
	.setLastActive(lastActive); //

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

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

Conversation

dragonls commented Sep 9, 2022 • edited Loading

Motivation

Modifications

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

liangyepianzhou left a comment

Choose a reason for hiding this comment

dragonls commented Sep 13, 2022 • edited Loading

codelipenghui left a comment

Choose a reason for hiding this comment

dragonls commented Sep 15, 2022

codelipenghui commented Sep 15, 2022

codelipenghui commented Sep 15, 2022

dragonls commented Sep 16, 2022

dragonls commented Sep 16, 2022 • edited Loading

codelipenghui commented Sep 16, 2022

dragonls commented Sep 9, 2022 •

edited

Loading

dragonls commented Sep 13, 2022 •

edited

Loading

dragonls commented Sep 16, 2022 •

edited

Loading