[fix][broker] Propagate topic policy initialization failures - #26259
Open
Praveenkumar76 wants to merge 1 commit into
Open
[fix][broker] Propagate topic policy initialization failures#26259Praveenkumar76 wants to merge 1 commit into
Praveenkumar76 wants to merge 1 commit into
Conversation
Praveenkumar76
force-pushed
the
fix/26137-topic-policy-load-failure
branch
from
August 1, 2026 06:41
b5987ce to
84bb5f1
Compare
Praveenkumar76
marked this pull request as ready for review
August 2, 2026 07:57
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.
Fixes #26137
Motivation
Topic policy loading is performed during topic initialization through
AbstractTopic.initTopicPolicy().Although
BrokerServicemay read topic policies earlier in the topic-loading flow, a separate policy read is performed later byinitTopicPolicy(). The earlier read can succeed while this later read fails because of a transient metadata-store error, a__change_eventsread failure, or a policy-cache initialization timeout.Previously,
PersistentTopic.initialize()andNonPersistentTopic.initialize()handled this later failure with a terminalexceptionally(...)stage that returnednull. This converted the failed initialization future into a successful one, allowing the topic to finish loading without its topic-level policies applied.The topic could then temporarily or indefinitely use namespace- or broker-level fallback values. This is particularly unsafe when a topic has longer retention or TTL settings than its namespace, because messages can become eligible for deletion or expiration earlier than configured.
This PR chooses fail-closed behavior for failures from
initTopicPolicy(): the failure is logged and propagated instead of allowing the topic to finish initialization with unapplied topic policies.Modifications
initTopicPolicy()inPersistentTopic.initialize()andNonPersistentTopic.initialize()withwhenComplete(...).initTopicPolicy()rather than unrelated initialization operations.Verifying this change
This change added tests and can be verified as follows:
PersistentTopicTest.testInitializeFailsWhenTopicPolicyLoadingFails.NonPersistentTopicTest.testInitializeFailsWhenTopicPolicyLoadingFails.Does this pull request potentially affect one of the following parts:
This changes topic-loading behavior when the later
initTopicPolicy()operation fails. Instead of loading the topic with namespace- or broker-level fallback policies, topic initialization now fails.