[fix][ml] Make managed ledger properties updates atomic

[void-ptr974]

Motivation

ManagedLedgerImpl initializes propertiesMap as a ConcurrentHashMap, but replaces it with a plain
HashMap when recovering persisted properties. In addition, getProperties() exposes the live internal
map, and property updates mutate the in-memory map before the metadata-store write succeeds.

As a result, callers can mutate managed-ledger state without using the property update APIs, readers can
observe a partially applied multi-property update, and a failed metadata-store update can leave memory
inconsistent with persisted metadata.

Modifications

• Keep recovered properties in a concurrent map and publish replacements through a volatile reference.
• Return detached snapshots from getProperties() and property-update callbacks.
• Defensively copy maps supplied to synchronous and asynchronous property update APIs.
• Build property updates on a private candidate map, including interceptor changes, and publish the new
  snapshot only after the metadata-store write succeeds.
• Release the metadata mutex before invoking user callbacks and isolate callback failures.

This change is limited to the managed-ledger properties API and does not change termination, migration,
or other managed-ledger metadata update flows.

Verifying this change

• Make sure that the change passes the CI checks.

This change added tests covering:

• detached property snapshots and callback maps;
• rollback of in-memory state after a metadata-store failure;
• atomic visibility while a multi-property metadata update is pending;
• metadata-mutex release when a callback throws;
• existing concurrent property updates and read-only managed-ledger compatibility.

Commands run locally:

• ./gradlew quickCheck
• targeted :managed-ledger:test properties, concurrency, and read-only tests
• ./gradlew :pulsar-broker:test --tests org.apache.pulsar.broker.intercept.ManagedLedgerInterceptorImplTest.testRecoveryIndex

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

ManagedLedger.getProperties() now explicitly returns a detached snapshot rather than the live backing
map. The threading change is limited to atomic publication of property snapshots; no executors or thread
pools are changed.

[lhotari]

Thanks for working on making managed-ledger property updates atomic. One concurrent migration path can still lose the durable migration marker when it overlaps a pending property update.

[lhotari]

[BUG] Publishing this snapshot can erase a concurrent migration marker

asyncMigrate() writes migrated=true directly to the currently published map at ManagedLedgerImpl.java:1400-1403 without taking metadataMutex. If a property update has already captured propertiesSnapshot, migration can add the marker while its BookKeeper close is pending, and this callback can then replace the map with the older snapshot. The close callback subsequently serializes that replacement using the updated ledgersStat, so both operations can succeed while the terminated metadata lacks the marker; reopening the ledger then reports isMigrated() as false. Please serialize the internal migration write with snapshot publication and add a deterministic regression test that gates the property metadata callback and BookKeeper close callback in this order.

[void-ptr974]

Thanks, good catch. Fixed in 38b196b.

asyncMigrate() now writes the migration marker through asyncSetProperty() and starts termination only after the property update succeeds. This serializes it with other property snapshot updates under metadataMutex.

I added a deterministic regression test that releases the pending property callback before the gated BookKeeper close callback, then reopens the ledger and verifies the migration marker is preserved. It fails on the previous revision and passes with the fix. quickCheck and the focused tests also pass.