vvcephei commented on a change in pull request #9148:
URL: https://github.com/apache/kafka/pull/9148#discussion_r467671774
##########
File path: streams/src/main/java/org/apache/kafka/streams/StreamsBuilder.java
##########
@@ -499,7 +500,7 @@ public synchronized StreamsBuilder addStateStore(final
StoreBuilder<?> builder)
topic,
new ConsumedInternal<>(consumed),
processorName,
- stateUpdateSupplier
+ () -> ProcessorAdapter.adapt(stateUpdateSupplier.get())
Review comment:
Adapt the old API to the new one.
##########
File path: streams/src/main/java/org/apache/kafka/streams/Topology.java
##########
@@ -721,7 +722,7 @@ public synchronized Topology addStateStore(final
StoreBuilder<?> storeBuilder,
valueDeserializer,
topic,
processorName,
- stateUpdateSupplier
+ () -> ProcessorAdapter.adapt(stateUpdateSupplier.get())
Review comment:
Also here, adapting the old API to the new one.
##########
File path:
streams/src/main/java/org/apache/kafka/streams/kstream/internals/InternalStreamsBuilder.java
##########
@@ -201,12 +200,12 @@ public synchronized void addStateStore(final
StoreBuilder<?> builder) {
addGraphNode(root, new StateStoreNode<>(builder));
}
- public synchronized <K, V> void addGlobalStore(final StoreBuilder<?>
storeBuilder,
- final String sourceName,
- final String topic,
- final ConsumedInternal<K,
V> consumed,
- final String processorName,
- final ProcessorSupplier<K,
V> stateUpdateSupplier) {
+ public synchronized <KIn, VIn> void addGlobalStore(final StoreBuilder<?>
storeBuilder,
+ final String sourceName,
+ final String topic,
+ final
ConsumedInternal<KIn, VIn> consumed,
+ final String
processorName,
+ final
org.apache.kafka.streams.processor.api.ProcessorSupplier<KIn, VIn, Void, Void>
stateUpdateSupplier) {
Review comment:
For the internal builder, just directly change to the new API.
##########
File path:
streams/src/main/java/org/apache/kafka/streams/kstream/internals/graph/TableSourceNode.java
##########
@@ -89,14 +90,16 @@ public void writeToTopology(final InternalTopologyBuilder
topologyBuilder) {
new
TimestampedKeyValueStoreMaterializer<>((MaterializedInternal<K, V,
KeyValueStore<Bytes, byte[]>>) materializedInternal).materialize();
if (isGlobalKTable) {
- topologyBuilder.addGlobalStore(storeBuilder,
- sourceName,
-
consumedInternal().timestampExtractor(),
-
consumedInternal().keyDeserializer(),
-
consumedInternal().valueDeserializer(),
- topicName,
- processorParameters.processorName(),
-
processorParameters.processorSupplier());
+ topologyBuilder.addGlobalStore(
+ storeBuilder,
+ sourceName,
+ consumedInternal().timestampExtractor(),
+ consumedInternal().keyDeserializer(),
+ consumedInternal().valueDeserializer(),
+ topicName,
+ processorParameters.processorName(),
+ () ->
ProcessorAdapter.adapt(processorParameters.processorSupplier().get())
Review comment:
For now, just adapt. Later this whole node will get converted to the new
API.
##########
File path:
streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopologyBuilder.java
##########
@@ -192,16 +192,23 @@ private boolean isWindowStore() {
}
private static class ProcessorNodeFactory<KIn, VIn, KOut, VOut> extends
NodeFactory<KIn, VIn, KOut, VOut> {
- private final ProcessorSupplier<KIn, VIn> supplier;
+ private final
org.apache.kafka.streams.processor.api.ProcessorSupplier<KIn, VIn, KOut, VOut>
supplier;
private final Set<String> stateStoreNames = new HashSet<>();
ProcessorNodeFactory(final String name,
final String[] predecessors,
- final ProcessorSupplier<KIn, VIn> supplier) {
+ final
org.apache.kafka.streams.processor.api.ProcessorSupplier<KIn, VIn, KOut, VOut>
supplier) {
Review comment:
We need to create a factory for the global node, so I'm adding the
constructor for the new API, and also converting the internals here.
##########
File path:
streams/src/test/java/org/apache/kafka/streams/processor/internals/GlobalStreamThreadTest.java
##########
@@ -96,7 +96,7 @@ public String newStoreName(final String prefix) {
null,
GLOBAL_STORE_TOPIC_NAME,
"processorName",
- new KTableSource<>(GLOBAL_STORE_NAME, GLOBAL_STORE_NAME));
+ () -> ProcessorAdapter.adapt(new KTableSource<>(GLOBAL_STORE_NAME,
GLOBAL_STORE_NAME).get()));
Review comment:
Adapting for now. Once the Source is converted, this line will revert to
the original.
##########
File path:
streams/src/test/java/org/apache/kafka/streams/processor/internals/InternalTopologyBuilderTest.java
##########
@@ -270,14 +271,15 @@ public void testPatternSourceTopicsWithGlobalTopics() {
builder.addSource(null, "source-1", null, null, null,
Pattern.compile("topic-1"));
builder.addSource(null, "source-2", null, null, null,
Pattern.compile("topic-2"));
builder.addGlobalStore(
- new MockKeyValueStoreBuilder("global-store",
false).withLoggingDisabled(),
- "globalSource",
- null,
- null,
- null,
- "globalTopic",
- "global-processor",
- new MockProcessorSupplier<>());
+ new MockKeyValueStoreBuilder("global-store",
false).withLoggingDisabled(),
+ "globalSource",
+ null,
+ null,
+ null,
+ "globalTopic",
+ "global-processor",
+ new MockApiProcessorSupplier<>()
Review comment:
Using the new API. some extra lines changed when I cleaned up the
indentation formatting.
##########
File path:
streams/src/test/java/org/apache/kafka/streams/integration/GlobalThreadShutDownOrderTest.java
##########
@@ -36,9 +36,8 @@
import org.apache.kafka.streams.state.Stores;
import org.apache.kafka.streams.state.internals.KeyValueStoreBuilder;
import org.apache.kafka.test.IntegrationTest;
-import org.apache.kafka.test.MockProcessorSupplier;
+import org.apache.kafka.test.MockApiProcessorSupplier;
Review comment:
I went ahead and just converted the mock processor to the new API. I
gave it a temporary name. Eventually, the old one will be unused, and we can
delete the old one and rename the new one back to `MockProcessorSupplier`.
##########
File path: streams/src/test/java/org/apache/kafka/test/MockApiProcessor.java
##########
@@ -0,0 +1,174 @@
+/*
+ * 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.test;
+
+import org.apache.kafka.streams.KeyValueTimestamp;
+import org.apache.kafka.streams.processor.Cancellable;
+import org.apache.kafka.streams.processor.PunctuationType;
+import org.apache.kafka.streams.processor.api.Processor;
+import org.apache.kafka.streams.processor.api.ProcessorContext;
+import org.apache.kafka.streams.state.ValueAndTimestamp;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.is;
+
+public class MockApiProcessor<KIn, VIn, KOut, VOut> implements Processor<KIn,
VIn, KOut, VOut> {
Review comment:
As noted earlier, I'm copying this class from `MockProcesor` to convert
it to the new API (to avoid disturbing too much code at once).
##########
File path:
streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorContextImplTest.java
##########
@@ -150,13 +149,21 @@ public void setup() {
((InternalProcessorContext) context).transitionToActive(task, null,
null);
EasyMock.expect(task.recordCollector()).andStubReturn(recordCollector);
- context.setCurrentNode(new ProcessorNode<String, Long, Object,
Object>("fake", null,
- new HashSet<>(asList(
- "LocalKeyValueStore",
- "LocalTimestampedKeyValueStore",
- "LocalWindowStore",
- "LocalTimestampedWindowStore",
- "LocalSessionStore"))));
+ context.setCurrentNode(
+ new ProcessorNode<>(
+ "fake",
+ (org.apache.kafka.streams.processor.api.Processor<String,
Long, Object, Object>) null,
Review comment:
This is all that changed. I had to cast the null to resolve the right
overload.
##########
File path:
streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorNode.java
##########
@@ -54,7 +54,19 @@
private Sensor createSensor;
public ProcessorNode(final String name) {
- this(name, null, null);
+ this(name, (Processor<KIn, VIn, KOut, VOut>) null, null);
+ }
+
+ public ProcessorNode(final String name,
+ final Processor<KIn, VIn, KOut, VOut> processor,
+ final Set<String> stateStores) {
+
+ this.name = name;
+ this.processor = processor;
+ this.children = new ArrayList<>();
+ this.childByName = new HashMap<>();
+ this.stateStores = stateStores;
+ this.time = new SystemTime();
Review comment:
Adding the constructor for the new API. The old constructor still exists
and still functions to adapt the old API. Eventually, the old constructor will
become unused.
##########
File path:
streams/streams-scala/src/main/scala/org/apache/kafka/streams/scala/StreamsBuilder.scala
##########
@@ -176,12 +176,33 @@ class StreamsBuilder(inner: StreamsBuilderJ = new
StreamsBuilderJ) {
*
* @see `org.apache.kafka.streams.StreamsBuilder#addGlobalStore`
*/
+ @deprecated(
+ "Use #addGlobalStore(StoreBuilder, String, Consumed,
org.apache.kafka.streams.processor.api.ProcessorSupplier) instead.",
+ "2.7.0"
+ )
def addGlobalStore[K, V](storeBuilder: StoreBuilder[_ <: StateStore],
topic: String,
consumed: Consumed[K, V],
stateUpdateSupplier: ProcessorSupplier[K, V]):
StreamsBuilderJ =
inner.addGlobalStore(storeBuilder, topic, consumed, stateUpdateSupplier)
+ /**
+ * Adds a global `StateStore` to the topology. Global stores should not be
added to `Processor`, `Transformer`,
+ * or `ValueTransformer` (in contrast to regular stores).
+ * <p>
+ * It is not required to connect a global store to `Processor`,
`Transformer`, or `ValueTransformer`;
+ * those have read-only access to all global stores by default.
+ *
+ * @see `org.apache.kafka.streams.StreamsBuilder#addGlobalStore`
+ */
+ def addGlobalStore[K, V](
Review comment:
Keeping the Scala API in sync with the Java one.
##########
File path:
streams/src/test/java/org/apache/kafka/test/MockApiProcessorSupplier.java
##########
@@ -0,0 +1,69 @@
+/*
+ * 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.test;
+
+import org.apache.kafka.streams.processor.api.Processor;
+import org.apache.kafka.streams.processor.api.ProcessorSupplier;
+import org.apache.kafka.streams.processor.PunctuationType;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.junit.Assert.assertEquals;
+
+public class MockApiProcessorSupplier<KIn, VIn, KOut, VOut> implements
ProcessorSupplier<KIn, VIn, KOut, VOut> {
Review comment:
As noted earlier, I'm copying this class from `MockProcesorSupplier` to
convert it to the new API (to avoid disturbing too much code at once).
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
