KAFKA-3625: Add public test utils for Kafka Streams #4402

mjsax · 2018-01-08T21:21:57Z

add new artifact test-utils
add TopologyTestDriver
add MockTime, TestRecord, add TestRecordFactory

This PR requires a KIP and is WIP. DO NOT MERGE.

Committer Checklist (excluded from commit message)

Verify design and implementation
Verify test coverage and CI build status
Verify documentation (including upgrade notes)

bbejeck · 2018-01-09T15:52:16Z

@mjsax failures look related

bbejeck · 2018-01-09T15:55:27Z

streams/test-utils/src/main/java/org/apache/kafka/streams/InternalTopologyBuilderAccessor.java

+ */
+@InterfaceStability.Unstable
+public class InternalTopologyBuilderAccessor {
+    public static InternalTopologyBuilder getInternalTopologyBuilder(Topology topology) {


bbejeck · 2018-01-09T16:03:35Z

streams/test-utils/src/main/java/org/apache/kafka/streams/test/MockTime.java

+    }
+
+    @Override
+    public void sleep(final long ms) {


IMHO sleep doesn't feel quite right, what about SMT along the lines of advance or process
EDIT: NM just saw that this implements Time.sleep I get how this is intended to work but I still find it somewhat confusing as it seems nothing is actually paused (I could be wrong).

In the mock impl nothing is paused, but to the runtime it's just Time object that could build on SystemTime and thus really sleep here. We could add a second method advance() that internally just calls sleep() though to make tests more readable. But having two methods doing the same thing might be confusion, too. Thus, I prefer to leave as is. WDYT?

Yeah, I'd have to agree it doesn't make sense to have two methods doing the same thing, I'd say leave as is.

bbejeck · 2018-01-09T16:17:51Z

streams/test-utils/src/main/java/org/apache/kafka/streams/test/MockTime.java

+    }
+
+    public void setCurrentTimeMs(final long newMs) {
+        long oldMs = timeMs.getAndSet(newMs);


bbejeck · 2018-01-09T16:18:06Z

streams/test-utils/src/main/java/org/apache/kafka/streams/test/MockTime.java

+    }
+
+    private void tick() {
+        for (MockTimeListener listener : listeners) {


bbejeck · 2018-01-09T17:14:03Z

left some initial comments.

I have two meta-comments regarding TestRecord and TestRecordFactory

IMHO if we add a builder to TestRecord (additionally the builder will have a copy constructor) it could help with using the TestRecordFactory I know this isn't entirely clear so I'll need to come up with an example.
What do you think of providing methods auto-generating simple key-value pairs of basic types such as KeyValue<Integer, String> etc

mjsax · 2018-01-12T23:59:28Z

@bbejeck @guozhangwang @dguy Updated PR according to KIP. Feedback is welcome.

dguy

Thanks @mjsax, made a pass and left some initial comments

dguy · 2018-01-15T12:30:29Z

streams/test-utils/src/main/java/org/apache/kafka/streams/MockTime.java

+ * This is an internal class and should not be used.
+ */
+@InterfaceStability.Unstable
+public class MockTime implements Time {


Why do we need another MockTime? We have one already in o.a.k.common.utils

MockTime in utils is part of the test package, and thus not public API. I don't think we should depend on this.

dguy · 2018-01-15T12:32:51Z

streams/test-utils/src/main/java/org/apache/kafka/streams/test/ConsumerRecordFactory.java

+                                 final long autoAdvanceMs) {
+        Objects.requireNonNull(keySerializer, "keySerializer cannot be null");
+        Objects.requireNonNull(valueSerializer, "valueSerializer cannot be null");
+        this.topicName = defaultTopicName;


should we also be checking that topicName is non-null?

I want to allow topicName=null for the case that somebody wants a factory without default topic name. We have a constructor without taking a defaultTopicName.

dguy · 2018-01-15T12:33:22Z

streams/test-utils/src/main/java/org/apache/kafka/streams/test/ConsumerRecordFactory.java

+                                                 final K key,
+                                                 final V value,
+                                                 final long timestampMs) {
+        if (topicName == null) {


Objects.requireNonNull(..) ?

We could. No strong opinion -- IllegalArgumentException seems more appropriate than NPE.

It would be consistent with what we have done elsewhere in the API

dguy · 2018-01-15T12:40:52Z

...ms/test-utils/src/test/java/org/apache/kafka/streams/test/ConsumerRecordFactoryConsumer.java

+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.fail;
+
+public class ConsumerRecordFactoryConsumer {


ConsumerRecordFactoryTest ?

dguy · 2018-01-15T12:41:29Z

...ms/test-utils/src/test/java/org/apache/kafka/streams/test/ConsumerRecordFactoryConsumer.java

+    }
+
+    @Test
+    public void shouldNotAllowToCreateTopicWithNullTopicName() {


i'd probably do this as separate tests for each method

dguy · 2018-01-15T12:41:38Z

...ms/test-utils/src/test/java/org/apache/kafka/streams/test/ConsumerRecordFactoryConsumer.java

+    }
+
+    @Test
+    public void shouldRequireCustomTopicNameIfNotDefaultFactoryTopicName() {


dguy · 2018-01-15T12:42:41Z

...ms/test-utils/src/test/java/org/apache/kafka/streams/test/ConsumerRecordFactoryConsumer.java

+
+    @Test
+    public void shouldCreateConsumerRecord() {
+        record = factory.create(rawKey, value);


why save record to a field? Can just inline it in call to verifyRecord - elsewhere too

dguy · 2018-01-15T12:45:04Z

streams/test-utils/src/test/java/org/apache/kafka/streams/test/TopologyTestDriverTest.java

+        assertEquals(1, processedRecords1.size());
+        assertEquals(0, processedRecords2.size());
+
+        Record record = processedRecords1.get(0);


if you added an equals(...) to Record you could use assertThat(expected, equalTo(..))

mjsax · 2018-01-16T22:09:05Z

Updates this.

mjsax · 2018-01-18T03:35:16Z

Add OutputVerfier as proposed in KIP. Added more tests for TopologyTestDriver. Remove "WIP" annotation. Please review and comment on KIP.

dguy · 2018-01-19T11:07:05Z

streams/test-utils/src/test/java/org/apache/kafka/streams/test/TopologyTestDriverTest.java

+        topology.addStateStore(
+            new KeyValueStoreBuilder<>(
+                Stores.inMemoryKeyValueStore("store"),
+                new Serdes.ByteArraySerde(),


Serdes.ByteArray()? other places too

mjsax · 2018-01-20T04:37:25Z

Updates this. (reworked JavaDocs and added update notes).

mjsax · 2018-01-20T20:46:42Z

Retest this please.

dguy

Just one minor comment otherwise LGTM

dguy · 2018-01-22T21:46:29Z

streams/test-utils/src/main/java/org/apache/kafka/streams/test/TopologyTestDriver.java

+    @SuppressWarnings("unchecked")
+    public <K, V> SessionStore<K, V> getSessionStore(final String name) {
+        final StateStore store = getStateStore(name);
+        return store instanceof SessionStore ? (SessionStore<K, V>) getStateStore(name) : null;


you could just return the store you already have rather than getting it again

guozhangwang

Left some comments. As pointed out that we should consider replacing the internal test drivers with this one as a way to "further unit testing" it and verify if there's any functionality gaps: this can be done in a follow-up PR, but we should consider doing it sooner.

guozhangwang · 2018-01-22T23:08:35Z

streams/test-utils/src/main/java/org/apache/kafka/streams/InternalTopologyBuilderAccessor.java

+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.streams;


This is a meta comment: should we consider using a sub-package name than o.a.k.streams? Perhaps o.a.k.streams.testutils?

For this specific class, we need it to put it in this package, because the InternalTopologyBuilder is accessible package private only and I don't think we should make it public.

The package testutils you suggest is the package that is already there but called test -- I can rename test to testutils and update the KIP of course. Thus, the classes in test/testutils are the actual public part. The two classed in this packaged (InternalTopologyBuilderAccessor and MockTime are actually internal -- as we need InternalTopologyBuilderAccessor to be in this package, the idea is to not create an additional internal package, but use this package for this purpose)

guozhangwang · 2018-01-22T23:10:35Z

streams/test-utils/src/main/java/org/apache/kafka/streams/test/ConsumerRecordFactory.java

+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.streams.test;


Hmm, the package name is different with other classes under the test-utils artifact, is this intentional? (see my other comment above)

See answer above.

guozhangwang · 2018-01-22T23:11:10Z

streams/test-utils/src/main/java/org/apache/kafka/streams/test/OutputVerifier.java

+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.streams.test;


Ditto here.

guozhangwang · 2018-01-22T23:13:30Z

streams/test-utils/src/main/java/org/apache/kafka/streams/test/TopologyTestDriver.java

+ * @see OutputVerifier
+ */
+@InterfaceStability.Evolving
+public class TopologyTestDriver {


As a follow-up PR: we should remove the KStreamTestDriver and ProcessorTopologyTestDriver and replace them with this public class in our internal unit tests as well. It also helps us to identify if there are any gaps that worth closing (i.e. enhancing the TopologyTestDriver in next release).

Completely agree.

guozhangwang · 2018-01-22T23:24:16Z

streams/test-utils/src/main/java/org/apache/kafka/streams/test/TopologyTestDriver.java

+        };
+
+        final MockConsumer<byte[], byte[]> consumer = new MockConsumer<>(OffsetResetStrategy.EARLIEST);
+        final StateDirectory stateDirectory = new StateDirectory(streamsConfig, mockTime);


If user do not modify the state directory path in their unit tests, does that mean two unit tests would conflict on the state directory path then? Should we just use a random directory than enforcing themselves to remember setting to a different path each time?

guozhangwang · 2018-01-22T23:28:47Z

streams/test-utils/src/test/java/org/apache/kafka/streams/test/TopologyTestDriverTest.java

+        }
+    };
+
+    private final static class Record {


Why not just using the ConsumerRecord directly?

ConsumerRecord does not implement equals or hashCode and we thus cannot use assertThat if we use ConsumerRecord.

We are also just interested in some fields, for example we don't need partition, headers, serializedKeySize, etc.

guozhangwang · 2018-01-22T23:30:18Z

streams/test-utils/src/test/java/org/apache/kafka/streams/test/TopologyTestDriverTest.java

+        }
+    }
+
+    private final class MockPunctuator implements Punctuator {


I'd recommend making MockPunctuator a separate class in o.a.k.test package under streams/tests folder, to be shared with others.

guozhangwang · 2018-01-22T23:31:35Z

streams/test-utils/src/test/java/org/apache/kafka/streams/test/TopologyTestDriverTest.java

+        }
+    }
+
+    private final class MockProcessor implements Processor {


We already have a MockProcessor in o.a.k.test, could we refactor that class and use it here? Similar for MockProcessorSupplier.

guozhangwang · 2018-01-22T23:32:17Z

streams/test-utils/src/test/java/org/apache/kafka/streams/test/TopologyTestDriverTest.java

+        }
+    }
+
+    private final static class Punctuation {


Please see my other comment for MockProcessor: it's better to just reuse the existing one with improved punctuation configuration.

Are you referring to the classes in test package? I think we should not depend on the test package, as test package is not public API and the new artifact should not depend on the test artifact from my point of view. We can refactor this later though, adding Mocks to this new test artifact and remove the mocks from our unit test to use the mocks from the public test artifact.

That makes sense.

mjsax · 2018-01-23T23:30:40Z

Retest this please.

guozhangwang · 2018-01-23T23:35:25Z

LGTM. Asking @norwood @xvrl for another round of reviews.

guozhangwang · 2018-01-23T23:35:47Z

retest this please

norwood · 2018-01-23T23:56:59Z

💝

guozhangwang · 2018-01-26T17:16:58Z

retest this please

guozhangwang · 2018-01-26T18:12:52Z

@mjsax could you rebase this PR (there is no conflicts but some PR that just committed in caused jenkins to fail)

- add new artifact test-utils - add TopologyTestDriver - add MockTime, TestRecord, add TestRecordFactory

Add more tests for TopplogyTestDriver

mjsax · 2018-01-26T23:19:34Z

Rebased.

guozhangwang · 2018-01-29T04:24:30Z

This seems relevant:

/home/jenkins/jenkins-slave/workspace/kafka-pr-jdk8-scala2.12/streams/test-utils/src/main/java/org/apache/kafka/streams/TopologyTestDriver.java:303: error: cannot find symbol
00:30:11             task.initialize();

mjsax · 2018-01-29T20:50:56Z

Updated.

guozhangwang · 2018-01-30T01:20:54Z

Unit test passed locally on my machine.

mjsax force-pushed the kafka-3625-test branch 5 times, most recently from a386e97 to a91611f Compare January 9, 2018 04:28

bbejeck reviewed Jan 9, 2018

View reviewed changes

mjsax force-pushed the kafka-3625-test branch 2 times, most recently from 435afb9 to a3fb054 Compare January 10, 2018 00:35

dguy reviewed Jan 15, 2018

View reviewed changes

mjsax changed the title ~~KAFKA-3625: Add public test utils for Kafka Streams [WIP]~~ KAFKA-3625: Add public test utils for Kafka Streams Jan 18, 2018

dguy reviewed Jan 19, 2018

View reviewed changes

mjsax force-pushed the kafka-3625-test branch from fb01fda to 8718f0c Compare January 20, 2018 04:40

dguy reviewed Jan 22, 2018

View reviewed changes

guozhangwang reviewed Jan 22, 2018

View reviewed changes

guozhangwang mentioned this pull request Jan 26, 2018

MINOR: Add attributes processedKeys and processedValues to MockProcessorSupplier #3999

Merged

mjsax added 11 commits January 26, 2018 15:18

KAFKA-3625: Add public test utils for Kafka Streams

2faea1d

- add new artifact test-utils - add TopologyTestDriver - add MockTime, TestRecord, add TestRecordFactory

Github comments

8de6114

Reworked after some initial discussion

ccea4ef

Github comments and KIP comments

f2b0cff

Add OutputVerifier

e0baec0

Add more tests for TopplogyTestDriver

Add getAllStateStores()

d5960c7

fix failing tests

ab49359

Github comment

b73621c

Updated JavaDocs

67bd7d1

Update docs

cebe815

Github comments / added 2 new tests

fab87fa

mjsax force-pushed the kafka-3625-test branch from 5764577 to fab87fa Compare January 26, 2018 23:19

Fixup after rebase

5e6d224

guozhangwang merged commit a78f66a into apache:trunk Jan 30, 2018

mjsax deleted the kafka-3625-test branch January 30, 2018 02:20

KAFKA-3625: Add public test utils for Kafka Streams #4402

KAFKA-3625: Add public test utils for Kafka Streams #4402

Conversation

mjsax commented Jan 8, 2018

Committer Checklist (excluded from commit message)

bbejeck commented Jan 9, 2018

Choose a reason for hiding this comment

bbejeck Jan 9, 2018 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

bbejeck commented Jan 9, 2018

mjsax commented Jan 12, 2018

dguy left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mjsax commented Jan 16, 2018

mjsax commented Jan 18, 2018

Choose a reason for hiding this comment

mjsax commented Jan 20, 2018

mjsax commented Jan 20, 2018

dguy left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

guozhangwang left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mjsax Jan 22, 2018 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mjsax commented Jan 23, 2018

guozhangwang commented Jan 23, 2018

guozhangwang commented Jan 23, 2018

norwood commented Jan 23, 2018

guozhangwang commented Jan 26, 2018

guozhangwang commented Jan 26, 2018

mjsax commented Jan 26, 2018

guozhangwang commented Jan 29, 2018

mjsax commented Jan 29, 2018

guozhangwang commented Jan 30, 2018

bbejeck Jan 9, 2018 •

edited

Loading

mjsax Jan 22, 2018 •

edited

Loading