KAFKA-10025: guard RocksDBMetricsRecorder value provider reads against store close

[Aangbaeck]

What

RocksDBMetricsRecorder reads native RocksDB value providers (RocksDB and Statistics)
in three places:

• record() — statistics.getAndResetTickerCount(...) / getHistogramData(...)
• the property gauges (gaugeToComputeSumOfProperties) — db.getAggregatedLongProperty(...)
• the block-cache gauges (gaugeToComputeBlockCacheMetrics) — db.getLongProperty(...)

These run with no mutual exclusion against removeValueProviders(...). RocksDBStore.close()
calls removeValueProviders(name) and then closes (frees) the native RocksDB and
Statistics. Because the reads and the removal are not mutually exclusive, a metrics read
that is in flight when a store is closed (during a rebalance / task migration) can
dereference a native handle that close() is concurrently freeing — a native
use-after-free / SIGSEGV.

Two observed crash frames, same root cause:

• record() path → Statistics::getAndResetTickerCount — this is KAFKA-10025 (open since 2020).
• gauge path → rocksdb::DBImpl::GetAggregatedIntProperty — observed in production under a
  from-zero state rebuild, where warmup/probing rebalances close stores continuously while a
  metrics reporter / JMX scrape evaluates the (INFO-level) RocksDB property gauges.

Note the gauge metrics are registered at RecordingLevel.INFO, so they are active and
scraped even when metrics.recording.level=INFO; only the record() (statistics) path is
gated to DEBUG. So the crash is reachable at INFO.

Why the current code is unsafe

storeToValueProviders is a ConcurrentHashMap, which makes the map operations
thread-safe, but does not prevent a reader that has already obtained a
DbAndCacheAndStatistics from calling a native method on its db/statistics after (or
while) RocksDBStore.close() frees them. There is no happens-before between "recorder reads
the provider" and "store closes the provider".

Fix

Introduce a single lock (valueProvidersLock) in RocksDBMetricsRecorder and hold it
around every read of the value providers (record(), both gauge lambdas) and around the
map mutations (addValueProviders, removeValueProviders). Since RocksDBStore.close()
already calls removeValueProviders(...) before it frees the native db/statistics,
removeValueProviders(...) acquiring the lock guarantees:

• any in-flight read completes before the segment is removed and the native handles freed, and
• any read that starts after removal no longer sees the segment,

so no read can ever dereference a freed handle.

No lock-ordering risk: RocksDBMetricsRecordingTrigger holds no lock while calling
record(), and the guarded reads never call back into RocksDBStore. RocksDBStore.close()
takes the store monitor then this lock; opens take the same order — consistent, no cycle.

Testing

• New test RocksDBMetricsRecorderGaugesTest#shouldNotRemoveValueProvidersWhileGaugeIsReadingThem:
  blocks a gauge evaluation inside getAggregatedLongProperty (holding the lock) and asserts
  removeValueProviders(...) cannot return until the read completes — i.e. the use-after-free
  window is closed. Verified it fails without the fix (AssertionError: …the use-after-free window is open) and passes with it.
• Existing RocksDBMetricsRecorderTest / RocksDBMetricsRecorderGaugesTest, checkstyle and
  spotbugs all pass.

End-to-end confirmation (outside this PR, in a standalone Docker harness): a real Kafka
Streams app on the released kafka-streams 8.2.1-ce, at metrics.recording.level=INFO, with
a JMX-style metrics scrape (reading the INFO-level RocksDB property gauges) plus forced
rebalances, SIGSEGVs in rocksdb::DBImpl::GetAggregatedIntProperty+0x83 on a scrape thread
after ~27M gauge reads. Running the same binary with only RocksDBMetricsRecorder replaced
by the patched class (classpath shadow, load-verified) survived 600s / ~336M gauge reads / 25
rebalances with zero crashes. The exact native crash was also reproduced in a pure rocksdbjni
harness by racing getAggregatedLongProperty against a concurrent DB close+reopen.