[fix][ml] Preserve cursor properties when recovering from an unreadable cursor ledger

[dao-jun]

Fixes #26483

Motivation

ManagedCursorImpl#recoverFromLedger() has three error paths that rebuild the cursor from
the metadata-store snapshot after failing to read the cursor ledger:

initialize(getRollbackPosition(info), Collections.emptyMap(), cursorProperties, callback);

• opening the cursor ledger fails with a non-recoverable BookKeeper error (or with
  ledgerForceRecovery enabled, any error)
• the cursor ledger is empty (lastAddConfirmed < 0)
• reading the last entry fails with a non-recoverable error

initialize() does not treat that map as in-memory state — it persists it:
persistPositionMetaStore() writes it into the durable ManagedCursorInfo.properties list,
so the properties are overwritten with an empty list. The position rollback itself is
defensible (getRollbackPosition() returns the snapshotted mark-delete position clamped up
to the first existing ledger), but the properties are not rolled back to an older value:
they are gone, which is not a rollback to a consistent earlier state.

These properties are load-bearing. Two in-tree consumers:

• Compactor.COMPACTED_TOPIC_LEDGER_PROPERTY is the only durable pointer to a topic's
  compacted ledger. If the key is absent after recovery, PulsarCompactorSubscription
  never registers the compacted ledger, so the topic serves uncompacted data until the next
  successful compaction — and the previously compacted ledger stays unreferenced in
  BookKeeper forever, since its only deletion site is the previousContext replacement.
• PersistentSubscription.REPLICATED_SUBSCRIPTION_PROPERTY: a replicated subscription
  silently stops being treated as replicated after recovery.

The sibling recovery branch already does the right thing: when cursorsLedgerId == -1,
the same method decodes and restores the properties from the ManagedCursorInfo. Only the
cursor-ledger error paths drop them. An ERROR is logged about the ledger, but nothing
indicates that the compacted-topic pointer or the replicated-subscription marker was just
discarded, so the loss is invisible in the logs.

Modifications

In ManagedCursorImpl#recoverFromLedger(), decode the properties from the same
ManagedCursorInfo snapshot that provides the rollback position, and pass them to
initialize() in all three error paths instead of Collections.emptyMap(). Position and
properties now come from the same snapshot, so the rollback is to a consistent earlier
state: if the last cursor-ledger entry carried newer values, they are rolled back together
with the position, which is safe — e.g. an older compacted-topic pointer references a
ledger that is only deleted once a newer pointer has been registered.

The decode is extracted into a small static recoverProperties(count, accessor) helper
(same shape as the existing recoverIndividualDeletedMessages(count, accessor)); the
cursorsLedgerId == -1 branch and the successful-recovery path now use it too — a purely
mechanical deduplication with no behavior change. When the snapshot contains no
properties, the helper returns Collections.emptyMap(), exactly as before, so
property-less cursors are unaffected.

Verifying this change

• Make sure that the change passes the CI checks.

(Please pick either of the following options)

This change is a trivial rework / code cleanup without any test coverage.

(or)

This change is already covered by existing tests, such as (please describe tests).

(or)

This change added tests and can be verified as follows:

(example:)

• Added integration tests for end-to-end deployment with large payloads (10MB)
• Extended integration test for recovery after broker failure

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

[void-ptr974]

Thanks for the fix! I am wondering whether CompactedTopicLedger needs additional consideration because it refers to an external BookKeeper ledger. For example:

1. The metadata-store snapshot contains position P1 and CompactedTopicLedger=L1.
2. A later compaction creates L2, persists P2/L2 in the cursor ledger, and deletes L1 after the mark-delete succeeds.
3. Before the metadata-store snapshot advances, the cursor ledger becomes unreadable.
4. This recovery path rolls back to P1 and restores CompactedTopicLedger=L1, although L1 has already been deleted.

On topic loading, PulsarCompactorSubscription may then try to open the missing L1. Since CompactedTopicImpl.newCompactedLedger() also installs the corresponding compaction horizon and failed future, readCompacted reads at or before that horizon may fail rather than fall back to the original topic data.

Is there an existing guarantee that keeps L1 alive while it is referenced by the metadata-store snapshot? If not, it may be helpful to cover this recovery sequence in a test.

[lhotari]

Preserving snapshot properties addresses the marker loss, but restoring a deleted compacted-ledger pointer can leave readCompacted reads failing. The rollover tests also need to wait for the intended snapshot before injecting recovery failures.