MINOR: remove unnecessary store info from TopologyBuilder by guozhangwang · Pull Request #2338 · apache/kafka

guozhangwang · 2017-01-10T06:25:17Z

This PR is extracted from #2333 as an incremental fix to ease the reviewing:

Removed storeToProcessorNodeMap from ProcessorTopology since it was previously used to set the context current record, and can now be replaced with the dirty entry in the named cache.
Replaced sourceStoreToSourceTopic from ProcessorTopology with storeToChangelogTopic map, which includes the corresponding changelog topic name for all stores that are changelog enabled.
Modified ProcessorStateManager to rely on sourceStoreToSourceTopic when retrieving the changelog topic; this makes the second parameter loggingEnabled in register not needed any more, and we can deprecate the old API with a new one.
Also fixed a minor issue in KStreamBuilder: if the storeName is not provided in the table(..) function, do not create the underlying materialized store. Modified the unit tests to cover this case.
Fixed a bunch of other unit tests failures that are exposed by this refactoring, in which we are not setting the applicationId correctly when constructing the mocking processor topology.

guozhangwang · 2017-01-10T06:25:32Z

ping @dguy @mjsax @enothereska for reviews.

asfbot · 2017-01-10T07:05:11Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/655/
Test PASSed (JDK 8 and Scala 2.12).

asfbot · 2017-01-10T07:05:59Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/656/
Test PASSed (JDK 8 and Scala 2.11).

asfbot · 2017-01-10T07:12:55Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.10/654/
Test PASSed (JDK 7 and Scala 2.10).

dguy

LGTM

mjsax · 2017-01-10T22:14:52Z

-        } finally {
-            currNode = current;
-
+        for (StateStore stateStore : context.allStateStores().values()) {


Why do we not need currNode to map to the store we are flushing any more? I don't see any change in StateStore#flush() -- or did we not need this in the first place and this is just a cleanup of unnecessary code?

This is a leftover work that was not introduced in this PR: before we use the topology.storeToProcessorNodeMap() to remember the linked processor node of the state store, so we can set it while flushing and possibly forwarding to children processor nodes. When we add KIP-63 we store that information in DirtyEntry so we do not need this any more, but forgot to remove it.

This cleanup is mainly for removing those not-needed-anymore maps from topology builder

mjsax · 2017-01-10T22:16:19Z

        final KStreamBuilder builder = new KStreamBuilder();
        builder.stream("topic").groupByKey().count("my-store");
-        final ProcessorTopology topology = builder.build(0);
+        final ProcessorTopology topology = builder.setApplicationId(applicationId).build(0);


Why was a missing application ID not a problem before the change? I don't see the connection.

Explained as above. We are now solely depending on the topology builder to give us the information about which state stores should be changelog enabled and what are their changelog topic names. Before that we have a separate loggingEnabled parameter in register to indicate that.

mjsax · 2017-01-10T22:19:20Z

+                    put(storeName1, storeChangelogTopicName1);
+                    put(storeName2, storeChangelogTopicName2);
+                }
+            });


I don't get this change? (same blow). Why do we to provide this now?

