Apache Kafka Trunk: Transaction Load Safety, Share Group DLQ Speedups and a Streams OOM Guard

Apache Kafka trunk saw 48 commits in the last week, touching 243 files. Most of that is test migration, but three changes matter to operators: a transaction coordinator correctness fix, a performance pass on the share group dead letter queue, and a memory guard in Kafka Streams.

The project lives at apache/kafka. Everything below is on trunk, not in a tagged release.

The most serious fix is KAFKA-21349. When a broker became leader of a __transaction_state partition, TransactionStateManager loaded the log up to the log end offset. A new leader can hold records above the high watermark. Those records can still be truncated if leadership moves again.

The commit message gives a concrete failure. The coordinator could send COMMIT markers for a PREPARE_COMMIT that the next leader never had. That leader later times out and aborts the transaction. The result is a transaction committed on some partitions and aborted on others.

Loading with FetchIsolation.HIGH_WATERMARK is not enough, because a fresh leader’s high watermark can lag the true one. The fix is different. If the high watermark is below the log end offset, the load registers a PartitionListener and runs once the high watermark reaches the offset captured at load start. It checks again after registering, to close the race with a concurrent update.

The tradeoff is latency. Coordinator loads can now wait for replication to catch up, and during that time clients see COORDINATOR_LOAD_IN_PROGRESS. That is the safer failure. The change is about 99 lines in TransactionStateManager.scala plus a larger test file.

A related fix, KAFKA-20734, restores noNode from KRaftMetadataCache for known partitions whose leader is offline. The refactor from Scala had returned empty. The tests added there cover the transaction marker channel manager.

KAFKA-21356 is a performance fix for the share group dead letter queue. The persister thread also completes write responses for every share partition, so DLQ work on it slows unrelated groups. The commit lists three causes:

  • isDlqEnabledOnTopic and dlqTopicMaxMessageBytes built a LogConfig on every DLQ request. That parses every topic config. They now read one key directly.
  • The source record read ran synchronously on the persister thread. It now runs on a single threaded fetch executor owned by BrokerServer.
  • ShareGroupDLQRecordFetcher.collectRecords visited every batch of a 1 MiB log read, loading file backed batches from disk. It now rejects batches whose base offset is past the DLQ end offset.

The first item applies whether record copying is on or off. The other two affect only the copy on case. The executor size comes from a new supplier, shareGroupDLQManagerFetcherNumThreads, wired in BrokerServer.scala at construction. No benchmark numbers were cited in the commit.

The config helper change has one subtle point. ConfigDef.parseType checks type only, so the code applies the topic config validator for max.message.bytes by hand. Invalid values fall back to the broker setting.

Shutdown order also moved. The share partition manager now closes first, then the persister and DLQ manager, and only then the share coordinator and replica manager. The comments in the diff explain why: late requests must not hit a stopped component.

KAFKA-21230 targets an OOM during long rebalances. StreamThread#maybeCommit forces a commit when uncommitted bytes exceed statestore.uncommitted.max.bytes. But TaskManager skips commits during a rebalance, and processing continued. RocksDB holds those bytes in a WriteBatch in native memory, outside the block cache and write buffers.

Now each loop iteration begins with maybePauseProcessingForUncommittedBytes(). While paused, the thread does not process records or run punctuators, but it keeps polling so the rebalance can finish. It resumes after a successful commit or a flush on partition revocation. Pause and resume are logged at INFO.

The cost is lower throughput during long rebalances. The commit message says so plainly. The alternative was a dead process.

MINOR: Reject negative maxRecords makes a ShareFetch v1+ request with a negative MaxRecords return INVALID_REQUEST before a session context is created. Zero stays valid, since clients use it for acknowledgement only requests. The check is in KafkaApis.scala.

The rest is housekeeping. The KAFKA-20684 series points client internals at RebalanceListener across several parts. TransactionsBounceTest moved from Scala to Java. Several integration tests were sped up. None of that changes runtime behavior.

  • Transaction coordinator failovers: expect a longer load window when the high watermark lags. Watch COORDINATOR_LOAD_IN_PROGRESS rates after upgrading.
  • Share group DLQ: if you test the feature, check persister thread latency with copying on and off. The new fetcher thread count setting is worth reading in ShareGroupConfigTest.java.
  • Streams apps with large RocksDB state: look for the new INFO pause messages. Frequent pauses mean rebalances are long or the limit is too low.