You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "srv9989 (via GitHub)" <gi...@apache.org> on 2026/10/29 06:41:24 UTC
[PR] KAFKA-21374: Prevent cross-thread header leakage in global state stores [kafka]
srv9989 opened a new pull request, #24008:
URL: https://github.com/apache/kafka/pull/24008
Summary
This PR addresses KAFKA-21374, where global state store reads can receive incorrect record headers from the global processing thread.
While investigating the issue, I found that the global processor context retains headers from the record being processed. Since the same context is used by the global state store, other threads can receive unrelated headers during serialization and deserialization.
This can cause problems for applications that depend on record headers for deserialization.
Changes Made
* Updated GlobalProcessorContextImpl to allow the owning global thread to access its record headers, while returning empty contextual headers for reads from other threads.
* Updated GlobalStateUpdateTask to clear the record context after processing, including when an exception occurs.
* Added regression tests to cover header leakage during normal processing and concurrent global store reads.
* Added tests to verify thread ownership and record context cleanup.
Testing
I first reproduced the issue with a failing regression test and confirmed that the deserializer was receiving headers from an unrelated global record.
After applying the fix, I ran the relevant Kafka Streams tests.
* 323 selected tests passed.
* Sequential and concurrent regression tests passed.
* Both at-least-once and exactly-once test scenarios passed.
* Checkstyle passed.
* Spotless passed.
* SpotBugs passed.
* git diff --check passed.
The concurrent scenario was tested using TopologyTestDriver with controlled thread synchronization. I have not run a separate real-broker concurrency integration test.
JIRA: https://issues.apache.org/jira/browse/KAFKA-21374
--
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] KAFKA-21374: Prevent cross-thread header leakage in global state stores [kafka]
Posted by "aliehsaeedii (via GitHub)" <gi...@apache.org>.
aliehsaeedii commented on code in PR #24008:
URL: https://github.com/apache/kafka/pull/24008#discussion_r4410797173
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStateUpdateTask.java:
##########
@@ -120,14 +120,18 @@ public void update(final ConsumerRecord<byte[], byte[]> record) {
deserialized.topic(),
deserialized.headers());
processorContext.setRecordContext(recordContext);
- processorContext.setCurrentNode(sourceNodeAndDeserializer.sourceNode());
- final Record<Object, Object> toProcess = new Record<>(
- deserialized.key(),
- deserialized.value(),
- processorContext.recordContext().timestamp(),
- processorContext.recordContext().headers()
- );
- ((SourceNode<Object, Object>) sourceNodeAndDeserializer.sourceNode()).process(toProcess);
+ try {
+ processorContext.setCurrentNode(sourceNodeAndDeserializer.sourceNode());
+ final Record<Object, Object> toProcess = new Record<>(
+ deserialized.key(),
+ deserialized.value(),
+ processorContext.recordContext().timestamp(),
+ processorContext.recordContext().headers()
+ );
+ ((SourceNode<Object, Object>) sourceNodeAndDeserializer.sourceNode()).process(toProcess);
+ } finally {
+ processorContext.setRecordContext(null);
Review Comment:
Clearing the context here breaks direct writes to a global store from TTD after a record was piped. For example, `put` on a cached `globalTable` store NPEs in `CachingKeyValueStore.putInternal` (`recordContext().headers()`), and tombstones on headers stores fail the same way. TTD already sets a dummy context for task stores in `getStateStore()` when none exists. Can you do the same for the global context and add a test?
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStateUpdateTask.java:
##########
@@ -120,14 +120,18 @@ public void update(final ConsumerRecord<byte[], byte[]> record) {
deserialized.topic(),
deserialized.headers());
processorContext.setRecordContext(recordContext);
- processorContext.setCurrentNode(sourceNodeAndDeserializer.sourceNode());
- final Record<Object, Object> toProcess = new Record<>(
- deserialized.key(),
- deserialized.value(),
- processorContext.recordContext().timestamp(),
- processorContext.recordContext().headers()
- );
- ((SourceNode<Object, Object>) sourceNodeAndDeserializer.sourceNode()).process(toProcess);
+ try {
+ processorContext.setCurrentNode(sourceNodeAndDeserializer.sourceNode());
+ final Record<Object, Object> toProcess = new Record<>(
+ deserialized.key(),
+ deserialized.value(),
+ processorContext.recordContext().timestamp(),
+ processorContext.recordContext().headers()
+ );
+ ((SourceNode<Object, Object>) sourceNodeAndDeserializer.sourceNode()).process(toProcess);
+ } finally {
Review Comment:
`GlobalStateManagerImpl.reprocessState()` also sets the record context for each restored record and never clears it, so store access in processors' `init()` still gets the last restored record's headers. Clear it there too?
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalProcessorContextImpl.java:
##########
@@ -51,6 +53,13 @@ public GlobalProcessorContextImpl(final StreamsConfig config,
super(new TaskId(-1, -1), config, metrics, cache);
stateManager = stateMgr;
this.time = time;
+ owningThread = Thread.currentThread();
Review Comment:
This assumes the context is built on the thread that later calls `update()`. That holds for `GlobalStreamThread`, but in TTD it's the thread that builds the driver, so a test that pipes from another thread (e.g. `assertTimeoutPreemptively`) silently gets empty headers on the global processor's own writes too. Please document this assumption here.
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/GlobalStateTaskTest.java:
##########
@@ -161,6 +162,8 @@ public void shouldProcessRecordsForTopic() {
globalStateTask.update(record(topic1, 1, 1, "foo".getBytes(), "bar".getBytes()));
assertEquals(1, sourceOne.numReceived);
assertEquals(0, sourceTwo.numReceived);
+ assertNull(context.recordContext());
+ assertEquals(new RecordHeaders(), context.headers());
Review Comment:
This passes on trunk too, since the test record has empty headers, and the `assertNull` above already covers it. Drop it?
--
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