You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "pegasas (via GitHub)" <gi...@apache.org> on 2024/08/05 07:30:09 UTC
[PR] KAFKA-12822 Remove Deprecated methods can classes of old Processor API [kafka]
pegasas opened a new pull request, #16791:
URL: https://github.com/apache/kafka/pull/16791
related to https://issues.apache.org/jira/browse/KAFKA-12829
remove:
org.apache.kafka.streams.Topology#addProcessor(java.lang.String, org.apache.kafka.streams.processor.ProcessorSupplier, java.lang.String...)
org.apache.kafka.streams.Topology#addGlobalStore(org.apache.kafka.streams.state.StoreBuilder<?>, java.lang.String,
org.apache.kafka.streams.Topology#addGlobalStore(org.apache.kafka.streams.state.StoreBuilder<?>, java.lang.String,
### 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
Re: [PR] KAFKA-12822 Remove Deprecated methods can classes of old Processor API [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #16791:
URL: https://github.com/apache/kafka/pull/16791#discussion_r1704715322
##########
streams/src/main/java/org/apache/kafka/streams/processor/ProcessorContext.java:
##########
@@ -292,7 +292,7 @@ Cancellable schedule(final Duration interval,
* (including the currently processed record), i.e., it can be considered a high-watermark.
* Stream-time is tracked on a per-task basis and is preserved across restarts and during task migration.
*
- * <p> Note: this method is not supported for global processors (cf. {@link Topology#addGlobalStore} (...)
Review Comment:
There is still a `addGlobalStore()` method we should link look
##########
streams/src/main/java/org/apache/kafka/streams/processor/api/ProcessingContext.java:
##########
@@ -201,7 +201,7 @@ Cancellable schedule(final Duration interval,
* (including the currently processed record), i.e., it can be considered a high-watermark.
* Stream-time is tracked on a per-task basis and is preserved across restarts and during task migration.
*
- * <p> Note: this method is not supported for global processors (cf. {@link Topology#addGlobalStore} (...)
+ * <p> Note: this method is not supported for global processors (...)
* and {@link StreamsBuilder#addGlobalStore} (...),
Review Comment:
As above.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-12829: Remove deprecated methods and classes of old Processor API [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on PR #16791:
URL: https://github.com/apache/kafka/pull/16791#issuecomment-2308985631
The build has checkstyle errors:
```
> Task :streams:checkstyleMain
--
| 2993 | 09:39:12 AM | [ant:checkstyle] [ERROR] /home/jenkins/jenkins-agent/workspace/Kafka_kafka-pr_PR-16791/streams/src/main/java/org/apache/kafka/streams/processor/ProcessorContext.java:23:8: Unused import - org.apache.kafka.streams.Topology. [UnusedImports]
| 2994 | 09:39:12 AM | [ant:checkstyle] [ERROR] /home/jenkins/jenkins-agent/workspace/Kafka_kafka-pr_PR-16791/streams/src/main/java/org/apache/kafka/streams/processor/api/ProcessingContext.java:22:8: Unused import - org.apache.kafka.streams.Topology. [UnusedImports]
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-12829: Remove deprecated methods and classes of old Processor API [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #16791:
URL: https://github.com/apache/kafka/pull/16791#discussion_r1730426306
##########
streams/src/main/java/org/apache/kafka/streams/processor/ProcessorContext.java:
##########
@@ -292,8 +292,7 @@ Cancellable schedule(final Duration interval,
* (including the currently processed record), i.e., it can be considered a high-watermark.
* Stream-time is tracked on a per-task basis and is preserved across restarts and during task migration.
*
- * <p> Note: this method is not supported for global processors (cf. {@link Topology#addGlobalStore} (...)
- * and {@link StreamsBuilder#addGlobalStore} (...),
+ * <p> Note: this method is not supported for global processors {@link StreamsBuilder#addGlobalStore} (...)
Review Comment:
```suggestion
* <p> Note: this method is not supported for global processors {@link Topology#addGlobalStore(StoreBuilder, String, Deserializer, Deserializer, String, String, ProcessorSupplier)}```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-12829: Remove deprecated methods and classes of old Processor API [kafka]
Posted by "pegasas (via GitHub)" <gi...@apache.org>.
pegasas commented on code in PR #16791:
URL: https://github.com/apache/kafka/pull/16791#discussion_r1730390328
##########
streams/src/main/java/org/apache/kafka/streams/Topology.java:
##########
@@ -820,110 +778,6 @@ public synchronized <KIn, VIn> Topology addReadOnlyStateStore(final StoreBuilder
);
}
- /**
- * Adds a global {@link StateStore} to the topology.
- * The {@link StateStore} sources its data from all partitions of the provided input topic.
- * There will be exactly one instance of this {@link StateStore} per Kafka Streams instance.
- * <p>
- * A {@link SourceNode} with the provided sourceName will be added to consume the data arriving from the partitions
- * of the input topic.
- * <p>
- * The provided {@link org.apache.kafka.streams.processor.ProcessorSupplier} will be used to create an {@link ProcessorNode} that will receive all
- * records forwarded from the {@link SourceNode}.
- * The supplier should always generate a new instance each time
- * {@link org.apache.kafka.streams.processor.ProcessorSupplier#get()} gets called. Creating a single
- * {@link org.apache.kafka.streams.processor.Processor} object and returning the same object reference in
- * {@link org.apache.kafka.streams.processor.ProcessorSupplier#get()} would be a violation of the supplier pattern
- * and leads to runtime exceptions.
- * This {@link ProcessorNode} should be used to keep the {@link StateStore} up-to-date.
- * The default {@link TimestampExtractor} as specified in the {@link StreamsConfig config} is used.
- *
- * @param storeBuilder user defined state store builder
- * @param sourceName name of the {@link SourceNode} that will be automatically added
- * @param keyDeserializer the {@link Deserializer} to deserialize keys with
- * @param valueDeserializer the {@link Deserializer} to deserialize values with
- * @param topic the topic to source the data from
- * @param processorName the name of the {@link org.apache.kafka.streams.processor.ProcessorSupplier}
- * @param stateUpdateSupplier the instance of {@link org.apache.kafka.streams.processor.ProcessorSupplier}
- * @return itself
- * @throws TopologyException if the processor of state is already registered
- * @deprecated Since 2.7.0. Use {@link #addGlobalStore(StoreBuilder, String, Deserializer, Deserializer, String, String, ProcessorSupplier)} instead.
- */
- @Deprecated
- public synchronized <K, V> Topology addGlobalStore(final StoreBuilder<?> storeBuilder,
- final String sourceName,
- final Deserializer<K> keyDeserializer,
- final Deserializer<V> valueDeserializer,
- final String topic,
- final String processorName,
- final org.apache.kafka.streams.processor.ProcessorSupplier<K, V> stateUpdateSupplier) {
- internalTopologyBuilder.addGlobalStore(
- new StoreBuilderWrapper(storeBuilder),
- sourceName,
- null,
- keyDeserializer,
- valueDeserializer,
- topic,
- processorName,
- () -> ProcessorAdapter.adapt(stateUpdateSupplier.get()),
- true
- );
- return this;
- }
-
- /**
- * Adds a global {@link StateStore} to the topology.
- * The {@link StateStore} sources its data from all partitions of the provided input topic.
- * There will be exactly one instance of this {@link StateStore} per Kafka Streams instance.
- * <p>
- * A {@link SourceNode} with the provided sourceName will be added to consume the data arriving from the partitions
- * of the input topic.
- * <p>
- * The provided {@link org.apache.kafka.streams.processor.ProcessorSupplier} will be used to create an {@link ProcessorNode} that will receive all
- * records forwarded from the {@link SourceNode}.
- * The supplier should always generate a new instance each time
- * {@link org.apache.kafka.streams.processor.ProcessorSupplier#get()} gets called. Creating a single
- * {@link org.apache.kafka.streams.processor.Processor} object and returning the same object reference in
- * {@link org.apache.kafka.streams.processor.ProcessorSupplier#get()} would be a violation of the supplier pattern
- * and leads to runtime exceptions.
- * This {@link ProcessorNode} should be used to keep the {@link StateStore} up-to-date.
- *
- * @param storeBuilder user defined key value store builder
- * @param sourceName name of the {@link SourceNode} that will be automatically added
- * @param timestampExtractor the stateless timestamp extractor used for this source,
- * if not specified the default extractor defined in the configs will be used
- * @param keyDeserializer the {@link Deserializer} to deserialize keys with
- * @param valueDeserializer the {@link Deserializer} to deserialize values with
- * @param topic the topic to source the data from
- * @param processorName the name of the {@link org.apache.kafka.streams.processor.ProcessorSupplier}
- * @param stateUpdateSupplier the instance of {@link org.apache.kafka.streams.processor.ProcessorSupplier}
- * @return itself
- * @throws TopologyException if the processor of state is already registered
- * @deprecated Since 2.7.0. Use {@link #addGlobalStore(StoreBuilder, String, TimestampExtractor, Deserializer, Deserializer, String, String, ProcessorSupplier)} instead.
- */
- @Deprecated
- public synchronized <K, V> Topology addGlobalStore(final StoreBuilder<?> storeBuilder,
- final String sourceName,
- final TimestampExtractor timestampExtractor,
- final Deserializer<K> keyDeserializer,
- final Deserializer<V> valueDeserializer,
- final String topic,
- final String processorName,
- final org.apache.kafka.streams.processor.ProcessorSupplier<K, V> stateUpdateSupplier) {
- internalTopologyBuilder.addGlobalStore(
- new StoreBuilderWrapper(storeBuilder),
- sourceName,
- timestampExtractor,
- keyDeserializer,
- valueDeserializer,
- topic,
- processorName,
- () -> ProcessorAdapter.adapt(stateUpdateSupplier.get()),
Review Comment:
I agree with that. it seems a lot of interfaces has example/code reference. we should divide and conquer.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-12829: Remove deprecated Topology#addGlobalStore of old Processor API [kafka]
Posted by "pegasas (via GitHub)" <gi...@apache.org>.
pegasas commented on PR #16791:
URL: https://github.com/apache/kafka/pull/16791#issuecomment-2309191228
> The build has checkstyle errors:
>
> ```
> > Task :streams:checkstyleMain
> --
> | 2993 | 09:39:12 AM | [ant:checkstyle] [ERROR] /home/jenkins/jenkins-agent/workspace/Kafka_kafka-pr_PR-16791/streams/src/main/java/org/apache/kafka/streams/processor/ProcessorContext.java:23:8: Unused import - org.apache.kafka.streams.Topology. [UnusedImports]
> | 2994 | 09:39:12 AM | [ant:checkstyle] [ERROR] /home/jenkins/jenkins-agent/workspace/Kafka_kafka-pr_PR-16791/streams/src/main/java/org/apache/kafka/streams/processor/api/ProcessingContext.java:22:8: Unused import - org.apache.kafka.streams.Topology. [UnusedImports]
> ```
fixed these checkstyle errors.
Is there any commands with gradle which I can run in local? like mvn spotless:apply or something?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-12829: Remove deprecated methods and classes of old Processor API [kafka]
Posted by "pegasas (via GitHub)" <gi...@apache.org>.
pegasas commented on PR #16791:
URL: https://github.com/apache/kafka/pull/16791#issuecomment-2308913465
> @pegasas -- Any updates on this PR?
org.apache.kafka.streams.Topology#addProcessor(java.lang.String, org.apache.kafka.streams.processor.ProcessorSupplier, java.lang.String...)
It looks like this method still uses by examples. shall we remove it in this loop?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-12829: Remove deprecated Topology#addGlobalStore of old Processor API [kafka]
Posted by "pegasas (via GitHub)" <gi...@apache.org>.
pegasas commented on PR #16791:
URL: https://github.com/apache/kafka/pull/16791#issuecomment-2316714262
> > It seems not work on my local
>
> Why not fix it manually?
Done.
It seems I can run `./gradlew --build-cache --info --scan check -x test` to check style.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-12822 Remove Deprecated methods can classes of old Processor API [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #16791:
URL: https://github.com/apache/kafka/pull/16791#discussion_r1704715322
##########
streams/src/main/java/org/apache/kafka/streams/processor/ProcessorContext.java:
##########
@@ -292,7 +292,7 @@ Cancellable schedule(final Duration interval,
* (including the currently processed record), i.e., it can be considered a high-watermark.
* Stream-time is tracked on a per-task basis and is preserved across restarts and during task migration.
*
- * <p> Note: this method is not supported for global processors (cf. {@link Topology#addGlobalStore} (...)
Review Comment:
There is still a `addGlobalStore()` method we should link look (the original link was broken anyway -- we should just fix it)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-12829: Remove deprecated Topology#addGlobalStore of old Processor API [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on PR #16791:
URL: https://github.com/apache/kafka/pull/16791#issuecomment-2316316508
> It seems not work on my local
Why not fix it manually?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-12829: Remove deprecated methods and classes of old Processor API [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on PR #16791:
URL: https://github.com/apache/kafka/pull/16791#issuecomment-2308982763
There is two overloads. The old one, using `org.apache.kafka.streams.processor.ProcessorSupplier<K, V>`
```
org.apache.kafka.streams.Topology#addProcessor(java.lang.String, org.apache.kafka.streams.processor.ProcessorSupplier, java.lang.String...)
```
And the new one using `org.apache.kafka.streams.processor.api.ProcessorSupplier<KIn, VIn, KOut, VOut>`
```
org.apache.kafka.streams.Topology#addProcessor(java.lang.String, org.apache.kafka.streams.processor.ProcessorSupplier<KIn, VIn, KOut, VOut>, java.lang.String...)
```
Note the different package names. Only the old one is deprecated is should be removed, and as far as I can tell, it's only used in some test (which can also be removed).
The screenshot you posted seem to refer to the new one though.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-12829: Remove deprecated methods and classes of old Processor API [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #16791:
URL: https://github.com/apache/kafka/pull/16791#discussion_r1730426457
##########
streams/src/main/java/org/apache/kafka/streams/processor/api/ProcessingContext.java:
##########
@@ -201,8 +201,7 @@ Cancellable schedule(final Duration interval,
* (including the currently processed record), i.e., it can be considered a high-watermark.
* Stream-time is tracked on a per-task basis and is preserved across restarts and during task migration.
*
- * <p> Note: this method is not supported for global processors (cf. {@link Topology#addGlobalStore} (...)
- * and {@link StreamsBuilder#addGlobalStore} (...),
+ * <p> Note: this method is not supported for global processors {@link StreamsBuilder#addGlobalStore} (...)
Review Comment:
```suggestion
* <p> Note: this method is not supported for global processors {@link Topology#addGlobalStore(StoreBuilder, String, Deserializer, Deserializer, String, String, ProcessorSupplier)}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-12829: Remove deprecated Topology#addGlobalStore of old Processor API [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on PR #16791:
URL: https://github.com/apache/kafka/pull/16791#issuecomment-2319467424
> This is because ci pending costs too much time.
Well, you should be able to reproduce checkstyle error locally -- "apply" means to auto-update the code, but just running `./gradlew checkstyleMain chechstyleTest spotbugsMain spotbugsTest` should give you all errors locally and you could update the code manually to address them.
Anyway. Seems you found a way and the last build did not show any error :)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-12829: Remove deprecated methods and classes of old Processor API [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #16791:
URL: https://github.com/apache/kafka/pull/16791#discussion_r1730426384
##########
streams/src/main/java/org/apache/kafka/streams/processor/ProcessorContext.java:
##########
@@ -292,8 +292,7 @@ Cancellable schedule(final Duration interval,
* (including the currently processed record), i.e., it can be considered a high-watermark.
* Stream-time is tracked on a per-task basis and is preserved across restarts and during task migration.
*
- * <p> Note: this method is not supported for global processors (cf. {@link Topology#addGlobalStore} (...)
- * and {@link StreamsBuilder#addGlobalStore} (...),
+ * <p> Note: this method is not supported for global processors {@link StreamsBuilder#addGlobalStore} (...)
Review Comment:
This not correct. Should be something like this ^
##########
streams/src/main/java/org/apache/kafka/streams/processor/ProcessorContext.java:
##########
@@ -292,8 +292,7 @@ Cancellable schedule(final Duration interval,
* (including the currently processed record), i.e., it can be considered a high-watermark.
* Stream-time is tracked on a per-task basis and is preserved across restarts and during task migration.
*
- * <p> Note: this method is not supported for global processors (cf. {@link Topology#addGlobalStore} (...)
- * and {@link StreamsBuilder#addGlobalStore} (...),
+ * <p> Note: this method is not supported for global processors {@link StreamsBuilder#addGlobalStore} (...)
Review Comment:
```suggestion
* <p> Note: this method is not supported for global processors {@link Topology#addGlobalStore(StoreBuilder, String, Deserializer, Deserializer, String, String, ProcessorSupplier)}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-12829: Remove deprecated Topology#addGlobalStore of old Processor API [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on PR #16791:
URL: https://github.com/apache/kafka/pull/16791#issuecomment-2313923638
From the build output:
```
FAILURE: Build completed with 2 failures.
--
| 3850 | 07:23:42 PM |
| 3851 | 07:23:42 PM | 1: Task failed with an exception.
| 3852 | 07:23:42 PM | -----------
| 3853 | 07:23:42 PM | * What went wrong:
| 3854 | 07:23:42 PM | Execution failed for task ':streams:spotlessJavaCheck'.
| 3855 | 07:23:42 PM | > The following files had format violations:
| 3856 | 07:23:42 PM | src/main/java/org/apache/kafka/streams/processor/api/ProcessingContext.java
| 3857 | 07:23:42 PM | @@ -17,7 +17,6 @@
| 3858 | 07:23:42 PM | package·org.apache.kafka.streams.processor.api;
| 3859 | 07:23:42 PM |
more_vert | 3860 | 07:23:42 PM | import·org.apache.kafka.common.serialization.Serde;
| 3861 | 07:23:42 PM | -import·org.apache.kafka.streams.StreamsBuilder;
| 3862 | 07:23:42 PM | import·org.apache.kafka.streams.StreamsMetrics;
| 3863 | 07:23:42 PM | import·org.apache.kafka.streams.Topology;
| 3864 | 07:23:42 PM | import·org.apache.kafka.streams.processor.Cancellable;
| 3865 | 07:23:42 PM | Run './gradlew :streams:spotlessApply' to fix these violations.
```
If the readme is outdated, could you do a new PR to update it?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-12829: Remove deprecated Topology#addGlobalStore of old Processor API [kafka]
Posted by "pegasas (via GitHub)" <gi...@apache.org>.
pegasas commented on PR #16791:
URL: https://github.com/apache/kafka/pull/16791#issuecomment-2315828618
> From the build output:
>
> ```
>
> FAILURE: Build completed with 2 failures.
> --
> | 3850 | 07:23:42 PM |
> | 3851 | 07:23:42 PM | 1: Task failed with an exception.
> | 3852 | 07:23:42 PM | -----------
> | 3853 | 07:23:42 PM | * What went wrong:
> | 3854 | 07:23:42 PM | Execution failed for task ':streams:spotlessJavaCheck'.
> | 3855 | 07:23:42 PM | > The following files had format violations:
> | 3856 | 07:23:42 PM | src/main/java/org/apache/kafka/streams/processor/api/ProcessingContext.java
> | 3857 | 07:23:42 PM | @@ -17,7 +17,6 @@
> | 3858 | 07:23:42 PM | package·org.apache.kafka.streams.processor.api;
> | 3859 | 07:23:42 PM |
> more_vert | 3860 | 07:23:42 PM | import·org.apache.kafka.common.serialization.Serde;
> | 3861 | 07:23:42 PM | -import·org.apache.kafka.streams.StreamsBuilder;
> | 3862 | 07:23:42 PM | import·org.apache.kafka.streams.StreamsMetrics;
> | 3863 | 07:23:42 PM | import·org.apache.kafka.streams.Topology;
> | 3864 | 07:23:42 PM | import·org.apache.kafka.streams.processor.Cancellable;
> | 3865 | 07:23:42 PM | Run './gradlew :streams:spotlessApply' to fix these violations.
> ```
>
> If the readme is outdated, could you do a new PR to update it?
It seems not work on my local. I will dig if there is other solution. It seems there is no special action in .github/build.yml
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-12829: Remove deprecated Topology#addGlobalStore of old Processor API [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #16791:
URL: https://github.com/apache/kafka/pull/16791#discussion_r1731824083
##########
streams/src/main/java/org/apache/kafka/streams/processor/ProcessorContext.java:
##########
@@ -292,8 +290,7 @@ Cancellable schedule(final Duration interval,
* (including the currently processed record), i.e., it can be considered a high-watermark.
* Stream-time is tracked on a per-task basis and is preserved across restarts and during task migration.
*
- * <p> Note: this method is not supported for global processors (cf. {@link Topology#addGlobalStore} (...)
- * and {@link StreamsBuilder#addGlobalStore} (...),
+ * <p> Note: this method is not supported for global processors
Review Comment:
```suggestion
* <p> Note: this method is not supported for global processors (cf. {@link Topology#addGlobalStore(StoreBuilder, String, Deserializer, Deserializer, String, String, ProcessorSupplier)})
```
##########
streams/src/main/java/org/apache/kafka/streams/processor/api/ProcessingContext.java:
##########
@@ -201,8 +201,7 @@ Cancellable schedule(final Duration interval,
* (including the currently processed record), i.e., it can be considered a high-watermark.
* Stream-time is tracked on a per-task basis and is preserved across restarts and during task migration.
*
- * <p> Note: this method is not supported for global processors (cf. {@link Topology#addGlobalStore} (...)
- * and {@link StreamsBuilder#addGlobalStore} (...),
+ * <p> Note: this method is not supported for global processors {@link Topology#addGlobalStore(StoreBuilder, String, Deserializer, Deserializer, String, String, ProcessorSupplier)}
Review Comment:
```suggestion
* <p> Note: this method is not supported for global processors (cf. {@link Topology#addGlobalStore(StoreBuilder, String, Deserializer, Deserializer, String, String, ProcessorSupplier)})
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-12829: Remove deprecated Topology#addGlobalStore of old Processor API [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on PR #16791:
URL: https://github.com/apache/kafka/pull/16791#issuecomment-2311081736
Yes, check the readme for gradle commands: https://github.com/apache/kafka
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-12829: Remove deprecated methods and classes of old Processor API [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on PR #16791:
URL: https://github.com/apache/kafka/pull/16791#issuecomment-2305508160
@pegasas -- Any updates on 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
Re: [PR] KAFKA-12822 Remove Deprecated methods can classes of old Processor API [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #16791:
URL: https://github.com/apache/kafka/pull/16791#discussion_r1704716420
##########
streams/src/main/java/org/apache/kafka/streams/Topology.java:
##########
@@ -820,110 +778,6 @@ public synchronized <KIn, VIn> Topology addReadOnlyStateStore(final StoreBuilder
);
}
- /**
- * Adds a global {@link StateStore} to the topology.
- * The {@link StateStore} sources its data from all partitions of the provided input topic.
- * There will be exactly one instance of this {@link StateStore} per Kafka Streams instance.
- * <p>
- * A {@link SourceNode} with the provided sourceName will be added to consume the data arriving from the partitions
- * of the input topic.
- * <p>
- * The provided {@link org.apache.kafka.streams.processor.ProcessorSupplier} will be used to create an {@link ProcessorNode} that will receive all
- * records forwarded from the {@link SourceNode}.
- * The supplier should always generate a new instance each time
- * {@link org.apache.kafka.streams.processor.ProcessorSupplier#get()} gets called. Creating a single
- * {@link org.apache.kafka.streams.processor.Processor} object and returning the same object reference in
- * {@link org.apache.kafka.streams.processor.ProcessorSupplier#get()} would be a violation of the supplier pattern
- * and leads to runtime exceptions.
- * This {@link ProcessorNode} should be used to keep the {@link StateStore} up-to-date.
- * The default {@link TimestampExtractor} as specified in the {@link StreamsConfig config} is used.
- *
- * @param storeBuilder user defined state store builder
- * @param sourceName name of the {@link SourceNode} that will be automatically added
- * @param keyDeserializer the {@link Deserializer} to deserialize keys with
- * @param valueDeserializer the {@link Deserializer} to deserialize values with
- * @param topic the topic to source the data from
- * @param processorName the name of the {@link org.apache.kafka.streams.processor.ProcessorSupplier}
- * @param stateUpdateSupplier the instance of {@link org.apache.kafka.streams.processor.ProcessorSupplier}
- * @return itself
- * @throws TopologyException if the processor of state is already registered
- * @deprecated Since 2.7.0. Use {@link #addGlobalStore(StoreBuilder, String, Deserializer, Deserializer, String, String, ProcessorSupplier)} instead.
- */
- @Deprecated
- public synchronized <K, V> Topology addGlobalStore(final StoreBuilder<?> storeBuilder,
- final String sourceName,
- final Deserializer<K> keyDeserializer,
- final Deserializer<V> valueDeserializer,
- final String topic,
- final String processorName,
- final org.apache.kafka.streams.processor.ProcessorSupplier<K, V> stateUpdateSupplier) {
- internalTopologyBuilder.addGlobalStore(
- new StoreBuilderWrapper(storeBuilder),
- sourceName,
- null,
- keyDeserializer,
- valueDeserializer,
- topic,
- processorName,
- () -> ProcessorAdapter.adapt(stateUpdateSupplier.get()),
- true
- );
- return this;
- }
-
- /**
- * Adds a global {@link StateStore} to the topology.
- * The {@link StateStore} sources its data from all partitions of the provided input topic.
- * There will be exactly one instance of this {@link StateStore} per Kafka Streams instance.
- * <p>
- * A {@link SourceNode} with the provided sourceName will be added to consume the data arriving from the partitions
- * of the input topic.
- * <p>
- * The provided {@link org.apache.kafka.streams.processor.ProcessorSupplier} will be used to create an {@link ProcessorNode} that will receive all
- * records forwarded from the {@link SourceNode}.
- * The supplier should always generate a new instance each time
- * {@link org.apache.kafka.streams.processor.ProcessorSupplier#get()} gets called. Creating a single
- * {@link org.apache.kafka.streams.processor.Processor} object and returning the same object reference in
- * {@link org.apache.kafka.streams.processor.ProcessorSupplier#get()} would be a violation of the supplier pattern
- * and leads to runtime exceptions.
- * This {@link ProcessorNode} should be used to keep the {@link StateStore} up-to-date.
- *
- * @param storeBuilder user defined key value store builder
- * @param sourceName name of the {@link SourceNode} that will be automatically added
- * @param timestampExtractor the stateless timestamp extractor used for this source,
- * if not specified the default extractor defined in the configs will be used
- * @param keyDeserializer the {@link Deserializer} to deserialize keys with
- * @param valueDeserializer the {@link Deserializer} to deserialize values with
- * @param topic the topic to source the data from
- * @param processorName the name of the {@link org.apache.kafka.streams.processor.ProcessorSupplier}
- * @param stateUpdateSupplier the instance of {@link org.apache.kafka.streams.processor.ProcessorSupplier}
- * @return itself
- * @throws TopologyException if the processor of state is already registered
- * @deprecated Since 2.7.0. Use {@link #addGlobalStore(StoreBuilder, String, TimestampExtractor, Deserializer, Deserializer, String, String, ProcessorSupplier)} instead.
- */
- @Deprecated
- public synchronized <K, V> Topology addGlobalStore(final StoreBuilder<?> storeBuilder,
- final String sourceName,
- final TimestampExtractor timestampExtractor,
- final Deserializer<K> keyDeserializer,
- final Deserializer<V> valueDeserializer,
- final String topic,
- final String processorName,
- final org.apache.kafka.streams.processor.ProcessorSupplier<K, V> stateUpdateSupplier) {
- internalTopologyBuilder.addGlobalStore(
- new StoreBuilderWrapper(storeBuilder),
- sourceName,
- timestampExtractor,
- keyDeserializer,
- valueDeserializer,
- topic,
- processorName,
- () -> ProcessorAdapter.adapt(stateUpdateSupplier.get()),
Review Comment:
Do we use `ProcessorAdapter.adapt(...)` elsewhere? If not, we can remove it, too. (Maybe if a follow up PR if this PR becomes too big -- it's simple, but it's already more than 700 LOC...)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-12829: Remove deprecated methods and classes of old Processor API [kafka]
Posted by "pegasas (via GitHub)" <gi...@apache.org>.
pegasas commented on code in PR #16791:
URL: https://github.com/apache/kafka/pull/16791#discussion_r1730392424
##########
streams/src/main/java/org/apache/kafka/streams/processor/ProcessorContext.java:
##########
@@ -292,7 +292,7 @@ Cancellable schedule(final Duration interval,
* (including the currently processed record), i.e., it can be considered a high-watermark.
* Stream-time is tracked on a per-task basis and is preserved across restarts and during task migration.
*
- * <p> Note: this method is not supported for global processors (cf. {@link Topology#addGlobalStore} (...)
Review Comment:
Done.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-12829: Remove deprecated Topology#addGlobalStore of old Processor API [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #16791:
URL: https://github.com/apache/kafka/pull/16791#discussion_r1733767187
##########
streams/src/main/java/org/apache/kafka/streams/processor/ProcessorContext.java:
##########
@@ -292,8 +290,7 @@ Cancellable schedule(final Duration interval,
* (including the currently processed record), i.e., it can be considered a high-watermark.
* Stream-time is tracked on a per-task basis and is preserved across restarts and during task migration.
*
- * <p> Note: this method is not supported for global processors (cf. {@link Topology#addGlobalStore} (...)
- * and {@link StreamsBuilder#addGlobalStore} (...),
+ * <p> Note: this method is not supported for global processors
Review Comment:
This PR remove the old `addGlobalStore(...)` but we should be able to link to the new `addGlobalStore(...)`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-12829: Remove deprecated Topology#addGlobalStore of old Processor API [kafka]
Posted by "pegasas (via GitHub)" <gi...@apache.org>.
pegasas commented on PR #16791:
URL: https://github.com/apache/kafka/pull/16791#issuecomment-2311398352
https://github.com/apache/kafka/blob/trunk/.github/workflows/build.yml#L74
It looks like origin command './gradlew checkstyleMain checkstyleTest spotlessCheck --info' not work now, and new py script also could not be executed in my linux notebook.
Not sure if it is in a stable status currently.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-12829: Remove deprecated Topology#addGlobalStore of old Processor API [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on PR #16791:
URL: https://github.com/apache/kafka/pull/16791#issuecomment-2311091019
There is still checkstyle errors.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-12829: Remove deprecated Topology#addGlobalStore of old Processor API [kafka]
Posted by "pegasas (via GitHub)" <gi...@apache.org>.
pegasas commented on code in PR #16791:
URL: https://github.com/apache/kafka/pull/16791#discussion_r1732026055
##########
streams/src/main/java/org/apache/kafka/streams/processor/ProcessorContext.java:
##########
@@ -292,8 +290,7 @@ Cancellable schedule(final Duration interval,
* (including the currently processed record), i.e., it can be considered a high-watermark.
* Stream-time is tracked on a per-task basis and is preserved across restarts and during task migration.
*
- * <p> Note: this method is not supported for global processors (cf. {@link Topology#addGlobalStore} (...)
- * and {@link StreamsBuilder#addGlobalStore} (...),
+ * <p> Note: this method is not supported for global processors
Review Comment:
this will break checkstyle, I revert this change.
due to Topology#addGlobalStore has been removed 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
Re: [PR] KAFKA-12829: Remove deprecated Topology#addGlobalStore of old Processor API [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on PR #16791:
URL: https://github.com/apache/kafka/pull/16791#issuecomment-2319484722
Thanks for the PR. Merged to `trunk`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-12829: Remove deprecated Topology#addGlobalStore of old Processor API [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax merged PR #16791:
URL: https://github.com/apache/kafka/pull/16791
--
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