[
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)
