Aangbaeck opened a new pull request, #22717:
URL: https://github.com/apache/kafka/pull/22717
## 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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
