[fix][ml][broker] Keep source ledger data on every shadow managed ledger trim, delete and offload path - #26746

Motivation

A shadow managed ledger reads the ledgers of its source managed ledger, but those ledgers, and any data offloaded from them, belong to the source. The source's own lifecycle manages them. Most shadow code paths already follow this model. This change makes the remaining managed ledger lifecycle paths follow it consistently: trimming, deletion, offloading and offload-lag trimming. It also makes the shadow source a persistent property of the shadow managed ledger instead of a config-only value.

Making the ownership explicit also covers these cases:

• Shadow managed ledgers that are deleted while not open, or opened without the shadow config. This includes partitions of partitioned shadow topics created by earlier versions, where the shadow source is recorded only in the partitioned topic metadata.

• The PULSAR.SHADOW_SOURCE property, which defines what a shadow topic is. It is kept stable for the lifetime of the topic.

  Removing the property never turned a shadow topic into a usable standalone topic, and it isn't part of the shadow topic lifecycle in PIP-180 or the admin API. The shadow's managed ledger metadata still lists the source's ledgers, and its cursors point into them. Without the property, the topic reloads as a regular managed ledger that treats those ledgers as its own, so trimming or deleting it would remove ledgers that belong to the source. The source can also trim those ledgers at any time on its own, so reads from such a topic would stop working in any case. That is why the property can no longer be removed or set to null. Detaching a shadow topic from its source would need a dedicated operation, for example one that drops the source ledgers from the shadow's ledger list, and is out of scope here.

• Pending ack stores of shadow topic subscriptions. These own their ledgers and are cleaned up with their own managed ledger config.

Modifications

• ManagedLedgerImpl: add ownsLedgerData(). It returns false when the config or the stored managed ledger properties contain PULSAR.SHADOW_SOURCE, and it is always false for ShadowManagedLedgerImpl. When a managed ledger does not own its ledger data, trimming keeps the ledgers, asyncOffloadPrefix fails, and automatic offloading is a no-op. The PULSAR.SHADOW_SOURCE managed ledger property cannot be removed.
• ShadowManagedLedgerImpl: store the shadow source in the managed ledger properties on initialization.
• ManagedLedgerFactoryImpl: deleting a managed ledger whose stored properties or supplied config contain the shadow source removes only its metadata. The shared metadata removal code is extracted into a helper method.
• BrokerService: when deleting the managed ledger of a partition, resolve the shadow source from the partitioned topic metadata and pass it in the deletion config.
• Admin API: removing the PULSAR.SHADOW_SOURCE topic property, or setting it to null, returns 412 Precondition Failed.
• PersistentTopic: when a subscription of a shadow topic is deleted, delete its pending ack store with the pending ack store's own managed ledger config. It keeps the topic's storage class and ledger offloader, so the offloader that wrote any offloaded pending ack data also cleans it up.

Verifying this change

• Make sure that the change passes the CI checks.

This change added tests and can be verified as follows:

• ShadowManagedLedgerImplTest:
  • testShadowTrimmingKeepsSourceLedgers, testShadowDeletionKeepsSourceLedgers, testClosedShadowDeletionKeepsSourceLedgers: trimming or deleting an open or closed shadow managed ledger leaves the source ledgers in place.
  • testShadowReopenedWithoutSourceKeepsSourceLedgers, testExistingManagedLedgerOpenedAsShadowStoresSourceProperty, testShadowWithoutStoredSourcePropertyDeletionKeepsSourceLedgers: the stored property, or the source supplied in the deletion config, is honoured when the shadow config is absent or the metadata was written by an earlier version.
  • testShadowSourcePropertyCannotBeRemoved.
  • testShadowOffloadKeepsSourceOffloadedData, testShadowAutomaticOffloadAndOffloadLagTrimmingKeepSourceData: manual offload, automatic offload and offload-lag trimming leave the source's ledgers and offloaded data in place. The automatic offload test uses a clock set past the source offload timestamp instead of sleeps.
• ShadowTopicTest:
  • testShadowSourcePropertyCannotBeRemoved, testShadowSourcePropertyCannotBeUpdatedToNull: the admin API returns 412 and the shadow source stays configured.
  • testPartitionedShadowDeletionWithoutStoredSourcePropertyKeepsSourceLedgers: deleting an unloaded partition of a partitioned shadow topic leaves the source ledgers in place.
  • testShadowTopicSubscriptionDeletionDeletesPendingAckStoreLedgers, testShadowTopicSubscriptionDeletionUsesTopicOffloaderForPendingAckStore: pending ack store ledgers are cleaned up, using the topic's offloader.

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

REST endpoints: removing the PULSAR.SHADOW_SOURCE topic property, or setting it to null, returns 412 Precondition Failed.