[fix][broker] Run the message expiry check off the topic policy update path - #26040
Merged
lhotari merged 2 commits intoJun 17, 2026
Merged
Conversation
… changed since the last run, schedule on messageExpiryMonitor thread
…expiry check checkMessageExpiry now runs asynchronously on the messageExpiryMonitor thread, so the synchronous verify(times(N)) right after the policy update could race the background execution. Use Mockito timeout verification to wait for the invocations, and drop the now-unused static times import. Assisted-by: Claude Opus 4.8
lhotari
requested review from
Technoboy-,
dao-jun,
merlimat,
nodece and
poorbarcode
June 17, 2026 20:49
merlimat
approved these changes
Jun 17, 2026
3 tasks
sandeep-ctds
pushed a commit
to datastax/pulsar
that referenced
this pull request
Jul 31, 2026
…e path (apache#26040) (cherry picked from commit 4763128)
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Main Issue: #26037
Motivation
When topic policies are applied (
PersistentTopic.onUpdate/onPoliciesUpdate→applyUpdatedTopicPolicies),the message-expiry check (
checkMessageExpiry) was run inline on the calling thread. That check can block: itwalks subscription cursors and can wait on metadata operations
(
...getNthEntry→counter.await(metadataOperationsTimeoutSeconds), default 60s).For topic-policy updates this inline work runs on the single, process-wide shared
broker-client-shared-internal-executorreader thread (the thread reported in #26037), so a single topic whoseoldest ledger is temporarily inaccessible could block topic-policy loading/updates. The check was also executed
on every policy update, even when the message TTL had not changed.
Modifications
applyUpdatedTopicPoliciesno longer runscheckMessageExpiryinline. Instead,maybeCheckMessageExpiryOnPolicyUpdateInBackgroundschedules it on the dedicatedmessageExpiryMonitorthread — the same thread the periodic expiry check already uses — so it never blocks applying or loading
topic policies. The check is idempotent and the periodic monitor still provides eventual coverage, so it is
no longer part of the futures returned by
applyUpdatedTopicPolicies(callers no longer wait for it).new
volatile int lastMessageTtlInSeconds), avoiding redundant scans on unrelated policy updates.BrokerService.messageExpiryMonitorvia@Getterso the topic can schedule on it.Verifying this change
This change is already covered by existing tests:
MessageTTLTest—testTTLPoliciesUpdatewas updated to wait for the now-asynchronous expiry check usingMockito
timeoutverification (the previous synchronousverify(times(N))would otherwise race thebackground execution).
PersistentMessageExpiryMonitorMockTest.Does this pull request potentially affect one of the following parts:
The message-expiry check triggered by a topic-policy update now runs asynchronously on the
messageExpiryMonitorthread instead of synchronously on the policy-application thread.