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 2020/07/09 22:42:38 UTC
[GitHub] [kafka] vvcephei opened a new pull request #9004: KAFKA-10261: Introduce the KIP-478 apis with shims
vvcephei opened a new pull request #9004:
URL: https://github.com/apache/kafka/pull/9004
Adds the new Processor and ProcessorContext interfaces
as proposed in KIP-478. To integrate in a staged fashion
with the code base, shims are included to convert back
and forth between the new and old APIs.
ProcessorNode is converted to the new APIs.
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [kafka] vvcephei commented on pull request #9004: KAFKA-10261: Introduce the KIP-478 apis with adapters
Posted by GitBox <gi...@apache.org>.
vvcephei commented on pull request #9004:
URL: https://github.com/apache/kafka/pull/9004#issuecomment-671134513
Merged to trunk. Thanks for the review @abbccdda !
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [kafka] abbccdda commented on a change in pull request #9004: KAFKA-10261: Introduce the KIP-478 apis with shims
Posted by GitBox <gi...@apache.org>.
abbccdda commented on a change in pull request #9004:
URL: https://github.com/apache/kafka/pull/9004#discussion_r460341215
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/api/ProcessorContext.java
##########
@@ -0,0 +1,240 @@
+/*
+ * 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.api;
+
+import org.apache.kafka.common.header.Headers;
+import org.apache.kafka.common.serialization.Serde;
+import org.apache.kafka.streams.StreamsMetrics;
+import org.apache.kafka.streams.errors.StreamsException;
+import org.apache.kafka.streams.processor.Cancellable;
+import org.apache.kafka.streams.processor.Processor;
+import org.apache.kafka.streams.processor.PunctuationType;
+import org.apache.kafka.streams.processor.Punctuator;
+import org.apache.kafka.streams.processor.StateRestoreCallback;
+import org.apache.kafka.streams.processor.StateStore;
+import org.apache.kafka.streams.processor.TaskId;
+import org.apache.kafka.streams.processor.TimestampExtractor;
+import org.apache.kafka.streams.processor.To;
+
+import java.io.File;
+import java.time.Duration;
+import java.util.Map;
+
+/**
+ * Processor context interface.
+ *
+ * @param <KForward>> a bound on the types of keys that may be forwarded
+ * @param <VForward>> a bound on the types of values that may be forwarded
+ */
+public interface ProcessorContext<KForward, VForward> {
Review comment:
Sounds good.
##########
File path: streams/src/test/java/org/apache/kafka/streams/processor/internals/GlobalProcessorContextImplTest.java
##########
@@ -82,20 +79,12 @@ public void setup() {
null,
null);
- final ProcessorNode<?, ?> processorNode = mock(ProcessorNode.class);
- globalContext.setCurrentNode(processorNode);
+ final ProcessorNode<Object, Object, Object, Object> processorNode = new ProcessorNode<>("testNode");
child = mock(ProcessorNode.class);
+ processorNode.addChild(child);
- expect(processorNode.children())
Review comment:
Why are we removing these checks?
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalApiProcessorContext.java
##########
@@ -0,0 +1,123 @@
+/*
+ * 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.internals;
+
+import org.apache.kafka.common.serialization.ByteArraySerializer;
+import org.apache.kafka.common.serialization.BytesSerializer;
+import org.apache.kafka.common.utils.Bytes;
+import org.apache.kafka.streams.processor.api.ProcessorContext;
+import org.apache.kafka.streams.processor.RecordContext;
+import org.apache.kafka.streams.processor.StateStore;
+import org.apache.kafka.streams.processor.internals.Task.TaskType;
+import org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl;
+import org.apache.kafka.streams.state.StoreBuilder;
+import org.apache.kafka.streams.state.internals.ThreadCache;
+import org.apache.kafka.streams.state.internals.ThreadCache.DirtyEntryFlushListener;
+
+/**
+ * For internal use so we can update the {@link RecordContext} and current
+ * {@link ProcessorNode} when we are forwarding items that have been evicted or flushed from
+ * {@link ThreadCache}
+ */
+public interface InternalApiProcessorContext<KForward, VForward> extends ProcessorContext<KForward, VForward> {
+ BytesSerializer BYTES_KEY_SERIALIZER = new BytesSerializer();
Review comment:
Seems both serializer are not used.
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorShim.java
##########
@@ -0,0 +1,53 @@
+/*
+ * 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.internals;
+
+
+import org.apache.kafka.streams.processor.api.Processor;
+import org.apache.kafka.streams.processor.api.ProcessorContext;
+
+public final class ProcessorShim<KIn, VIn, KOut, VOut> implements Processor<KIn, VIn, KOut, VOut> {
Review comment:
As I'm not a native speaker, is it common to the word `shim` to represent such conversion? I was thinking whether a `ProcessorConverter` would also do the job, or I'm missing the term meaning here.
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopologyBuilder.java
##########
@@ -880,40 +880,41 @@ public synchronized ProcessorTopology buildGlobalStateTopology() {
return globalGroups;
}
+ @SuppressWarnings("unchecked")
private ProcessorTopology build(final Set<String> nodeGroup) {
Objects.requireNonNull(applicationId, "topology has not completed optimization");
- final Map<String, ProcessorNode<?, ?>> processorMap = new LinkedHashMap<>();
- final Map<String, SourceNode<?, ?>> topicSourceMap = new HashMap<>();
- final Map<String, SinkNode<?, ?>> topicSinkMap = new HashMap<>();
+ final Map<String, ProcessorNode<?, ?, ?, ?>> processorMap = new LinkedHashMap<>();
+ final Map<String, SourceNode<?, ?, ?, ?>> topicSourceMap = new HashMap<>();
+ final Map<String, SinkNode<?, ?, ?, ?>> topicSinkMap = new HashMap<>();
final Map<String, StateStore> stateStoreMap = new LinkedHashMap<>();
final Set<String> repartitionTopics = new HashSet<>();
// create processor nodes in a topological order ("nodeFactories" is already topologically sorted)
// also make sure the state store map values following the insertion ordering
- for (final NodeFactory<?, ?> factory : nodeFactories.values()) {
+ for (final NodeFactory<?, ?, ?, ?> factory : nodeFactories.values()) {
if (nodeGroup == null || nodeGroup.contains(factory.name)) {
- final ProcessorNode<?, ?> node = factory.build();
+ final ProcessorNode<?, ?, ?, ?> node = factory.build();
processorMap.put(node.name(), node);
if (factory instanceof ProcessorNodeFactory) {
buildProcessorNode(processorMap,
stateStoreMap,
- (ProcessorNodeFactory<?, ?>) factory,
- node);
+ (ProcessorNodeFactory<?, ?, ?, ?>) factory,
+ (ProcessorNode<Object, Object, Object, Object>) node);
} else if (factory instanceof SourceNodeFactory) {
buildSourceNode(topicSourceMap,
repartitionTopics,
- (SourceNodeFactory<?, ?>) factory,
- (SourceNode<?, ?>) node);
+ (SourceNodeFactory<?, ?, ?, ?>) factory,
+ (SourceNode<?, ?, ?, ?>) node);
} else if (factory instanceof SinkNodeFactory) {
buildSinkNode(processorMap,
topicSinkMap,
repartitionTopics,
- (SinkNodeFactory<?, ?>) factory,
- (SinkNode<?, ?>) node);
+ (SinkNodeFactory<?, ?, ?, ?>) factory,
+ (SinkNode<Object, Object, Object, Object>) node);
Review comment:
Why do we use `Object` here instead of `?`?
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorNode.java
##########
@@ -57,9 +57,9 @@ public ProcessorNode(final String name) {
this(name, null, null);
}
- public ProcessorNode(final String name, final Processor<K, V> processor, final Set<String> stateStores) {
+ public ProcessorNode(final String name, final org.apache.kafka.streams.processor.Processor<KIn, VIn> processor, final Set<String> stateStores) {
Review comment:
nit: breakdown the line
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalApiProcessorContext.java
##########
@@ -0,0 +1,123 @@
+/*
+ * 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.internals;
+
+import org.apache.kafka.common.serialization.ByteArraySerializer;
+import org.apache.kafka.common.serialization.BytesSerializer;
+import org.apache.kafka.common.utils.Bytes;
+import org.apache.kafka.streams.processor.api.ProcessorContext;
+import org.apache.kafka.streams.processor.RecordContext;
+import org.apache.kafka.streams.processor.StateStore;
+import org.apache.kafka.streams.processor.internals.Task.TaskType;
+import org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl;
+import org.apache.kafka.streams.state.StoreBuilder;
+import org.apache.kafka.streams.state.internals.ThreadCache;
+import org.apache.kafka.streams.state.internals.ThreadCache.DirtyEntryFlushListener;
+
+/**
+ * For internal use so we can update the {@link RecordContext} and current
+ * {@link ProcessorNode} when we are forwarding items that have been evicted or flushed from
+ * {@link ThreadCache}
+ */
+public interface InternalApiProcessorContext<KForward, VForward> extends ProcessorContext<KForward, VForward> {
+ BytesSerializer BYTES_KEY_SERIALIZER = new BytesSerializer();
+ ByteArraySerializer BYTEARRAY_VALUE_SERIALIZER = new ByteArraySerializer();
+
+ @Override
+ StreamsMetricsImpl metrics();
+
+ /**
+ * @param timeMs current wall-clock system timestamp in milliseconds
+ */
+ void setSystemTimeMs(long timeMs);
+
+ /**
+ * @retun the current wall-clock system timestamp in milliseconds
Review comment:
typo
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/api/ProcessorContext.java
##########
@@ -0,0 +1,240 @@
+/*
+ * 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.api;
+
+import org.apache.kafka.common.header.Headers;
+import org.apache.kafka.common.serialization.Serde;
+import org.apache.kafka.streams.StreamsMetrics;
+import org.apache.kafka.streams.errors.StreamsException;
+import org.apache.kafka.streams.processor.Cancellable;
+import org.apache.kafka.streams.processor.Processor;
+import org.apache.kafka.streams.processor.PunctuationType;
+import org.apache.kafka.streams.processor.Punctuator;
+import org.apache.kafka.streams.processor.StateRestoreCallback;
+import org.apache.kafka.streams.processor.StateStore;
+import org.apache.kafka.streams.processor.TaskId;
+import org.apache.kafka.streams.processor.TimestampExtractor;
+import org.apache.kafka.streams.processor.To;
+
+import java.io.File;
+import java.time.Duration;
+import java.util.Map;
+
+/**
+ * Processor context interface.
+ *
+ * @param <KForward>> a bound on the types of keys that may be forwarded
Review comment:
duplicate >
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorContextReverseShim.java
##########
@@ -0,0 +1,244 @@
+/*
+ * 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.internals;
+
+import org.apache.kafka.common.header.Headers;
+import org.apache.kafka.common.serialization.Serde;
+import org.apache.kafka.common.utils.Bytes;
+import org.apache.kafka.streams.processor.Cancellable;
+import org.apache.kafka.streams.processor.PunctuationType;
+import org.apache.kafka.streams.processor.Punctuator;
+import org.apache.kafka.streams.processor.StateRestoreCallback;
+import org.apache.kafka.streams.processor.StateStore;
+import org.apache.kafka.streams.processor.TaskId;
+import org.apache.kafka.streams.processor.To;
+import org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl;
+import org.apache.kafka.streams.state.StoreBuilder;
+import org.apache.kafka.streams.state.internals.ThreadCache;
+
+import java.io.File;
+import java.time.Duration;
+import java.util.Map;
+
+public final class ProcessorContextReverseShim implements InternalProcessorContext {
Review comment:
We are missing the implementation for `transitionToActive`
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [kafka] vvcephei commented on pull request #9004: KAFKA-10261: Introduce the KIP-478 apis with adapters
Posted by GitBox <gi...@apache.org>.
vvcephei commented on pull request #9004:
URL: https://github.com/apache/kafka/pull/9004#issuecomment-671134339
Just one unrelated test failure:
```
org.apache.kafka.streams.integration.EosBetaUpgradeIntegrationTest.shouldUpgradeFromEosAlphaToEosBeta[true] failed, log available in /home/jenkins/jenkins-slave/workspace/kafka-pr-jdk11-scala2.13/streams/build/reports/testOutput/org.apache.kafka.streams.integration.EosBetaUpgradeIntegrationTest.shouldUpgradeFromEosAlphaToEosBeta[true].test.stdout
org.apache.kafka.streams.integration.EosBetaUpgradeIntegrationTest > shouldUpgradeFromEosAlphaToEosBeta[true] FAILED
java.lang.AssertionError:
Expected: <[KeyValue(2, 0), KeyValue(2, 1), KeyValue(2, 3), KeyValue(2, 6), KeyValue(2, 10), KeyValue(2, 15), KeyValue(2, 21), KeyValue(2, 28), KeyValue(2, 36), KeyValue(2, 45), KeyValue(2, 55), KeyValue(2, 66), KeyValue(2, 78), KeyValue(2, 91), KeyValue(2, 105), KeyValue(2, 120), KeyValue(2, 136), KeyValue(2, 153), KeyValue(2, 171), KeyValue(2, 190), KeyValue(2, 120), KeyValue(2, 136), KeyValue(2, 153), KeyValue(2, 171), KeyValue(2, 190), KeyValue(2, 210), KeyValue(2, 231), KeyValue(2, 253), KeyValue(2, 276), KeyValue(2, 300), KeyValue(2, 325), KeyValue(2, 351), KeyValue(2, 378), KeyValue(2, 406), KeyValue(2, 435), KeyValue(2, 210), KeyValue(2, 231), KeyValue(2, 253), KeyValue(2, 276), KeyValue(2, 300), KeyValue(2, 325), KeyValue(2, 351), KeyValue(2, 378), KeyValue(2, 406), KeyValue(2, 435), KeyValue(2, 465), KeyValue(2, 496), KeyValue(2, 528), KeyValue(2, 561), KeyValue(2, 595)]>
but: was <[KeyValue(2, 0), KeyValue(2, 1), KeyValue(2, 3), KeyValue(2, 6), KeyValue(2, 10), KeyValue(2, 15), KeyValue(2, 21), KeyValue(2, 28), KeyValue(2, 36), KeyValue(2, 45), KeyValue(2, 55), KeyValue(2, 66), KeyValue(2, 78), KeyValue(2, 91), KeyValue(2, 105), KeyValue(2, 120), KeyValue(2, 136), KeyValue(2, 153), KeyValue(2, 171), KeyValue(2, 190), KeyValue(2, 120), KeyValue(2, 136), KeyValue(2, 153), KeyValue(2, 171), KeyValue(2, 190), KeyValue(2, 210), KeyValue(2, 231), KeyValue(2, 253), KeyValue(2, 276), KeyValue(2, 300), KeyValue(2, 325), KeyValue(2, 351), KeyValue(2, 378), KeyValue(2, 406), KeyValue(2, 435), KeyValue(2, 210), KeyValue(2, 231), KeyValue(2, 253), KeyValue(2, 276), KeyValue(2, 300), KeyValue(2, 325), KeyValue(2, 351), KeyValue(2, 378), KeyValue(2, 406), KeyValue(2, 435), KeyValue(2, 465), KeyValue(2, 496), KeyValue(2, 528), KeyValue(2, 561), KeyValue(2, 595), KeyValue(2, 465), KeyValue(2, 496)]>
at org.hamcrest.MatcherAssert.assertThat(MatcherAssert.java:20)
at org.hamcrest.MatcherAssert.assertThat(MatcherAssert.java:6)
at org.apache.kafka.streams.integration.EosBetaUpgradeIntegrationTest.checkResultPerKey(EosBetaUpgradeIntegrationTest.java:1042)
at org.apache.kafka.streams.integration.EosBetaUpgradeIntegrationTest.verifyUncommitted(EosBetaUpgradeIntegrationTest.java:1005)
at org.apache.kafka.streams.integration.EosBetaUpgradeIntegrationTest.shouldUpgradeFromEosAlphaToEosBeta(EosBetaUpgradeIntegrationTest.java:738)
```
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [kafka] vvcephei commented on a change in pull request #9004: KAFKA-10261: Introduce the KIP-478 apis with shims
Posted by GitBox <gi...@apache.org>.
vvcephei commented on a change in pull request #9004:
URL: https://github.com/apache/kafka/pull/9004#discussion_r452588767
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorContextReverseShim.java
##########
@@ -0,0 +1,244 @@
+/*
+ * 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.internals;
+
+import org.apache.kafka.common.header.Headers;
+import org.apache.kafka.common.serialization.Serde;
+import org.apache.kafka.common.utils.Bytes;
+import org.apache.kafka.streams.processor.Cancellable;
+import org.apache.kafka.streams.processor.PunctuationType;
+import org.apache.kafka.streams.processor.Punctuator;
+import org.apache.kafka.streams.processor.StateRestoreCallback;
+import org.apache.kafka.streams.processor.StateStore;
+import org.apache.kafka.streams.processor.TaskId;
+import org.apache.kafka.streams.processor.To;
+import org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl;
+import org.apache.kafka.streams.state.StoreBuilder;
+import org.apache.kafka.streams.state.internals.ThreadCache;
+
+import java.io.File;
+import java.time.Duration;
+import java.util.Map;
+
+public final class ProcessorContextReverseShim implements InternalProcessorContext {
+ final InternalApiProcessorContext<Object, Object> delegate;
+
+ static InternalProcessorContext shim(final InternalApiProcessorContext<Object, Object> delegate) {
+ if (delegate instanceof ProcessorContextShim) {
+ return ((ProcessorContextShim<Object, Object>) delegate).delegate;
Review comment:
You'll see this block in all the shims. There are times when the internal code would wind up converting new to old and then back to new. This block prevents us from jumping though multiple layers in that case.
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/api/ProcessorContext.java
##########
@@ -0,0 +1,240 @@
+/*
+ * 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.api;
+
+import org.apache.kafka.common.header.Headers;
+import org.apache.kafka.common.serialization.Serde;
+import org.apache.kafka.streams.StreamsMetrics;
+import org.apache.kafka.streams.errors.StreamsException;
+import org.apache.kafka.streams.processor.Cancellable;
+import org.apache.kafka.streams.processor.Processor;
+import org.apache.kafka.streams.processor.PunctuationType;
+import org.apache.kafka.streams.processor.Punctuator;
+import org.apache.kafka.streams.processor.StateRestoreCallback;
+import org.apache.kafka.streams.processor.StateStore;
+import org.apache.kafka.streams.processor.TaskId;
+import org.apache.kafka.streams.processor.TimestampExtractor;
+import org.apache.kafka.streams.processor.To;
+
+import java.io.File;
+import java.time.Duration;
+import java.util.Map;
+
+/**
+ * Processor context interface.
+ *
+ * @param <KForward>> a bound on the types of keys that may be forwarded
+ * @param <VForward>> a bound on the types of values that may be forwarded
+ */
+public interface ProcessorContext<KForward, VForward> {
Review comment:
I'll have to update the KIP. Replacing ProcessorContext instead of just adding the generic parameters is going to avoid the Scala compatibility issue we faced last time.
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorShim.java
##########
@@ -0,0 +1,53 @@
+/*
+ * 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.internals;
+
+
+import org.apache.kafka.streams.processor.api.Processor;
+import org.apache.kafka.streams.processor.api.ProcessorContext;
+
+public final class ProcessorShim<KIn, VIn, KOut, VOut> implements Processor<KIn, VIn, KOut, VOut> {
Review comment:
We also need shims for the processors.
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalProcessorContextImpl.java
##########
@@ -57,12 +57,12 @@ public StateStore getStateStore(final String name) {
@SuppressWarnings("unchecked")
@Override
- public <K, V> void forward(final K key, final V value) {
- final ProcessorNode<?, ?> previousNode = currentNode();
+ public <KIn, VIn> void forward(final KIn key, final VIn value) {
+ final ProcessorNode<?, ?, ?, ?> previousNode = currentNode();
try {
- for (final ProcessorNode<?, ?> child : currentNode().children()) {
+ for (final ProcessorNode<?, ?, ?, ?> child : currentNode().children()) {
setCurrentNode(child);
- ((ProcessorNode<K, V>) child).process(key, value);
+ ((ProcessorNode<KIn, VIn, KIn, VIn>) child).process(key, value); // FIXME
Review comment:
Oh, just saw this again. The thing I have to fix is the assumption that the child's result arguments are also KIn and VIn.
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorContextReverseShim.java
##########
@@ -0,0 +1,244 @@
+/*
+ * 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.internals;
+
+import org.apache.kafka.common.header.Headers;
+import org.apache.kafka.common.serialization.Serde;
+import org.apache.kafka.common.utils.Bytes;
+import org.apache.kafka.streams.processor.Cancellable;
+import org.apache.kafka.streams.processor.PunctuationType;
+import org.apache.kafka.streams.processor.Punctuator;
+import org.apache.kafka.streams.processor.StateRestoreCallback;
+import org.apache.kafka.streams.processor.StateStore;
+import org.apache.kafka.streams.processor.TaskId;
+import org.apache.kafka.streams.processor.To;
+import org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl;
+import org.apache.kafka.streams.state.StoreBuilder;
+import org.apache.kafka.streams.state.internals.ThreadCache;
+
+import java.io.File;
+import java.time.Duration;
+import java.util.Map;
+
+public final class ProcessorContextReverseShim implements InternalProcessorContext {
Review comment:
This is a shim converting the new internal context back to the old one, so that it can be injected to the old Processor instances, which need the old interface.
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalApiProcessorContext.java
##########
@@ -0,0 +1,123 @@
+/*
+ * 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.internals;
+
+import org.apache.kafka.common.serialization.ByteArraySerializer;
+import org.apache.kafka.common.serialization.BytesSerializer;
+import org.apache.kafka.common.utils.Bytes;
+import org.apache.kafka.streams.processor.api.ProcessorContext;
+import org.apache.kafka.streams.processor.RecordContext;
+import org.apache.kafka.streams.processor.StateStore;
+import org.apache.kafka.streams.processor.internals.Task.TaskType;
+import org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl;
+import org.apache.kafka.streams.state.StoreBuilder;
+import org.apache.kafka.streams.state.internals.ThreadCache;
+import org.apache.kafka.streams.state.internals.ThreadCache.DirtyEntryFlushListener;
+
+/**
+ * For internal use so we can update the {@link RecordContext} and current
+ * {@link ProcessorNode} when we are forwarding items that have been evicted or flushed from
+ * {@link ThreadCache}
+ */
+public interface InternalApiProcessorContext<KForward, VForward> extends ProcessorContext<KForward, VForward> {
Review comment:
This is the "InternalProcessorContext" equivalent for the new ProcessorContext API. It's also identical to the old one, except for the new arguments.
Unlike the old public APIs, once I finish up the KIP implementation, we'll be able to delete the old InternalProcessorContext, at which point, I'll probably rename this class back to InternalProcessorContext.
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/api/ProcessorContext.java
##########
@@ -0,0 +1,240 @@
+/*
+ * 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.api;
+
+import org.apache.kafka.common.header.Headers;
+import org.apache.kafka.common.serialization.Serde;
+import org.apache.kafka.streams.StreamsMetrics;
+import org.apache.kafka.streams.errors.StreamsException;
+import org.apache.kafka.streams.processor.Cancellable;
+import org.apache.kafka.streams.processor.Processor;
+import org.apache.kafka.streams.processor.PunctuationType;
+import org.apache.kafka.streams.processor.Punctuator;
+import org.apache.kafka.streams.processor.StateRestoreCallback;
+import org.apache.kafka.streams.processor.StateStore;
+import org.apache.kafka.streams.processor.TaskId;
+import org.apache.kafka.streams.processor.TimestampExtractor;
+import org.apache.kafka.streams.processor.To;
+
+import java.io.File;
+import java.time.Duration;
+import java.util.Map;
+
+/**
+ * Processor context interface.
+ *
+ * @param <KForward>> a bound on the types of keys that may be forwarded
+ * @param <VForward>> a bound on the types of values that may be forwarded
+ */
+public interface ProcessorContext<KForward, VForward> {
+
+ /**
+ * Returns the application id.
+ *
+ * @return the application id
+ */
+ String applicationId();
+
+ /**
+ * Returns the task id.
+ *
+ * @return the task id
+ */
+ TaskId taskId();
+
+ /**
+ * Returns the default key serde.
+ *
+ * @return the key serializer
+ */
+ Serde<?> keySerde();
+
+ /**
+ * Returns the default value serde.
+ *
+ * @return the value serializer
+ */
+ Serde<?> valueSerde();
+
+ /**
+ * Returns the state directory for the partition.
+ *
+ * @return the state directory
+ */
+ File stateDir();
+
+ /**
+ * Returns Metrics instance.
+ *
+ * @return StreamsMetrics
+ */
+ StreamsMetrics metrics();
+
+ /**
+ * Registers and possibly restores the specified storage engine.
+ *
+ * @param store the storage engine
+ * @param stateRestoreCallback the restoration callback logic for log-backed state stores upon restart
+ *
+ * @throws IllegalStateException If store gets registered after initialized is already finished
+ * @throws StreamsException if the store's change log does not contain the partition
+ */
+ void register(final StateStore store,
+ final StateRestoreCallback stateRestoreCallback);
+
+ /**
+ * Get the state store given the store name.
+ *
+ * @param name The store name
+ * @return The state store instance
+ */
+ StateStore getStateStore(final String name);
+
+ /**
+ * Schedules a periodic operation for processors. A processor may call this method during
+ * {@link Processor#init(org.apache.kafka.streams.processor.ProcessorContext) initialization} or
+ * {@link Processor#process(Object, Object) processing} to
+ * schedule a periodic callback — called a punctuation — to {@link Punctuator#punctuate(long)}.
+ * The type parameter controls what notion of time is used for punctuation:
+ * <ul>
+ * <li>{@link PunctuationType#STREAM_TIME} — uses "stream time", which is advanced by the processing of messages
+ * in accordance with the timestamp as extracted by the {@link TimestampExtractor} in use.
+ * The first punctuation will be triggered by the first record that is processed.
+ * <b>NOTE:</b> Only advanced if messages arrive</li>
+ * <li>{@link PunctuationType#WALL_CLOCK_TIME} — uses system time (the wall-clock time),
+ * which is advanced independent of whether new messages arrive.
+ * The first punctuation will be triggered after interval has elapsed.
+ * <b>NOTE:</b> This is best effort only as its granularity is limited by how long an iteration of the
+ * processing loop takes to complete</li>
+ * </ul>
+ *
+ * <b>Skipping punctuations:</b> Punctuations will not be triggered more than once at any given timestamp.
+ * This means that "missed" punctuation will be skipped.
+ * It's possible to "miss" a punctuation if:
+ * <ul>
+ * <li>with {@link PunctuationType#STREAM_TIME}, when stream time advances more than interval</li>
+ * <li>with {@link PunctuationType#WALL_CLOCK_TIME}, on GC pause, too short interval, ...</li>
+ * </ul>
+ *
+ * @param interval the time interval between punctuations (supported minimum is 1 millisecond)
+ * @param type one of: {@link PunctuationType#STREAM_TIME}, {@link PunctuationType#WALL_CLOCK_TIME}
+ * @param callback a function consuming timestamps representing the current stream or system time
+ * @return a handle allowing cancellation of the punctuation schedule established by this method
+ */
+ Cancellable schedule(final Duration interval,
+ final PunctuationType type,
+ final Punctuator callback);
+
+ /**
+ * Forwards a key/value pair to all downstream processors.
+ * Used the input record's timestamp as timestamp for the output record.
+ *
+ * @param key key
+ * @param value value
+ */
+ <K extends KForward, V extends VForward> void forward(final K key, final V value);
Review comment:
Since this is a new class, I've dropped the deprecated members. Everything else from the old ProcessorContext is preserved.
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractProcessorContext.java
##########
@@ -42,7 +42,7 @@
private final Serde<?> valueSerde;
private boolean initialized;
protected ProcessorRecordContext recordContext;
- protected ProcessorNode<?, ?> currentNode;
+ protected ProcessorNode<?, ?, ?, ?> currentNode;
Review comment:
As you're about to find out, this kind of thing is the bulk of the changes.
Since I've only converted the ProcessorNode (an internal class) in this PR, you'll see the arguments right now are overwhelmingly wildcards and Object. ProcessorNode is almost exclusively used in the "machine room" parts of Streams where we don't have access to, or any benefit from, the actual generic type parameters of the Processors.
In the follow-on PRs, where I convert the actual Processors to the new API is when we'll start to see the benefits.
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [kafka] vvcephei commented on a change in pull request #9004: KAFKA-10261: Introduce the KIP-478 apis with shims
Posted by GitBox <gi...@apache.org>.
vvcephei commented on a change in pull request #9004:
URL: https://github.com/apache/kafka/pull/9004#discussion_r461271276
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopologyBuilder.java
##########
@@ -880,40 +880,41 @@ public synchronized ProcessorTopology buildGlobalStateTopology() {
return globalGroups;
}
+ @SuppressWarnings("unchecked")
private ProcessorTopology build(final Set<String> nodeGroup) {
Objects.requireNonNull(applicationId, "topology has not completed optimization");
- final Map<String, ProcessorNode<?, ?>> processorMap = new LinkedHashMap<>();
- final Map<String, SourceNode<?, ?>> topicSourceMap = new HashMap<>();
- final Map<String, SinkNode<?, ?>> topicSinkMap = new HashMap<>();
+ final Map<String, ProcessorNode<?, ?, ?, ?>> processorMap = new LinkedHashMap<>();
+ final Map<String, SourceNode<?, ?, ?, ?>> topicSourceMap = new HashMap<>();
+ final Map<String, SinkNode<?, ?, ?, ?>> topicSinkMap = new HashMap<>();
final Map<String, StateStore> stateStoreMap = new LinkedHashMap<>();
final Set<String> repartitionTopics = new HashSet<>();
// create processor nodes in a topological order ("nodeFactories" is already topologically sorted)
// also make sure the state store map values following the insertion ordering
- for (final NodeFactory<?, ?> factory : nodeFactories.values()) {
+ for (final NodeFactory<?, ?, ?, ?> factory : nodeFactories.values()) {
if (nodeGroup == null || nodeGroup.contains(factory.name)) {
- final ProcessorNode<?, ?> node = factory.build();
+ final ProcessorNode<?, ?, ?, ?> node = factory.build();
processorMap.put(node.name(), node);
if (factory instanceof ProcessorNodeFactory) {
buildProcessorNode(processorMap,
stateStoreMap,
- (ProcessorNodeFactory<?, ?>) factory,
- node);
+ (ProcessorNodeFactory<?, ?, ?, ?>) factory,
+ (ProcessorNode<Object, Object, Object, Object>) node);
} else if (factory instanceof SourceNodeFactory) {
buildSourceNode(topicSourceMap,
repartitionTopics,
- (SourceNodeFactory<?, ?>) factory,
- (SourceNode<?, ?>) node);
+ (SourceNodeFactory<?, ?, ?, ?>) factory,
+ (SourceNode<?, ?, ?, ?>) node);
} else if (factory instanceof SinkNodeFactory) {
buildSinkNode(processorMap,
topicSinkMap,
repartitionTopics,
- (SinkNodeFactory<?, ?>) factory,
- (SinkNode<?, ?>) node);
+ (SinkNodeFactory<?, ?, ?, ?>) factory,
+ (SinkNode<Object, Object, Object, Object>) node);
Review comment:
They have subtly different meanings, which I'm not 100% clear on all the time. I'm not sure if I had to change this one, of if it was an accident. I'll give it a closer look.
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorShim.java
##########
@@ -0,0 +1,53 @@
+/*
+ * 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.internals;
+
+
+import org.apache.kafka.streams.processor.api.Processor;
+import org.apache.kafka.streams.processor.api.ProcessorContext;
+
+public final class ProcessorShim<KIn, VIn, KOut, VOut> implements Processor<KIn, VIn, KOut, VOut> {
Review comment:
I think "adapter" is the standard design pattern name for this type of thing. Not sure why I thought "shim" was a good choice in the heat of the moment. Maybe because I'm kind of slipping these classes in the middle to make everything line up? I can change them to "adapter".
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopologyBuilder.java
##########
@@ -880,40 +880,41 @@ public synchronized ProcessorTopology buildGlobalStateTopology() {
return globalGroups;
}
+ @SuppressWarnings("unchecked")
private ProcessorTopology build(final Set<String> nodeGroup) {
Objects.requireNonNull(applicationId, "topology has not completed optimization");
- final Map<String, ProcessorNode<?, ?>> processorMap = new LinkedHashMap<>();
- final Map<String, SourceNode<?, ?>> topicSourceMap = new HashMap<>();
- final Map<String, SinkNode<?, ?>> topicSinkMap = new HashMap<>();
+ final Map<String, ProcessorNode<?, ?, ?, ?>> processorMap = new LinkedHashMap<>();
+ final Map<String, SourceNode<?, ?, ?, ?>> topicSourceMap = new HashMap<>();
+ final Map<String, SinkNode<?, ?, ?, ?>> topicSinkMap = new HashMap<>();
final Map<String, StateStore> stateStoreMap = new LinkedHashMap<>();
final Set<String> repartitionTopics = new HashSet<>();
// create processor nodes in a topological order ("nodeFactories" is already topologically sorted)
// also make sure the state store map values following the insertion ordering
- for (final NodeFactory<?, ?> factory : nodeFactories.values()) {
+ for (final NodeFactory<?, ?, ?, ?> factory : nodeFactories.values()) {
if (nodeGroup == null || nodeGroup.contains(factory.name)) {
- final ProcessorNode<?, ?> node = factory.build();
+ final ProcessorNode<?, ?, ?, ?> node = factory.build();
processorMap.put(node.name(), node);
if (factory instanceof ProcessorNodeFactory) {
buildProcessorNode(processorMap,
stateStoreMap,
- (ProcessorNodeFactory<?, ?>) factory,
- node);
+ (ProcessorNodeFactory<?, ?, ?, ?>) factory,
+ (ProcessorNode<Object, Object, Object, Object>) node);
} else if (factory instanceof SourceNodeFactory) {
buildSourceNode(topicSourceMap,
repartitionTopics,
- (SourceNodeFactory<?, ?>) factory,
- (SourceNode<?, ?>) node);
+ (SourceNodeFactory<?, ?, ?, ?>) factory,
+ (SourceNode<?, ?, ?, ?>) node);
} else if (factory instanceof SinkNodeFactory) {
buildSinkNode(processorMap,
topicSinkMap,
repartitionTopics,
- (SinkNodeFactory<?, ?>) factory,
- (SinkNode<?, ?>) node);
+ (SinkNodeFactory<?, ?, ?, ?>) factory,
+ (SinkNode<Object, Object, Object, Object>) node);
Review comment:
They have subtly different meanings, which I'm not 100% clear on all the time. Usually, the reason I switched to Object because the wildcard makes the type system want to bind the type to something unfortunate, and it can't prove that the usage is actually ok.
I'm not sure if I had to change this one, of if it was an accident. I'll give it a closer look.
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorContextReverseShim.java
##########
@@ -0,0 +1,244 @@
+/*
+ * 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.internals;
+
+import org.apache.kafka.common.header.Headers;
+import org.apache.kafka.common.serialization.Serde;
+import org.apache.kafka.common.utils.Bytes;
+import org.apache.kafka.streams.processor.Cancellable;
+import org.apache.kafka.streams.processor.PunctuationType;
+import org.apache.kafka.streams.processor.Punctuator;
+import org.apache.kafka.streams.processor.StateRestoreCallback;
+import org.apache.kafka.streams.processor.StateStore;
+import org.apache.kafka.streams.processor.TaskId;
+import org.apache.kafka.streams.processor.To;
+import org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl;
+import org.apache.kafka.streams.state.StoreBuilder;
+import org.apache.kafka.streams.state.internals.ThreadCache;
+
+import java.io.File;
+import java.time.Duration;
+import java.util.Map;
+
+public final class ProcessorContextReverseShim implements InternalProcessorContext {
Review comment:
Thanks. Maybe it's not in the place you were looking for it. I see it on line 133.
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopologyBuilder.java
##########
@@ -880,40 +880,41 @@ public synchronized ProcessorTopology buildGlobalStateTopology() {
return globalGroups;
}
+ @SuppressWarnings("unchecked")
private ProcessorTopology build(final Set<String> nodeGroup) {
Objects.requireNonNull(applicationId, "topology has not completed optimization");
- final Map<String, ProcessorNode<?, ?>> processorMap = new LinkedHashMap<>();
- final Map<String, SourceNode<?, ?>> topicSourceMap = new HashMap<>();
- final Map<String, SinkNode<?, ?>> topicSinkMap = new HashMap<>();
+ final Map<String, ProcessorNode<?, ?, ?, ?>> processorMap = new LinkedHashMap<>();
+ final Map<String, SourceNode<?, ?, ?, ?>> topicSourceMap = new HashMap<>();
+ final Map<String, SinkNode<?, ?, ?, ?>> topicSinkMap = new HashMap<>();
final Map<String, StateStore> stateStoreMap = new LinkedHashMap<>();
final Set<String> repartitionTopics = new HashSet<>();
// create processor nodes in a topological order ("nodeFactories" is already topologically sorted)
// also make sure the state store map values following the insertion ordering
- for (final NodeFactory<?, ?> factory : nodeFactories.values()) {
+ for (final NodeFactory<?, ?, ?, ?> factory : nodeFactories.values()) {
if (nodeGroup == null || nodeGroup.contains(factory.name)) {
- final ProcessorNode<?, ?> node = factory.build();
+ final ProcessorNode<?, ?, ?, ?> node = factory.build();
processorMap.put(node.name(), node);
if (factory instanceof ProcessorNodeFactory) {
buildProcessorNode(processorMap,
stateStoreMap,
- (ProcessorNodeFactory<?, ?>) factory,
- node);
+ (ProcessorNodeFactory<?, ?, ?, ?>) factory,
+ (ProcessorNode<Object, Object, Object, Object>) node);
} else if (factory instanceof SourceNodeFactory) {
buildSourceNode(topicSourceMap,
repartitionTopics,
- (SourceNodeFactory<?, ?>) factory,
- (SourceNode<?, ?>) node);
+ (SourceNodeFactory<?, ?, ?, ?>) factory,
+ (SourceNode<?, ?, ?, ?>) node);
} else if (factory instanceof SinkNodeFactory) {
buildSinkNode(processorMap,
topicSinkMap,
repartitionTopics,
- (SinkNodeFactory<?, ?>) factory,
- (SinkNode<?, ?>) node);
+ (SinkNodeFactory<?, ?, ?, ?>) factory,
+ (SinkNode<Object, Object, Object, Object>) node);
Review comment:
Ah, yes, indeed. It's because inside buildSinkNode, we are calling `getProcessor(...).addChild(node)`, and the type system is unable to prove that the forward types of the parent match the input types of the child, because in this internal layer, we have already lost all the type information of the nodes.
It doesn't really matter in the internals anyway (note that this class always just had wildcards/Object as the generic parameters. The real benefit of KIP-478 is for users of the PAPI (including the DSL implementation), not this internal plumbing logic.
##########
File path: streams/src/test/java/org/apache/kafka/streams/processor/internals/GlobalProcessorContextImplTest.java
##########
@@ -82,20 +79,12 @@ public void setup() {
null,
null);
- final ProcessorNode<?, ?> processorNode = mock(ProcessorNode.class);
- globalContext.setCurrentNode(processorNode);
+ final ProcessorNode<Object, Object, Object, Object> processorNode = new ProcessorNode<>("testNode");
child = mock(ProcessorNode.class);
+ processorNode.addChild(child);
- expect(processorNode.children())
Review comment:
Ah, it's because I swapped out the mock of the ProcessorNode for a real processor node (on line 82). Note that these were not checks, they were just dummy returns, since we never verified them, and it was a nice mock.
Now, all the tests behave the same, and we don't need as much boilerplace test setup.
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalApiProcessorContext.java
##########
@@ -0,0 +1,123 @@
+/*
+ * 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.internals;
+
+import org.apache.kafka.common.serialization.ByteArraySerializer;
+import org.apache.kafka.common.serialization.BytesSerializer;
+import org.apache.kafka.common.utils.Bytes;
+import org.apache.kafka.streams.processor.api.ProcessorContext;
+import org.apache.kafka.streams.processor.RecordContext;
+import org.apache.kafka.streams.processor.StateStore;
+import org.apache.kafka.streams.processor.internals.Task.TaskType;
+import org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl;
+import org.apache.kafka.streams.state.StoreBuilder;
+import org.apache.kafka.streams.state.internals.ThreadCache;
+import org.apache.kafka.streams.state.internals.ThreadCache.DirtyEntryFlushListener;
+
+/**
+ * For internal use so we can update the {@link RecordContext} and current
+ * {@link ProcessorNode} when we are forwarding items that have been evicted or flushed from
+ * {@link ThreadCache}
+ */
+public interface InternalApiProcessorContext<KForward, VForward> extends ProcessorContext<KForward, VForward> {
+ BytesSerializer BYTES_KEY_SERIALIZER = new BytesSerializer();
+ ByteArraySerializer BYTEARRAY_VALUE_SERIALIZER = new ByteArraySerializer();
+
+ @Override
+ StreamsMetricsImpl metrics();
+
+ /**
+ * @param timeMs current wall-clock system timestamp in milliseconds
+ */
+ void setSystemTimeMs(long timeMs);
+
+ /**
+ * @retun the current wall-clock system timestamp in milliseconds
Review comment:
Good eye. I'll also fix it in InternalProcessorContext.
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalApiProcessorContext.java
##########
@@ -0,0 +1,123 @@
+/*
+ * 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.internals;
+
+import org.apache.kafka.common.serialization.ByteArraySerializer;
+import org.apache.kafka.common.serialization.BytesSerializer;
+import org.apache.kafka.common.utils.Bytes;
+import org.apache.kafka.streams.processor.api.ProcessorContext;
+import org.apache.kafka.streams.processor.RecordContext;
+import org.apache.kafka.streams.processor.StateStore;
+import org.apache.kafka.streams.processor.internals.Task.TaskType;
+import org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl;
+import org.apache.kafka.streams.state.StoreBuilder;
+import org.apache.kafka.streams.state.internals.ThreadCache;
+import org.apache.kafka.streams.state.internals.ThreadCache.DirtyEntryFlushListener;
+
+/**
+ * For internal use so we can update the {@link RecordContext} and current
+ * {@link ProcessorNode} when we are forwarding items that have been evicted or flushed from
+ * {@link ThreadCache}
+ */
+public interface InternalApiProcessorContext<KForward, VForward> extends ProcessorContext<KForward, VForward> {
+ BytesSerializer BYTES_KEY_SERIALIZER = new BytesSerializer();
Review comment:
Good eye :)
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [kafka] vvcephei merged pull request #9004: KAFKA-10261: Introduce the KIP-478 apis with adapters
Posted by GitBox <gi...@apache.org>.
vvcephei merged pull request #9004:
URL: https://github.com/apache/kafka/pull/9004
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [kafka] vvcephei commented on pull request #9004: KAFKA-10261: Introduce the KIP-478 apis with shims
Posted by GitBox <gi...@apache.org>.
vvcephei commented on pull request #9004:
URL: https://github.com/apache/kafka/pull/9004#issuecomment-656446145
Note, I didn't bother testing the shims directly, since they are very effectively tested by literally all the tests in Streams.
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [kafka] vvcephei commented on pull request #9004: KAFKA-10261: Introduce the KIP-478 apis with shims
Posted by GitBox <gi...@apache.org>.
vvcephei commented on pull request #9004:
URL: https://github.com/apache/kafka/pull/9004#issuecomment-658804914
Retest this please
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [kafka] vvcephei commented on pull request #9004: KAFKA-10261: Introduce the KIP-478 apis with shims
Posted by GitBox <gi...@apache.org>.
vvcephei commented on pull request #9004:
URL: https://github.com/apache/kafka/pull/9004#issuecomment-664740837
Thanks for the review, @abbccdda ! I've addressed your feedback.
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [kafka] abbccdda commented on a change in pull request #9004: KAFKA-10261: Introduce the KIP-478 apis with adapters
Posted by GitBox <gi...@apache.org>.
abbccdda commented on a change in pull request #9004:
URL: https://github.com/apache/kafka/pull/9004#discussion_r462057744
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopologyBuilder.java
##########
@@ -764,12 +764,12 @@ private void connectProcessorAndStateStore(final String processorName,
if (!sourceTopics.isEmpty()) {
stateStoreNameToSourceTopics.put(stateStoreName,
- Collections.unmodifiableSet(sourceTopics));
+ Collections.unmodifiableSet(sourceTopics));
Review comment:
format looks weird, maybe just do 4 spaces
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalProcessorContextImpl.java
##########
@@ -57,12 +57,12 @@ public StateStore getStateStore(final String name) {
@SuppressWarnings("unchecked")
@Override
- public <K, V> void forward(final K key, final V value) {
- final ProcessorNode<?, ?> previousNode = currentNode();
+ public <KIn, VIn> void forward(final KIn key, final VIn value) {
+ final ProcessorNode<?, ?, ?, ?> previousNode = currentNode();
try {
- for (final ProcessorNode<?, ?> child : currentNode().children()) {
+ for (final ProcessorNode<?, ?, ?, ?> child : currentNode().children()) {
setCurrentNode(child);
- ((ProcessorNode<K, V>) child).process(key, value);
+ ((ProcessorNode<KIn, VIn, KIn, VIn>) child).process(key, value); // FIXME
Review comment:
Could we leave a more clear comment on what needs to be fixed?
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/api/ProcessorContext.java
##########
@@ -0,0 +1,240 @@
+/*
+ * 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.api;
+
+import org.apache.kafka.common.header.Headers;
+import org.apache.kafka.common.serialization.Serde;
+import org.apache.kafka.streams.StreamsMetrics;
+import org.apache.kafka.streams.errors.StreamsException;
+import org.apache.kafka.streams.processor.Cancellable;
+import org.apache.kafka.streams.processor.Processor;
+import org.apache.kafka.streams.processor.PunctuationType;
+import org.apache.kafka.streams.processor.Punctuator;
+import org.apache.kafka.streams.processor.StateRestoreCallback;
+import org.apache.kafka.streams.processor.StateStore;
+import org.apache.kafka.streams.processor.TaskId;
+import org.apache.kafka.streams.processor.TimestampExtractor;
+import org.apache.kafka.streams.processor.To;
+
+import java.io.File;
+import java.time.Duration;
+import java.util.Map;
+
+/**
+ * Processor context interface.
+ *
+ * @param <KForward> a bound on the types of keys that may be forwarded
+ * @param <VForward> a bound on the types of values that may be forwarded
+ */
+public interface ProcessorContext<KForward, VForward> {
+
+ /**
+ * Returns the application id.
+ *
+ * @return the application id
+ */
+ String applicationId();
+
+ /**
+ * Returns the task id.
+ *
+ * @return the task id
+ */
+ TaskId taskId();
+
+ /**
+ * Returns the default key serde.
+ *
+ * @return the key serializer
+ */
+ Serde<?> keySerde();
+
+ /**
+ * Returns the default value serde.
+ *
+ * @return the value serializer
+ */
+ Serde<?> valueSerde();
+
+ /**
+ * Returns the state directory for the partition.
+ *
+ * @return the state directory
+ */
+ File stateDir();
+
+ /**
+ * Returns Metrics instance.
+ *
+ * @return StreamsMetrics
+ */
+ StreamsMetrics metrics();
+
+ /**
+ * Registers and possibly restores the specified storage engine.
+ *
+ * @param store the storage engine
+ * @param stateRestoreCallback the restoration callback logic for log-backed state stores upon restart
+ *
+ * @throws IllegalStateException If store gets registered after initialized is already finished
+ * @throws StreamsException if the store's change log does not contain the partition
+ */
+ void register(final StateStore store,
+ final StateRestoreCallback stateRestoreCallback);
+
+ /**
+ * Get the state store given the store name.
+ *
+ * @param name The store name
+ * @return The state store instance
+ */
+ StateStore getStateStore(final String name);
+
+ /**
+ * Schedules a periodic operation for processors. A processor may call this method during
+ * {@link Processor#init(org.apache.kafka.streams.processor.ProcessorContext) initialization} or
+ * {@link Processor#process(Object, Object) processing} to
+ * schedule a periodic callback — called a punctuation — to {@link Punctuator#punctuate(long)}.
+ * The type parameter controls what notion of time is used for punctuation:
+ * <ul>
+ * <li>{@link PunctuationType#STREAM_TIME} — uses "stream time", which is advanced by the processing of messages
+ * in accordance with the timestamp as extracted by the {@link TimestampExtractor} in use.
+ * The first punctuation will be triggered by the first record that is processed.
+ * <b>NOTE:</b> Only advanced if messages arrive</li>
+ * <li>{@link PunctuationType#WALL_CLOCK_TIME} — uses system time (the wall-clock time),
+ * which is advanced independent of whether new messages arrive.
+ * The first punctuation will be triggered after interval has elapsed.
+ * <b>NOTE:</b> This is best effort only as its granularity is limited by how long an iteration of the
+ * processing loop takes to complete</li>
+ * </ul>
+ *
+ * <b>Skipping punctuations:</b> Punctuations will not be triggered more than once at any given timestamp.
+ * This means that "missed" punctuation will be skipped.
+ * It's possible to "miss" a punctuation if:
+ * <ul>
+ * <li>with {@link PunctuationType#STREAM_TIME}, when stream time advances more than interval</li>
+ * <li>with {@link PunctuationType#WALL_CLOCK_TIME}, on GC pause, too short interval, ...</li>
+ * </ul>
+ *
+ * @param interval the time interval between punctuations (supported minimum is 1 millisecond)
+ * @param type one of: {@link PunctuationType#STREAM_TIME}, {@link PunctuationType#WALL_CLOCK_TIME}
+ * @param callback a function consuming timestamps representing the current stream or system time
+ * @return a handle allowing cancellation of the punctuation schedule established by this method
+ */
+ Cancellable schedule(final Duration interval,
+ final PunctuationType type,
+ final Punctuator callback);
+
+ /**
+ * Forwards a key/value pair to all downstream processors.
+ * Used the input record's timestamp as timestamp for the output record.
+ *
+ * @param key key
+ * @param value value
+ */
+ <K extends KForward, V extends VForward> void forward(final K key, final V value);
+
+ /**
+ * Forwards a key/value pair to the specified downstream processors.
+ * Can be used to set the timestamp of the output record.
+ *
+ * @param key key
+ * @param value value
+ * @param to the options to use when forwarding
+ */
+ <K extends KForward, V extends VForward> void forward(final K key, final V value, final To to);
+
+ /**
+ * Requests a commit.
+ */
+ void commit();
+
+ /**
+ * Returns the topic name of the current input record; could be null if it is not
+ * available (for example, if this method is invoked from the punctuate call).
+ *
+ * @return the topic name
+ */
+ String topic();
+
+ /**
+ * Returns the partition id of the current input record; could be -1 if it is not
+ * available (for example, if this method is invoked from the punctuate call).
+ *
+ * @return the partition id
+ */
+ int partition();
+
+ /**
+ * Returns the offset of the current input record; could be -1 if it is not
+ * available (for example, if this method is invoked from the punctuate call).
+ *
+ * @return the offset
+ */
+ long offset();
+
+ /**
+ * Returns the headers of the current input record; could be null if it is not
+ * available (for example, if this method is invoked from the punctuate call).
+ *
+ * @return the headers
+ */
+ Headers headers();
+
+ /**
+ * Returns the current timestamp.
+ *
+ * <p> If it is triggered while processing a record streamed from the source processor,
+ * timestamp is defined as the timestamp of the current input record; the timestamp is extracted from
+ * {@link org.apache.kafka.clients.consumer.ConsumerRecord ConsumerRecord} by {@link TimestampExtractor}.
+ *
+ * <p> If it is triggered while processing a record generated not from the source processor (for example,
+ * if this method is invoked from the punctuate call), timestamp is defined as the current
+ * task's stream time, which is defined as the largest timestamp of any record processed by the task.
+ *
+ * @return the timestamp
+ */
+ long timestamp();
+
+ /**
+ * Returns all the application config properties as key/value pairs.
+ *
+ * <p> The config properties are defined in the {@link org.apache.kafka.streams.StreamsConfig}
+ * object and associated to the ProcessorContext.
+ *
+ * <p> The type of the values is dependent on the {@link org.apache.kafka.common.config.ConfigDef.Type type} of the property
+ * (e.g. the value of {@link org.apache.kafka.streams.StreamsConfig#DEFAULT_KEY_SERDE_CLASS_CONFIG DEFAULT_KEY_SERDE_CLASS_CONFIG}
+ * will be of type {@link Class}, even if it was specified as a String to
+ * {@link org.apache.kafka.streams.StreamsConfig#StreamsConfig(Map) StreamsConfig(Map)}).
+ *
+ * @return all the key/values from the StreamsConfig properties
+ */
+ Map<String, Object> appConfigs();
+
+ /**
+ * Returns all the application config properties with the given key prefix, as key/value pairs
+ * stripping the prefix.
+ *
+ * <p> The config properties are defined in the {@link org.apache.kafka.streams.StreamsConfig}
+ * object and associated to the ProcessorContext.
+ *
+ * @param prefix the properties prefix
+ * @return the key/values matching the given prefix from the StreamsConfig properties.
+ */
+ Map<String, Object> appConfigsWithPrefix(final String prefix);
+
Review comment:
nit: remove extra line
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorContextAdapter.java
##########
@@ -0,0 +1,230 @@
+/*
+ * 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.internals;
+
+import org.apache.kafka.common.header.Headers;
+import org.apache.kafka.common.serialization.Serde;
+import org.apache.kafka.common.utils.Bytes;
+import org.apache.kafka.streams.processor.Cancellable;
+import org.apache.kafka.streams.processor.PunctuationType;
+import org.apache.kafka.streams.processor.Punctuator;
+import org.apache.kafka.streams.processor.StateRestoreCallback;
+import org.apache.kafka.streams.processor.StateStore;
+import org.apache.kafka.streams.processor.TaskId;
+import org.apache.kafka.streams.processor.To;
+import org.apache.kafka.streams.processor.api.ProcessorContext;
+import org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl;
+import org.apache.kafka.streams.state.StoreBuilder;
+import org.apache.kafka.streams.state.internals.ThreadCache;
+
+import java.io.File;
+import java.time.Duration;
+import java.util.Map;
+
+public final class ProcessorContextAdapter<KForward, VForward>
+ implements ProcessorContext<KForward, VForward>, InternalApiProcessorContext<KForward, VForward> {
+
+ final InternalProcessorContext delegate;
Review comment:
nit: could make access private and get an accessor.
----------------------------------------------------------------
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:
users@infra.apache.org