[fix][broker] Propagate topic policy initialization failures

[Praveenkumar76]

Fixes #26137

Motivation

Topic policies are loaded during topic initialization through AbstractTopic.initTopicPolicy().

Previously, both PersistentTopic.initialize() and NonPersistentTopic.initialize() handled failures from initTopicPolicy() with exceptionally(...) and returned null. This converted the failed future into a successful one, allowing the topic to finish loading without its topic-level policies being applied.

As a result, the topic could temporarily or indefinitely use namespace-level or broker-level fallback settings. This is unsafe for policies such as retention and TTL because the effective behavior may differ from the policy configured directly on the topic.

For non-persistent topics, propagating the initialization failure also requires complete cleanup. The topic-policy listener is registered before the initial policy fetch finishes. If topic creation fails and the topic is only removed from the broker cache without being closed, the listener can continue retaining the failed topic instance. Retrying the same topic could then accumulate stale listeners.

Modifications

• Updated PersistentTopic.initialize() to preserve failures from initTopicPolicy() instead of converting them into successful initialization.
• Updated NonPersistentTopic.initialize() to preserve failures from initTopicPolicy() instead of ignoring them.
• Kept policy-loading failure logging while allowing the original exception to propagate.
• Preserved the existing reset of isEncryptionRequired when persistent-topic policy initialization fails.
• Updated BrokerService#createNonPersistentTopic() to fully close a partially initialized non-persistent topic when creation fails.
• Replaced the limited stopReplProducers() cleanup with close(true, true) so registered topic-policy listeners and other topic resources are released.
• Preserved the original topic-creation failure even if cleanup also fails.
• Added tests for persistent and non-persistent topic initialization failures.
• Added a BrokerService-level test that verifies:
  • the failed topic is removed from the topic cache;
  • its topic-policy listener is unregistered;
  • retrying the same topic succeeds;
  • only one listener remains after the retry.

Verifying this change

This change added tests and can be verified as follows:

• PersistentTopicTest.testInitializeFailsWhenTopicPolicyLoadingFails
• NonPersistentTopicTest.testInitializeFailsWhenTopicPolicyLoadingFails
• BrokerServiceTest.testNonPersistentTopicPolicyLoadFailureCleansUpListener

The following related tests also pass:

• BrokerServiceTest.testGetTopicWhenTopicPoliciesFail
• BrokerServiceTest.testMetricsNonPersistentTopicLoadFails

./gradlew :pulsar-broker:test \
  --tests "org.apache.pulsar.broker.service.PersistentTopicTest.testInitializeFailsWhenTopicPolicyLoadingFails" \
  --tests "org.apache.pulsar.broker.service.nonpersistent.NonPersistentTopicTest.testInitializeFailsWhenTopicPolicyLoadingFails" \
  --tests "org.apache.pulsar.broker.service.BrokerServiceTest.testNonPersistentTopicPolicyLoadFailureCleansUpListener" \
  --tests "org.apache.pulsar.broker.service.BrokerServiceTest.testGetTopicWhenTopicPoliciesFail" \
  --tests "org.apache.pulsar.broker.service.BrokerServiceTest.testMetricsNonPersistentTopicLoadFails" \
  -PskipJavaVersionCheck \
  -DtestRetryCount=0 \
  --no-daemon \
  --no-configuration-cache

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

• 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