You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "bbejeck (via GitHub)" <gi...@apache.org> on 2026/06/09 22:57:41 UTC
[PR] MINOR: Add additional documentation for KIP-1035 [kafka]
bbejeck opened a new pull request, #22526:
URL: https://github.com/apache/kafka/pull/22526
Docs-only change documenting the KIP-1035 RocksDB flush behavior in 4.3: the changelog offset is now durable only on an organic memtable flush (write_buffer_size) or clean close (no more per-commit
force-flush), so a low-traffic store + unclean exit can leave a stale offset that surfaces as `OffsetOutOfRangeException/TaskCorruptedException` on restart. Extends the upgrade guide's KIP-1035 entry
(incl. a note that 4.3.0's contributing defects are fixed in 4.3.1, KAFKA-20663) and adds a "Changelog offset durability and flush frequency" subsection to the RocksDB memory-management guide with
write_buffer_size tuning and clean-shutdown guidance.
No code or public API changes.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Add additional documentation for KIP-1035 [kafka]
Posted by "bbejeck (via GitHub)" <gi...@apache.org>.
bbejeck commented on PR #22526:
URL: https://github.com/apache/kafka/pull/22526#issuecomment-4672152664
@mjsax updated per comments
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Add additional documentation for KIP-1035 [kafka]
Posted by "bbejeck (via GitHub)" <gi...@apache.org>.
bbejeck commented on PR #22526:
URL: https://github.com/apache/kafka/pull/22526#issuecomment-4673562646
cherry-picked to 4.3
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Add additional documentation for KIP-1035 [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #22526:
URL: https://github.com/apache/kafka/pull/22526#discussion_r3384589547
##########
docs/streams/developer-guide/memory-mgmt.md:
##########
@@ -130,6 +130,28 @@ To avoid reading stale data, you can `flush()` the store before creating the ite
Each instance of RocksDB allocates off-heap memory for a block cache, index and filter blocks, and memtable (write buffer). Critical configs (for RocksDB version 4.1.0) include `block_cache_size`, `write_buffer_size` and `max_write_buffer_number`. These can be specified through the `rocksdb.config.setter` configuration.
+### Changelog offset durability and flush frequency {#rocksdb-offset-durability}
+
+Since 4.3 ([KIP-1035](https://cwiki.apache.org/confluence/display/KAFKA/KIP-1035%3A+StateStore+managed+changelog+offsets)), Kafka Streams stores each persistent store's changelog offset inside RocksDB rather than in a per-task `.checkpoint` file. Kafka Streams runs RocksDB with the write-ahead log (WAL) disabled — the changelog topic is the durable log of record — so data and the changelog offset become durable on disk only when a memtable is **flushed** to an SST file. A flush happens when the memtable fills `write_buffer_size` (16 MB by default) or when the store is **closed cleanly** (`KafkaStreams#close`). Unlike earlier releases, Kafka Streams no longer force-flushes RocksDB on every commit.
+
+The practical implication is that the on-disk changelog offset is only as fresh as the last organic flush or clean close. For a high-throughput store the memtable fills frequently, so this is not a concern. For a **low-traffic store** whose memtable may take a long time to fill, the persisted offset can lag the store's actual position for an extended period. If the application then exits uncleanly (for example `SIGKILL`, an OOM-kill, or a `KafkaStreams#close` that does not complete within the process/pod shutdown grace period), only the last-flushed offset survives. Should the changelog topic's log-start offset have advanced past that stale offset (via retention or compaction) in the meantime, the restore consumer seeks out of range on the next restart, and the task is re-initialized from the changelog (logged as `OffsetOutOfRangeException` / `TaskCorruptedException`). This recovers automatically with no data loss, but causes a full re-restore of the task.
+
+To reduce the likelihood of this for an affected low-traffic store, you can make it flush more often by lowering its `write_buffer_size` (and adjusting `max_write_buffer_number`) through a custom `RocksDBConfigSetter`:
+
+ public static class CustomRocksDBConfig implements RocksDBConfigSetter {
+ @Override
+ public void setConfig(final String storeName, final Options options, final Map<String, Object> configs) {
+ // smaller write buffer => more frequent flushes => fresher on-disk changelog offset,
+ // at the cost of more (smaller) SST files and more compaction work
+ options.setWriteBufferSize(4 * 1024 * 1024L);
+ }
+
+ @Override
+ public void close(final String storeName, final Options options) {}
+ }
+
+This is a trade-off: a smaller write buffer bounds how stale the persisted offset can get, but increases the number of SST files and the compaction load. So tune only the specific low-traffic stores that need it, and size the buffer to the store's write rate. Note that flushing is **volume-based, not time-based**. A store that receives only a trickle of writes may still not flush until it is closed, so the most reliable protection is a **clean shutdown**. Register a shutdown hook that calls `KafkaStreams#close`, and ensure it is allowed to complete — on Kubernetes, set the close timeout comfortably below the pod termination grace period (default 30s) so the process is not `SIGKILL`ed mid-close.
Review Comment:
Should we add a note, that the buffer fills up with the distinct keys, but writes to the same key don't necessarily increase the used bytes, as it would overwrite the existing entry in-place?
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Add additional documentation for KIP-1035 [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #22526:
URL: https://github.com/apache/kafka/pull/22526#discussion_r3390695248
##########
docs/streams/developer-guide/memory-mgmt.md:
##########
@@ -130,6 +130,28 @@ To avoid reading stale data, you can `flush()` the store before creating the ite
Each instance of RocksDB allocates off-heap memory for a block cache, index and filter blocks, and memtable (write buffer). Critical configs (for RocksDB version 4.1.0) include `block_cache_size`, `write_buffer_size` and `max_write_buffer_number`. These can be specified through the `rocksdb.config.setter` configuration.
+### Changelog offset durability and flush frequency {#rocksdb-offset-durability}
+
+Since 4.3 ([KIP-1035](https://cwiki.apache.org/confluence/display/KAFKA/KIP-1035%3A+StateStore+managed+changelog+offsets)), Kafka Streams stores each persistent store's changelog offset inside RocksDB rather than in a per-task `.checkpoint` file. Kafka Streams runs RocksDB with the write-ahead log (WAL) disabled — the changelog topic is the durable log of record — so data and the changelog offset become durable on disk only when a memtable is **flushed** to an SST file. A flush happens when the memtable fills `write_buffer_size` (16 MB by default) or when the store is **closed cleanly** (`KafkaStreams#close`). Unlike earlier releases, Kafka Streams no longer force-flushes RocksDB on every commit.
+
+The practical implication is that the on-disk changelog offset is only as fresh as the last organic flush or clean close. For a high-throughput store the memtable fills frequently, so this is not a concern. For a **low-traffic store** whose memtable may take a long time to fill, the persisted offset can lag the store's actual position for an extended period. If the application then exits uncleanly (for example `SIGKILL`, an OOM-kill, or a `KafkaStreams#close` that does not complete within the process/pod shutdown grace period), only the last-flushed offset survives. Should the changelog topic's log-start offset have advanced past that stale offset (via retention or compaction) in the meantime, the restore consumer seeks out of range on the next restart, and the task is re-initialized from the changelog (logged as `OffsetOutOfRangeException` / `TaskCorruptedException`). This recovers automatically with no data loss, but causes a full re-restore of the task.
+
+To reduce the likelihood of this for an affected low-traffic store, you can make it flush more often by lowering its `write_buffer_size` (and adjusting `max_write_buffer_number`) through a custom `RocksDBConfigSetter`:
+
+ public static class CustomRocksDBConfig implements RocksDBConfigSetter {
+ @Override
+ public void setConfig(final String storeName, final Options options, final Map<String, Object> configs) {
+ // smaller write buffer => more frequent flushes => fresher on-disk changelog offset,
+ // at the cost of more (smaller) SST files and more compaction work
+ options.setWriteBufferSize(4 * 1024 * 1024L);
+ }
+
+ @Override
+ public void close(final String storeName, final Options options) {}
+ }
+
+This is a trade-off: a smaller write buffer bounds how stale the persisted offset can get, but increases the number of SST files and the compaction load. So tune only the specific low-traffic stores that need it, and size the buffer to the store's write rate. Note that flushing is **volume-based, not time-based**. A store that receives only a trickle of writes may still not flush until it is closed, so the most reliable protection is a **clean shutdown**. Register a shutdown hook that calls `KafkaStreams#close`, and ensure it is allowed to complete — on Kubernetes, set the close timeout comfortably below the pod termination grace period (default 30s) so the process is not `SIGKILL`ed mid-close. Note that the record cache (`statestore.cache.max.bytes`) in front of the store coalesces repeated updates to the same key in place before they are written to RocksDB. So for an update-heavy, small-keyspace workload even fewer bytes reach the memtable per commit, making it fill more slow
ly.
Review Comment:
> Note that the record cache (`statestore.cache.max.bytes`) in front of the store coalesces repeated updates to the same key in place before they are written to RocksDB
Even w/o the KS cache, the same happens in the RocksDB memtable IIRC -- In contrast to SST files, memtable is not append-only I believe?
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Add additional documentation for KIP-1035 [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #22526:
URL: https://github.com/apache/kafka/pull/22526#discussion_r3384563641
##########
docs/streams/upgrade-guide.md:
##########
@@ -77,6 +77,8 @@ Kafka Streams now allows to purge local state directories and checkpoint files d
Kafka Streams now persists state store changelog offsets inside each state store rather than in a single per-task `.checkpoint` file ([KIP-1035](https://cwiki.apache.org/confluence/display/KAFKA/KIP-1035%3A+StateStore+managed+changelog+offsets)). This is an internal infrastructure change and is transparent to most users — existing per-task `.checkpoint` files are migrated automatically on first startup, and no application or operator action is required. EOS crash behavior is unchanged in 4.3: state stores are still wiped and fully restored from the changelog. KIP-1035 is a prerequisite for [KIP-892: Transactional Semantics for StateStores](https://cwiki.apache.org/confluence/display/KAFKA/KIP-892%3A+Transactional+Semantics+for+StateStores), which will use these per-store offsets to make EOS state writes transactional and skip the full restore. Authors of custom `StateStore` implementations may opt-in to managing their own offsets via `managesOffsets()`, `commit(Map<TopicPartition,
Long>)`, and `committedOffset(TopicPartition)`; see KIP-1035 for the API. For downgrade implications, see [Notable compatibility changes in past releases](#notable-compatibility-changes-in-past-releases).
+As part of KIP-1035, the per-store changelog offset is written into RocksDB on each commit and is made durable on disk only when RocksDB flushes its memtable to an SST file — either organically once the memtable fills `write_buffer_size` (16 MB by default), or on a clean store close. Earlier releases force-flushed RocksDB on every commit. A consequence is that for a **low-traffic store** whose memtable rarely fills, the on-disk offset can lag the store's actual position until the next clean shutdown. For the durability model and guidance on tuning flush frequency for low-traffic stores, see [Memory Management: RocksDB](/{{version}}/documentation/streams/developer-guide/memory-mgmt.html#rocksdb-offset-durability).
Review Comment:
Maybe insert one sentence why this could become a problem?
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Add additional documentation for KIP-1035 [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #22526:
URL: https://github.com/apache/kafka/pull/22526#discussion_r3390764327
##########
docs/streams/developer-guide/memory-mgmt.md:
##########
@@ -130,6 +130,28 @@ To avoid reading stale data, you can `flush()` the store before creating the ite
Each instance of RocksDB allocates off-heap memory for a block cache, index and filter blocks, and memtable (write buffer). Critical configs (for RocksDB version 4.1.0) include `block_cache_size`, `write_buffer_size` and `max_write_buffer_number`. These can be specified through the `rocksdb.config.setter` configuration.
+### Changelog offset durability and flush frequency {#rocksdb-offset-durability}
+
+Since 4.3 ([KIP-1035](https://cwiki.apache.org/confluence/display/KAFKA/KIP-1035%3A+StateStore+managed+changelog+offsets)), Kafka Streams stores each persistent store's changelog offset inside RocksDB rather than in a per-task `.checkpoint` file. Kafka Streams runs RocksDB with the write-ahead log (WAL) disabled — the changelog topic is the durable log of record — so data and the changelog offset become durable on disk only when a memtable is **flushed** to an SST file. A flush happens when the memtable fills `write_buffer_size` (16 MB by default) or when the store is **closed cleanly** (`KafkaStreams#close`). Unlike earlier releases, Kafka Streams no longer force-flushes RocksDB on every commit.
+
+The practical implication is that the on-disk changelog offset is only as fresh as the last organic flush or clean close. For a high-throughput store the memtable fills frequently, so this is not a concern. For a **low-traffic store** whose memtable may take a long time to fill, the persisted offset can lag the store's actual position for an extended period. If the application then exits uncleanly (for example `SIGKILL`, an OOM-kill, or a `KafkaStreams#close` that does not complete within the process/pod shutdown grace period), only the last-flushed offset survives. Should the changelog topic's log-start offset have advanced past that stale offset (via retention or compaction) in the meantime, the restore consumer seeks out of range on the next restart, and the task is re-initialized from the changelog (logged as `OffsetOutOfRangeException` / `TaskCorruptedException`). This recovers automatically with no data loss, but causes a full re-restore of the task.
+
+To reduce the likelihood of this for an affected low-traffic store, you can make it flush more often by lowering its `write_buffer_size` (and adjusting `max_write_buffer_number`) through a custom `RocksDBConfigSetter`:
+
+ public static class CustomRocksDBConfig implements RocksDBConfigSetter {
+ @Override
+ public void setConfig(final String storeName, final Options options, final Map<String, Object> configs) {
+ // smaller write buffer => more frequent flushes => fresher on-disk changelog offset,
+ // at the cost of more (smaller) SST files and more compaction work
+ options.setWriteBufferSize(4 * 1024 * 1024L);
+ }
+
+ @Override
+ public void close(final String storeName, final Options options) {}
+ }
+
+This is a trade-off: a smaller write buffer bounds how stale the persisted offset can get, but increases the number of SST files and the compaction load. So tune only the specific low-traffic stores that need it, and size the buffer to the store's write rate. Note that flushing is **volume-based, not time-based**. A store that receives only a trickle of writes may still not flush until it is closed, so the most reliable protection is a **clean shutdown**. Register a shutdown hook that calls `KafkaStreams#close`, and ensure it is allowed to complete — on Kubernetes, set the close timeout comfortably below the pod termination grace period (default 30s) so the process is not `SIGKILL`ed mid-close. Note that the record cache (`statestore.cache.max.bytes`) in front of the store coalesces repeated updates to the same key in place before they are written to RocksDB. So for an update-heavy, small-keyspace workload even fewer bytes reach the memtable per commit, making it fill more slow
ly.
Review Comment:
Seems I was wrong...
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Add additional documentation for KIP-1035 [kafka]
Posted by "bbejeck (via GitHub)" <gi...@apache.org>.
bbejeck commented on PR #22526:
URL: https://github.com/apache/kafka/pull/22526#issuecomment-4673485226
merged #22526 into trunk
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Add additional documentation for KIP-1035 [kafka]
Posted by "bbejeck (via GitHub)" <gi...@apache.org>.
bbejeck merged PR #22526:
URL: https://github.com/apache/kafka/pull/22526
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org