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: us...@infra.apache.org
