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 2021/08/13 19:04:13 UTC
[GitHub] [kafka] jeqo opened a new pull request #11213: KAFKA-13201: Convert KTable aggregations to new PAPI
jeqo opened a new pull request #11213:
URL: https://github.com/apache/kafka/pull/11213
As part of the migration of KStream/KTable operations to the new Processor API https://issues.apache.org/jira/browse/KAFKA-8410, this PR includes the migration of KTable suppress.
Testing strategy: operations should keep the same tests as new processor should be compatible.
### 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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] jeqo commented on a change in pull request #11213: KAFKA-13201: Convert KTable suppress to new PAPI
Posted by GitBox <gi...@apache.org>.
jeqo commented on a change in pull request #11213:
URL: https://github.com/apache/kafka/pull/11213#discussion_r703713785
##########
File path: streams/src/main/java/org/apache/kafka/streams/kstream/internals/suppress/KTableSuppressProcessorSupplier.java
##########
@@ -198,7 +201,7 @@ private void emit(final TimeOrderedKeyValueBuffer.Eviction<K, V> toEmit) {
final ProcessorRecordContext prevRecordContext = internalProcessorContext.recordContext();
internalProcessorContext.setRecordContext(toEmit.recordContext());
try {
- internalProcessorContext.forward(toEmit.key(), toEmit.value());
+ internalProcessorContext.forward(toEmit.record());
Review comment:
It does, yes. I just tested in one of the methods. Adding the change and updating a test to cover this.
--
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] jeqo commented on a change in pull request #11213: KAFKA-13201: Convert KTable suppress to new PAPI
Posted by GitBox <gi...@apache.org>.
jeqo commented on a change in pull request #11213:
URL: https://github.com/apache/kafka/pull/11213#discussion_r703713785
##########
File path: streams/src/main/java/org/apache/kafka/streams/kstream/internals/suppress/KTableSuppressProcessorSupplier.java
##########
@@ -198,7 +201,7 @@ private void emit(final TimeOrderedKeyValueBuffer.Eviction<K, V> toEmit) {
final ProcessorRecordContext prevRecordContext = internalProcessorContext.recordContext();
internalProcessorContext.setRecordContext(toEmit.recordContext());
try {
- internalProcessorContext.forward(toEmit.key(), toEmit.value());
+ internalProcessorContext.forward(toEmit.record());
Review comment:
It does, yet. I just tested in one of the methods. Adding the change and updating a test to cover this.
--
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] jeqo commented on a change in pull request #11213: KAFKA-13201: Convert KTable suppress to new PAPI
Posted by GitBox <gi...@apache.org>.
jeqo commented on a change in pull request #11213:
URL: https://github.com/apache/kafka/pull/11213#discussion_r693488664
##########
File path: streams/src/test/java/org/apache/kafka/test/MockInternalNewProcessorContext.java
##########
@@ -0,0 +1,172 @@
+/*
+ * 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.common.utils.Bytes;
+import org.apache.kafka.streams.processor.MockProcessorContext;
Review comment:
> Will check if setting the timestamp when forwarding be a better approach.
It's not. Forward is called "too late" to check it in this suppress test cases. Staying with `setTimestamp` could be good enough then.
--
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] vvcephei commented on a change in pull request #11213: KAFKA-13201: Convert KTable suppress to new PAPI
Posted by GitBox <gi...@apache.org>.
vvcephei commented on a change in pull request #11213:
URL: https://github.com/apache/kafka/pull/11213#discussion_r703630483
##########
File path: streams/src/main/java/org/apache/kafka/streams/kstream/internals/suppress/KTableSuppressProcessorSupplier.java
##########
@@ -198,7 +201,7 @@ private void emit(final TimeOrderedKeyValueBuffer.Eviction<K, V> toEmit) {
final ProcessorRecordContext prevRecordContext = internalProcessorContext.recordContext();
internalProcessorContext.setRecordContext(toEmit.recordContext());
try {
- internalProcessorContext.forward(toEmit.key(), toEmit.value());
+ internalProcessorContext.forward(toEmit.record());
Review comment:
I think this might actually be a bug. IIRC, the Record forward call overrides the context, so you might have to actually set all the fields in Record from the context when you construct it in `toEmit.record()`. Specifically, I think we might be dropping the headers here.
I might also be wrong, so you might want to double-check me first.
It looks like we were never testing whether we propagate headers through Suppress. If it turns out that we never were, then there's also no problem. The case to look out for is if we _were_, but now we no longer are.
--
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] vvcephei commented on pull request #11213: KAFKA-13201: Convert KTable suppress to new PAPI
Posted by GitBox <gi...@apache.org>.
vvcephei commented on pull request #11213:
URL: https://github.com/apache/kafka/pull/11213#issuecomment-914663785
Java 17 errored out:
```
[2021-09-07T21:03:23.475Z] * What went wrong:
[2021-09-07T21:03:23.475Z] Execution failed for task ':core:integrationTest'.
[2021-09-07T21:03:23.475Z] > Process 'Gradle Test Executor 129' finished with non-zero exit value 1
[2021-09-07T21:03:23.475Z] This problem might be caused by incorrect test process configuration.
[2021-09-07T21:03:23.475Z] Please refer to the test execution section in the User Manual at https://docs.gradle.org/7.2/userguide/java_testing.html#sec:test_execution
```
But Java 8 and 11 passed, so I'll go ahead and merge (since Java 17 itself is new to this project).
--
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] vvcephei commented on a change in pull request #11213: KAFKA-13201: Convert KTable aggregations to new PAPI
Posted by GitBox <gi...@apache.org>.
vvcephei commented on a change in pull request #11213:
URL: https://github.com/apache/kafka/pull/11213#discussion_r690642563
##########
File path: streams/src/test/java/org/apache/kafka/test/MockInternalNewProcessorContext.java
##########
@@ -0,0 +1,172 @@
+/*
+ * 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.common.utils.Bytes;
+import org.apache.kafka.streams.processor.MockProcessorContext;
+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.Record;
+import org.apache.kafka.streams.processor.api.RecordMetadata;
+import org.apache.kafka.streams.processor.internals.InternalProcessorContext;
+import org.apache.kafka.streams.processor.internals.ProcessorNode;
+import org.apache.kafka.streams.processor.internals.ProcessorRecordContext;
+import org.apache.kafka.streams.processor.internals.RecordCollector;
+import org.apache.kafka.streams.processor.internals.StreamTask;
+import org.apache.kafka.streams.processor.internals.Task.TaskType;
+import org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl;
+import org.apache.kafka.streams.state.internals.ThreadCache;
+import org.apache.kafka.streams.state.internals.ThreadCache.DirtyEntryFlushListener;
+
+import java.io.File;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Properties;
+
+public class MockInternalNewProcessorContext<KOut, VOut> extends MockProcessorContext implements InternalProcessorContext<KOut, VOut> {
Review comment:
It looks like this is copied and modified from https://github.com/apache/kafka/blob/trunk/streams/src/test/java/org/apache/kafka/test/MockInternalProcessorContext.java . Is that right?
We also have https://github.com/apache/kafka/blob/trunk/streams/src/test/java/org/apache/kafka/test/InternalMockProcessorContext.java , and it's been a longstanding thorn in our side that there's a proliferation of these context implementations.
I'm hoping that by the time we're done with all these migrations, we can actually converge on this new class and delete the other two.
##########
File path: streams/src/test/java/org/apache/kafka/test/MockInternalNewProcessorContext.java
##########
@@ -0,0 +1,172 @@
+/*
+ * 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.common.utils.Bytes;
+import org.apache.kafka.streams.processor.MockProcessorContext;
Review comment:
This is the one defined in `test-utils`, right? Should we be using the new PAPI one (https://github.com/apache/kafka/blob/trunk/streams/test-utils/src/main/java/org/apache/kafka/streams/processor/api/MockProcessorContext.java) instead?
##########
File path: streams/src/main/java/org/apache/kafka/streams/kstream/internals/suppress/KTableSuppressProcessorSupplier.java
##########
@@ -150,16 +153,16 @@ public void init(final org.apache.kafka.streams.processor.ProcessorContext conte
}
@Override
- public void process(final K key, final Change<V> value) {
+ public void process(final Record<K, Change<V>> record) {
observedStreamTime = Math.max(observedStreamTime, internalProcessorContext.timestamp());
- buffer(key, value);
+ buffer(record);
enforceConstraints();
}
- private void buffer(final K key, final Change<V> value) {
- final long bufferTime = bufferTimeDefinition.time(internalProcessorContext, key);
+ private void buffer(final Record<K, Change<V>> record) {
+ final long bufferTime = bufferTimeDefinition.time(internalProcessorContext, record.key());
- buffer.put(bufferTime, key, value, internalProcessorContext.recordContext());
+ buffer.put(bufferTime, record.key(), record.value(), internalProcessorContext.recordContext());
Review comment:
It seems like these calls could be updated to use the `Record` itself instead of the key, value, and `InternalProcesorContext#recordContext`.
--
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] jeqo commented on a change in pull request #11213: KAFKA-13201: Convert KTable suppress to new PAPI
Posted by GitBox <gi...@apache.org>.
jeqo commented on a change in pull request #11213:
URL: https://github.com/apache/kafka/pull/11213#discussion_r693487470
##########
File path: streams/src/test/java/org/apache/kafka/test/MockInternalNewProcessorContext.java
##########
@@ -0,0 +1,172 @@
+/*
+ * 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.common.utils.Bytes;
+import org.apache.kafka.streams.processor.MockProcessorContext;
Review comment:
Good catch. One of the things I found migrating this mock was that timestamp is not defined.
At the moment I just added a new `setTimestamp` method, but seems rare to set timestamp for the context and set the timestamp for the record.
Will check if setting the timestamp when forwarding be a better approach.
--
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] vvcephei merged pull request #11213: KAFKA-13201: Convert KTable suppress to new PAPI
Posted by GitBox <gi...@apache.org>.
vvcephei merged pull request #11213:
URL: https://github.com/apache/kafka/pull/11213
--
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] vvcephei commented on a change in pull request #11213: KAFKA-13201: Convert KTable suppress to new PAPI
Posted by GitBox <gi...@apache.org>.
vvcephei commented on a change in pull request #11213:
URL: https://github.com/apache/kafka/pull/11213#discussion_r703627436
##########
File path: streams/src/test/java/org/apache/kafka/test/MockInternalNewProcessorContext.java
##########
@@ -0,0 +1,172 @@
+/*
+ * 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.common.utils.Bytes;
+import org.apache.kafka.streams.processor.MockProcessorContext;
Review comment:
Sounds good. I think we can afford to cross that bridge later on as well.
--
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] jeqo commented on a change in pull request #11213: KAFKA-13201: Convert KTable suppress to new PAPI
Posted by GitBox <gi...@apache.org>.
jeqo commented on a change in pull request #11213:
URL: https://github.com/apache/kafka/pull/11213#discussion_r693484448
##########
File path: streams/src/main/java/org/apache/kafka/streams/kstream/internals/suppress/KTableSuppressProcessorSupplier.java
##########
@@ -150,16 +153,16 @@ public void init(final org.apache.kafka.streams.processor.ProcessorContext conte
}
@Override
- public void process(final K key, final Change<V> value) {
+ public void process(final Record<K, Change<V>> record) {
observedStreamTime = Math.max(observedStreamTime, internalProcessorContext.timestamp());
- buffer(key, value);
+ buffer(record);
enforceConstraints();
}
- private void buffer(final K key, final Change<V> value) {
- final long bufferTime = bufferTimeDefinition.time(internalProcessorContext, key);
+ private void buffer(final Record<K, Change<V>> record) {
+ final long bufferTime = bufferTimeDefinition.time(internalProcessorContext, record.key());
- buffer.put(bufferTime, key, value, internalProcessorContext.recordContext());
+ buffer.put(bufferTime, record.key(), record.value(), internalProcessorContext.recordContext());
Review comment:
key and value, sure. For record context, I'm not really sure how to replace it. Its usage its propagated down to `BufferValue` and not only record metadata, but timestamp, and --most importantly--ser/des are used at the bottom.
Replacing it will involve refactoring all the buffering.
--
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] jeqo commented on a change in pull request #11213: KAFKA-13201: Convert KTable suppress to new PAPI
Posted by GitBox <gi...@apache.org>.
jeqo commented on a change in pull request #11213:
URL: https://github.com/apache/kafka/pull/11213#discussion_r693484714
##########
File path: streams/src/test/java/org/apache/kafka/test/MockInternalNewProcessorContext.java
##########
@@ -0,0 +1,172 @@
+/*
+ * 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.common.utils.Bytes;
+import org.apache.kafka.streams.processor.MockProcessorContext;
+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.Record;
+import org.apache.kafka.streams.processor.api.RecordMetadata;
+import org.apache.kafka.streams.processor.internals.InternalProcessorContext;
+import org.apache.kafka.streams.processor.internals.ProcessorNode;
+import org.apache.kafka.streams.processor.internals.ProcessorRecordContext;
+import org.apache.kafka.streams.processor.internals.RecordCollector;
+import org.apache.kafka.streams.processor.internals.StreamTask;
+import org.apache.kafka.streams.processor.internals.Task.TaskType;
+import org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl;
+import org.apache.kafka.streams.state.internals.ThreadCache;
+import org.apache.kafka.streams.state.internals.ThreadCache.DirtyEntryFlushListener;
+
+import java.io.File;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Properties;
+
+public class MockInternalNewProcessorContext<KOut, VOut> extends MockProcessorContext implements InternalProcessorContext<KOut, VOut> {
Review comment:
Agree. There will be space for cleaning up once things are fully migrated.
--
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] vvcephei commented on a change in pull request #11213: KAFKA-13201: Convert KTable suppress to new PAPI
Posted by GitBox <gi...@apache.org>.
vvcephei commented on a change in pull request #11213:
URL: https://github.com/apache/kafka/pull/11213#discussion_r703626660
##########
File path: streams/src/main/java/org/apache/kafka/streams/kstream/internals/suppress/KTableSuppressProcessorSupplier.java
##########
@@ -150,16 +153,16 @@ public void init(final org.apache.kafka.streams.processor.ProcessorContext conte
}
@Override
- public void process(final K key, final Change<V> value) {
+ public void process(final Record<K, Change<V>> record) {
observedStreamTime = Math.max(observedStreamTime, internalProcessorContext.timestamp());
- buffer(key, value);
+ buffer(record);
enforceConstraints();
}
- private void buffer(final K key, final Change<V> value) {
- final long bufferTime = bufferTimeDefinition.time(internalProcessorContext, key);
+ private void buffer(final Record<K, Change<V>> record) {
+ final long bufferTime = bufferTimeDefinition.time(internalProcessorContext, record.key());
- buffer.put(bufferTime, key, value, internalProcessorContext.recordContext());
+ buffer.put(bufferTime, record.key(), record.value(), internalProcessorContext.recordContext());
Review comment:
Thanks, @jeqo , I can see that my tone in that comment didn't encompass the scope of the change I was proposing.
I'm fine if we want to delay that refactor for later work and focus on the Processor API itself in this PR.
--
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] vvcephei commented on a change in pull request #11213: KAFKA-13201: Convert KTable aggregations to new PAPI
Posted by GitBox <gi...@apache.org>.
vvcephei commented on a change in pull request #11213:
URL: https://github.com/apache/kafka/pull/11213#discussion_r690642563
##########
File path: streams/src/test/java/org/apache/kafka/test/MockInternalNewProcessorContext.java
##########
@@ -0,0 +1,172 @@
+/*
+ * 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.common.utils.Bytes;
+import org.apache.kafka.streams.processor.MockProcessorContext;
+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.Record;
+import org.apache.kafka.streams.processor.api.RecordMetadata;
+import org.apache.kafka.streams.processor.internals.InternalProcessorContext;
+import org.apache.kafka.streams.processor.internals.ProcessorNode;
+import org.apache.kafka.streams.processor.internals.ProcessorRecordContext;
+import org.apache.kafka.streams.processor.internals.RecordCollector;
+import org.apache.kafka.streams.processor.internals.StreamTask;
+import org.apache.kafka.streams.processor.internals.Task.TaskType;
+import org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl;
+import org.apache.kafka.streams.state.internals.ThreadCache;
+import org.apache.kafka.streams.state.internals.ThreadCache.DirtyEntryFlushListener;
+
+import java.io.File;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Properties;
+
+public class MockInternalNewProcessorContext<KOut, VOut> extends MockProcessorContext implements InternalProcessorContext<KOut, VOut> {
Review comment:
It looks like this is copied and modified from https://github.com/apache/kafka/blob/trunk/streams/src/test/java/org/apache/kafka/test/MockInternalProcessorContext.java . Is that right?
We also have https://github.com/apache/kafka/blob/trunk/streams/src/test/java/org/apache/kafka/test/InternalMockProcessorContext.java , and it's been a longstanding thorn in our side that there's a proliferation of these context implementations.
I'm hoping that by the time we're done with all these migrations, we can actually converge on this new class and delete the other two.
##########
File path: streams/src/test/java/org/apache/kafka/test/MockInternalNewProcessorContext.java
##########
@@ -0,0 +1,172 @@
+/*
+ * 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.common.utils.Bytes;
+import org.apache.kafka.streams.processor.MockProcessorContext;
Review comment:
This is the one defined in `test-utils`, right? Should we be using the new PAPI one (https://github.com/apache/kafka/blob/trunk/streams/test-utils/src/main/java/org/apache/kafka/streams/processor/api/MockProcessorContext.java) instead?
##########
File path: streams/src/main/java/org/apache/kafka/streams/kstream/internals/suppress/KTableSuppressProcessorSupplier.java
##########
@@ -150,16 +153,16 @@ public void init(final org.apache.kafka.streams.processor.ProcessorContext conte
}
@Override
- public void process(final K key, final Change<V> value) {
+ public void process(final Record<K, Change<V>> record) {
observedStreamTime = Math.max(observedStreamTime, internalProcessorContext.timestamp());
- buffer(key, value);
+ buffer(record);
enforceConstraints();
}
- private void buffer(final K key, final Change<V> value) {
- final long bufferTime = bufferTimeDefinition.time(internalProcessorContext, key);
+ private void buffer(final Record<K, Change<V>> record) {
+ final long bufferTime = bufferTimeDefinition.time(internalProcessorContext, record.key());
- buffer.put(bufferTime, key, value, internalProcessorContext.recordContext());
+ buffer.put(bufferTime, record.key(), record.value(), internalProcessorContext.recordContext());
Review comment:
It seems like these calls could be updated to use the `Record` itself instead of the key, value, and `InternalProcesorContext#recordContext`.
--
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