KAFKA-7021: Reuse source based on config #5163

guozhangwang · 2018-06-07T23:41:30Z

This PR actually contains two changes:

leverage on the TOPOLOGY_OPTIMIZATION config to "adjust" the topology internally to reuse the source topic.
fixed a long dangling bug that whenever source topic is reused as changelog topic, write the checkpoint file for the consumed offset, this is done by union the ackedOffset from the producer, plus the consumed offset from the consumer, note we will priori ackedOffset since the same topic may show up in both (think about repartition topic), by doing this the consumed offset from source topics can be treated as checkpointed offset when reuse happens.
added a few unit and integration tests with / wo the reusing, and make sure the restoration, standby task, and internal topic creation behaviors are all correct.

Committer Checklist (excluded from commit message)

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

guozhangwang · 2018-06-07T23:42:28Z

@abbccdda @Ishiihara

@vvcephei @mjsax @bbejeck

mjsax

Couple of comments/question.

I think I understand to overall idea and it makes sense -- need think about some details a little more to really fully understand the change.

mjsax · 2018-06-08T00:26:46Z

streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java

+    /**
+     * Config value for parameter (@link #TOPOLOGY_OPTIMIZATION "topology.optimization" for enabling topology optimization
+     */
+    public static final String OPTIMIZE_AT_20 = "2.0";


Why do we need this? Should we not use "all" ? Also, this would be public API change -- we can still update the corresponding KIP, but I don't think we need this, do we?

I agree. Since this our first optimization, "all" should be sufficient. But this raises another good point if we plan to track optimization versions are we setting ourselves up for a KIP for each optimization release?

Sounds good. I'm reverting to use all.

Moving forward, not needing one KIP for each release. We can use a single KIP adding the mechanism, and if two versions are compatible (say if no new incompatible optimizations introduced from X to Y), we can just document it in upgrade section.

mjsax · 2018-06-08T00:27:33Z

streams/src/main/java/org/apache/kafka/streams/Topology.java

+        final boolean enableOptimization20 = config.getString(StreamsConfig.TOPOLOGY_OPTIMIZATION).equals(StreamsConfig.OPTIMIZE_AT_20);
+
+        if (enableOptimization20) {
+            for (Map.Entry<StoreBuilder, String> entry : internalTopologyBuilder.storeToSourceChangelogTopic.entrySet()) {


nit: add final

mjsax · 2018-06-08T00:35:11Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopologyBuilder.java

        if (storeToChangelogTopic.containsKey(sourceStoreName)) {
            throw new TopologyException("Source store " + sourceStoreName + " is already added.");
        }
        storeToChangelogTopic.put(sourceStoreName, topic);
    }

-    // TODO: this method is only used by DSL and we might want to refactor this part


Why did you remove those comments? I still think, we should refactor some parts here (also applies to other TODOs below).

After reviewed Bill's part II PR for optimization, which enhanced the InternalStreamsBuilder for logical plan generation, I think it makes less sense to refactor these into it.

mjsax · 2018-06-08T00:37:31Z

streams/src/test/java/org/apache/kafka/streams/integration/RestoreIntegrationTest.java

@@ -112,24 +121,106 @@ public void shutdown() {
    }

    @Test
-    public void shouldRestoreState() throws ExecutionException, InterruptedException {
+    public void shouldRestoreStateFromSourceTopic() throws InterruptedException, IOException {


nit: simplify InterruptedException, IOException to Exception

I disagree. It can be nicer for the reader to know which checked exceptions we're potentially going to throw.

But it doesn't matter much, so do what you want ;)

For API calls I agree, but for testing, I think to collapse to Exception is ok.

For regular API calls (ie, "main code") we should list them. For a test, no exception should ever be thrown, and if, the test fails. Which exception is irrelevant IMHO.

mjsax · 2018-06-08T00:39:15Z

streams/src/test/java/org/apache/kafka/streams/integration/TableTableJoinIntegrationTest.java

-        leftTable = builder.table(INPUT_TOPIC_LEFT);
-        rightTable = builder.table(INPUT_TOPIC_RIGHT);
+        leftTable = builder.table(INPUT_TOPIC_LEFT, Materialized.<Long, String, KeyValueStore<Bytes, byte[]>>as("left").withLoggingDisabled());
+        rightTable = builder.table(INPUT_TOPIC_RIGHT, Materialized.<Long, String, KeyValueStore<Bytes, byte[]>>as("right").withLoggingDisabled());


Do we need to disable logging explicitly here?

Yes. We want to not restore the stores in between tests (it is tricker to delete the changelogs since appId changes for each case, so I think this is easier).

mjsax · 2018-06-08T00:41:43Z

streams/src/main/java/org/apache/kafka/streams/kstream/internals/InternalStreamsBuilder.java

@@ -72,11 +72,7 @@ public InternalStreamsBuilder(final InternalTopologyBuilder internalTopologyBuil
    public <K, V> KTable<K, V> table(final String topic,
                                     final ConsumedInternal<K, V> consumed,
                                     final MaterializedInternal<K, V, KeyValueStore<Bytes, byte[]>> materialized) {
-        // explicitly disable logging for source table materialized stores
-        materialized.withLoggingDisabled();


Why do we remove this?

The actual fix works as following: in the parsing phase, we create the materialized store normally, which means not enforcing disable logging. And then in adjust we will modify the store, to 1) disable logging, 2) change its changelog topic to the source topic.

As I mentioned in another HOTFIX PR, note that the term "changelog" topic now has two semantics meaning: 1) for restoring, 2) for appending updates to. For example with optimizations turned on, the changelog for 1) will be the source topic, and for 2) it will be none. There are some tricky separate code paths handling those cases now.

