KAFKA-6474: Rewrite tests to use new public TopologyTestDriver [partial] #4832

h314to · 2018-04-06T00:48:33Z

* Remove ProcessorTopologyTestDriver from TopologyTest
* Fix ProcessorTopologyTest
* Remove ProcessorTopologyTestDriver and InternalTopologyAccessor
* Partially refactored StreamsBuilderTest but missing one test
* Refactor KStreamBuilderTest
* Refactor AbstractStreamTest
* Further cleanup of AbstractStreamTest
* Refactor GlobalKTableJoinsTest
* Refactor InternalStreamsBuilderTest
* Fix circular dependency in build.gradle
* Refactor KGroupedStreamImplTest
* Partial modifications to KGroupedTableImplTest
* Refactor KGroupedTableImplTest
* Refactor KStreamBranchTest
* Refactor KStreamFilterTest
* Refactor KStreamFlatMapTest KStreamFlatMapValuesTest
* Refactor KStreamForeachTest
* Refactor KStreamGlobalKTableJoinTest
* Refactor KStreamGlobalKTableLeftJoinTest
* Refactor KStreamImplTest
* Refactor KStreamImplTest
* Refactor KStreamKStreamJoinTest
* Refactor KStreamKStreamLeftJoinTest
* Refactor KStreamKTableJoinTest
* Refactor KStreamKTableLeftJoinTest
* Refactor KStreamMapTest and KStreamMapValuesTest
* Refactor KStreamPeekTest and KStreamTransformTest
* Refactor KStreamSelectKeyTest
* Refactor KStreamTransformValuesTest
* Refactor KStreamWindowAggregateTest
* Add Depercation anotation to KStreamTestDriver and rollback failing tests in StreamsBuilderTest and KTableAggregateTest
* Refactor KTableFilterTest
* Refactor KTableForeachTest
* Add getter for ProcessorTopology, and simplify tests in StreamsBuilderTest
* Refactor KTableImplTest
* Remove unused imports
* Refactor KTableAggregateTest
* Fix style errors
* Fix gradle build
* Address reviewer comments:
  - Remove properties new instance
  - Remove extraneous line
  - Remove unnecessary TopologyTestDriver instances from StreamsBuilderTest
  - Move props.clear() to @After
  - Clarify use of timestamp in KStreamFlatMapValuesTest
  - Keep test using old Punctuator in KStreamTransformTest
  - Add comment to clarify clock advances in KStreamTransformTest
  - Add TopologyTestDriverWrapper class to access the protected constructor of TopologyTestDriver
  - Revert KTableImplTest.testRepartition to KStreamTestDriver to avoid exposing the TopologyTestDriver processor topology
  - Revert partially migrated classes: KTableAggregateTest, KTableFilterTest, and KTableImplTest
* Rebase on current trunk and fix conflicts

mjsax · 2018-04-08T01:14:06Z

\cc @bbejeck @vvcephei

mjsax · 2018-04-08T01:14:52Z

@h314to Thanks a lot for this PR -- it's quite big -- I was wondering if we could break it down into multiple smaller PRs? This would make reviewing simpler.

mjsax

Thanks a lot for the PR! Very nice work.

With regard to my last comment -- maybe we can have one PR with all "straight forward" changes that we can merge quickly, and discuss the open issues on a reduced PR. This would help a lot and increase turn around time for reviews.

Thx.

mjsax · 2018-04-06T03:53:14Z

streams/src/test/java/org/apache/kafka/streams/StreamsBuilderTest.java

