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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