mjsax · 2018-06-08T00:42:58Z

streams/src/main/java/org/apache/kafka/streams/Topology.java

+    // Adjust the generated topology based on the configs.
+    // Not exposed as public API and should be removed post 2.0
+    Topology adjust(final StreamsConfig config) {
+        final boolean enableOptimization20 = config.getString(StreamsConfig.TOPOLOGY_OPTIMIZATION).equals(StreamsConfig.OPTIMIZE_AT_20);


revert the check to avoid NPE: StreamsConfig.OPTIMIZE_AT_20.equals(config.getString(StreamsConfig.TOPOLOGY_OPTIMIZATION))

I'm not sure I'm reading this right... If I say optimize:=all, we won't do this optimization, right? It seems like you want to check for 2.0 or all

mjsax · 2018-06-08T00:44:37Z

streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java

@@ -535,7 +535,7 @@ public void onRestoreEnd(final TopicPartition topicPartition, final String store
     */
    public KafkaStreams(final Topology topology,
                        final Properties props) {
-        this(topology.internalTopologyBuilder, new StreamsConfig(props), new DefaultKafkaClientSupplier());
+        this(topology.adjust(new StreamsConfig(props)).internalTopologyBuilder, new StreamsConfig(props), new DefaultKafkaClientSupplier());


why do we call adjust in each constructor? Doing it once in the private constructor should be sufficient?

I could be missing something, but the private constructor takes an InternalTopologyBuilder instance and adjust needs to execute to enable source topic reuse before returning the InternalTopologyBuilder.
But maybe we could shift the adjust method to the InternalTopologyBuilder thus only need to call once in the private constructor.

What I originally thought is exactly as @bbejeck explained. I think changing to InternalTopologyBuilder would be a better idea, will try it out.

mjsax · 2018-06-08T00:57:23Z

streams/src/main/java/org/apache/kafka/streams/Topology.java

+                // update store map to disable logging for this store
+                storeBuilder.withLoggingDisabled();
+                internalTopologyBuilder.addStateStore(storeBuilder, true);
+                internalTopologyBuilder.connectSourceStoreAndTopic(storeBuilder.name(), topicName);


Just for my understanding: we are overwriting the changelog-topic name with the source-topic name here?

mjsax · 2018-06-08T00:59:23Z

streams/src/main/java/org/apache/kafka/streams/Topology.java

+
+                // update store map to disable logging for this store
+                storeBuilder.withLoggingDisabled();
+                internalTopologyBuilder.addStateStore(storeBuilder, true);


why do we need to add the store again? Isn't storeBuilder the same object that is already present and we overwrite it with itself here? storeBuilder.withLoggingDisabled() is a mutable call, right?

Unfortunately not.. note that addStateStore directly constructs the StoreFactory and put that into the map stateFactories. And StoreFactory is immutable. So I have to overwrite the key with a new StoreFactory in this way.

vvcephei · 2018-06-08T14:50:57Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopologyBuilder.java

@@ -121,6 +121,9 @@

    private Map<Integer, Set<String>> nodeGroups = null;

+    // this is only temporary for 2.0 and should be removed


Maybe mark it TODO so we can search for it later.

vvcephei · 2018-06-08T14:58:56Z

I left a few comments, and I was also curious about several of @mjsax 's concerns, but assuming you hash those out, I'm +1 overall.

bbejeck

Overall I agree with the approach. Just left a couple of minor comments. I'll probably take another pass over the tests later to confirm my understanding of the offset restore fix

bbejeck · 2018-06-08T16:34:35Z

streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java

@@ -535,7 +535,7 @@ public void onRestoreEnd(final TopicPartition topicPartition, final String store
     */
    public KafkaStreams(final Topology topology,
                        final Properties props) {
-        this(topology.internalTopologyBuilder, new StreamsConfig(props), new DefaultKafkaClientSupplier());
+        this(topology.adjust(new StreamsConfig(props)).internalTopologyBuilder, new StreamsConfig(props), new DefaultKafkaClientSupplier());


I could be missing something, but the private constructor takes an InternalTopologyBuilder instance and adjust needs to execute to enable source topic reuse before returning the InternalTopologyBuilder.
But maybe we could shift the adjust method to the InternalTopologyBuilder thus only need to call once in the private constructor.

bbejeck · 2018-06-08T16:37:29Z

streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java

+    /**
+     * Config value for parameter (@link #TOPOLOGY_OPTIMIZATION "topology.optimization" for enabling topology optimization
+     */
+    public static final String OPTIMIZE_AT_20 = "2.0";


I agree. Since this our first optimization, "all" should be sufficient. But this raises another good point if we plan to track optimization versions are we setting ourselves up for a KIP for each optimization release?

bbejeck · 2018-06-08T16:45:25Z

streams/src/test/java/org/apache/kafka/streams/integration/RestoreIntegrationTest.java

@@ -112,24 +121,106 @@ public void shutdown() {
    }

    @Test
-    public void shouldRestoreState() throws ExecutionException, InterruptedException {
+    public void shouldRestoreStateFromSourceTopic() throws InterruptedException, IOException {


For API calls I agree, but for testing, I think to collapse to Exception is ok.

guozhangwang · 2018-06-08T00:08:43Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractTask.java

@@ -167,7 +167,7 @@ public String toString(final String indent) {
        return sb.toString();
    }

-    protected Map<TopicPartition, Long> recordCollectorOffsets() {
+    protected Map<TopicPartition, Long> activeTaskCheckpointableOffsets() {


nit renaming.

guozhangwang · 2018-06-08T00:09:41Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java

-        return recordCollector.offsets();
+    protected Map<TopicPartition, Long> activeTaskCheckpointableOffsets() {
+        final Map<TopicPartition, Long> checkpointableOffsets = recordCollector.offsets();
+        for (Map.Entry<TopicPartition, Long> entry : consumedOffsets.entrySet()) {


This is the actual fix.

nit: add final

…reuse-source-based-on-config

mjsax · 2018-06-08T19:58:16Z

@guozhangwang Build failed with checkstyle error

guozhangwang · 2018-06-08T23:46:18Z

Found the issue with failed system test, filed another PR: #5170

mjsax · 2018-06-10T20:40:46Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java

-        return recordCollector.offsets();
+    protected Map<TopicPartition, Long> activeTaskCheckpointableOffsets() {
+        final Map<TopicPartition, Long> checkpointableOffsets = recordCollector.offsets();
+        for (Map.Entry<TopicPartition, Long> entry : consumedOffsets.entrySet()) {


nit: add final

mjsax · 2018-06-10T20:46:01Z

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

+
+        assertThat(internalTopologyBuilder.build().storeToChangelogTopic(), equalTo(Collections.singletonMap("store", "appId-store-changelog")));
+
+        assertThat(internalTopologyBuilder.getStateStores().keySet(), equalTo(Collections.singleton("store")));


Should we add this check to shouldReuseSourceTopicAsChangelogsWithOptimization20, too?

mjsax · 2018-06-10T20:47:24Z

streams/src/test/java/org/apache/kafka/streams/integration/RestoreIntegrationTest.java

+        final int offsetLimitDelta = 1000;
+        final int offsetCheckpointed = 1000;
+        createStateForRestoration(INPUT_STREAM);
+        setCommittedOffset(INPUT_STREAM, offsetLimitDelta);


Why do we use a delta, here, instead of the actual offset we want to commit (ie, 4000) ? Might be simpler to understand the test?

The reason is that in line 150, I need to check exactly how many data were processed, which should be the diff between committed offset to log end offset.

However, for log end offset it is not always numKeys / 2 (num.partitions). In my tests I saw for example 5005 and 4995. So I have to use the limit to say commit at the log end offset minus that delta.

mjsax · 2018-06-10T20:48:53Z

streams/src/test/java/org/apache/kafka/streams/integration/RestoreIntegrationTest.java

+        final CountDownLatch shutdownLatch = new CountDownLatch(1);
+
+        builder.table(INPUT_STREAM, Consumed.with(Serdes.Integer(), Serdes.Integer()), Materialized.as("store"))
+                .toStream()


nit: indention should only be 4 spaces?

mjsax · 2018-06-10T20:51:14Z

streams/src/test/java/org/apache/kafka/streams/integration/RestoreIntegrationTest.java

-        final List<TopicPartition> partitions = Arrays.asList(new TopicPartition(INPUT_STREAM, 0),
-                                                              new TopicPartition(INPUT_STREAM, 1));
+        final List<TopicPartition> partitions = Arrays.asList(new TopicPartition(topic, 0),
+                new TopicPartition(topic, 1));

        consumer.assign(partitions);
        consumer.seekToEnd(partitions);


If we rewrite to use "absolute position" instead of delta, we can remove this.

…reuse-source-based-on-config

mjsax

LGTM. Thanks for the patch.

guozhangwang · 2018-06-11T23:09:01Z

Thanks for the reviews, I've merged it to trunk.

Will provide a separate PR for upgrade docs, and have separate PRs for older branches.

This PR actually contains two changes: 1. leverage on the TOPOLOGY_OPTIMIZATION config to "adjust" the topology internally to reuse the source topic. 2. fixed a long dangling bug that whenever source topic is reused as changelog topic, write the checkpoint file for the consumed offset, this is done by union the ackedOffset from the producer, plus the consumed offset from the consumer, note we will priori ackedOffset since the same topic may show up in both (think about repartition topic), by doing this the consumed offset from source topics can be treated as checkpointed offset when reuse happens. 3. added a few unit and integration tests with / wo the reusing, and make sure the restoration, standby task, and internal topic creation behaviors are all correct. Reviewers: John Roesler <[email protected]>, Bill Bejeck <[email protected]>, Matthias J. Sax <[email protected]>

guozhangwang · 2018-06-12T21:39:38Z

Cherry-picked to 2.0

This PR actually contains two changes: 1. leverage on the TOPOLOGY_OPTIMIZATION config to "adjust" the topology internally to reuse the source topic. 2. fixed a long dangling bug that whenever source topic is reused as changelog topic, write the checkpoint file for the consumed offset, this is done by union the ackedOffset from the producer, plus the consumed offset from the consumer, note we will priori ackedOffset since the same topic may show up in both (think about repartition topic), by doing this the consumed offset from source topics can be treated as checkpointed offset when reuse happens. 3. added a few unit and integration tests with / wo the reusing, and make sure the restoration, standby task, and internal topic creation behaviors are all correct. Reviewers: John Roesler <[email protected]>, Bill Bejeck <[email protected]>, Matthias J. Sax <[email protected]>

#6021) Updating the documentation for table operation because I believe it is incorrect. In PR #5163 the table operation stopped disabling the changelog topic by default and instead moved that optimization to a configuration that is not enabled by default. This PR updates the documentation to reflect the change in behavior and point to the new configuration for optimization. Reviewers: Bill Bejeck <[email protected]>, Guozhang Wang <[email protected]>

apache#6021) Updating the documentation for table operation because I believe it is incorrect. In PR apache#5163 the table operation stopped disabling the changelog topic by default and instead moved that optimization to a configuration that is not enabled by default. This PR updates the documentation to reflect the change in behavior and point to the new configuration for optimization. Reviewers: Bill Bejeck <[email protected]>, Guozhang Wang <[email protected]>

guozhangwang added 3 commits June 7, 2018 12:59

add adjust logic

8a15f25

add unit tests for the change

6fcb570

remove imports

6d2efab

nits

bf776a2

mjsax added the streams label Jun 8, 2018

mjsax reviewed Jun 8, 2018

View reviewed changes

vvcephei reviewed Jun 8, 2018

View reviewed changes

bbejeck reviewed Jun 8, 2018

View reviewed changes

guozhangwang commented Jun 8, 2018

View reviewed changes

guozhangwang changed the title ~~HOTFIX: Reuse source based on config~~ KAFKA-7021: Reuse source based on config Jun 8, 2018

guozhangwang added 2 commits June 8, 2018 11:23

Merge branch 'trunk' of https://github.com/apache/kafka into KHotfix-…

b274c88

…reuse-source-based-on-config

address github comments

f606736

remove imports

549cc40

remove imports

1007cb0

mjsax reviewed Jun 10, 2018

View reviewed changes

guozhangwang added 4 commits June 11, 2018 09:37

Merge branch 'trunk' of https://github.com/apache/kafka into KHotfix-…

1b3ccbc

…reuse-source-based-on-config

github comments

478cad9

remove imports

c9d80b9

remove imports

ca40f17

vvcephei approved these changes Jun 11, 2018

View reviewed changes

mjsax approved these changes Jun 11, 2018

View reviewed changes

bbejeck approved these changes Jun 11, 2018

View reviewed changes

guozhangwang merged commit d98ec33 into apache:trunk Jun 11, 2018

guozhangwang deleted the KHotfix-reuse-source-based-on-config branch June 11, 2018 23:09

bbejeck mentioned this pull request Jun 12, 2018

KAFKA-6761: Construct Physical Plan using Graph, Reduce streams footprint part III #5201

Merged

3 tasks

cwildman mentioned this pull request Dec 10, 2018

MINOR: Update documentation for internal changelog when using table(). #6021

Merged

3 tasks

		@@ -121,6 +121,9 @@

		private Map<Integer, Set<String>> nodeGroups = null;

		// this is only temporary for 2.0 and should be removed


		assertThat(internalTopologyBuilder.build().storeToChangelogTopic(), equalTo(Collections.singletonMap("store", "appId-store-changelog")));

		assertThat(internalTopologyBuilder.getStateStores().keySet(), equalTo(Collections.singleton("store")));

KAFKA-7021: Reuse source based on config #5163

KAFKA-7021: Reuse source based on config #5163

Conversation

guozhangwang commented Jun 7, 2018 • edited Loading

Committer Checklist (excluded from commit message)

guozhangwang commented Jun 7, 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

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

vvcephei commented Jun 8, 2018

bbejeck 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

mjsax commented Jun 8, 2018

guozhangwang commented Jun 8, 2018

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

guozhangwang Jun 11, 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

mjsax left a comment

Choose a reason for hiding this comment

guozhangwang commented Jun 11, 2018

guozhangwang commented Jun 12, 2018

guozhangwang commented Jun 7, 2018 •

edited

Loading

guozhangwang Jun 11, 2018 •

edited

Loading