KAFKA-8515: (POC) add state store type in KTable for correct materialization #6885

abbccdda · 2019-06-05T02:18:22Z

More detailed description of your change,
if necessary. The PR title and PR message become
the squashed commit message, so use a separate
comment to ping reviewers.

Summary of testing strategy (including rationale)
for the feature or bug fix. Unit and/or integration
tests are expected for any behaviour change and
system tests should be considered for larger changes.

Committer Checklist (excluded from commit message)

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

mjsax · 2019-06-05T06:03:56Z

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

@@ -114,7 +114,7 @@
                                queryableStoreName,
                                aggregateSupplier,
                                statefulProcessorNode,
-                                builder);
+                                builder, );


Seems you missed passing in the store type? Similar in other classes.

mjsax · 2019-06-05T17:57:54Z

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

@@ -176,9 +170,10 @@ public String queryableStoreName() {

    @Override
    public KTable<K, V> filter(final Predicate<? super K, ? super V> predicate,
-                               final Materialized<K, V, KeyValueStore<Bytes, byte[]>> materialized) {
+                               final Materialized<K, V, KeyValueStore<Bytes, byte[]>> materialized) throws UnsupportedDataTypeException {


This does still enforce passing in a KeyValueStore -- hence, if the KTable is windowed, validateStateStoreMatchingDataType() would fail but there is no alternative to pass in a WindowStore.

It would be a "fix" to avoid unbounded growth of the result table, but still does not allow to materialize a windowed-KTable filter result.

abbccdda · 2019-06-07T22:17:26Z

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

@@ -1002,7 +1003,7 @@ public void kTableNamedMaterializedMapValuesShouldPreserveTopologyStructure() {
        final KTable<Object, Object> table = builder.table("input-topic");
        table.mapValues(
            (readOnlyKey, value) -> null,
-            Materialized.<Object, Object, KeyValueStore<Bytes, byte[]>>as("store-name").withKeySerde(null).withValueSerde(null));


This is the place where we need to remove type assertion for example

abbccdda · 2019-06-10T01:51:04Z

...ms/src/test/java/org/apache/kafka/streams/kstream/internals/TimeWindowedKStreamImplTest.java

@@ -244,6 +246,45 @@ public void shouldMaterializeAggregated() {
        }
    }

+    @Test


This is a sample usage to materialize the windowed KTable. The syntax is still not ideal here, but could work.

abbccdda · 2019-06-10T01:52:04Z

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

                    .withValueSerde(Serdes.Integer()));
        final KTableImpl<String, String, Integer> table3 =
            (KTableImpl<String, String, Integer>) table1.mapValues(
                value -> new Integer(value) * (-1),
-                Materialized.<String, Integer, KeyValueStore<Bytes, byte[]>>as(storeName3)
+                Materialized.<String, Integer, StateStore>as(storeName3)


In terms of refactoring, user needs to use weak type here.

abbccdda

Some of the trade-offs we take during this POC: ProcessorSupplier type weakened from <K, V> to <V> because we need the capability to return both K and Windowed<K> for KTable materialization.

abbccdda · 2019-06-10T02:09:10Z

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

    }

    @Override
-    public Processor<K, Change<V>> get() {
-        return new KTableFilterProcessor();
+    public Processor<?, Change<V>> get() {


Use KTableFilter as an example. We need to return different processor upon the state store type for all the supplier get function calls. This is because the assumption of getting a key-value store is no longer valid.

abbccdda · 2019-06-10T19:48:53Z

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

@@ -1138,12 +1137,19 @@
     */
    <VO, VR> KTable<K, VR> outerJoin(final KTable<K, VO> other,
                                     final ValueJoiner<? super V, ? super VO, ? extends VR> joiner,
-                                     final Materialized<K, VR, KeyValueStore<Bytes, byte[]>> materialized);
+                                     final Materialized<K, VR, StateStore> materialized);


The whole KTable API is generified by reducing KeyValueStore to StateStore. Instead of compile time type checking, we pushed the type check internally

abbccdda · 2019-06-10T19:49:55Z

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


    private final String queryableStoreName;

+    private final StateStoreType stateStoreType;
+
+    // Needed if given KTable needs to be materialized as window store.


Put timeWindow/sessionWindow as optional fields with KTable API. This gives us access to utilize those information to materialize the table.

abbccdda · 2019-06-10T19:50:15Z

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

@@ -22,18 +22,22 @@
 public class WindowedSerdes {

    static public class TimeWindowedSerde<T> extends Serdes.WrapperSerde<Windowed<T>> {
+        final Serde<T> inner;


Make windowed serde capable of returning raw serde type

abbccdda · 2019-06-10T19:51:04Z

...ms/src/main/java/org/apache/kafka/streams/state/internals/MeteredTimestampedWindowStore.java

@@ -50,9 +51,18 @@
    @SuppressWarnings("unchecked")
    @Override
    void initStoreSerde(final ProcessorContext context) {
+        final Serde<K> rawKeySerde;


Get raw keySerde back when we are materializing window store.

abbccdda · 2019-06-10T19:54:38Z

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

@@ -318,15 +318,6 @@ public void shouldNotAcceptInvalidStoreNameWhenReducingSessionWindows() {
            .reduce(MockReducer.STRING_ADDER, Materialized.as(INVALID_STORE_NAME));
    }

-    @Test(expected = NullPointerException.class)


This will fail at compile time because of vague type.

mjsax reviewed Jun 5, 2019

View reviewed changes

add state store type

db7aecd

abbccdda force-pushed the flagged_table branch from f945644 to db7aecd Compare June 7, 2019 21:56

abbccdda commented Jun 7, 2019

View reviewed changes

abbccdda force-pushed the flagged_table branch from b6e39de to 2165d09 Compare June 8, 2019 00:26

need auditoring

8330b63

abbccdda force-pushed the flagged_table branch from 2165d09 to 8330b63 Compare June 8, 2019 01:02

demo

d22edfc

abbccdda force-pushed the flagged_table branch 2 times, most recently from ed66614 to 68217af Compare June 8, 2019 15:52

abbccdda added 2 commits June 8, 2019 13:50

change base type

a642240

predicate

7fc1bc1

abbccdda force-pushed the flagged_table branch from 68217af to 7fc1bc1 Compare June 8, 2019 21:10

abbccdda added 2 commits June 8, 2019 17:09

try name

612da44

stats

f9ff670

abbccdda changed the title ~~POC: add state store type to avoid user using wrong state store type~~ KAFKA-8515: (POC) add state store type in KTable for correct materialization Jun 10, 2019

working example

b92f467

abbccdda force-pushed the flagged_table branch from 9f95ce1 to b92f467 Compare June 10, 2019 01:48

abbccdda commented Jun 10, 2019

View reviewed changes

add session stores

ed4f82e

abbccdda force-pushed the flagged_table branch from 1f003a9 to ed4f82e Compare June 10, 2019 06:10

abbccdda commented Jun 10, 2019

View reviewed changes

style

960f57f

abbccdda closed this Jun 21, 2019

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

KAFKA-8515: (POC) add state store type in KTable for correct materialization #6885

KAFKA-8515: (POC) add state store type in KTable for correct materialization #6885

abbccdda commented Jun 5, 2019

mjsax Jun 5, 2019

mjsax Jun 5, 2019

abbccdda Jun 7, 2019

abbccdda Jun 10, 2019

abbccdda Jun 10, 2019

abbccdda left a comment

abbccdda Jun 10, 2019

abbccdda Jun 10, 2019

abbccdda Jun 10, 2019

abbccdda Jun 10, 2019

abbccdda Jun 10, 2019

abbccdda Jun 10, 2019

@@ @@ -244,6 +246,45 @@ public void shouldMaterializeAggregated() { @@
                       }
                   }
+                  @Test

KAFKA-8515: (POC) add state store type in KTable for correct materialization #6885

KAFKA-8515: (POC) add state store type in KTable for correct materialization #6885

Conversation

abbccdda commented Jun 5, 2019

Committer Checklist (excluded from commit message)

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

abbccdda 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