You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by GitBox <gi...@apache.org> on 2022/10/13 09:14:56 UTC
[GitHub] [kafka] calmera opened a new pull request, #12742: KAFKA-10892: Shared Readonly State Stores ( revisited )
calmera opened a new pull request, #12742:
URL: https://github.com/apache/kafka/pull/12742
This PR implements KIP-813: support for shareable state stores.
Tests were created to validate the structure of the resulting topology.
Certain parts (MockProcessor eg.) have been migrated to the newer PAPI as well. Tests have been ran against the whole project to be sure this migration doesn't cause any side-effects.
The contribution is my original work and I license the work to the project under the project's open source license
This PR is a split of the previous PR #12188 which contained changes to the Topology as well as the PAPI test classes.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] calmera commented on a diff in pull request #12742: KAFKA-10892: Shared Readonly State Stores ( revisited )
Posted by GitBox <gi...@apache.org>.
calmera commented on code in PR #12742:
URL: https://github.com/apache/kafka/pull/12742#discussion_r1063323844
##########
streams/src/main/java/org/apache/kafka/streams/Topology.java:
##########
@@ -737,6 +737,91 @@ public synchronized Topology addStateStore(final StoreBuilder<?> storeBuilder,
return this;
}
+ /**
+ * Adds a Read Only {@link StateStore} to the topology.
+ *
+ * A Read Only StateStore can use any compacted topic as a changelog.
+ * <p>
+ * A {@link SourceNode} will be added to consume the data arriving from the partitions of the input topic.
+ * <p>
+ * The provided {@link ProcessorSupplier} will be used to create an {@link ProcessorNode} that will receive all
+ * records forwarded from the {@link SourceNode}.
+ * This {@link ProcessorNode} should be used to keep the {@link StateStore} up-to-date.
+ *
+ * @param storeBuilder user defined key value store builder
+ * @param sourceName name of the {@link SourceNode} that will be automatically added
+ * @param timestampExtractor the stateless timestamp extractor used for this source,
+ * if not specified the default extractor defined in the configs will be used
+ * @param keyDeserializer the {@link Deserializer} to deserialize keys with
+ * @param valueDeserializer the {@link Deserializer} to deserialize values with
+ * @param topic the topic to source the data from
+ * @param processorName the name of the {@link ProcessorSupplier}
+ * @param stateUpdateSupplier the instance of {@link ProcessorSupplier}
+ * @return itself
+ * @throws TopologyException if the processor of state is already registered
+ */
Review Comment:
Updated the javadocs to reflect that we hardcode the offset reset to earliest. We might want to create another method to allow to override that, but I can't really come up with cases where you want to make it anything else than earliest.
We can add it later on if needed.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] calmera commented on a diff in pull request #12742: KAFKA-10892: Shared Readonly State Stores ( revisited )
Posted by GitBox <gi...@apache.org>.
calmera commented on code in PR #12742:
URL: https://github.com/apache/kafka/pull/12742#discussion_r1063310192
##########
docs/streams/developer-guide/processor-api.html:
##########
@@ -396,6 +397,21 @@ <h2>
</ul>
</p>
</div>
+ <div class="section" id="readonly-state-stores">
+ <span id="streams-developer-guide-state-store-readonly"></span><h3><a class="toc-backref" href="#id12">ReadOnly State Stores</a><a class="headerlink" href="#readonly-state-stores" title="Permalink to this headline"></a></h3>
+ <p>The changelog topic for a fault-tolerant state store can also be used to perform lookups against. It is
+ crucial to understand that there can be only one writer to this state store, but there can be multiple
+ readers.</p>
+
+ <p>This allows you to use a readonly state store for looking up reference data managed by another processor
+ or even another process all together. The messages within the compacted topic may be projected (subset of data extracted)
+ before going into the readonly state store. Therefor we can optimize what the data within our readonly state
+ store should look like, depending on the use-case it needs to serve.</p>
+
+ <p><b>note:</b> beware of the partitioning requirements when using readonly statestores for lookups during
+ processing. You might want to make sure the original changelog topic is co-partitioned with the processors
+ reading the readonly statestore.</p>
Review Comment:
Eventually this would be needed, but I think it is indeed best to do that as a followup tbh.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] calmera commented on a diff in pull request #12742: KAFKA-10892: Shared Readonly State Stores ( revisited )
Posted by "calmera (via GitHub)" <gi...@apache.org>.
calmera commented on code in PR #12742:
URL: https://github.com/apache/kafka/pull/12742#discussion_r1102355418
##########
clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java:
##########
@@ -54,7 +54,7 @@
* The consumer configuration keys
*/
public class ConsumerConfig extends AbstractConfig {
- private static final ConfigDef CONFIG;
+ protected static final ConfigDef CONFIG;
Review Comment:
Hmm looks like my reverse consumer work got committed by accident. will clean it and go through your comments
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-10892: Shared Readonly State Stores ( revisited ) [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on PR #12742:
URL: https://github.com/apache/kafka/pull/12742#issuecomment-1986241410
Merged to `trunk`. Thanks for the KIP and PR @calmera!
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-10892: Shared Readonly State Stores ( revisited ) [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on PR #12742:
URL: https://github.com/apache/kafka/pull/12742#issuecomment-1981929378
@calmera -- As discussed in person (a long time back...) I took the liberty to update this PR. Hope we can get this merged soon.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] calmera commented on pull request #12742: KAFKA-10892: Shared Readonly State Stores ( revisited )
Posted by GitBox <gi...@apache.org>.
calmera commented on PR #12742:
URL: https://github.com/apache/kafka/pull/12742#issuecomment-1373454834
I agree we need to have some tests for that but that will require a bit more time and effort. I'll try to allocate some time to get that done, but if there is someone else who wants to jump in to do that, that would be great!
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] calmera commented on a diff in pull request #12742: KAFKA-10892: Shared Readonly State Stores ( revisited )
Posted by GitBox <gi...@apache.org>.
calmera commented on code in PR #12742:
URL: https://github.com/apache/kafka/pull/12742#discussion_r1062611257
##########
docs/streams/developer-guide/processor-api.html:
##########
@@ -396,6 +397,21 @@ <h2>
</ul>
</p>
</div>
+ <div class="section" id="readonly-state-stores">
+ <span id="streams-developer-guide-state-store-readonly"></span><h3><a class="toc-backref" href="#id12">ReadOnly State Stores</a><a class="headerlink" href="#readonly-state-stores" title="Permalink to this headline"></a></h3>
+ <p>The changelog topic for a fault-tolerant state store can also be used to perform lookups against. It is
+ crucial to understand that there can be only one writer to this state store, but there can be multiple
+ readers.</p>
+
+ <p>This allows you to use a readonly state store for looking up reference data managed by another processor
+ or even another process all together. The messages within the compacted topic may be projected (subset of data extracted)
Review Comment:
yeah, pretty sure eventually we came to that conclusion as well. Too bad though, would have been nice for re-usability purposes.
I will change the description and that the projected stuff out.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] calmera commented on a diff in pull request #12742: KAFKA-10892: Shared Readonly State Stores ( revisited )
Posted by GitBox <gi...@apache.org>.
calmera commented on code in PR #12742:
URL: https://github.com/apache/kafka/pull/12742#discussion_r1063271295
##########
streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamBranchTest.java:
##########
@@ -59,7 +59,7 @@ public void testKStreamBranch() {
assertEquals(3, branches.length);
- final MockProcessorSupplier<Integer, String> supplier = new MockProcessorSupplier<>();
+ final MockProcessorSupplier<Integer, String, Void, Void> supplier = new MockProcessorSupplier<>();
Review Comment:
np, made it a bit easier ;)
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-10892: Shared Readonly State Stores ( revisited ) [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #12742:
URL: https://github.com/apache/kafka/pull/12742#discussion_r1516564608
##########
streams/src/test/java/org/apache/kafka/streams/processor/ReadOnlyStoreTest.java:
##########
@@ -0,0 +1,132 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.streams.processor;
+
+import org.apache.kafka.common.serialization.IntegerDeserializer;
+import org.apache.kafka.common.serialization.IntegerSerializer;
+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.KeyValue;
+import org.apache.kafka.streams.TestInputTopic;
+import org.apache.kafka.streams.TestOutputTopic;
+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.KeyValueIterator;
+import org.apache.kafka.streams.state.KeyValueStore;
+import org.apache.kafka.streams.state.Stores;
+import org.junit.Test;
+
+import java.util.LinkedList;
+import java.util.List;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.equalTo;
+
+public class ReadOnlyStoreTest {
+
+ @Test
+ public void shouldLoadDataIntoReadOnlyStoreAndAllowAccessFromProcessor() {
Review Comment:
It's not about the error path -- it's really just regular processing. Happy to rename. Suggestions? -- In the end, this test verifies that we can read data from the input topic and put into the store, plus, plus that some other Processor can access the store (we don't really have a "read-only" guard and cannot enforce it; and we can only verify if we can "connect" some other Processor to the read-only store and get a handle on it -- and we test read, as write is not allowed, even if we cannot enforce it...)
The error path, ie, store recovery from the input topic (as we don't have a changelog) is tested via new test in `RestoreIntegrationTest`.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-10892: Shared Readonly State Stores ( revisited ) [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax merged PR #12742:
URL: https://github.com/apache/kafka/pull/12742
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-10892: Shared Readonly State Stores ( revisited ) [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on PR #12742:
URL: https://github.com/apache/kafka/pull/12742#issuecomment-1760672364
@calmera -- sorry to dropping the ball on this one -- can we resurrect it? -- First step would be rebase to resolve the conflict.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] mjsax commented on a diff in pull request #12742: KAFKA-10892: Shared Readonly State Stores ( revisited )
Posted by GitBox <gi...@apache.org>.
mjsax commented on code in PR #12742:
URL: https://github.com/apache/kafka/pull/12742#discussion_r1068740650
##########
streams/src/main/java/org/apache/kafka/streams/Topology.java:
##########
@@ -737,6 +737,85 @@ public synchronized Topology addStateStore(final StoreBuilder<?> storeBuilder,
return this;
}
+ /**
+ * Adds a read-only {@link StateStore} to the topology.
+ * <p>
+ * A read-only StateStore can use any compacted topic as a changelog. The <code>auto.offset.reset</code> property
+ * will be set to <code>earliest</code> for this topic.
+ * <p>
+ * The provided {@link ProcessorSupplier} will be used to create a processor for all messages received
+ * from the given topic. This processor should contain logic to keep the {@link StateStore} up-to-date.
+ *
+ * @param storeBuilder user defined store builder
+ * @param sourceName name of the {@link SourceNode} that will be automatically added
+ * @param timestampExtractor the stateless timestamp extractor used for this source,
+ * if not specified the default extractor defined in the configs will be used
+ * @param keyDeserializer the {@link Deserializer} to deserialize keys with
+ * @param valueDeserializer the {@link Deserializer} to deserialize values with
+ * @param topic the topic to source the data from
+ * @param processorName the name of the {@link ProcessorSupplier}
+ * @param stateUpdateSupplier the instance of {@link ProcessorSupplier}
+ * @return itself
+ * @throws TopologyException if the processor of state is already registered
+ */
+ public synchronized <KIn, VIn> Topology addReadOnlyStateStore(final StoreBuilder<?> storeBuilder,
+ final String sourceName,
+ final TimestampExtractor timestampExtractor,
+ final Deserializer<KIn> keyDeserializer,
+ final Deserializer<VIn> valueDeserializer,
+ final String topic,
+ final String processorName,
+ final ProcessorSupplier<KIn, VIn, Void, Void> stateUpdateSupplier) {
+ // -- disabling logging. We might want to print some logging.
+ storeBuilder.withLoggingDisabled();
+
+ internalTopologyBuilder.addSource(AutoOffsetReset.EARLIEST, sourceName, timestampExtractor, keyDeserializer, valueDeserializer, topic);
+ internalTopologyBuilder.addProcessor(processorName, stateUpdateSupplier, sourceName);
+ internalTopologyBuilder.addStateStore(storeBuilder, processorName);
+ internalTopologyBuilder.connectSourceStoreAndTopic(storeBuilder.name(), topic);
+
+ return this;
+ }
+
+ /**
+ * Adds a Read Only {@link StateStore} to the topology.
Review Comment:
```suggestion
* Adds a read-only {@link StateStore} to the topology.
```
##########
clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java:
##########
@@ -54,7 +54,7 @@
* The consumer configuration keys
*/
public class ConsumerConfig extends AbstractConfig {
- private static final ConfigDef CONFIG;
+ protected static final ConfigDef CONFIG;
Review Comment:
Why this change? Also what is all this `ReverseXxx` stuff below? Did you add something that does not belong to this PR by accident?
##########
streams/src/main/java/org/apache/kafka/streams/Topology.java:
##########
@@ -737,6 +737,85 @@ public synchronized Topology addStateStore(final StoreBuilder<?> storeBuilder,
return this;
}
+ /**
+ * Adds a read-only {@link StateStore} to the topology.
+ * <p>
+ * A read-only StateStore can use any compacted topic as a changelog. The <code>auto.offset.reset</code> property
Review Comment:
```suggestion
* A read-only {@link StateStore} can use any compacted topic as a changelog. The <code>auto.offset.reset</code> property
```
##########
streams/src/main/java/org/apache/kafka/streams/Topology.java:
##########
@@ -737,6 +737,85 @@ public synchronized Topology addStateStore(final StoreBuilder<?> storeBuilder,
return this;
}
+ /**
+ * Adds a read-only {@link StateStore} to the topology.
+ * <p>
+ * A read-only StateStore can use any compacted topic as a changelog. The <code>auto.offset.reset</code> property
+ * will be set to <code>earliest</code> for this topic.
+ * <p>
+ * The provided {@link ProcessorSupplier} will be used to create a processor for all messages received
+ * from the given topic. This processor should contain logic to keep the {@link StateStore} up-to-date.
+ *
+ * @param storeBuilder user defined store builder
+ * @param sourceName name of the {@link SourceNode} that will be automatically added
+ * @param timestampExtractor the stateless timestamp extractor used for this source,
+ * if not specified the default extractor defined in the configs will be used
+ * @param keyDeserializer the {@link Deserializer} to deserialize keys with
+ * @param valueDeserializer the {@link Deserializer} to deserialize values with
+ * @param topic the topic to source the data from
+ * @param processorName the name of the {@link ProcessorSupplier}
+ * @param stateUpdateSupplier the instance of {@link ProcessorSupplier}
+ * @return itself
+ * @throws TopologyException if the processor of state is already registered
+ */
+ public synchronized <KIn, VIn> Topology addReadOnlyStateStore(final StoreBuilder<?> storeBuilder,
+ final String sourceName,
+ final TimestampExtractor timestampExtractor,
+ final Deserializer<KIn> keyDeserializer,
+ final Deserializer<VIn> valueDeserializer,
+ final String topic,
+ final String processorName,
+ final ProcessorSupplier<KIn, VIn, Void, Void> stateUpdateSupplier) {
+ // -- disabling logging. We might want to print some logging.
+ storeBuilder.withLoggingDisabled();
+
+ internalTopologyBuilder.addSource(AutoOffsetReset.EARLIEST, sourceName, timestampExtractor, keyDeserializer, valueDeserializer, topic);
+ internalTopologyBuilder.addProcessor(processorName, stateUpdateSupplier, sourceName);
+ internalTopologyBuilder.addStateStore(storeBuilder, processorName);
+ internalTopologyBuilder.connectSourceStoreAndTopic(storeBuilder.name(), topic);
+
+ return this;
+ }
+
+ /**
+ * Adds a Read Only {@link StateStore} to the topology.
+ *
Review Comment:
```suggestion
* <p>
```
##########
streams/src/main/java/org/apache/kafka/streams/Topology.java:
##########
@@ -737,6 +737,85 @@ public synchronized Topology addStateStore(final StoreBuilder<?> storeBuilder,
return this;
}
+ /**
+ * Adds a read-only {@link StateStore} to the topology.
+ * <p>
+ * A read-only StateStore can use any compacted topic as a changelog. The <code>auto.offset.reset</code> property
+ * will be set to <code>earliest</code> for this topic.
+ * <p>
+ * The provided {@link ProcessorSupplier} will be used to create a processor for all messages received
+ * from the given topic. This processor should contain logic to keep the {@link StateStore} up-to-date.
+ *
+ * @param storeBuilder user defined store builder
+ * @param sourceName name of the {@link SourceNode} that will be automatically added
+ * @param timestampExtractor the stateless timestamp extractor used for this source,
+ * if not specified the default extractor defined in the configs will be used
+ * @param keyDeserializer the {@link Deserializer} to deserialize keys with
+ * @param valueDeserializer the {@link Deserializer} to deserialize values with
+ * @param topic the topic to source the data from
+ * @param processorName the name of the {@link ProcessorSupplier}
+ * @param stateUpdateSupplier the instance of {@link ProcessorSupplier}
+ * @return itself
+ * @throws TopologyException if the processor of state is already registered
+ */
+ public synchronized <KIn, VIn> Topology addReadOnlyStateStore(final StoreBuilder<?> storeBuilder,
+ final String sourceName,
+ final TimestampExtractor timestampExtractor,
+ final Deserializer<KIn> keyDeserializer,
+ final Deserializer<VIn> valueDeserializer,
+ final String topic,
+ final String processorName,
+ final ProcessorSupplier<KIn, VIn, Void, Void> stateUpdateSupplier) {
+ // -- disabling logging. We might want to print some logging.
+ storeBuilder.withLoggingDisabled();
+
+ internalTopologyBuilder.addSource(AutoOffsetReset.EARLIEST, sourceName, timestampExtractor, keyDeserializer, valueDeserializer, topic);
+ internalTopologyBuilder.addProcessor(processorName, stateUpdateSupplier, sourceName);
+ internalTopologyBuilder.addStateStore(storeBuilder, processorName);
+ internalTopologyBuilder.connectSourceStoreAndTopic(storeBuilder.name(), topic);
+
+ return this;
+ }
+
+ /**
+ * Adds a Read Only {@link StateStore} to the topology.
+ *
+ * A Read Only StateStore can use any compacted topic as a changelog. The <code>auto.offset.reset</code> property
Review Comment:
```suggestion
* A read-only {@link StateStore} can use any compacted topic as a changelog. The <code>auto.offset.reset</code> property
```
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] calmera commented on a diff in pull request #12742: KAFKA-10892: Shared Readonly State Stores ( revisited )
Posted by GitBox <gi...@apache.org>.
calmera commented on code in PR #12742:
URL: https://github.com/apache/kafka/pull/12742#discussion_r1062606439
##########
streams/src/main/java/org/apache/kafka/streams/Topology.java:
##########
@@ -737,6 +737,91 @@ public synchronized Topology addStateStore(final StoreBuilder<?> storeBuilder,
return this;
}
+ /**
+ * Adds a Read Only {@link StateStore} to the topology.
+ *
+ * A Read Only StateStore can use any compacted topic as a changelog.
+ * <p>
+ * A {@link SourceNode} will be added to consume the data arriving from the partitions of the input topic.
+ * <p>
+ * The provided {@link ProcessorSupplier} will be used to create an {@link ProcessorNode} that will receive all
+ * records forwarded from the {@link SourceNode}.
+ * This {@link ProcessorNode} should be used to keep the {@link StateStore} up-to-date.
+ *
+ * @param storeBuilder user defined key value store builder
+ * @param sourceName name of the {@link SourceNode} that will be automatically added
+ * @param timestampExtractor the stateless timestamp extractor used for this source,
+ * if not specified the default extractor defined in the configs will be used
+ * @param keyDeserializer the {@link Deserializer} to deserialize keys with
+ * @param valueDeserializer the {@link Deserializer} to deserialize values with
+ * @param topic the topic to source the data from
+ * @param processorName the name of the {@link ProcessorSupplier}
+ * @param stateUpdateSupplier the instance of {@link ProcessorSupplier}
+ * @return itself
+ * @throws TopologyException if the processor of state is already registered
+ */
+ public synchronized <KIn, VIn> Topology addReadOnlyStateStore(final StoreBuilder<?> storeBuilder,
+ final String sourceName,
+ final TimestampExtractor timestampExtractor,
+ final Deserializer<KIn> keyDeserializer,
+ final Deserializer<VIn> valueDeserializer,
+ final String topic,
+ final String processorName,
+ final ProcessorSupplier<KIn, VIn, Void, Void> stateUpdateSupplier) {
+ if (storeBuilder.loggingEnabled()) {
+ // -- disabling logging. We might want to print some logging.
Review Comment:
agreed, will change accordingly
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] calmera commented on a diff in pull request #12742: KAFKA-10892: Shared Readonly State Stores ( revisited )
Posted by GitBox <gi...@apache.org>.
calmera commented on code in PR #12742:
URL: https://github.com/apache/kafka/pull/12742#discussion_r1063273259
##########
streams/src/test/java/org/apache/kafka/test/MockProcessor.java:
##########
@@ -28,9 +29,11 @@
import java.util.List;
import java.util.Map;
-@SuppressWarnings("deprecation") // Old PAPI. Needs to be migrated.
-public class MockProcessor<K, V> extends org.apache.kafka.streams.processor.AbstractProcessor<K, V> {
- private final MockApiProcessor<K, V, Object, Object> delegate;
+public class MockProcessor<KIn, VIn, KOut, VOut> implements Processor<KIn, VIn, KOut, VOut> {
+ private final MockApiProcessor<KIn, VIn, KOut, VOut> delegate;
+
+ private ProcessorContext<KOut, VOut> context;
Review Comment:
I didn't look at the MockApiProcessor internals, but you are right, I can use the context from the delegate instead of tracking the context myself.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-10892: Shared Readonly State Stores ( revisited ) [kafka]
Posted by "wcarlson5 (via GitHub)" <gi...@apache.org>.
wcarlson5 commented on code in PR #12742:
URL: https://github.com/apache/kafka/pull/12742#discussion_r1515301279
##########
streams/src/test/java/org/apache/kafka/streams/processor/ReadOnlyStoreTest.java:
##########
@@ -0,0 +1,132 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.streams.processor;
+
+import org.apache.kafka.common.serialization.IntegerDeserializer;
+import org.apache.kafka.common.serialization.IntegerSerializer;
+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.KeyValue;
+import org.apache.kafka.streams.TestInputTopic;
+import org.apache.kafka.streams.TestOutputTopic;
+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.KeyValueIterator;
+import org.apache.kafka.streams.state.KeyValueStore;
+import org.apache.kafka.streams.state.Stores;
+import org.junit.Test;
+
+import java.util.LinkedList;
+import java.util.List;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.equalTo;
+
+public class ReadOnlyStoreTest {
+
+ @Test
+ public void shouldLoadDataIntoReadOnlyStoreAndAllowAccessFromProcessor() {
Review Comment:
This kind of sounds like its testing the error path if the processor shouldn't have access to it tries to write/access that. Do we have a test for that? It seems that's what's special about a read only store
##########
streams/src/test/java/org/apache/kafka/streams/processor/ReadOnlyStoreTest.java:
##########
@@ -0,0 +1,132 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.streams.processor;
+
+import org.apache.kafka.common.serialization.IntegerDeserializer;
+import org.apache.kafka.common.serialization.IntegerSerializer;
+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.KeyValue;
+import org.apache.kafka.streams.TestInputTopic;
+import org.apache.kafka.streams.TestOutputTopic;
+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.KeyValueIterator;
+import org.apache.kafka.streams.state.KeyValueStore;
+import org.apache.kafka.streams.state.Stores;
+import org.junit.Test;
+
+import java.util.LinkedList;
+import java.util.List;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.equalTo;
+
+public class ReadOnlyStoreTest {
+
+ @Test
+ public void shouldLoadDataIntoReadOnlyStoreAndAllowAccessFromProcessor() {
+ final Topology topology = new Topology();
+ topology.addReadOnlyStateStore(
+ Stores.keyValueStoreBuilder(
+ Stores.inMemoryKeyValueStore("readOnlyStore"),
+ new Serdes.IntegerSerde(),
+ new Serdes.StringSerde()
+ ),
+ "readOnlySource",
+ new IntegerDeserializer(),
+ new StringDeserializer(),
+ "storeTopic",
+ "readOnlyProcessor",
+ () -> new Processor<Integer, String, Void, Void>() {
+ KeyValueStore<Integer, String> store;
+
+ @Override
+ public void init(final ProcessorContext<Void, Void> context) {
+ store = context.getStateStore("readOnlyStore");
+ }
+ @Override
+ public void process(final Record<Integer, String> record) {
+ store.put(record.key(), record.value());
+ }
+ }
+ );
+
+ topology.addSource("source", new IntegerDeserializer(), new StringDeserializer(), "inputTopic");
+ topology.addProcessor(
+ "processor",
+ () -> new Processor<Integer, String, Integer, String>() {
+ ProcessorContext<Integer, String> context;
+ KeyValueStore<Integer, String> store;
+
+ @Override
+ public void init(final ProcessorContext<Integer, String> context) {
+ this.context = context;
+ store = context.getStateStore("readOnlyStore");
+ }
+
+ @Override
+ public void process(final Record<Integer, String> record) {
+ context.forward(record.withValue(
+ record.value() + " -- " + store.get(record.key())
+ ));
+ }
+ },
+ "source"
+ );
+ topology.connectProcessorAndStateStores("processor", "readOnlyStore");
+ topology.addSink("sink", "outputTopic", new IntegerSerializer(), new StringSerializer(), "processor");
+
+ try (final TopologyTestDriver driver = new TopologyTestDriver(topology)) {
+ final TestInputTopic<Integer, String> changelog =
Review Comment:
Naming this changelog and then writing to it was about 5% confusing.
super nit but maybe 'readOnlyStoreTopicChangelog'. Idk jut something that isn't an already used term
##########
streams/src/test/java/org/apache/kafka/test/MockProcessor.java:
##########
@@ -43,12 +42,12 @@ public MockProcessor() {
}
@Override
- public void init(ProcessorContext<KOut, VOut> context) {
+ public void init(final ProcessorContext<KOut, VOut> context) {
Review Comment:
Thanks for cleaning up
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-10892: Shared Readonly State Stores ( revisited ) [kafka]
Posted by "calmera (via GitHub)" <gi...@apache.org>.
calmera commented on PR #12742:
URL: https://github.com/apache/kafka/pull/12742#issuecomment-1987893614
Nice job @mjsax @wcarlson5! Sorry I couldn't spend more time on 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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-10892: Shared Readonly State Stores ( revisited ) [kafka]
Posted by "wcarlson5 (via GitHub)" <gi...@apache.org>.
wcarlson5 commented on code in PR #12742:
URL: https://github.com/apache/kafka/pull/12742#discussion_r1516823620
##########
streams/src/test/java/org/apache/kafka/streams/processor/ReadOnlyStoreTest.java:
##########
@@ -0,0 +1,132 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.streams.processor;
+
+import org.apache.kafka.common.serialization.IntegerDeserializer;
+import org.apache.kafka.common.serialization.IntegerSerializer;
+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.KeyValue;
+import org.apache.kafka.streams.TestInputTopic;
+import org.apache.kafka.streams.TestOutputTopic;
+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.KeyValueIterator;
+import org.apache.kafka.streams.state.KeyValueStore;
+import org.apache.kafka.streams.state.Stores;
+import org.junit.Test;
+
+import java.util.LinkedList;
+import java.util.List;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.equalTo;
+
+public class ReadOnlyStoreTest {
+
+ @Test
+ public void shouldLoadDataIntoReadOnlyStoreAndAllowAccessFromProcessor() {
Review Comment:
Maybe shouldLoadDataIntoReadOnlyStoreFromConnectedProcessor or shouldConnectProcessorAndWriteDataToReadOnlyStore.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] mjsax commented on a diff in pull request #12742: KAFKA-10892: Shared Readonly State Stores ( revisited )
Posted by GitBox <gi...@apache.org>.
mjsax commented on code in PR #12742:
URL: https://github.com/apache/kafka/pull/12742#discussion_r1059135143
##########
streams/src/main/java/org/apache/kafka/streams/Topology.java:
##########
@@ -737,6 +737,91 @@ public synchronized Topology addStateStore(final StoreBuilder<?> storeBuilder,
return this;
}
+ /**
+ * Adds a Read Only {@link StateStore} to the topology.
+ *
+ * A Read Only StateStore can use any compacted topic as a changelog.
+ * <p>
+ * A {@link SourceNode} will be added to consume the data arriving from the partitions of the input topic.
Review Comment:
Seems to be too detailed. Also `SourceNode` is an internal class and I think we should not refer to it in the JavaDocs.
##########
streams/src/main/java/org/apache/kafka/streams/Topology.java:
##########
@@ -737,6 +737,91 @@ public synchronized Topology addStateStore(final StoreBuilder<?> storeBuilder,
return this;
}
+ /**
+ * Adds a Read Only {@link StateStore} to the topology.
Review Comment:
```suggestion
* Adds a read-only {@link StateStore} to the topology.
```
##########
streams/src/main/java/org/apache/kafka/streams/Topology.java:
##########
@@ -737,6 +737,91 @@ public synchronized Topology addStateStore(final StoreBuilder<?> storeBuilder,
return this;
}
+ /**
+ * Adds a Read Only {@link StateStore} to the topology.
+ *
+ * A Read Only StateStore can use any compacted topic as a changelog.
Review Comment:
```suggestion
* A read-only StateStore can use any compacted topic as a changelog.
```
##########
streams/src/main/java/org/apache/kafka/streams/Topology.java:
##########
@@ -737,6 +737,91 @@ public synchronized Topology addStateStore(final StoreBuilder<?> storeBuilder,
return this;
}
+ /**
+ * Adds a Read Only {@link StateStore} to the topology.
+ *
Review Comment:
```suggestion
* <p>
```
##########
streams/src/main/java/org/apache/kafka/streams/Topology.java:
##########
@@ -737,6 +737,91 @@ public synchronized Topology addStateStore(final StoreBuilder<?> storeBuilder,
return this;
}
+ /**
+ * Adds a Read Only {@link StateStore} to the topology.
+ *
+ * A Read Only StateStore can use any compacted topic as a changelog.
+ * <p>
+ * A {@link SourceNode} will be added to consume the data arriving from the partitions of the input topic.
+ * <p>
+ * The provided {@link ProcessorSupplier} will be used to create an {@link ProcessorNode} that will receive all
+ * records forwarded from the {@link SourceNode}.
+ * This {@link ProcessorNode} should be used to keep the {@link StateStore} up-to-date.
+ *
+ * @param storeBuilder user defined key value store builder
Review Comment:
Why "key value" ?
##########
streams/src/main/java/org/apache/kafka/streams/Topology.java:
##########
@@ -737,6 +737,91 @@ public synchronized Topology addStateStore(final StoreBuilder<?> storeBuilder,
return this;
}
+ /**
+ * Adds a Read Only {@link StateStore} to the topology.
+ *
+ * A Read Only StateStore can use any compacted topic as a changelog.
+ * <p>
+ * A {@link SourceNode} will be added to consume the data arriving from the partitions of the input topic.
+ * <p>
+ * The provided {@link ProcessorSupplier} will be used to create an {@link ProcessorNode} that will receive all
+ * records forwarded from the {@link SourceNode}.
+ * This {@link ProcessorNode} should be used to keep the {@link StateStore} up-to-date.
+ *
+ * @param storeBuilder user defined key value store builder
+ * @param sourceName name of the {@link SourceNode} that will be automatically added
Review Comment:
as above
##########
streams/src/main/java/org/apache/kafka/streams/Topology.java:
##########
@@ -737,6 +737,91 @@ public synchronized Topology addStateStore(final StoreBuilder<?> storeBuilder,
return this;
}
+ /**
+ * Adds a Read Only {@link StateStore} to the topology.
+ *
+ * A Read Only StateStore can use any compacted topic as a changelog.
+ * <p>
+ * A {@link SourceNode} will be added to consume the data arriving from the partitions of the input topic.
+ * <p>
+ * The provided {@link ProcessorSupplier} will be used to create an {@link ProcessorNode} that will receive all
+ * records forwarded from the {@link SourceNode}.
+ * This {@link ProcessorNode} should be used to keep the {@link StateStore} up-to-date.
+ *
+ * @param storeBuilder user defined key value store builder
+ * @param sourceName name of the {@link SourceNode} that will be automatically added
+ * @param timestampExtractor the stateless timestamp extractor used for this source,
+ * if not specified the default extractor defined in the configs will be used
+ * @param keyDeserializer the {@link Deserializer} to deserialize keys with
+ * @param valueDeserializer the {@link Deserializer} to deserialize values with
+ * @param topic the topic to source the data from
+ * @param processorName the name of the {@link ProcessorSupplier}
+ * @param stateUpdateSupplier the instance of {@link ProcessorSupplier}
+ * @return itself
+ * @throws TopologyException if the processor of state is already registered
+ */
+ public synchronized <KIn, VIn> Topology addReadOnlyStateStore(final StoreBuilder<?> storeBuilder,
+ final String sourceName,
+ final TimestampExtractor timestampExtractor,
+ final Deserializer<KIn> keyDeserializer,
+ final Deserializer<VIn> valueDeserializer,
+ final String topic,
+ final String processorName,
+ final ProcessorSupplier<KIn, VIn, Void, Void> stateUpdateSupplier) {
+ if (storeBuilder.loggingEnabled()) {
+ // -- disabling logging. We might want to print some logging.
Review Comment:
For globlaKTable we also just disable logging without logging any warning -- guess we could mention it in the JavaDocs.
Also, we don't really need the `if` and can call `withLoggingDisabled()` blindly.
##########
streams/src/main/java/org/apache/kafka/streams/Topology.java:
##########
@@ -737,6 +737,91 @@ public synchronized Topology addStateStore(final StoreBuilder<?> storeBuilder,
return this;
}
+ /**
+ * Adds a Read Only {@link StateStore} to the topology.
+ *
+ * A Read Only StateStore can use any compacted topic as a changelog.
+ * <p>
+ * A {@link SourceNode} will be added to consume the data arriving from the partitions of the input topic.
+ * <p>
+ * The provided {@link ProcessorSupplier} will be used to create an {@link ProcessorNode} that will receive all
Review Comment:
Similar for `ProcessorNode`
##########
streams/src/main/java/org/apache/kafka/streams/Topology.java:
##########
@@ -737,6 +737,91 @@ public synchronized Topology addStateStore(final StoreBuilder<?> storeBuilder,
return this;
}
+ /**
+ * Adds a Read Only {@link StateStore} to the topology.
+ *
+ * A Read Only StateStore can use any compacted topic as a changelog.
+ * <p>
+ * A {@link SourceNode} will be added to consume the data arriving from the partitions of the input topic.
+ * <p>
+ * The provided {@link ProcessorSupplier} will be used to create an {@link ProcessorNode} that will receive all
+ * records forwarded from the {@link SourceNode}.
+ * This {@link ProcessorNode} should be used to keep the {@link StateStore} up-to-date.
+ *
+ * @param storeBuilder user defined key value store builder
+ * @param sourceName name of the {@link SourceNode} that will be automatically added
+ * @param timestampExtractor the stateless timestamp extractor used for this source,
+ * if not specified the default extractor defined in the configs will be used
+ * @param keyDeserializer the {@link Deserializer} to deserialize keys with
+ * @param valueDeserializer the {@link Deserializer} to deserialize values with
+ * @param topic the topic to source the data from
+ * @param processorName the name of the {@link ProcessorSupplier}
+ * @param stateUpdateSupplier the instance of {@link ProcessorSupplier}
+ * @return itself
+ * @throws TopologyException if the processor of state is already registered
+ */
Review Comment:
Should we mention anything about `auto.offset.reset` in the JavaDocs? Seems we don't allow to pass in a config, and we might want to hard-code it to "earliest"? (Cannot remember if/what we discussion on the KIP about it?)
##########
streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamBranchTest.java:
##########
@@ -59,7 +59,7 @@ public void testKStreamBranch() {
assertEquals(3, branches.length);
- final MockProcessorSupplier<Integer, String> supplier = new MockProcessorSupplier<>();
+ final MockProcessorSupplier<Integer, String, Void, Void> supplier = new MockProcessorSupplier<>();
Review Comment:
Thanks for the cleanup!
##########
streams/src/test/java/org/apache/kafka/streams/TopologyTest.java:
##########
@@ -2303,6 +2304,63 @@ private void addGlobalStoreToTopologyAndExpectedDescription(final String globalS
expectedDescription.addGlobalStore(expectedGlobalStore);
}
+ @Test
+ public void readonlyStateStoresShouldHaveTheirOwnSubTopology() {
Review Comment:
```suggestion
public void readOnlyStateStoresShouldHaveTheirOwnSubTopology() {
```
##########
docs/streams/developer-guide/processor-api.html:
##########
@@ -396,6 +397,21 @@ <h2>
</ul>
</p>
</div>
+ <div class="section" id="readonly-state-stores">
+ <span id="streams-developer-guide-state-store-readonly"></span><h3><a class="toc-backref" href="#id12">ReadOnly State Stores</a><a class="headerlink" href="#readonly-state-stores" title="Permalink to this headline"></a></h3>
+ <p>The changelog topic for a fault-tolerant state store can also be used to perform lookups against. It is
+ crucial to understand that there can be only one writer to this state store, but there can be multiple
+ readers.</p>
+
+ <p>This allows you to use a readonly state store for looking up reference data managed by another processor
+ or even another process all together. The messages within the compacted topic may be projected (subset of data extracted)
+ before going into the readonly state store. Therefor we can optimize what the data within our readonly state
+ store should look like, depending on the use-case it needs to serve.</p>
+
+ <p><b>note:</b> beware of the partitioning requirements when using readonly statestores for lookups during
Review Comment:
```suggestion
<p><b>note:</b> beware of the partitioning requirements when using read-only state stores for lookups during
```
##########
docs/streams/developer-guide/processor-api.html:
##########
@@ -396,6 +397,21 @@ <h2>
</ul>
</p>
</div>
+ <div class="section" id="readonly-state-stores">
+ <span id="streams-developer-guide-state-store-readonly"></span><h3><a class="toc-backref" href="#id12">ReadOnly State Stores</a><a class="headerlink" href="#readonly-state-stores" title="Permalink to this headline"></a></h3>
+ <p>The changelog topic for a fault-tolerant state store can also be used to perform lookups against. It is
Review Comment:
This sentence is somewhat confusing. Maybe we should start with something like this:
> A read-only state store materialized the data from its input topic. It also uses the input topic for fault-tolerance, and thus does not have an additional changelog topic (the input topic is re-used as changelog). Thus, the input topic should be configured with log compaction (add link to log compaction). Note that no other processor should modify the content of the state store, and the only writer should be the associated "state update processor"; other processor may read the content of the read-only store.
##########
streams/src/test/java/org/apache/kafka/streams/TopologyTest.java:
##########
@@ -2303,6 +2304,63 @@ private void addGlobalStoreToTopologyAndExpectedDescription(final String globalS
expectedDescription.addGlobalStore(expectedGlobalStore);
}
+ @Test
+ public void readonlyStateStoresShouldHaveTheirOwnSubTopology() {
+ final String sourceName = "source";
+ final String storeName = "store";
+ final String topicName = "topic";
+ final String processorName = "processor";
+
+ final KeyValueStoreBuilder<?, ?> storeBuilder = mock(KeyValueStoreBuilder.class);
+ when(storeBuilder.name()).thenReturn(storeName);
+ topology.addReadOnlyStateStore(
+ storeBuilder,
+ sourceName,
+ null,
+ null,
+ null,
+ topicName,
+ processorName,
+ new MockProcessorSupplier<>());
+
+ final TopologyDescription.Source expectedSource = new InternalTopologyBuilder.Source(sourceName, Sets.newSet(topicName), null);
+ final TopologyDescription.Processor expectedProcessor = new InternalTopologyBuilder.Processor(processorName, Sets.newSet(storeName));
+
+ ((InternalTopologyBuilder.AbstractNode) expectedSource).addSuccessor(expectedProcessor);
+ ((InternalTopologyBuilder.AbstractNode) expectedProcessor).addPredecessor(expectedSource);
+
+ final Set<TopologyDescription.Node> allNodes = new HashSet<>();
+ allNodes.add(expectedSource);
+ allNodes.add(expectedProcessor);
+ expectedDescription.addSubtopology(new SubtopologyDescription(0, allNodes));
+
+ assertThat(topology.describe(), equalTo(expectedDescription));
+ assertThat(topology.describe().hashCode(), equalTo(expectedDescription.hashCode()));
+ }
+
+ @Test
+ public void readonlyStateStoresShouldNotLog() {
Review Comment:
```suggestion
public void readOnlyStateStoresShouldNotLog() {
```
##########
docs/streams/developer-guide/processor-api.html:
##########
@@ -396,6 +397,21 @@ <h2>
</ul>
</p>
</div>
+ <div class="section" id="readonly-state-stores">
+ <span id="streams-developer-guide-state-store-readonly"></span><h3><a class="toc-backref" href="#id12">ReadOnly State Stores</a><a class="headerlink" href="#readonly-state-stores" title="Permalink to this headline"></a></h3>
+ <p>The changelog topic for a fault-tolerant state store can also be used to perform lookups against. It is
+ crucial to understand that there can be only one writer to this state store, but there can be multiple
+ readers.</p>
+
+ <p>This allows you to use a readonly state store for looking up reference data managed by another processor
Review Comment:
```suggestion
<p>This allows you to use a read-only state store for looking up reference data managed by another processor
```
##########
streams/src/test/java/org/apache/kafka/test/MockProcessor.java:
##########
@@ -28,9 +29,11 @@
import java.util.List;
import java.util.Map;
-@SuppressWarnings("deprecation") // Old PAPI. Needs to be migrated.
-public class MockProcessor<K, V> extends org.apache.kafka.streams.processor.AbstractProcessor<K, V> {
- private final MockApiProcessor<K, V, Object, Object> delegate;
+public class MockProcessor<KIn, VIn, KOut, VOut> implements Processor<KIn, VIn, KOut, VOut> {
+ private final MockApiProcessor<KIn, VIn, KOut, VOut> delegate;
+
+ private ProcessorContext<KOut, VOut> context;
Review Comment:
Why do we add this? We inherit `context` from the super class anyway.
##########
docs/streams/developer-guide/processor-api.html:
##########
@@ -396,6 +397,21 @@ <h2>
</ul>
</p>
</div>
+ <div class="section" id="readonly-state-stores">
+ <span id="streams-developer-guide-state-store-readonly"></span><h3><a class="toc-backref" href="#id12">ReadOnly State Stores</a><a class="headerlink" href="#readonly-state-stores" title="Permalink to this headline"></a></h3>
+ <p>The changelog topic for a fault-tolerant state store can also be used to perform lookups against. It is
+ crucial to understand that there can be only one writer to this state store, but there can be multiple
+ readers.</p>
+
+ <p>This allows you to use a readonly state store for looking up reference data managed by another processor
+ or even another process all together. The messages within the compacted topic may be projected (subset of data extracted)
+ before going into the readonly state store. Therefor we can optimize what the data within our readonly state
+ store should look like, depending on the use-case it needs to serve.</p>
+
+ <p><b>note:</b> beware of the partitioning requirements when using readonly statestores for lookups during
+ processing. You might want to make sure the original changelog topic is co-partitioned with the processors
+ reading the readonly statestore.</p>
Review Comment:
This reminds me: should we allow to pass in a custom partitioner? 🤔 (Could also be follow up work though...)
##########
docs/streams/developer-guide/processor-api.html:
##########
@@ -396,6 +397,21 @@ <h2>
</ul>
</p>
</div>
+ <div class="section" id="readonly-state-stores">
+ <span id="streams-developer-guide-state-store-readonly"></span><h3><a class="toc-backref" href="#id12">ReadOnly State Stores</a><a class="headerlink" href="#readonly-state-stores" title="Permalink to this headline"></a></h3>
+ <p>The changelog topic for a fault-tolerant state store can also be used to perform lookups against. It is
+ crucial to understand that there can be only one writer to this state store, but there can be multiple
+ readers.</p>
+
+ <p>This allows you to use a readonly state store for looking up reference data managed by another processor
+ or even another process all together. The messages within the compacted topic may be projected (subset of data extracted)
Review Comment:
> may be projected
I don't think that works? When we restore the state, we would by-pass the processor and put the bytes from the input topic into the store as-is...
##########
docs/streams/developer-guide/processor-api.html:
##########
@@ -396,6 +397,21 @@ <h2>
</ul>
</p>
</div>
+ <div class="section" id="readonly-state-stores">
+ <span id="streams-developer-guide-state-store-readonly"></span><h3><a class="toc-backref" href="#id12">ReadOnly State Stores</a><a class="headerlink" href="#readonly-state-stores" title="Permalink to this headline"></a></h3>
+ <p>The changelog topic for a fault-tolerant state store can also be used to perform lookups against. It is
+ crucial to understand that there can be only one writer to this state store, but there can be multiple
+ readers.</p>
+
+ <p>This allows you to use a readonly state store for looking up reference data managed by another processor
+ or even another process all together. The messages within the compacted topic may be projected (subset of data extracted)
+ before going into the readonly state store. Therefor we can optimize what the data within our readonly state
+ store should look like, depending on the use-case it needs to serve.</p>
+
+ <p><b>note:</b> beware of the partitioning requirements when using readonly statestores for lookups during
+ processing. You might want to make sure the original changelog topic is co-partitioned with the processors
+ reading the readonly statestore.</p>
Review Comment:
```suggestion
reading the read-only state store.</p>
```
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] calmera commented on a diff in pull request #12742: KAFKA-10892: Shared Readonly State Stores ( revisited )
Posted by GitBox <gi...@apache.org>.
calmera commented on code in PR #12742:
URL: https://github.com/apache/kafka/pull/12742#discussion_r1063270375
##########
streams/src/main/java/org/apache/kafka/streams/Topology.java:
##########
@@ -737,6 +737,91 @@ public synchronized Topology addStateStore(final StoreBuilder<?> storeBuilder,
return this;
}
+ /**
+ * Adds a Read Only {@link StateStore} to the topology.
+ *
+ * A Read Only StateStore can use any compacted topic as a changelog.
+ * <p>
+ * A {@link SourceNode} will be added to consume the data arriving from the partitions of the input topic.
+ * <p>
+ * The provided {@link ProcessorSupplier} will be used to create an {@link ProcessorNode} that will receive all
+ * records forwarded from the {@link SourceNode}.
+ * This {@link ProcessorNode} should be used to keep the {@link StateStore} up-to-date.
+ *
+ * @param storeBuilder user defined key value store builder
+ * @param sourceName name of the {@link SourceNode} that will be automatically added
Review Comment:
I think this was copied from the description of `addGlobalStore`.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] calmera commented on a diff in pull request #12742: KAFKA-10892: Shared Readonly State Stores ( revisited )
Posted by GitBox <gi...@apache.org>.
calmera commented on code in PR #12742:
URL: https://github.com/apache/kafka/pull/12742#discussion_r1063269411
##########
streams/src/main/java/org/apache/kafka/streams/Topology.java:
##########
@@ -737,6 +737,91 @@ public synchronized Topology addStateStore(final StoreBuilder<?> storeBuilder,
return this;
}
+ /**
+ * Adds a Read Only {@link StateStore} to the topology.
+ *
+ * A Read Only StateStore can use any compacted topic as a changelog.
+ * <p>
+ * A {@link SourceNode} will be added to consume the data arriving from the partitions of the input topic.
+ * <p>
+ * The provided {@link ProcessorSupplier} will be used to create an {@link ProcessorNode} that will receive all
+ * records forwarded from the {@link SourceNode}.
+ * This {@link ProcessorNode} should be used to keep the {@link StateStore} up-to-date.
+ *
+ * @param storeBuilder user defined key value store builder
Review Comment:
This indeed should just be any store builder
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] calmera commented on a diff in pull request #12742: KAFKA-10892: Shared Readonly State Stores ( revisited )
Posted by GitBox <gi...@apache.org>.
calmera commented on code in PR #12742:
URL: https://github.com/apache/kafka/pull/12742#discussion_r1063316877
##########
docs/streams/developer-guide/processor-api.html:
##########
@@ -396,6 +397,21 @@ <h2>
</ul>
</p>
</div>
+ <div class="section" id="readonly-state-stores">
+ <span id="streams-developer-guide-state-store-readonly"></span><h3><a class="toc-backref" href="#id12">ReadOnly State Stores</a><a class="headerlink" href="#readonly-state-stores" title="Permalink to this headline"></a></h3>
+ <p>The changelog topic for a fault-tolerant state store can also be used to perform lookups against. It is
Review Comment:
a lot better indeed. Modified 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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org