We are now passing the map from store names to its changlog topics as the last parameter into ProcessorStateManager, since we can actually gather that information beforehand in the topology builder already (any stores that are not changelog enabled won't have an entry in the map). So in ProcessorStateManager it only needs to access this map to decide whether we need to get a changelog for the state store and what would be the changelog name.

In these tests since we are adding two stores persistentXX and nonPersistentXX which are all changelogged, we need to pass in this map.

Yes. But in the original version of this test we pass in two empty maps (sourceStoreToSourceTopics and storeToProcessorNodeMap) -- so I am wondering why we do need to provide this information now but not before (or where did this information come from before). Still confused.

mjsax · 2017-01-10T22:22:12Z

        Set<TopicPartition> sourcePartitions = Utils.mkSet(new TopicPartition(storeTopicName3, 1));

-        ProcessorStateManager stateMgr = new ProcessorStateManager(applicationId, taskId, sourcePartitions, restoreConsumer, true, stateDirectory, null, Collections.<StateStore, ProcessorNode>emptyMap()); // standby
+        ProcessorStateManager stateMgr = new ProcessorStateManager(applicationId, taskId, sourcePartitions, restoreConsumer, true, stateDirectory, storeToChangelogTopic); // standby


The test worked with an empty map before -- why do we need to provide the mapping now? (same below)

Ditto above.

mjsax · 2017-01-10T22:24:53Z

+        final KTableImpl<K, ?, V> kTable = new KTableImpl<>(this, name, processorSupplier, Collections.singleton(source), storeName);
+
+        // only materialize the KTable into a state store if the storeName is not null
+        if (storeName != null) {


I don't see a new test for this.

My bad, this is not committed.

…into KMinor-refactor-state-to-changelogtopic

asfbot · 2017-01-11T20:15:15Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/756/
Test PASSed (JDK 8 and Scala 2.12).

asfbot · 2017-01-11T20:16:31Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/758/
Test PASSed (JDK 8 and Scala 2.11).

asfbot · 2017-01-11T20:25:03Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.10/756/
Test FAILed (JDK 7 and Scala 2.10).

asfbot · 2017-01-12T00:02:21Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/768/
Test FAILed (JDK 8 and Scala 2.11).

asfbot · 2017-01-12T00:10:13Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/766/
Test FAILed (JDK 8 and Scala 2.12).

asfbot · 2017-01-12T02:27:17Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.10/766/
Test FAILed (JDK 7 and Scala 2.10).

guozhangwang · 2017-01-12T04:37:51Z

retest this please.

asfbot · 2017-01-12T05:16:59Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/776/
Test FAILed (JDK 8 and Scala 2.12).

asfbot · 2017-01-12T05:21:35Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/778/
Test PASSed (JDK 8 and Scala 2.11).

asfbot · 2017-01-12T05:28:06Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.10/776/
Test FAILed (JDK 7 and Scala 2.10).

guozhangwang · 2017-01-12T06:59:15Z

retest this please

asfbot · 2017-01-12T07:44:17Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/779/
Test PASSed (JDK 8 and Scala 2.12).

asfbot · 2017-01-12T07:55:41Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/781/
Test PASSed (JDK 8 and Scala 2.11).

asfbot · 2017-01-12T09:59:45Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.10/779/
Test FAILed (JDK 7 and Scala 2.10).

asfbot · 2017-01-13T19:20:38Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.10/851/
Test FAILed (JDK 7 and Scala 2.10).

ewencp

@guozhangwang One trivial typo, but LGTM

ewencp · 2017-01-13T19:18:21Z

    private final Map<String, Set<String>> stateStoreNameToSourceTopics = new HashMap<>();

-    // map from state store names that are directly associated with source processors to their subscribed topics,
+    // map from state store names to ths state store's corresponding changelog topic if possible,


nit: typo "ths"

guozhangwang · 2017-01-13T19:29:03Z

Merged to trunk.

asfbot · 2017-01-13T19:48:42Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/853/
Test PASSed (JDK 8 and Scala 2.11).

asfbot · 2017-01-13T20:11:07Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/853/
Test FAILed (JDK 8 and Scala 2.12).

asfbot · 2017-01-13T20:14:30Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/855/
Test PASSed (JDK 8 and Scala 2.11).

asfbot · 2017-01-13T20:46:12Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.10/853/
Test FAILed (JDK 7 and Scala 2.10).

This PR is extracted from apache#2333 as an incremental fix to ease the reviewing: 1. Removed `storeToProcessorNodeMap` from ProcessorTopology since it was previously used to set the context current record, and can now be replaced with the dirty entry in the named cache. 2. Replaced `sourceStoreToSourceTopic` from ProcessorTopology with `storeToChangelogTopic` map, which includes the corresponding changelog topic name for all stores that are changelog enabled. 3. Modified `ProcessorStateManager` to rely on `sourceStoreToSourceTopic` when retrieving the changelog topic; this makes the second parameter `loggingEnabled` in `register` not needed any more, and we can deprecate the old API with a new one. 4. Also fixed a minor issue in `KStreamBuilder`: if the storeName is not provided in the `table(..)` function, do not create the underlying materialized store. Modified the unit tests to cover this case. 5. Fixed a bunch of other unit tests failures that are exposed by this refactoring, in which we are not setting the applicationId correctly when constructing the mocking processor topology. Author: Guozhang Wang <wangguoz@gmail.com> Reviewers: Damian Guy, Matthias J. Sax, Ewen Cheslack-Postava Closes apache#2338 from guozhangwang/KMinor-refactor-state-to-changelogtopic

guozhangwang added 2 commits January 9, 2017 21:31

refactor topology builder

2073cd2

fix unit tests

675caca

dguy approved these changes Jan 10, 2017

View reviewed changes

mjsax reviewed Jan 10, 2017

View reviewed changes

guozhangwang added 3 commits January 10, 2017 16:46

Merge branch 'trunk' of https://git-wip-us.apache.org/repos/asf/kafka …

c33475b

…into KMinor-refactor-state-to-changelogtopic

add unit tests

dba8b5d

minor fix

3d05f01

rebase on apache trunk head

46f362b

guozhangwang mentioned this pull request Jan 13, 2017

KAFKA-3452: follow up - fix state store restoration for session and window stores #2359

Closed

rebase from apache trunk

d2d588d

mjsax approved these changes Jan 13, 2017

View reviewed changes

ewencp approved these changes Jan 13, 2017

View reviewed changes

fix typo

07baf06

asfgit closed this in e90db3e Jan 13, 2017

guozhangwang deleted the KMinor-refactor-state-to-changelogtopic branch July 15, 2017 22:07

Conversation

guozhangwang commented Jan 10, 2017

Uh oh!

guozhangwang commented Jan 10, 2017

Uh oh!

asfbot commented Jan 10, 2017

Uh oh!

asfbot commented Jan 10, 2017

Uh oh!

asfbot commented Jan 10, 2017

Uh oh!

dguy left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

mjsax Jan 11, 2017 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

asfbot commented Jan 11, 2017

Uh oh!

asfbot commented Jan 11, 2017

Uh oh!

asfbot commented Jan 11, 2017

Uh oh!

asfbot commented Jan 12, 2017

Uh oh!

asfbot commented Jan 12, 2017

Uh oh!

asfbot commented Jan 12, 2017

Uh oh!

guozhangwang commented Jan 12, 2017

Uh oh!

asfbot commented Jan 12, 2017

Uh oh!

asfbot commented Jan 12, 2017

Uh oh!

asfbot commented Jan 12, 2017

Uh oh!

guozhangwang commented Jan 12, 2017

Uh oh!

asfbot commented Jan 12, 2017

Uh oh!

asfbot commented Jan 12, 2017

Uh oh!

asfbot commented Jan 12, 2017

Uh oh!

asfbot commented Jan 13, 2017

Uh oh!

ewencp left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

guozhangwang commented Jan 13, 2017

Uh oh!

asfbot commented Jan 13, 2017

Uh oh!

asfbot commented Jan 13, 2017

mjsax Jan 11, 2017 •

edited

Loading