kafka-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From guozh...@apache.org
Subject kafka git commit: KAFKA-3817: KTableRepartitionMap publish old Change first, for non-count aggregates
Date Fri, 05 Aug 2016 17:03:26 GMT
Repository: kafka
Updated Branches:
  refs/heads/trunk 6fb33afff -> 3dafb81da


KAFKA-3817: KTableRepartitionMap publish old Change first, for non-count aggregates

I affirm that the contribution is my original work and that I license the work to the project
under the project's open source license.

This cleans up misbehaviour that was introduce while fixing KAFKA-3817. It is impossible for
a non-count aggregate to be build, when the addition happens before the removal. IMHO making
sure that these details are correct is very important.

This PR has local test errors. It somehow fails the ResetIntegrationTest. It doesn't quite
appear to me why but it looks like this PR breaks it, especially because the error appears
with the ordering of the events. Still I am unable to find where I could have broken it. Maybe
not seems to fail on trunk aswell.

Author: jfilipiak <Jan.Filipiak@trivago.com>

Reviewers: Guozhang Wang <wangguoz@gmail.com>

Closes #1705 from Kaiserchen/KAFKA-3817-preserve-order-for-aggreagators


Project: http://git-wip-us.apache.org/repos/asf/kafka/repo
Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/3dafb81d
Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/3dafb81d
Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/3dafb81d

Branch: refs/heads/trunk
Commit: 3dafb81da788294d4c2e9811f49437608e5b9ce8
Parents: 6fb33af
Author: Jan Filipiak <Jan.Filipiak@trivago.com>
Authored: Fri Aug 5 10:03:22 2016 -0700
Committer: Guozhang Wang <wangguoz@gmail.com>
Committed: Fri Aug 5 10:03:22 2016 -0700

----------------------------------------------------------------------
 .../kstream/internals/KTableRepartitionMap.java |  9 ++-
 .../kstream/internals/KTableAggregateTest.java  | 81 +++++++++++++++++---
 2 files changed, 77 insertions(+), 13 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/kafka/blob/3dafb81d/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableRepartitionMap.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableRepartitionMap.java
b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableRepartitionMap.java
index ac7c00e..939a1df 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableRepartitionMap.java
+++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableRepartitionMap.java
@@ -81,12 +81,15 @@ public class KTableRepartitionMap<K, V, K1, V1> implements KTableProcessorSuppli
             KeyValue<K1, V1> oldPair = change.oldValue == null ? null : mapper.apply(key,
change.oldValue);
 
             // if the selected repartition key or value is null, skip
-            if (newPair != null && newPair.key != null && newPair.value !=
null) {
-                context().forward(newPair.key, new Change<>(newPair.value, null));
-            }
+            // forward oldPair first, to be consistent with reduce and aggregate
             if (oldPair != null && oldPair.key != null && oldPair.value !=
null) {
                 context().forward(oldPair.key, new Change<>(null, oldPair.value));
             }
+
+            if (newPair != null && newPair.key != null && newPair.value !=
null) {
+                context().forward(newPair.key, new Change<>(newPair.value, null));
+            }
+            
         }
     }
 

http://git-wip-us.apache.org/repos/asf/kafka/blob/3dafb81d/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableAggregateTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableAggregateTest.java
b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableAggregateTest.java
index 26e6a0f..a405da4 100644
--- a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableAggregateTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableAggregateTest.java
@@ -21,6 +21,8 @@ import org.apache.kafka.common.serialization.Serde;
 import org.apache.kafka.common.serialization.Serdes;
 import org.apache.kafka.common.utils.Utils;
 import org.apache.kafka.streams.KeyValue;
+import org.apache.kafka.streams.kstream.Aggregator;
+import org.apache.kafka.streams.kstream.Initializer;
 import org.apache.kafka.streams.kstream.KStreamBuilder;
 import org.apache.kafka.streams.kstream.KTable;
 import org.apache.kafka.streams.kstream.KeyValueMapper;
@@ -34,6 +36,7 @@ import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
 
+
 import java.io.File;
 import java.io.IOException;
 
@@ -91,12 +94,12 @@ public class KTableAggregateTest {
         assertEquals(Utils.mkList(
                 "A:0+1",
                 "B:0+2",
-                "A:0+1+3", "A:0+1+3-1",
-                "B:0+2+4", "B:0+2+4-2",
+                "A:0+1-1", "A:0+1-1+3",
+                "B:0+2-2", "B:0+2-2+4",
                 "C:0+5",
                 "D:0+6",
-                "B:0+2+4-2+7", "B:0+2+4-2+7-4",
-                "C:0+5+8", "C:0+5+8-5"), proc.processed);
+                "B:0+2-2+4-4", "B:0+2-2+4-4+7",
+                "C:0+5-5", "C:0+5-5+8"), proc.processed);
     }
 
     @Test
@@ -144,11 +147,12 @@ public class KTableAggregateTest {
                 "1:0+1",
                 "1:0+1-1",
                 "1:0+1-1+1",
-                "2:0+2",
-                "4:0+4",
-                "2:0+2-2",
-                "7:0+7",
-                "4:0+4-4"), proc.processed);
+                "2:0+2", 
+                  //noop
+                "2:0+2-2", "4:0+4",
+                  //noop
+                "4:0+4-4", "7:0+7"
+                ), proc.processed);
     }
 
     @Test
@@ -171,6 +175,63 @@ public class KTableAggregateTest {
         driver.process(input, "C", "yellow");
         driver.process(input, "D", "green");
 
-        assertEquals(Utils.mkList("green:1", "green:2", "blue:1", "green:1", "yellow:1",
"green:2"), proc.processed);
+        assertEquals(Utils.mkList(
+                 "green:1",
+                 "green:2",
+                 "green:1", "blue:1",
+                 "yellow:1",
+                 "green:2"
+                 ), proc.processed);
+    }
+    
+    @Test
+    public void testRemoveOldBeforeAddNew() throws IOException {
+        final KStreamBuilder builder = new KStreamBuilder();
+        final String input = "count-test-input";
+        final MockProcessorSupplier<String, String> proc = new MockProcessorSupplier<>();
+
+        builder.table(Serdes.String(), Serdes.String(), input, "anyStoreName")
+                .groupBy(new KeyValueMapper<String, String, KeyValue<String, String>>()
{
+
+                    @Override
+                    public KeyValue<String, String> apply(String key, String value)
{
+                        return KeyValue.pair(String.valueOf(key.charAt(0)), String.valueOf(key.charAt(1)));
+                    }
+                }, stringSerde, stringSerde)
+                .aggregate(new Initializer<String>() {
+
+                    @Override
+                    public String apply() {
+                        return "";
+                    }
+                }, new Aggregator<String, String, String>() {
+                    
+                    @Override
+                    public String apply(String aggKey, String value, String aggregate) {
+                        return aggregate + value;
+                    } 
+                }, new Aggregator<String, String, String>() {
+
+                    @Override
+                    public String apply(String key, String value, String aggregate) {
+                        return aggregate.replaceAll(value, "");
+                    }
+                }, "someStore")
+                .toStream()
+                .process(proc);
+
+        final KStreamTestDriver driver = new KStreamTestDriver(builder, stateDir);
+
+        driver.process(input, "11", "A");
+        driver.process(input, "12", "B");
+        driver.process(input, "11", null);
+        driver.process(input, "12", "C");
+
+        assertEquals(Utils.mkList(
+                 "1:1",
+                 "1:12",
+                 "1:2",
+                 "1:", "1:2"
+                 ), proc.processed);
     }
 }


Mime
View raw message