+
+    @Before
+    public void setup() {
+        Properties props = new Properties();


nit: add final

Yes, will do. This pattern is repeated a bunch of times. I'll add final to all.

mjsax · 2018-04-08T01:21:24Z

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

+     * @param builder builder for the topology to be tested
+     * @param config the configuration for the topology
+     */
+    public TopologyTestDriver(final InternalTopologyBuilder builder,


This class is part of public API -- thus, we cannot change any public interface without a KIP. I am also not sure, if we should change this class in the first place. What limitations did you find rewriting the tests? We should try to rewrite out test without any change to this class IMHO.

For this particular change, we should not expose InternalTopologyBuilder because it's an internal class.

Ah, ok. Sorry for changing it without the KIP. I needed it, for instance, for the ProcessorTopologyTest, in order to instantiate a TopologyTestDriver with the internalTopologyDriver of a org.apache.kafka.streams.processor.TopologyBuilder. Maybe there's some cleaner way to do it that I am missing.

I see. I would suggest the following workaround. Make this constructor package-private and create a test class in the same package go be able to call it. Use this new class to create the TopologyTestDriver in ProcessorTopologyTest -- note, that TopologyBuilder is deprecated and will be removed eventually. Similar idea to the InternalTopologyAccessor class.

Does this make sense?

Alternatively, we can avoid using this driver for that test, or to rewrite the test to give a Topology instead of an InternalTopologyBuilder.

IMO, the function of this class should be strictly limited to what makes sense for users to test their topologies. If we bolt stuff on for testing internals, it paves the way to pollute this class's purpose.

I'm all for using this class for internal tests, but if we need something that users wouldn't need, then we should use an internal driver instead of adding backdoors for internal testing to this class.

I followed @mjsax 's suggestion and implemented a wrapper class which exposes the protected constructor. It is only used in a few places, and the tests where it is used are for deprecated classes should be removed in the future. This way we can still use TopologyTestDriver for these internal tests, while keeping the public interface unchanged.

But I agree with @vvcephei . While the Jira ticket asks for a removal of the KStreamsTestDriver class, I think it is wise to keep it around for those cases where we need the added functionality, to avoid polluting the TopologyTestDriver.

mjsax · 2018-04-08T01:22:08Z

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

+     * Processor topology getter
+     * @return the processor topology
+     */
+    public ProcessorTopology getProcessorTopology() {


As above: ProcessorTopology is an internal class and should not be exposed to the user.

This one I needed for StreamsBuilderTest, in order to access the ProcessorTopology.processorConnectedStateStores and in KTableImplTest to get processor by name. Any suggestions on how to get these without exposing the processor topology?

I haven't been around here long enough to be confident about this, but here goes...

It looks like those tests are making assertions about the structure of the runtime topology that results from different DSL operations. To test that, it should be sufficient to directly call the relevant build operations to get a ProcessorTopology from a Topology. It doesn't seem to need the TopologyTestDriver to be involved.

Nevertheless, this illustrates the tension between an internal test driver and an external one. IMO, this class should be only an external test driver, so if there's some stuff that we need for internal testing purposes, then we need to keep the internal test driver around.

So I guess my suggestions are:

try to rewrite the those tests to avoid using a topology test driver at all

if that fails, just use the internal test driver instead

I reverted back this back to KStreamTestDriver to avoid having to expose the TopologyTestDriver internal class

mjsax · 2018-04-08T01:30:12Z

streams/src/test/java/org/apache/kafka/streams/StreamsBuilderTest.java

@@ -70,13 +99,12 @@ public void shouldAllowJoinUnmaterializedFilteredKTable() {
        final KTable<Bytes, String> filteredKTable = builder.<Bytes, String>table("table-topic").filter(MockPredicate.<Bytes, String>allGoodPredicate());
        builder.<Bytes, String>stream("stream-topic").join(filteredKTable, MockValueJoiner.TOSTRING_JOINER);

-        driver.setUp(builder, TestUtils.tempDirectory());


I think, we can just remove this line. Calling the (old) driver is not required in this test.

yes, it's gone

mjsax · 2018-04-08T01:31:13Z

streams/src/test/java/org/apache/kafka/streams/StreamsBuilderTest.java

@@ -70,13 +99,12 @@ public void shouldAllowJoinUnmaterializedFilteredKTable() {
        final KTable<Bytes, String> filteredKTable = builder.<Bytes, String>table("table-topic").filter(MockPredicate.<Bytes, String>allGoodPredicate());
        builder.<Bytes, String>stream("stream-topic").join(filteredKTable, MockValueJoiner.TOSTRING_JOINER);

-        driver.setUp(builder, TestUtils.tempDirectory());
-
-        ProcessorTopology topology = builder.internalTopologyBuilder.build();


We can keep this line and also don't need changes below. Why did you move this to TopologyTestDriver? Seems to be an unnecessary redirection?

Yes, it is unnecessary. Thanks!

mjsax · 2018-04-08T01:53:38Z

streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamMapTest.java

+            driver.close();
+        }
+        driver = null;
+


nit: remove empty line

Yes, will do.

mjsax · 2018-04-08T01:56:33Z

streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamTransformTest.java

@@ -52,7 +82,13 @@ public void testTransform() {
                        private int total = 0;

                        @Override
-                        public void init(ProcessorContext context) {
+                        public void init(final ProcessorContext context) {
+                            context.schedule(1, PunctuationType.WALL_CLOCK_TIME, new Punctuator() {


I am wondering, if we should rewrite this (I understand that we must if we want to us TopologyTestDriver that only supports the new punctuation API) because we should have tests for the old deprecated API as long as the API is not removed... \cc @guozhangwang @bbejeck @vvcephei WDYT?

It's a +1 from me to update this test to cover the deprecated API. IMHO, as long as it's still in the code base we shouldn't remove any tests covering the old API.

Yes, it is wise to keep the test for the old API. I would keep the old test, which uses the old punctuation api with KStreamTestDriver, and add the above test which covers the new punctuation and TopologyTestDriver. Is that ok?

Sounds right to me. It's probably deserving of a comment to explain the rationale, though.

mjsax · 2018-04-08T01:57:36Z

streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamTransformTest.java

-        driver.punctuate(2);
-        driver.punctuate(3);
+        driver.advanceWallClockTime(2);
+        driver.advanceWallClockTime(1);


There are no assertion between both calls -- can we do a single call and pass in 3 instead?

I think we must because of the punctuation. If we just do a single call and pass 3 a value will be missing from the test results.

Ack. Makes sense. Maybe add a comment?

The time in this changeset is 1 instead of 3. I imagine you meant to leave it as 3.

Maybe there is some condition you could assert in between these two ticks?

mjsax · 2018-04-08T01:59:23Z

streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableAggregateTest.java

    @Rule
-    public final KStreamTestDriver driver = new KStreamTestDriver();
+    //TODO: replace this driver with TopologyTestDriver when cache.max.bytes.buffering is working


Can you explain this?

I may be doing something wrong, but setting CACHE_MAX_BYTES_BUFFERING_CONFIG had no effect and I kept getting the same results as I would if the cache was disabled, so tests which rely on caching were failing for TopologyTestDriver due to results mismatch.

Hmmm. Not entirely sure without digging into details -- maybe the old driver passed this config forward while the new does not (or not correctly) -- might be a bug in TopologyTestDriver. Let us know if you need more input on this to figure out the root cause.

mjsax · 2018-04-08T02:01:01Z

streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableFilterTest.java

    @Rule
-    public final KStreamTestDriver driver = new KStreamTestDriver();
-    private File stateDir = null;
+    //TODO: remove this driver when we can pass processor context to the getters using the TopologyTestDriver


Can you provide some details for this issue?

KTableValueGetter must be inited with the processor context, and I didn't find any easy way to do it, so I just skipped it and work on the remaining classes, and left this to workout at the end of the refactor. If anyone has some tips on how to do it the are very welcome.

h314to · 2018-04-09T12:52:55Z

@mjsax Thank you very much for your thorough review. Sorry for the large PR, but a lot of stuff depends on KStreamTestDriver, which trying to remove. I like your suggestion: I'll move all the straightforward stuff into a separate PR, and take care the remaining stuff subsequent PRs. I'll also be addressing your comments above.

bbejeck

@h314to thanks for the PR.

I've made an initial pass, and I've left one comment regarding tests of deprecated methods. I'll take a deeper look when the PR splitting mentioned gets pushed.

bbejeck · 2018-04-09T15:43:31Z

streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamTransformTest.java

@@ -52,7 +82,13 @@ public void testTransform() {
                        private int total = 0;

                        @Override
-                        public void init(ProcessorContext context) {
+                        public void init(final ProcessorContext context) {
+                            context.schedule(1, PunctuationType.WALL_CLOCK_TIME, new Punctuator() {


It's a +1 from me to update this test to cover the deprecated API. IMHO, as long as it's still in the code base we shouldn't remove any tests covering the old API.

vvcephei · 2018-04-11T18:01:47Z

@h314to , Thanks for the PR! One strategy you might consider for sending smaller PRs is not to try and remove the new driver, but just to stop using it one test at a time.

This way, you can have an arbitrary number of PRs where you whittle down the usage or the current driver in favor of TopologyTestDriver and then one last PR where you remove the now-unused classes.

vvcephei

Hey @h314to ,

Thanks so much for the PR. I see that you're planning to break it up into smaller PRs, but I went ahead and made some comments that might be helpful.

-John

vvcephei · 2018-04-11T17:45:14Z

build.gradle

@@ -929,6 +929,8 @@ project(':streams') {
    testCompile libs.easymock
    testCompile libs.bcpkix

+    // testRuntimeOnly makes dependencies available at test runtime, while preventing the cyclic dependency as noted above
+    testRuntimeOnly project(':streams:test-utils')


👍 I also noticed this was needed.

vvcephei · 2018-04-11T17:48:56Z

streams/src/test/java/org/apache/kafka/streams/StreamsBuilderTest.java

+        if (driver != null) {
+            driver.close();
+        }
+        driver = null;


Alternatively, we could ditch the driver field and just make it a local variable in every test.

Having it as a field made more sense when it was setUp in each methods, but now that it's instantiated in each test, it would be simpler to limit the scope to a local variable. Each test would need to call close at the end, but that's fine with me. If anything, it demonstrates proper use of the driver.

vvcephei · 2018-04-11T18:03:52Z

streams/src/test/java/org/apache/kafka/streams/kstream/KStreamBuilderTest.java

        builder.setApplicationId(APP_ID);
+        Properties props = new Properties();


If it's the same for every test, you could also just create a getProperties() method for the tests to call when they need the properties. This way, you won't need a field to store the properties in.

vvcephei · 2018-04-11T18:13:41Z

streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamTransformTest.java

@@ -52,7 +82,13 @@ public void testTransform() {
                        private int total = 0;

                        @Override
-                        public void init(ProcessorContext context) {
+                        public void init(final ProcessorContext context) {
+                            context.schedule(1, PunctuationType.WALL_CLOCK_TIME, new Punctuator() {


Sounds right to me. It's probably deserving of a comment to explain the rationale, though.

vvcephei · 2018-04-11T18:15:34Z

streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamTransformTest.java

-        driver.punctuate(2);
-        driver.punctuate(3);
+        driver.advanceWallClockTime(2);
+        driver.advanceWallClockTime(1);


The time in this changeset is 1 instead of 3. I imagine you meant to leave it as 3.

Maybe there is some condition you could assert in between these two ticks?

vvcephei · 2018-04-11T18:25:29Z

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

+     * @param builder builder for the topology to be tested
+     * @param config the configuration for the topology
+     */
+    public TopologyTestDriver(final InternalTopologyBuilder builder,


Alternatively, we can avoid using this driver for that test, or to rewrite the test to give a Topology instead of an InternalTopologyBuilder.

IMO, the function of this class should be strictly limited to what makes sense for users to test their topologies. If we bolt stuff on for testing internals, it paves the way to pollute this class's purpose.

I'm all for using this class for internal tests, but if we need something that users wouldn't need, then we should use an internal driver instead of adding backdoors for internal testing to this class.

vvcephei · 2018-04-11T18:35:43Z

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

+     * Processor topology getter
+     * @return the processor topology
+     */
+    public ProcessorTopology getProcessorTopology() {


I haven't been around here long enough to be confident about this, but here goes...

It looks like those tests are making assertions about the structure of the runtime topology that results from different DSL operations. To test that, it should be sufficient to directly call the relevant build operations to get a ProcessorTopology from a Topology. It doesn't seem to need the TopologyTestDriver to be involved.

Nevertheless, this illustrates the tension between an internal test driver and an external one. IMO, this class should be only an external test driver, so if there's some stuff that we need for internal testing purposes, then we need to keep the internal test driver around.

So I guess my suggestions are:

try to rewrite the those tests to avoid using a topology test driver at all

if that fails, just use the internal test driver instead

mjsax · 2018-04-17T07:55:00Z

@h314to Are you working on the second PR to extract the "straight forward changes"? Or did I miss that you opened a PR for that already?

h314to · 2018-04-23T14:50:49Z

In order to keep comment history, which might make further code review simpler, I removed the thorny parts from this PR, addressed your previous comments, and rebased to the current trunk. I hope that is ok. I'll be working on the remaining classes and will submit smaller PR with the remaining stuff when they are done.

bbejeck

Thanks for the PR! Overall this looks good. I have a couple of general meta-comments.

Most of the tests use the same properties except for the application-id. I'm wondering if we want to add a method StreamsTestUtils where you pass in the application id and get back the properties as configured in the tests.

Or since it looks like all the tests use the same setup and cleanup methods do we want an abstract base test and provide a method to define the application id?

WDYT?
\cc @guozhangwang @mjsax @vvcephei

bbejeck · 2018-04-23T16:20:33Z

retest this please

@after

* Remove ProcessorTopologyTestDriver from TopologyTest * Fix ProcessorTopologyTest * Remove ProcessorTopologyTestDriver and InternalTopologyAccessor * Partially refactored StreamsBuilderTest but missing one test * Refactor KStreamBuilderTest * Refactor AbstractStreamTest * Further cleanup of AbstractStreamTest * Refactor GlobalKTableJoinsTest * Refactor InternalStreamsBuilderTest * Fix circular dependency in build.gradle * Refactor KGroupedStreamImplTest * Partial modifications to KGroupedTableImplTest * Refactor KGroupedTableImplTest * Refactor KStreamBranchTest * Refactor KStreamFilterTest * Refactor KStreamFlatMapTest KStreamFlatMapValuesTest * Refactor KStreamForeachTest * Refactor KStreamGlobalKTableJoinTest * Refactor KStreamGlobalKTableLeftJoinTest * Refactor KStreamImplTest * Refactor KStreamImplTest * Refactor KStreamKStreamJoinTest * Refactor KStreamKStreamLeftJoinTest * Refactor KStreamKTableJoinTest * Refactor KStreamKTableLeftJoinTest * Refactor KStreamMapTest and KStreamMapValuesTest * Refactor KStreamPeekTest and KStreamTransformTest * Refactor KStreamSelectKeyTest * Refactor KStreamTransformValuesTest * Refactor KStreamWindowAggregateTest * Add Depercation anotation to KStreamTestDriver and rollback failing tests in StreamsBuilderTest and KTableAggregateTest * Refactor KTableFilterTest * Refactor KTableForeachTest * Add getter for ProcessorTopology, and simplify tests in StreamsBuilderTest * Refactor KTableImplTest * Remove unused imports * Refactor KTableAggregateTest * Fix style errors * Fix gradle build * Address reviewer comments: - Remove properties new instance - Remove extraneous line - Remove unnecessary TopologyTestDriver instances from StreamsBuilderTest - Move props.clear() to @after - Clarify use of timestamp in KStreamFlatMapValuesTest - Keep test using old Punctuator in KStreamTransformTest - Add comment to clarify clock advances in KStreamTransformTest - Add TopologyTestDriverWrapper class to access the protected constructor of TopologyTestDriver - Revert KTableImplTest.testRepartition to KStreamTestDriver to avoid exposing the TopologyTestDriver processor topology - Revert partially migrated classes: KTableAggregateTest, KTableFilterTest, and KTableImplTest * Rebase on current trunk an fix conflicts

h314to · 2018-04-26T11:00:55Z

There were some conflicts due to recent commits in trunk. I rebased and fixed them.

guozhangwang · 2018-04-26T18:02:15Z

Most of the tests use the same properties except for the application-id. I'm wondering if we want to add a method StreamsTestUtils where you pass in the application id and get back the properties as configured in the tests.

Or since it looks like all the tests use the same setup and cleanup methods do we want an abstract base test and provide a method to define the application id?

@bbejeck good suggestion. I'll suggest filing a JIRA and address it in a different PR :)

guozhangwang

LGTM! Thanks @h314to

guozhangwang · 2018-04-26T18:06:07Z

streams/src/test/java/org/apache/kafka/streams/TopologyTestDriverWrapper.java

+ *  This class allows the instantiation of a {@link TopologyTestDriver} using a
+ *  {@link InternalTopologyBuilder} by exposing a protected constructor.
+ *
+ *  It should be used only for testing, and should be removed once the deprecated


This is not a suggestion for change: while working on removing KStreamBuilder and TopologyBuilder I realized in some unit tests we may still need this class to access the internal topology builder. So probably we cannot remove it even after that, but we can discuss this later in the cleanup PR.

guozhangwang · 2018-04-26T18:31:07Z

@h314to thanks for your contribution! Please ping us when you have the second PR ready for review.

vvcephei

Hey @h314to ,

I see @guozhangwang already merged this, but do you mind considering my comments and possibly incorporating them in a follow-up?

Thanks,
-John

vvcephei · 2018-04-26T18:10:15Z

streams/src/test/java/org/apache/kafka/streams/StreamsBuilderTest.java

+        props.setProperty(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9091");
+        props.setProperty(StreamsConfig.STATE_DIR_CONFIG, TestUtils.tempDirectory().getAbsolutePath());
+        props.setProperty(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
+        props.setProperty(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());


There's now a Utils.mkProperties method you can use (in conjunction with mkMap) to set these at the declaration site instead of setting them (redundantly) before every test. Then you won't need the @Before at all.

Yes, that will make it a bit cleaner

vvcephei · 2018-04-26T18:18:33Z

streams/src/test/java/org/apache/kafka/streams/StreamsBuilderTest.java

+            driver.close();
+        }
+        driver = null;
+    }


This cleanup seems a bit awkward. It assumes that tests will initialize the driver but not close it, which seems like a strange abdication of responsibility.

I think it would be cleaner and clearer to get rid of the driver field entirely. Tests that need the driver already initialize it; they can declare it as a local variable as well. Then, they clearly need to close it as well.

Since TopologyTestDriver is AutoCloseable, one option is to declare the driver in try-with-resources style:

@Test public void myTest() { try (final TopologyTestDriver driver) { // the test code } }

Yes, this is definitely better. Using try-with-resources and Utils.mkProperties we won't need @Before and @After

vvcephei · 2018-04-26T18:22:20Z

streams/src/test/java/org/apache/kafka/streams/TopologyTestDriverWrapper.java

 */
-public class InternalTopologyAccessor {
+public class TopologyTestDriverWrapper extends TopologyTestDriver {


Nice solution to this problem.

Thanks. Just following a nice tip from @mjsax .

vvcephei · 2018-04-26T18:23:44Z

streams/src/test/java/org/apache/kafka/streams/kstream/KStreamBuilderTest.java

+        props.setProperty(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9091");
+        props.setProperty(StreamsConfig.STATE_DIR_CONFIG, TestUtils.tempDirectory().getAbsolutePath());
+        props.setProperty(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
+        props.setProperty(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());


Please see my comment in StreamsBuilderTest about mkProperties

vvcephei · 2018-04-26T18:24:13Z

streams/src/test/java/org/apache/kafka/streams/kstream/KStreamBuilderTest.java

+        if (driver != null) {
+            driver.close();
+        }
+        driver = null;


Please see my comment in StreamsBuilderTest about making the driver local.

vvcephei · 2018-04-26T19:03:38Z

...ams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamWindowAggregateTest.java

+        if (driver != null) {
+            driver.close();
+        }
+        driver = null;


ditto on removing before/after.

vvcephei · 2018-04-26T19:04:15Z

streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableForeachTest.java

+        if (driver != null) {
+            driver.close();
+        }
+        driver = null;


ditto on removing before/after.

vvcephei · 2018-04-26T19:05:06Z

streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorTopologyTest.java

    }

    @After
    public void cleanup() {
+        props.clear();


ditto on removing before/after.

vvcephei · 2018-04-26T19:06:02Z

streams/src/test/java/org/apache/kafka/test/KStreamTestDriver.java

@@ -44,6 +44,7 @@
 import java.util.Set;
 import java.util.regex.Pattern;

+@Deprecated


Thanks; do you mind adding a comment recommending to use TopologyTestDriver instead, just to help out new devs coming to the project?

Yes, I'll add it.

vvcephei · 2018-04-26T19:10:17Z

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

+     * @param builder builder for the topology to be tested
+     * @param config the configuration for the topology
+     */
+    protected TopologyTestDriver(final InternalTopologyBuilder builder,


I think this can be package-private, since your wrapper is in the same package.

Yes, it can. I'll change it.

h314to · 2018-04-27T12:47:35Z

Thank you all for your helpful reviews! @vvcephei left some nice tips after this was merged (thanks!), so if that is ok I'd like to address his comments in a followup cleanup PR.

vvcephei · 2018-04-27T15:24:53Z

Hi Filipe, Yeah, I think that's a good solution. Sorry I was too slow on my review! Thanks, -John

…

On Fri, Apr 27, 2018 at 7:47 AM, Filipe Agapito ***@***.***> wrote: Thank you all for your helpful reviews! @vvcephei <https://github.com/vvcephei> left some nice tips after this was merged (thanks!), so if that is ok I'd like to address his comments in a followup cleanup PR. — You are receiving this because you were mentioned. Reply to this email directly, view it on GitHub <#4832 (comment)>, or mute the thread <https://github.com/notifications/unsubscribe-auth/AAy1EwXuf-pSSMrtl-KlrIDwXRqbxyNIks5tsxN-gaJpZM4TJXJ_> .

vvcephei · 2018-04-27T15:27:45Z

Ah, oh well, nevermind then. I missed that nuance.

…

On Fri, Apr 27, 2018 at 10:26 AM, Filipe Agapito ***@***.***> wrote: ***@***.**** commented on this pull request. ------------------------------ In streams/src/test/java/org/apache/kafka/streams/kstream/internals/ KStreamKStreamJoinTest.java <#4832 (comment)>: > @@ -327,17 +341,15 @@ public void testWindowing() { assertEquals(1, copartitionGroups.size()); assertEquals(new HashSet<>(Arrays.asList(topic1, topic2)), copartitionGroups.iterator().next()); - driver.setUp(builder, stateDir); - + driver = new TopologyTestDriver(builder.build(), props, time); While I was changing it I realised that, although it would be cleaner to remove time and use advanceTimeMs instead, it is actually easier to use time because the tests require the time to advance and recede. — You are receiving this because you were mentioned. Reply to this email directly, view it on GitHub <#4832 (comment)>, or mute the thread <https://github.com/notifications/unsubscribe-auth/AAy1Eweeb-r26r6ub5VdmfzpBVF815Ntks5tsziMgaJpZM4TJXJ_> .

@after

…al] (apache#4832) * Remove ProcessorTopologyTestDriver from TopologyTest * Fix ProcessorTopologyTest * Remove ProcessorTopologyTestDriver and InternalTopologyAccessor * Partially refactored StreamsBuilderTest but missing one test * Refactor KStreamBuilderTest * Refactor AbstractStreamTest * Further cleanup of AbstractStreamTest * Refactor GlobalKTableJoinsTest * Refactor InternalStreamsBuilderTest * Fix circular dependency in build.gradle * Refactor KGroupedStreamImplTest * Partial modifications to KGroupedTableImplTest * Refactor KGroupedTableImplTest * Refactor KStreamBranchTest * Refactor KStreamFilterTest * Refactor KStreamFlatMapTest KStreamFlatMapValuesTest * Refactor KStreamForeachTest * Refactor KStreamGlobalKTableJoinTest * Refactor KStreamGlobalKTableLeftJoinTest * Refactor KStreamImplTest * Refactor KStreamImplTest * Refactor KStreamKStreamJoinTest * Refactor KStreamKStreamLeftJoinTest * Refactor KStreamKTableJoinTest * Refactor KStreamKTableLeftJoinTest * Refactor KStreamMapTest and KStreamMapValuesTest * Refactor KStreamPeekTest and KStreamTransformTest * Refactor KStreamSelectKeyTest * Refactor KStreamTransformValuesTest * Refactor KStreamWindowAggregateTest * Add Depercation anotation to KStreamTestDriver and rollback failing tests in StreamsBuilderTest and KTableAggregateTest * Refactor KTableFilterTest * Refactor KTableForeachTest * Add getter for ProcessorTopology, and simplify tests in StreamsBuilderTest * Refactor KTableImplTest * Remove unused imports * Refactor KTableAggregateTest * Fix style errors * Fix gradle build * Address reviewer comments: - Remove properties new instance - Remove extraneous line - Remove unnecessary TopologyTestDriver instances from StreamsBuilderTest - Move props.clear() to @after - Clarify use of timestamp in KStreamFlatMapValuesTest - Keep test using old Punctuator in KStreamTransformTest - Add comment to clarify clock advances in KStreamTransformTest - Add TopologyTestDriverWrapper class to access the protected constructor of TopologyTestDriver - Revert KTableImplTest.testRepartition to KStreamTestDriver to avoid exposing the TopologyTestDriver processor topology - Revert partially migrated classes: KTableAggregateTest, KTableFilterTest, and KTableImplTest * Rebase on current trunk an fix conflicts Reviewers: Matthias J Sax <matthias@confluentio>, Bill Bejeck <bill@confluent.io>, John Roesler <john@confluent.io>

@after

…al] (apache#4832) * Remove ProcessorTopologyTestDriver from TopologyTest * Fix ProcessorTopologyTest * Remove ProcessorTopologyTestDriver and InternalTopologyAccessor * Partially refactored StreamsBuilderTest but missing one test * Refactor KStreamBuilderTest * Refactor AbstractStreamTest * Further cleanup of AbstractStreamTest * Refactor GlobalKTableJoinsTest * Refactor InternalStreamsBuilderTest * Fix circular dependency in build.gradle * Refactor KGroupedStreamImplTest * Partial modifications to KGroupedTableImplTest * Refactor KGroupedTableImplTest * Refactor KStreamBranchTest * Refactor KStreamFilterTest * Refactor KStreamFlatMapTest KStreamFlatMapValuesTest * Refactor KStreamForeachTest * Refactor KStreamGlobalKTableJoinTest * Refactor KStreamGlobalKTableLeftJoinTest * Refactor KStreamImplTest * Refactor KStreamImplTest * Refactor KStreamKStreamJoinTest * Refactor KStreamKStreamLeftJoinTest * Refactor KStreamKTableJoinTest * Refactor KStreamKTableLeftJoinTest * Refactor KStreamMapTest and KStreamMapValuesTest * Refactor KStreamPeekTest and KStreamTransformTest * Refactor KStreamSelectKeyTest * Refactor KStreamTransformValuesTest * Refactor KStreamWindowAggregateTest * Add Depercation anotation to KStreamTestDriver and rollback failing tests in StreamsBuilderTest and KTableAggregateTest * Refactor KTableFilterTest * Refactor KTableForeachTest * Add getter for ProcessorTopology, and simplify tests in StreamsBuilderTest * Refactor KTableImplTest * Remove unused imports * Refactor KTableAggregateTest * Fix style errors * Fix gradle build * Address reviewer comments: - Remove properties new instance - Remove extraneous line - Remove unnecessary TopologyTestDriver instances from StreamsBuilderTest - Move props.clear() to @after - Clarify use of timestamp in KStreamFlatMapValuesTest - Keep test using old Punctuator in KStreamTransformTest - Add comment to clarify clock advances in KStreamTransformTest - Add TopologyTestDriverWrapper class to access the protected constructor of TopologyTestDriver - Revert KTableImplTest.testRepartition to KStreamTestDriver to avoid exposing the TopologyTestDriver processor topology - Revert partially migrated classes: KTableAggregateTest, KTableFilterTest, and KTableImplTest * Rebase on current trunk an fix conflicts Reviewers: Matthias J Sax <matthias@confluentio>, Bill Bejeck <bill@confluent.io>, John Roesler <john@confluent.io>

mjsax added the streams label Apr 8, 2018

mjsax requested review from dguy, mjsax and guozhangwang April 8, 2018 01:13

mjsax requested changes Apr 8, 2018

View reviewed changes

bbejeck reviewed Apr 9, 2018

View reviewed changes

vvcephei requested changes Apr 11, 2018

View reviewed changes

h314to force-pushed the fix/KAFKA-6474 branch from dadcf57 to df0a2b2 Compare April 23, 2018 14:30

h314to changed the title ~~KAFKA-6474: [WIP] Rewrite tests to use new public TopologyTestDriver~~ KAFKA-6474: Rewrite tests to use new public TopologyTestDriver [partial] Apr 23, 2018

bbejeck reviewed Apr 23, 2018

View reviewed changes

h314to force-pushed the fix/KAFKA-6474 branch from df0a2b2 to bc5df4c Compare April 26, 2018 11:00

guozhangwang approved these changes Apr 26, 2018

View reviewed changes

guozhangwang merged commit 885abbf into apache:trunk Apr 26, 2018

vvcephei reviewed Apr 26, 2018

View reviewed changes

h314to deleted the fix/KAFKA-6474 branch April 27, 2018 10:17

h314to mentioned this pull request Apr 27, 2018

KAFKA-6474: Rewrite tests to use new public TopologyTestDriver [cleanup] #4939

Merged

		builder.setApplicationId(APP_ID);
		Properties props = new Properties();

KAFKA-6474: Rewrite tests to use new public TopologyTestDriver [partial] #4832

KAFKA-6474: Rewrite tests to use new public TopologyTestDriver [partial] #4832

Conversation

h314to commented Apr 6, 2018 • edited Loading

mjsax commented Apr 8, 2018

mjsax commented Apr 8, 2018

mjsax 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

h314to Apr 23, 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

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

h314to commented Apr 9, 2018

bbejeck left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

vvcephei commented Apr 11, 2018

vvcephei 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

mjsax commented Apr 17, 2018

h314to commented Apr 23, 2018

bbejeck left a comment

Choose a reason for hiding this comment

bbejeck commented Apr 23, 2018

h314to commented Apr 26, 2018

guozhangwang commented Apr 26, 2018

guozhangwang left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

guozhangwang commented Apr 26, 2018

vvcephei left a comment • 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

Choose a reason for hiding this comment

h314to commented Apr 6, 2018 •

edited

Loading

h314to Apr 23, 2018 •

edited

Loading

vvcephei left a comment •

edited

Loading