From 53faeb85ef0a83f4afd8e92e5f48d4207d0817ff Mon Sep 17 00:00:00 2001 From: "Matthias J. Sax" Date: Thu, 26 Jun 2025 10:06:46 -0700 Subject: [PATCH] MINOR: Improve ProcessorContext JavaDocs (#20042) Clarify that state stores are sharded, and shards cannot be shared across Processors. Reviewers: Lucas Brutschy --- .../apache/kafka/streams/processor/ProcessorContext.java | 7 +++++++ .../kafka/streams/processor/api/ProcessingContext.java | 6 ++++++ 2 files changed, 13 insertions(+) diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/ProcessorContext.java b/streams/src/main/java/org/apache/kafka/streams/processor/ProcessorContext.java index d65244bcc86..45ab411d62e 100644 --- a/streams/src/main/java/org/apache/kafka/streams/processor/ProcessorContext.java +++ b/streams/src/main/java/org/apache/kafka/streams/processor/ProcessorContext.java @@ -105,6 +105,13 @@ public interface ProcessorContext { /** * Get the state store given the store name. * + *

The returned state store represent one shard of the overall state, which belongs to the current task. + * The returned shard of the state store may only be used by the current + * {@link org.apache.kafka.streams.kstream.Transformer}, {@link org.apache.kafka.streams.kstream.ValueTransformer}, + * or {@link org.apache.kafka.streams.kstream.ValueTransformerWithKey} instance. + * Sharing a shard across different transformers (ie, from different "sibling" tasks; same sub-topology but different + * partition) may lead to data corruption and/or data loss. + * * @param name The store name * @param The type or interface of the store to return * @return The state store instance diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/api/ProcessingContext.java b/streams/src/main/java/org/apache/kafka/streams/processor/api/ProcessingContext.java index 78aa2c8489b..7f6874c8c19 100644 --- a/streams/src/main/java/org/apache/kafka/streams/processor/api/ProcessingContext.java +++ b/streams/src/main/java/org/apache/kafka/streams/processor/api/ProcessingContext.java @@ -107,6 +107,12 @@ public interface ProcessingContext { /** * Get the state store given the store name. * + *

The returned state store represent one shard of the overall state, which belongs to the current task. + * The returned shard of the state store may only be used by the current {@link Processor} or + * {@link FixedKeyProcessor} instance. + * Sharing a shard across different processors (ie, from different "sibling" tasks; same sub-topology but different + * partition) may lead to data corruption and/or data loss. + * * @param name The store name * @param The type or interface of the store to return * @return The state store instance