Eswarar Siva created KAFKA-21374:
------------------------------------
Summary: 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.3.2, 4.4.0
Reporter: Eswarar Siva
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)
