arnabnandy7 opened a new pull request, #23352: URL: https://github.com/apache/kafka/pull/23352
## Summary KAFKA-20856: Adds per-buffer control over headers support for the Kafka Streams `suppress()` operator. Suppression buffers currently determine whether headers should be stored exclusively from the global `dsl.store.format` configuration. Unlike state stores configured through `Materialized`, applications cannot override that behavior for an individual suppression operator. This change adds the following methods to `Suppressed.BufferConfig`: - `withHeadersEnabled()` - `withHeadersDisabled()` The per-buffer setting takes precedence over `dsl.store.format`. When neither method is called, the suppression buffer continues to follow the global configuration, preserving existing behavior. The headers setting is retained when chaining other buffer configuration methods, including size limits, logging options, and buffer-full strategies. ## Implementation The optional headers override is carried through the immutable suppression buffer configuration classes and passed to `InMemoryTimeOrderedKeyValueChangeBuffer`. During store initialization, the buffer resolves headers support in this order: 1. Use the per-buffer override when explicitly configured. 2. Otherwise, fall back to the global `dsl.store.format` setting. This preserves the existing global configuration behavior while providing the per-operator control available to other stateful DSL operations. ## Documentation Updated the following documentation: - Streams configuration guide - Streams DSL API guide - Streams upgrade guide The documentation describes the new methods, their precedence over `dsl.store.format`, and the unchanged default behavior. ## Testing Added a focused unit test verifying that enabled and disabled headers settings are preserved across immutable buffer-configuration chains, including: - Buffer-size changes - Buffer-full strategy changes - Logging configuration changes The following are targeted verification: ```shell ./gradlew streams:test --tests org.apache.kafka.streams.kstream.SuppressedTest ./gradlew streams:test --tests org.apache.kafka.streams.kstream.internals.SuppressHeadersScenarioTest ./gradlew streams:checkstyleMain streams:checkstyleTest streams:spotlessCheck ``` Reviewers: Matthias J. Sax (@mjsax ) -- 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]
