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 09:59:23 UTC
[GitHub] [kafka] soarez opened a new pull request #9000: KAFKA-10036 Improve handling and documentation of Suppliers
soarez opened a new pull request #9000:
URL: https://github.com/apache/kafka/pull/9000
Following up on #8752 which seems to have gone stale.
@mjsax can you continue the review?
----------------------------------------------------------------
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] mjsax commented on pull request #9000: KAFKA-10036 Improve handling and documentation of Suppliers
Posted by GitBox <gi...@apache.org>.
mjsax commented on pull request #9000:
URL: https://github.com/apache/kafka/pull/9000#issuecomment-660280418
@soarez Is this PR a replacement of #8752 and we should close the other 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] mjsax commented on a change in pull request #9000: KAFKA-10036 Improve handling and documentation of Suppliers
Posted by GitBox <gi...@apache.org>.
mjsax commented on a change in pull request #9000:
URL: https://github.com/apache/kafka/pull/9000#discussion_r464749727
##########
File path: streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamUtil.java
##########
@@ -0,0 +1,38 @@
+/*
+ * 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.kstream.internals;
+
+import org.apache.kafka.streams.kstream.TransformerSupplier;
+
+/**
+ * Shared functions to handle verifications of a valid {@link org.apache.kafka.streams.kstream.KStream}.
+ */
+final class KStreamUtil {
+
+ private KStreamUtil() {}
+
+ /**
+ * @throws IllegalArgumentException if the same transformer instance is obtained each time
+ */
+ static void checkTransformerSupplier(final TransformerSupplier<?, ?, ?> supplier) {
+ if (supplier.get() == supplier.get()) {
+ throw new IllegalArgumentException("TransformerSupplier generates single transformer reference. Supplier " +
+ "pattern violated.");
Review comment:
I think we can be more elaborate and add something like `TransformerSupplier#get() must return a new Transformer object each time it is called.`
##########
File path: streams/src/test/java/org/apache/kafka/streams/processor/internals/InternalTopologyBuilderTest.java
##########
@@ -175,6 +177,12 @@ public void testAddProcessorWithNullParents() {
builder.addProcessor("processor", new MockProcessorSupplier<>(), (String) null);
}
+ @Test(expected = TopologyException.class)
+ public void testAddProcessorWithBadSupplier() {
Review comment:
We should also test `addGlobalStateStore()`
##########
File path: streams/src/main/java/org/apache/kafka/streams/StreamsBuilder.java
##########
@@ -521,6 +521,10 @@ public synchronized StreamsBuilder addStateStore(final StoreBuilder<?> builder)
* <p>
* It is not required to connect a global store to {@link Processor Processors}, {@link Transformer Transformers},
* or {@link ValueTransformer ValueTransformer}; those have read-only access to all global stores by default.
+ * <p>
+ * The supplier should always generate a new instance each time invoking {@link ProcessorSupplier#get()}. Creating
Review comment:
nit `each time {@link ProcessorSupplier#get()} is called.` (similar elsewhere -- please fix throughout the whole PR)
##########
File path: streams/src/main/java/org/apache/kafka/streams/StreamsBuilder.java
##########
@@ -521,6 +521,10 @@ public synchronized StreamsBuilder addStateStore(final StoreBuilder<?> builder)
* <p>
* It is not required to connect a global store to {@link Processor Processors}, {@link Transformer Transformers},
* or {@link ValueTransformer ValueTransformer}; those have read-only access to all global stores by default.
+ * <p>
+ * The supplier should always generate a new instance each time invoking {@link ProcessorSupplier#get()}. Creating
+ * a single Processor object and returning the same object reference in {@link ProcessorSupplier#get()} would be
Review comment:
`{@link Processor}`(similar elsewhere -- please fix throughout the whole PR)
##########
File path: streams/src/main/java/org/apache/kafka/streams/kstream/KStream.java
##########
@@ -2522,8 +2522,12 @@ void to(final TopicNameExtractor<K, V> topicExtractor,
* If in {@link Transformer#transform(Object, Object) Transformer#transform()} multiple records need to be emitted
* for each input record, it is recommended to use {@link #flatTransform(TransformerSupplier, String...)
* flatTransform()}.
+ * The supplier should always generate a new instance each time invoking {@link TransformerSupplier#get()}. Creating
+ * a single Transformer object and returning the same object reference in {@link TransformerSupplier#get()} would be
Review comment:
`{@link Transformer}` (also elsewhere)
##########
File path: streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamImplTest.java
##########
@@ -1865,6 +1865,17 @@ public void shouldNotAllowNullNamedOnTransformWithStoreName() {
assertThat(exception.getMessage(), equalTo("named can't be null"));
}
+ @Test
+ public void shouldNotAllowBadTransformerSupplierOnFlatTransform() {
Review comment:
It might be nice to replicate this test for all "siblings" of `flatTransform` (including the "Value" variants). -- Also note that there is `KStream#process()` and `StreamsBuilder#addGlobalStateStore), too, that should also be tested.
##########
File path: streams/src/test/java/org/apache/kafka/streams/processor/internals/InternalTopologyBuilderTest.java
##########
@@ -175,6 +177,12 @@ public void testAddProcessorWithNullParents() {
builder.addProcessor("processor", new MockProcessorSupplier<>(), (String) null);
}
+ @Test(expected = TopologyException.class)
+ public void testAddProcessorWithBadSupplier() {
+ final Processor<Object, Object> processor = new MockProcessor<>();
+ builder.addProcessor("processor", () -> processor, (String) null);
Review comment:
We should use `aasertThrows` and verify the error message similar to the `TransformerSupplier` test
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/TopologyUtil.java
##########
@@ -0,0 +1,39 @@
+/*
+ * 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.errors.TopologyException;
+import org.apache.kafka.streams.processor.ProcessorSupplier;
+
+/**
+ * Shared functions to handle verifications of a valid {@link org.apache.kafka.streams.Topology}.
+ */
+public final class TopologyUtil {
+
+ private TopologyUtil() {}
+
+ /**
+ * @throws TopologyException if the same processor instance is obtained each time
+ */
+ public static void checkProcessorSupplier(final ProcessorSupplier<?, ?> supplier) {
+ if (supplier.get() == supplier.get()) {
+ throw new TopologyException("ProcessorSupplier generates single processor reference. Supplier pattern" +
+ " violated.");
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] soarez commented on a change in pull request #9000: KAFKA-10036 Improve handling and documentation of Suppliers
Posted by GitBox <gi...@apache.org>.
soarez commented on a change in pull request #9000:
URL: https://github.com/apache/kafka/pull/9000#discussion_r466080451
##########
File path: streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamImplTest.java
##########
@@ -1865,6 +1865,17 @@ public void shouldNotAllowNullNamedOnTransformWithStoreName() {
assertThat(exception.getMessage(), equalTo("named can't be null"));
}
+ @Test
+ public void shouldNotAllowBadTransformerSupplierOnFlatTransform() {
Review comment:
The "Value" variants use a `ValueTransformerSupplier` instead of a `TransformerSupplier`. Does it makes sense to extend the supplier usage verification to `ValueTransformerSupplier` 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] soarez commented on pull request #9000: KAFKA-10036 Improve handling and documentation of Suppliers
Posted by GitBox <gi...@apache.org>.
soarez commented on pull request #9000:
URL: https://github.com/apache/kafka/pull/9000#issuecomment-721637971
> Took way to long to get this merged.
My fault for screwing up several times. Thanks for not giving up on this and all the time dedicated to reviewing.
----------------------------------------------------------------
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] mjsax merged pull request #9000: KAFKA-10036 Improve handling and documentation of Suppliers
Posted by GitBox <gi...@apache.org>.
mjsax merged pull request #9000:
URL: https://github.com/apache/kafka/pull/9000
----------------------------------------------------------------
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] mjsax commented on pull request #9000: KAFKA-10036 Improve handling and documentation of Suppliers
Posted by GitBox <gi...@apache.org>.
mjsax commented on pull request #9000:
URL: https://github.com/apache/kafka/pull/9000#issuecomment-713781998
@soarez Sorry but you will need to rebase your PR to get https://github.com/apache/kafka/commit/2db67db8e1329cb2e047322cff81d97ff98b4328 -- otherwise, Jenkins does fail...
----------------------------------------------------------------
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] soarez commented on pull request #9000: KAFKA-10036 Improve handling and documentation of Suppliers
Posted by GitBox <gi...@apache.org>.
soarez commented on pull request #9000:
URL: https://github.com/apache/kafka/pull/9000#issuecomment-713147482
Seems I had forgotten to run `:streams:test-utils:unitTest`. Tests should be passing now. Please take another look @mjsax
----------------------------------------------------------------
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] soarez commented on pull request #9000: KAFKA-10036 Improve handling and documentation of Suppliers
Posted by GitBox <gi...@apache.org>.
soarez commented on pull request #9000:
URL: https://github.com/apache/kafka/pull/9000#issuecomment-660299315
@mjsax yes, if we can continue the current work here, then we should close it. I don’t think I’m able to add commits to the original 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] soarez commented on a change in pull request #9000: KAFKA-10036 Improve handling and documentation of Suppliers
Posted by GitBox <gi...@apache.org>.
soarez commented on a change in pull request #9000:
URL: https://github.com/apache/kafka/pull/9000#discussion_r479832011
##########
File path: streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamUtil.java
##########
@@ -0,0 +1,59 @@
+/*
+ * 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.kstream.internals;
+
+import org.apache.kafka.streams.kstream.TransformerSupplier;
+import org.apache.kafka.streams.kstream.ValueTransformerSupplier;
+import org.apache.kafka.streams.kstream.ValueTransformerWithKeySupplier;
+
+/**
+ * Shared functions to handle verifications of a valid {@link org.apache.kafka.streams.kstream.KStream}.
+ */
+final class KStreamUtil {
+
+ private KStreamUtil() {}
+
+ /**
+ * @throws IllegalArgumentException if the same transformer instance is obtained each time
+ */
+ static void checkSupplier(final TransformerSupplier<?, ?, ?> supplier) {
Review comment:
Good idea. `supplier.getClass().getName()` doesn't really work, but we get the name by looking through `supplier.getClass().getInterfaces()`.
----------------------------------------------------------------
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] soarez commented on pull request #9000: KAFKA-10036 Improve handling and documentation of Suppliers
Posted by GitBox <gi...@apache.org>.
soarez commented on pull request #9000:
URL: https://github.com/apache/kafka/pull/9000#issuecomment-660178140
@mjsax can something be done to move this forward?
----------------------------------------------------------------
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] mjsax commented on a change in pull request #9000: KAFKA-10036 Improve handling and documentation of Suppliers
Posted by GitBox <gi...@apache.org>.
mjsax commented on a change in pull request #9000:
URL: https://github.com/apache/kafka/pull/9000#discussion_r517048264
##########
File path: streams/src/main/java/org/apache/kafka/streams/internals/ApiUtils.java
##########
@@ -75,4 +81,31 @@ public static long validateMillisecondInstant(final Instant instant, final Strin
public static String prepareMillisCheckFailMsgPrefix(final Object value, final String name) {
return format(MILLISECOND_VALIDATION_FAIL_MSG_FRMT, name, value);
}
+
+ /**
+ * @throws IllegalArgumentException if the same instance is obtained each time
+ */
+ public static void checkSupplier(final Supplier<?> supplier) {
+ if (supplier.get() == supplier.get()) {
+ final String supplierClass = getAllImplementedInterfaces(supplier.getClass()).stream()
Review comment:
> But maybe just using the implementing class name is fine.
That was my though, too.
----------------------------------------------------------------
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] mjsax commented on a change in pull request #9000: KAFKA-10036 Improve handling and documentation of Suppliers
Posted by GitBox <gi...@apache.org>.
mjsax commented on a change in pull request #9000:
URL: https://github.com/apache/kafka/pull/9000#discussion_r474886689
##########
File path: streams/src/test/java/org/apache/kafka/streams/processor/internals/InternalTopologyBuilderTest.java
##########
@@ -175,6 +177,12 @@ public void testAddProcessorWithNullParents() {
builder.addProcessor("processor", new MockProcessorSupplier<>(), (String) null);
}
+ @Test(expected = TopologyException.class)
+ public void testAddProcessorWithBadSupplier() {
Review comment:
Yes. Sorry for using the wrong name...
----------------------------------------------------------------
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] mjsax commented on pull request #9000: KAFKA-10036 Improve handling and documentation of Suppliers
Posted by GitBox <gi...@apache.org>.
mjsax commented on pull request #9000:
URL: https://github.com/apache/kafka/pull/9000#issuecomment-661472161
There is a checkstyle error:
```
Task :streams:checkstyleTest
12:00:03 [ant:checkstyle] [ERROR] /home/jenkins/jenkins-slave/workspace/kafka-pr-jdk11-scala2.13/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamImplTest.java:1873: 'lambda arguments' has incorrect indentation level 16, expected level should be 12. [Indentation]
```
If the other PR is void, can you please close it? Thanks.
----------------------------------------------------------------
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] mjsax commented on pull request #9000: KAFKA-10036 Improve handling and documentation of Suppliers
Posted by GitBox <gi...@apache.org>.
mjsax commented on pull request #9000:
URL: https://github.com/apache/kafka/pull/9000#issuecomment-701584842
@soarez Sorry for dropping the ball on this PR. Seems there are new conflicts. Can you rebase again?
----------------------------------------------------------------
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] mjsax commented on a change in pull request #9000: KAFKA-10036 Improve handling and documentation of Suppliers
Posted by GitBox <gi...@apache.org>.
mjsax commented on a change in pull request #9000:
URL: https://github.com/apache/kafka/pull/9000#discussion_r516239935
##########
File path: streams/src/main/java/org/apache/kafka/streams/kstream/ValueTransformerSupplier.java
##########
@@ -33,12 +38,17 @@
* @see TransformerSupplier
* @see KStream#transform(TransformerSupplier, String...)
*/
-public interface ValueTransformerSupplier<V, VR> extends ConnectedStoreProvider {
+public interface ValueTransformerSupplier<V, VR> extends ConnectedStoreProvider, Supplier<ValueTransformer<V, VR>> {
Review comment:
This seems to be a public API change that we cannot do without a KIP. Seem you added it so you can pass the different suppliers into `checkSupplier` ? Also not sure if `checkSupplier` must be as "complicated" as proposed.
##########
File path: streams/src/main/java/org/apache/kafka/streams/internals/ApiUtils.java
##########
@@ -75,4 +81,31 @@ public static long validateMillisecondInstant(final Instant instant, final Strin
public static String prepareMillisCheckFailMsgPrefix(final Object value, final String name) {
return format(MILLISECOND_VALIDATION_FAIL_MSG_FRMT, name, value);
}
+
+ /**
+ * @throws IllegalArgumentException if the same instance is obtained each time
+ */
+ public static void checkSupplier(final Supplier<?> supplier) {
+ if (supplier.get() == supplier.get()) {
+ final String supplierClass = getAllImplementedInterfaces(supplier.getClass()).stream()
Review comment:
Do we really need to try to extract the concrete interface name?
----------------------------------------------------------------
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] soarez commented on pull request #9000: KAFKA-10036 Improve handling and documentation of Suppliers
Posted by GitBox <gi...@apache.org>.
soarez commented on pull request #9000:
URL: https://github.com/apache/kafka/pull/9000#issuecomment-683488002
Squashed and rebased as there were conflicts. Please take another look @mjsax .
----------------------------------------------------------------
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] soarez commented on pull request #9000: KAFKA-10036 Improve handling and documentation of Suppliers
Posted by GitBox <gi...@apache.org>.
soarez commented on pull request #9000:
URL: https://github.com/apache/kafka/pull/9000#issuecomment-714636553
@mjsax rebased and fixed an error. Can we try running the tests again?
----------------------------------------------------------------
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] mjsax commented on a change in pull request #9000: KAFKA-10036 Improve handling and documentation of Suppliers
Posted by GitBox <gi...@apache.org>.
mjsax commented on a change in pull request #9000:
URL: https://github.com/apache/kafka/pull/9000#discussion_r517048148
##########
File path: streams/src/main/java/org/apache/kafka/streams/kstream/ValueTransformerSupplier.java
##########
@@ -33,12 +38,17 @@
* @see TransformerSupplier
* @see KStream#transform(TransformerSupplier, String...)
*/
-public interface ValueTransformerSupplier<V, VR> extends ConnectedStoreProvider {
+public interface ValueTransformerSupplier<V, VR> extends ConnectedStoreProvider, Supplier<ValueTransformer<V, VR>> {
Review comment:
Yeah. Even if it's "compatible" and does not break anything, it's still fall into the "public api change" category...
----------------------------------------------------------------
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] mjsax commented on a change in pull request #9000: KAFKA-10036 Improve handling and documentation of Suppliers
Posted by GitBox <gi...@apache.org>.
mjsax commented on a change in pull request #9000:
URL: https://github.com/apache/kafka/pull/9000#discussion_r474887535
##########
File path: streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamImplTest.java
##########
@@ -1865,6 +1865,17 @@ public void shouldNotAllowNullNamedOnTransformWithStoreName() {
assertThat(exception.getMessage(), equalTo("named can't be null"));
}
+ @Test
+ public void shouldNotAllowBadTransformerSupplierOnFlatTransform() {
Review comment:
Yes, those suppliers must return a new instance on `get()`, too, so the new check should be done for them 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] soarez commented on a change in pull request #9000: KAFKA-10036 Improve handling and documentation of Suppliers
Posted by GitBox <gi...@apache.org>.
soarez commented on a change in pull request #9000:
URL: https://github.com/apache/kafka/pull/9000#discussion_r474950526
##########
File path: streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamImplTest.java
##########
@@ -1865,6 +1865,17 @@ public void shouldNotAllowNullNamedOnTransformWithStoreName() {
assertThat(exception.getMessage(), equalTo("named can't be null"));
}
+ @Test
+ public void shouldNotAllowBadTransformerSupplierOnFlatTransform() {
Review comment:
Ok. I just added equivalent documentation changes, checks and tests for `ValueTransformerSupplier` and `ValueTransformerWithKeySupplier` as well in 567d5e0 .
----------------------------------------------------------------
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] soarez commented on pull request #9000: KAFKA-10036 Improve handling and documentation of Suppliers
Posted by GitBox <gi...@apache.org>.
soarez commented on pull request #9000:
URL: https://github.com/apache/kafka/pull/9000#issuecomment-721129774
Thanks for giving this another look @mjsax
----------------------------------------------------------------
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] mjsax commented on pull request #9000: KAFKA-10036 Improve handling and documentation of Suppliers
Posted by GitBox <gi...@apache.org>.
mjsax commented on pull request #9000:
URL: https://github.com/apache/kafka/pull/9000#issuecomment-721462247
Thanks for the PR @soarez! Merged to `trunk`.
Took way to long to get this merged. Sorry for the delay.
----------------------------------------------------------------
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] mjsax commented on pull request #9000: KAFKA-10036 Improve handling and documentation of Suppliers
Posted by GitBox <gi...@apache.org>.
mjsax commented on pull request #9000:
URL: https://github.com/apache/kafka/pull/9000#issuecomment-678452210
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] soarez commented on pull request #9000: KAFKA-10036 Improve handling and documentation of Suppliers
Posted by GitBox <gi...@apache.org>.
soarez commented on pull request #9000:
URL: https://github.com/apache/kafka/pull/9000#issuecomment-709937235
@mjsax rebased now. Can you give this another look?
----------------------------------------------------------------
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] soarez commented on pull request #9000: KAFKA-10036 Improve handling and documentation of Suppliers
Posted by GitBox <gi...@apache.org>.
soarez commented on pull request #9000:
URL: https://github.com/apache/kafka/pull/9000#issuecomment-661501744
Thanks @mjsax . I've addressed the Checkstyle issue.
I don't think I'm able to close the original PR. In this Github repository I'm only able to close PRs that I've authored.
@sneakyburro maybe you can close #8752? Or finish the work there and I'll close this one.
----------------------------------------------------------------
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] soarez commented on pull request #9000: KAFKA-10036 Improve handling and documentation of Suppliers
Posted by GitBox <gi...@apache.org>.
soarez commented on pull request #9000:
URL: https://github.com/apache/kafka/pull/9000#issuecomment-715213289
Yes, sorry for that @mjsax . I thought I had already pushed the fix. Not enough ☕
Fixed it but seems one test still failed for JDK 8 - `org.apache.kafka.streams.integration.EosBetaUpgradeIntegrationTest.shouldUpgradeFromEosAlphaToEosBeta` - but it seems unrelated to this patch and [known to be flaky](https://github.com/apache/kafka/pull/9466#issuecomment-713246622).
----------------------------------------------------------------
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] soarez commented on a change in pull request #9000: KAFKA-10036 Improve handling and documentation of Suppliers
Posted by GitBox <gi...@apache.org>.
soarez commented on a change in pull request #9000:
URL: https://github.com/apache/kafka/pull/9000#discussion_r466081647
##########
File path: streams/src/test/java/org/apache/kafka/streams/processor/internals/InternalTopologyBuilderTest.java
##########
@@ -175,6 +177,12 @@ public void testAddProcessorWithNullParents() {
builder.addProcessor("processor", new MockProcessorSupplier<>(), (String) null);
}
+ @Test(expected = TopologyException.class)
+ public void testAddProcessorWithBadSupplier() {
Review comment:
Assuming you mean `.addGlobalStore()`, as there's no `.addGlobalStateStore()` in the builder. There was a supplier usage check missing there too.
----------------------------------------------------------------
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] mjsax commented on a change in pull request #9000: KAFKA-10036 Improve handling and documentation of Suppliers
Posted by GitBox <gi...@apache.org>.
mjsax commented on a change in pull request #9000:
URL: https://github.com/apache/kafka/pull/9000#discussion_r477620408
##########
File path: streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamUtil.java
##########
@@ -0,0 +1,59 @@
+/*
+ * 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.kstream.internals;
+
+import org.apache.kafka.streams.kstream.TransformerSupplier;
+import org.apache.kafka.streams.kstream.ValueTransformerSupplier;
+import org.apache.kafka.streams.kstream.ValueTransformerWithKeySupplier;
+
+/**
+ * Shared functions to handle verifications of a valid {@link org.apache.kafka.streams.kstream.KStream}.
+ */
+final class KStreamUtil {
+
+ private KStreamUtil() {}
+
+ /**
+ * @throws IllegalArgumentException if the same transformer instance is obtained each time
+ */
+ static void checkSupplier(final TransformerSupplier<?, ?, ?> supplier) {
Review comment:
I am wondering if we actually need three methods? Could we use `java.util.function.Supplier` instead (we don't really care about generic types.
To customize the error message we just pass an additional `String` or use `supplier.getClass().getName()` ?
----------------------------------------------------------------
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] mjsax commented on pull request #9000: KAFKA-10036 Improve handling and documentation of Suppliers
Posted by GitBox <gi...@apache.org>.
mjsax commented on pull request #9000:
URL: https://github.com/apache/kafka/pull/9000#issuecomment-713022002
@soarez Thanks for the update. It seems that `TopologyTestDriverTest.shouldCloseProcessor` is broken. Can you fix it and update the 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] soarez commented on pull request #9000: KAFKA-10036 Improve handling and documentation of Suppliers
Posted by GitBox <gi...@apache.org>.
soarez commented on pull request #9000:
URL: https://github.com/apache/kafka/pull/9000#issuecomment-675134899
Pinging @mjsax. Don't forget to review 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] mjsax commented on pull request #9000: KAFKA-10036 Improve handling and documentation of Suppliers
Posted by GitBox <gi...@apache.org>.
mjsax commented on pull request #9000:
URL: https://github.com/apache/kafka/pull/9000#issuecomment-713726509
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] soarez commented on pull request #9000: KAFKA-10036 Improve handling and documentation of Suppliers
Posted by GitBox <gi...@apache.org>.
soarez commented on pull request #9000:
URL: https://github.com/apache/kafka/pull/9000#issuecomment-665668720
@mjsax what can we do to proceed?
----------------------------------------------------------------
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] soarez commented on pull request #9000: KAFKA-10036 Improve handling and documentation of Suppliers
Posted by GitBox <gi...@apache.org>.
soarez commented on pull request #9000:
URL: https://github.com/apache/kafka/pull/9000#issuecomment-669622098
Thanks for the review @mjsax. I think I've addressed your feedback and I have a question about whether or not we should extend this to `ValueTransformerSupplier`.
----------------------------------------------------------------
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] mjsax commented on pull request #9000: KAFKA-10036 Improve handling and documentation of Suppliers
Posted by GitBox <gi...@apache.org>.
mjsax commented on pull request #9000:
URL: https://github.com/apache/kafka/pull/9000#issuecomment-714688801
The build did run, but failed with a compile error: Maybe something wrong with the rebase you did?
```
/home/jenkins/jenkins-agent/workspace/Kafka_kafka-pr_PR-9000@2/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableTransformValuesTest.java:209: error: cannot find symbol
new KTableTransformValues<>(parent, new SingletonNoOpValueTransformer<>(), null).enableSendingOldValues(true);
```
----------------------------------------------------------------
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] soarez commented on a change in pull request #9000: KAFKA-10036 Improve handling and documentation of Suppliers
Posted by GitBox <gi...@apache.org>.
soarez commented on a change in pull request #9000:
URL: https://github.com/apache/kafka/pull/9000#discussion_r516680210
##########
File path: streams/src/main/java/org/apache/kafka/streams/kstream/ValueTransformerSupplier.java
##########
@@ -33,12 +38,17 @@
* @see TransformerSupplier
* @see KStream#transform(TransformerSupplier, String...)
*/
-public interface ValueTransformerSupplier<V, VR> extends ConnectedStoreProvider {
+public interface ValueTransformerSupplier<V, VR> extends ConnectedStoreProvider, Supplier<ValueTransformer<V, VR>> {
Review comment:
Indeed, that was why. Since it already conforms to the interface, I didn't realize this would be a public API change. Will revert and overload `checkSupplier` instead.
##########
File path: streams/src/main/java/org/apache/kafka/streams/internals/ApiUtils.java
##########
@@ -75,4 +81,31 @@ public static long validateMillisecondInstant(final Instant instant, final Strin
public static String prepareMillisCheckFailMsgPrefix(final Object value, final String name) {
return format(MILLISECOND_VALIDATION_FAIL_MSG_FRMT, name, value);
}
+
+ /**
+ * @throws IllegalArgumentException if the same instance is obtained each time
+ */
+ public static void checkSupplier(final Supplier<?> supplier) {
+ if (supplier.get() == supplier.get()) {
+ final String supplierClass = getAllImplementedInterfaces(supplier.getClass()).stream()
Review comment:
We don't _really_ need to. I thought it could make for a better, and also predictable, error message. But maybe just using the implementing class name is fine.
----------------------------------------------------------------
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