KAFKA-5651: [FOLLOW-UP] add with method to Materialized #4009

dguy · 2017-10-03T16:17:34Z

Add a with(Serde keySerde, Serde valSerde) to Materialized for cases where people don't care about the state store name.

dguy · 2017-10-03T16:18:11Z

@mjsax, @guozhangwang, @bbejeck
If we are ok with this then i'll create a JIRA and add an addendum to the KIP

mjsax · 2017-10-03T16:47:41Z

@dguy As it's part of a KIP, I think a Jira is appropriate.

dguy · 2017-10-03T16:49:34Z

@mjsax, yes i was going to create one if we think this is what we want...

guozhangwang

Left a meta comment as another way to address this.

guozhangwang · 2017-10-04T01:29:47Z

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

@@ -24,15 +24,25 @@

 public class KeyValueStoreMaterializer<K, V> {
    private final MaterializedInternal<K, V, KeyValueStore<Bytes, byte[]>> materialized;
+    private final InternalNameProvider streamsBuilder;


nit: variable name to internalNameProvider?

guozhangwang · 2017-10-04T01:32:11Z

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

+package org.apache.kafka.streams.kstream.internals;
+
+public interface InternalNameProvider {
+    String newName(String prefix);


nit: newProcessorName?

guozhangwang · 2017-10-04T01:34:16Z

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

@@ -139,7 +140,7 @@ public InternalStreamsBuilder(final InternalTopologyBuilder internalTopologyBuil
        Objects.requireNonNull(materialized, "materialized can't be null");
        // explicitly disable logging for global stores
        materialized.withLoggingDisabled();
-        final StoreBuilder storeBuilder = new KeyValueStoreMaterializer<>(materialized).materialize();
+        final StoreBuilder storeBuilder = new KeyValueStoreMaterializer<>(materialized, this).materialize(KTableImpl.SOURCE_NAME);


hmm... shouldn't we add KTableImpl.STATE_STORE_NAME also?

EDIT: actually, how about the following: add a protected storeName setter in Materialized, and then we can just add the additional logic in this class by checking the MaterializedInternal object has either storeName or StoreSupplier, and if none is provided call the setter with the store name from newStoreName(SOURCE_NAME).

Then we would not need any other classes' modifications.

I'm not sure i follow

I think it might be better to pass the InternalNameProvider and prefix into MaterializedInternal at construction as this will be required in other places. That way at least the logic will all be within the MaterializedInternal ctor

ijuma · 2017-10-04T10:41:28Z

Btw, you can reuse the same JIRA for the original KIP as well (i.e. a new JIRA is not necessarily required).

mjsax

One nit.

Also, can you update the JavaDoc in KGroupedTable#aggregate(Initializer, Aggregator, Aggregator, Serde); and KGroupedStream#aggregate(Initializer, Aggregator, Serde)

mjsax · 2017-10-04T22:03:48Z

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

        super(materialized);
-        this.queryable = queryable;
+        if (storeName() == null) {
+            this.queryable = false;


Nit: remove this (same below)

mjsax · 2017-10-04T22:35:18Z

streams/src/main/java/org/apache/kafka/streams/kstream/KGroupedStream.java

@@ -1509,7 +1509,7 @@
     * @return a windowed {@link KTable} that contains "update" records with unmodified keys, and values that represent
     * the latest (rolling) aggregate for each key within a window
     * @deprecated use {@link #windowedBy(SessionWindows) windowedBy(sessionWindows)} followed by
-     * {@link SessionWindowedKStream#aggregate(Initializer, Aggregator, Merger, Materialized) aggregate(initializer, aggregator, sessionMerger, Materialized.as("someStoreName").withValueSerde(aggValueSerde))}
+     * {@link SessionWindowedKStream#aggregate(Initializer, Aggregator, Merger, Materialized) aggregate(initializer, aggregator, sessionMerger, Materialized.with(keySerde, aggValueSerde))}


This should be Materialized.with(null, aggValueSerde) -- there is no keySerde parameter in the deprecated method. The idea is to show a 1:1 translation to the new API.

And I forgot, there are three aggregate methods in this class that need this update.

Yep, in that case we also need to set the keySerde on the Materialized to the one provided in KGroupedStream etc. Otherwise it is going to revert to the defaults, which may not be correct

Yes. Using null implies fall back to default -- that is the same behavior as the deprecated method has (the deprecate method does not take a keySerde argument and thus used the default one)

mjsax · 2017-10-04T22:35:36Z

streams/src/main/java/org/apache/kafka/streams/kstream/KGroupedTable.java

@@ -855,7 +855,7 @@
     * @param <VR>          the value type of the aggregated {@link KTable}
     * @return a {@link KTable} that contains "update" records with unmodified keys, and values that represent the
     * latest (rolling) aggregate for each key
-     * @deprecated use {@link #aggregate(Initializer, Aggregator, Aggregator, Materialized) aggregate(initializer, adder, subtractor, Materialized.as("someStoreName").withValueSerde(aggValueSerde))}
+     * @deprecated use {@link #aggregate(Initializer, Aggregator, Aggregator, Materialized) aggregate(initializer, adder, subtractor, Materialized.with(keySerdes, aggValueSerde))}


as above (it's only one method in this class)

mjsax

LGTM

dguy · 2017-10-05T15:31:51Z

retest this please

guozhangwang · 2017-10-05T20:05:32Z

Hmm.. this one looks new to me:

java.lang.AssertionError: Key not found forest
Stacktrace

java.lang.AssertionError: Key not found forest
	at org.junit.Assert.fail(Assert.java:88)
	at org.apache.kafka.streams.integration.QueryableStateIntegrationTest.verifyGreaterOrEqual(QueryableStateIntegrationTest.java:898)
	at org.apache.kafka.streams.integration.QueryableStateIntegrationTest.concurrentAccesses(QueryableStateIntegrationTest.java:400)
	at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
	at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
	at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
	at java.lang.reflect.Method.invoke(Method.java:498)

does that looks familiar to you guys? If we believe it is not related is there a JIRA for it?

dguy · 2017-10-05T20:21:02Z

We've seen that fail periodically over the past couple of months

dguy · 2017-10-05T20:21:34Z

retest this please

mjsax · 2017-10-05T20:21:57Z

There is an old JIRA with incomplete description... https://issues.apache.org/jira/browse/KAFKA-4263 -- pretty sure, it's unrelated to this PR.

dguy · 2017-10-06T17:25:41Z

retest this please

guozhangwang · 2017-10-06T19:16:17Z

retest this please

guozhangwang · 2017-10-06T19:38:40Z

retest this please

guozhangwang · 2017-10-06T22:38:52Z

LGTM. Merged to trunk and 1.0.

Add a `with(Serde keySerde, Serde valSerde)` to `Materialized` for cases where people don't care about the state store name. Author: Damian Guy <damian.guy@gmail.com> Reviewers: Guozhang Wang <wangguoz@gmail.com>, Ismael Juma <ismael@juma.me.uk>, Matthias J. Sax <matthias@confluent.io> Closes #4009 from dguy/materialized (cherry picked from commit 23a0140) Signed-off-by: Guozhang Wang <wangguoz@gmail.com>

add with to Materialized

958f31c

guozhangwang reviewed Oct 4, 2017

View reviewed changes

dguy changed the title ~~MINOR: add with method to Materialized~~ KAFKA-5651: [FOLLOW-UP] add with method to Materialized Oct 4, 2017

dguy added 2 commits October 4, 2017 14:39

move store naming logic into MaterializedInternal

dc6571c

merge trunk

3a2a92c

mjsax reviewed Oct 4, 2017

View reviewed changes

comments

0819626

mjsax reviewed Oct 4, 2017

View reviewed changes

dguy added 2 commits October 4, 2017 16:19

comments

c33e614

address comments

dc8f9f9

mjsax approved these changes Oct 5, 2017

View reviewed changes

mjsax mentioned this pull request Oct 5, 2017

Improving EventDeduplicationLambdaIntegrationTest confluentinc/kafka-streams-examples#18

Closed

asfgit closed this in 23a0140 Oct 6, 2017

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

KAFKA-5651: [FOLLOW-UP] add with method to Materialized #4009

KAFKA-5651: [FOLLOW-UP] add with method to Materialized #4009

dguy commented Oct 3, 2017

dguy commented Oct 3, 2017

mjsax commented Oct 3, 2017

dguy commented Oct 3, 2017

guozhangwang left a comment

guozhangwang Oct 4, 2017

guozhangwang Oct 4, 2017

guozhangwang Oct 4, 2017

guozhangwang Oct 4, 2017

dguy Oct 4, 2017

dguy Oct 4, 2017

ijuma commented Oct 4, 2017

mjsax left a comment

mjsax Oct 4, 2017

mjsax Oct 4, 2017

mjsax Oct 4, 2017

dguy Oct 4, 2017

mjsax Oct 4, 2017

mjsax Oct 4, 2017

mjsax left a comment

dguy commented Oct 5, 2017

guozhangwang commented Oct 5, 2017

dguy commented Oct 5, 2017

dguy commented Oct 5, 2017

mjsax commented Oct 5, 2017 •

edited

Loading

dguy commented Oct 6, 2017

guozhangwang commented Oct 6, 2017

guozhangwang commented Oct 6, 2017

guozhangwang commented Oct 6, 2017

KAFKA-5651: [FOLLOW-UP] add with method to Materialized #4009

KAFKA-5651: [FOLLOW-UP] add with method to Materialized #4009

Conversation

dguy commented Oct 3, 2017

dguy commented Oct 3, 2017

mjsax commented Oct 3, 2017

dguy commented Oct 3, 2017

guozhangwang 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

ijuma commented Oct 4, 2017

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

mjsax left a comment

Choose a reason for hiding this comment

dguy commented Oct 5, 2017

guozhangwang commented Oct 5, 2017

dguy commented Oct 5, 2017

dguy commented Oct 5, 2017

mjsax commented Oct 5, 2017 • edited Loading

dguy commented Oct 6, 2017

guozhangwang commented Oct 6, 2017

guozhangwang commented Oct 6, 2017

guozhangwang commented Oct 6, 2017

mjsax commented Oct 5, 2017 •

edited

Loading