[jira] [Commented] (KAFKA-21374) Streams: reads of a global store pass the headers of the global thread's last record to the value deserializer

["Matthias J. Sax (Jira)" <ji...@apache.org>]

    [ https://issues.apache.org/jira/browse/KAFKA-21374?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18133801#comment-18133801 ] 

Matthias J. Sax commented on KAFKA-21374:
-----------------------------------------

{quote}Resetting the global processor context's record context after each update would also stop the headers from leaking past the update.
{quote}
At first glance, this seems to be the right fix? \cc [~alisa23] 

> Streams: reads of a global store pass the headers of the global thread's last record to the value deserializer
> --------------------------------------------------------------------------------------------------------------
>
>                 Key: KAFKA-21374
>                 URL: https://issues.apache.org/jira/browse/KAFKA-21374
>             Project: Kafka
>          Issue Type: Bug
>          Components: streams
>    Affects Versions: 4.4.0, 4.3.2
>            Reporter: Eswarar Siva
>            Priority: Major
>
> Since KAFKA-20173 the metered stores pass internalContext.headers() to the serdes. For a global store that context is the global processor context, so every read of a global store gives the value deserializer the headers of the last record the global thread processed:
> -> a processor in a stream task that reads a global store gets the global thread's last record headers, not the headers of the record it is processing. A task store read in the same processor gets the right ones.
> -> an interactive query on a global store gets them too, while the comment in MeteredKeyValueStore (511-513, 534-536) says the IQ path gets new RecordHeaders() since there is no record context.
> 4.2.1 passed no headers on these reads. Line numbers below are at 4.3.2, the same code is in 4.4.0 and on trunk (not run on trunk).
> Why:
> -> GlobalStateManagerImpl.initialize inits every global store with globalProcessorContext (189).
> -> ProcessorContextImpl.getStateStore gives a stream task the same store instance, read only (167-170).
> -> MeteredKeyValueStore keeps that context from init (135) and passes internalContext.headers() to the serdes on get and put (449-461).
> -> GlobalStateUpdateTask.update sets the record context for each record (115-122) and nothing resets it.
> So what a stream thread's read gets depends on how far the global thread has got. That record context is written by the global thread and read by the stream threads without any synchronization.
> It matters for value serdes that look at headers. Confluent's Avro deserializer for example, by default (DualSchemaIdDeserializer in 8.2.2 and 8.3.0), takes the schema id from a __value_schema_id header when there is one and then reads the whole value as the payload. A global store value stored with the id in its prefix is then decoded with another record's schema id whenever the global thread's last record carried that header.
> Repro: TopologyTestDriver, in-memory stores, a value deserializer that records the "origin" header it is given. A global record with origin=global-record is piped first, then a stream record with origin=stream-record whose processor reads a task store and the global store, then an interactive query on the global store.
> Results:
> ||Read||4.2.1||4.3.2||4.4.0||
> |task store read in the stream task processor|called without headers|origin=stream-record|origin=stream-record|
> |global store read in the same processor|called without headers|origin=global-record|origin=global-record|
> |interactive query on the global store|called without headers|origin=global-record|origin=global-record|
> TopologyTestDriver runs the global and the stream processing on one thread. I also saw it on 4.3.2 with KafkaStreams against a real broker in one of our integration tests, where the global thread is its own thread.
> {code:java}
> import org.apache.kafka.common.header.Headers;
> import org.apache.kafka.common.header.internals.RecordHeaders;
> import org.apache.kafka.common.serialization.Deserializer;
> import org.apache.kafka.common.serialization.Serde;
> import org.apache.kafka.common.serialization.Serdes;
> import org.apache.kafka.common.serialization.StringDeserializer;
> import org.apache.kafka.common.serialization.StringSerializer;
> import org.apache.kafka.streams.StreamsConfig;
> import org.apache.kafka.streams.Topology;
> import org.apache.kafka.streams.TopologyTestDriver;
> import org.apache.kafka.streams.processor.api.Processor;
> import org.apache.kafka.streams.processor.api.ProcessorContext;
> import org.apache.kafka.streams.processor.api.Record;
> import org.apache.kafka.streams.state.KeyValueStore;
> import org.apache.kafka.streams.state.Stores;
> import org.apache.kafka.streams.test.TestRecord;
> import java.nio.charset.StandardCharsets;
> import java.util.ArrayList;
> import java.util.List;
> import java.util.Properties;
> public class GlobalStoreHeadersRepro {
>     static final List<String> SEEN = new ArrayList<>();
>     static final class RecordingDeserializer implements Deserializer<String> {
>         @Override
>         public String deserialize(String topic, byte[] data) {
>             SEEN.add("called without headers");
>             return data == null ? null : new String(data, StandardCharsets.UTF_8);
>         }
>         @Override
>         public String deserialize(String topic, Headers headers, byte[] data) {
>             var origin = headers == null ? null : headers.lastHeader("origin");
>             SEEN.add(origin == null ? "empty headers" : "origin=" + new String(origin.value(), StandardCharsets.UTF_8));
>             return data == null ? null : new String(data, StandardCharsets.UTF_8);
>         }
>     }
>     static Serde<String> recordingSerde() {
>         return Serdes.serdeFrom(new StringSerializer(), new RecordingDeserializer());
>     }
>     static RecordHeaders origin(String value) {
>         var headers = new RecordHeaders();
>         headers.add("origin", value.getBytes(StandardCharsets.UTF_8));
>         return headers;
>     }
>     public static void main(String[] args) {
>         var topology = new Topology();
>         topology.addGlobalStore(
>                 Stores.keyValueStoreBuilder(Stores.inMemoryKeyValueStore("global-store"), Serdes.String(),
>                         recordingSerde()).withLoggingDisabled(),
>                 "global-source", new StringDeserializer(), new StringDeserializer(), "global-topic",
>                 "global-processor", () -> new Processor<String, String, Void, Void>() {
>                     private KeyValueStore<String, String> store;
>                     @Override
>                     public void init(ProcessorContext<Void, Void> context) {
>                         store = context.getStateStore("global-store");
>                     }
>                     @Override
>                     public void process(Record<String, String> record) {
>                         store.put(record.key(), record.value());
>                     }
>                 });
>         topology.addSource("stream-source", new StringDeserializer(), new StringDeserializer(), "stream-topic");
>         topology.addProcessor("reader", () -> new Processor<String, String, Void, Void>() {
>             private KeyValueStore<String, String> globalStore;
>             private KeyValueStore<String, String> taskStore;
>             @Override
>             public void init(ProcessorContext<Void, Void> context) {
>                 globalStore = context.getStateStore("global-store");
>                 taskStore = context.getStateStore("task-store");
>             }
>             @Override
>             public void process(Record<String, String> record) {
>                 taskStore.put("k", "task-value");
>                 SEEN.clear();
>                 taskStore.get("k");
>                 System.out.println("stream thread, task store read: " + SEEN);
>                 SEEN.clear();
>                 globalStore.get("k");
>                 System.out.println("stream thread, global store read: " + SEEN);
>             }
>         }, "stream-source");
>         topology.addStateStore(
>                 Stores.keyValueStoreBuilder(Stores.inMemoryKeyValueStore("task-store"), Serdes.String(),
>                         recordingSerde()).withLoggingDisabled(),
>                 "reader");
>         var props = new Properties();
>         props.put(StreamsConfig.APPLICATION_ID_CONFIG, "global-store-headers-repro");
>         props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "dummy:1234");
>         try (var driver = new TopologyTestDriver(topology, props)) {
>             var globalInput = driver.createInputTopic("global-topic", new StringSerializer(), new StringSerializer());
>             var streamInput = driver.createInputTopic("stream-topic", new StringSerializer(), new StringSerializer());
>             globalInput.pipeInput(new TestRecord<>("k", "global-value", origin("global-record")));
>             System.out.println("piped the stream record with origin=stream-record");
>             streamInput.pipeInput(new TestRecord<>("x", "y", origin("stream-record")));
>             SEEN.clear();
>             KeyValueStore<String, String> queried = driver.getKeyValueStore("global-store");
>             queried.get("k");
>             System.out.println("interactive query, global store read: " + SEEN);
>         }
>     }
> }
> {code}
> What I think could be done: give reads from outside the global thread (stream task processors and interactive queries) either empty headers or the headers of the reading task's current record. Not sure which one fits best. Resetting the global processor context's record context after each update would also stop the headers from leaking past the update.



--
This message was sent by Atlassian Jira
(v10.3.26#10030026)