nicktelford opened a new pull request, #22716:
URL: https://github.com/apache/kafka/pull/22716
When `enable.transactional.statestores=true`, the KIP-1035 open/closed
status marker written to the RocksDB offsets column family on `open()` and
`close()` was being routed through `TransactionalDBAccessor.put()`, which
stages writes in the in-memory `RocksDBTransactionBuffer`. The buffer is
discarded without committing on store close, so the CLOSED marker was silently
lost. The on-disk status remained permanently at OPEN, causing every subsequent
`open()` under EOS (`ignoreInvalidState=false`) to throw
`ProcessorStateException("Invalid state during store open...")`, which
escalated to `TaskCorruptedException`. Because the wipe-and-restore cycle would
reproduce the same condition, this created an unrecoverable loop (KAFKA-20753).
The fix suppresses status marker writes in `AbstractColumnFamilyAccessor`
when the store is transactional (`isTransactional=true`), while preserving the
read-and-check on `open()`. This means a stale OPEN marker left by a prior
non-transactional crash — e.g. after toggling
`enable.transactional.statestores` — is still detected and triggers the correct
EOS wipe. Writing the marker is unnecessary for transactional stores because
KIP-892's atomic `WriteBatch` commit ensures uncommitted data is never
persisted to disk; the original safety guarantee the marker provided no longer
applies.
The `isTransactional` flag is set from configs in `RocksDBStore.openDB()`
before `openRocksDB()` is called, so all `ColumnFamilyAccessor` subclasses
(`SingleColumnFamilyAccessor`, `DualColumnFamilyAccessor`) pick it up at
construction time. Three new tests are added to
`AbstractColumnFamilyAccessorTest` covering: no OPEN marker written on open, no
CLOSED marker written on close, and detection of a stale OPEN from a prior
non-transactional crash.
## Test plan
- `./gradlew :streams:test --tests '*AbstractColumnFamilyAccessorTest*'
--tests '*DualColumnFamilyAccessorTest*'`
- `./gradlew :streams:test --tests '*RocksDBStoreCloseLeakTest*'`
- `StandbyToActiveRecycleTransactionalStoreIntegrationTest` (KAFKA-20753
reproducer, ~2 min)
--
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]
