kafka-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From bbej...@apache.org
Subject [kafka] branch 2.2 updated: KAFKA-8289: Fix Session Expiration and Suppression (#6654) (#6670)
Date Fri, 03 May 2019 20:24:13 GMT
This is an automated email from the ASF dual-hosted git repository.

bbejeck pushed a commit to branch 2.2
in repository https://gitbox.apache.org/repos/asf/kafka.git


The following commit(s) were added to refs/heads/2.2 by this push:
     new bfad7f9  KAFKA-8289: Fix Session Expiration and Suppression (#6654) (#6670)
bfad7f9 is described below

commit bfad7f99f7b27348245189f8cd451250e4fab3e6
Author: John Roesler <vvcephei@users.noreply.github.com>
AuthorDate: Fri May 3 15:23:59 2019 -0500

    KAFKA-8289: Fix Session Expiration and Suppression (#6654) (#6670)
    
    Fix two problems in Streams:
    * Session windows expired prematurely (off-by-one error), since the window end is inclusive,
unlike other windows
    * Suppress duration for sessions incorrectly waited only the grace period, but session
windows aren't closed until gracePeriod + sessionGap
    
    cherry-pick of 6654 into trunk
    
    Reviewers: Bill Bejeck <bill@confluent.io>
---
 .../internals/KStreamSessionWindowAggregate.java   | 28 +++----
 .../internals/graph/GraphGraceSearchUtil.java      |  2 +-
 ...KStreamSessionWindowAggregateProcessorTest.java | 96 ++++++++++++++++++++--
 .../kstream/internals/SuppressScenarioTest.java    | 41 ++++-----
 .../internals/graph/GraphGraceSearchUtilTest.java  | 10 +--
 5 files changed, 128 insertions(+), 49 deletions(-)

diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamSessionWindowAggregate.java
b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamSessionWindowAggregate.java
index 2c98c10..aedf7f2 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamSessionWindowAggregate.java
+++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamSessionWindowAggregate.java
@@ -133,19 +133,7 @@ public class KStreamSessionWindowAggregate<K, V, Agg> implements
KStreamAggProce
                 }
             }
 
-            if (mergedWindow.end() > closeTime) {
-                if (!mergedWindow.equals(newSessionWindow)) {
-                    for (final KeyValue<Windowed<K>, Agg> session : merged) {
-                        store.remove(session.key);
-                        tupleForwarder.maybeForward(session.key, null, sendOldValues ? session.value
: null);
-                    }
-                }
-
-                agg = aggregator.apply(key, value, agg);
-                final Windowed<K> sessionKey = new Windowed<>(key, mergedWindow);
-                store.put(sessionKey, agg);
-                tupleForwarder.maybeForward(sessionKey, agg, null);
-            } else {
+            if (mergedWindow.end() < closeTime) {
                 LOG.debug(
                     "Skipping record for expired window. " +
                         "key=[{}] " +
@@ -153,7 +141,7 @@ public class KStreamSessionWindowAggregate<K, V, Agg> implements
KStreamAggProce
                         "partition=[{}] " +
                         "offset=[{}] " +
                         "timestamp=[{}] " +
-                        "window=[{},{}) " +
+                        "window=[{},{}] " +
                         "expiration=[{}] " +
                         "streamTime=[{}]",
                     key,
@@ -167,6 +155,18 @@ public class KStreamSessionWindowAggregate<K, V, Agg> implements
KStreamAggProce
                     observedStreamTime
                 );
                 lateRecordDropSensor.record();
+            } else {
+                if (!mergedWindow.equals(newSessionWindow)) {
+                    for (final KeyValue<Windowed<K>, Agg> session : merged) {
+                        store.remove(session.key);
+                        tupleForwarder.maybeForward(session.key, null, sendOldValues ? session.value
: null);
+                    }
+                }
+
+                agg = aggregator.apply(key, value, agg);
+                final Windowed<K> sessionKey = new Windowed<>(key, mergedWindow);
+                store.put(sessionKey, agg);
+                tupleForwarder.maybeForward(sessionKey, agg, null);
             }
         }
     }
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/graph/GraphGraceSearchUtil.java
b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/graph/GraphGraceSearchUtil.java
index 306ddf5..2fb28dd 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/graph/GraphGraceSearchUtil.java
+++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/graph/GraphGraceSearchUtil.java
@@ -78,7 +78,7 @@ public final class GraphGraceSearchUtil {
             } else if (processorSupplier instanceof KStreamSessionWindowAggregate) {
                 final KStreamSessionWindowAggregate kStreamSessionWindowAggregate = (KStreamSessionWindowAggregate)
processorSupplier;
                 final SessionWindows windows = kStreamSessionWindowAggregate.windows();
-                return windows.gracePeriodMs();
+                return windows.gracePeriodMs() + windows.inactivityGap();
             } else {
                 return null;
             }
diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamSessionWindowAggregateProcessorTest.java
b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamSessionWindowAggregateProcessorTest.java
index 8503980..f53d9f3 100644
--- a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamSessionWindowAggregateProcessorTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamSessionWindowAggregateProcessorTest.java
@@ -329,11 +329,11 @@ public class KStreamSessionWindowAggregateProcessorTest {
     }
 
     @Test
-    public void shouldLogAndMeterWhenSkippingLateRecord() {
+    public void shouldLogAndMeterWhenSkippingLateRecordWithZeroGrace() {
         LogCaptureAppender.setClassLoggerToDebug(KStreamSessionWindowAggregate.class);
         final LogCaptureAppender appender = LogCaptureAppender.createAndRegister();
         final Processor<String, String> processor = new KStreamSessionWindowAggregate<>(
-            SessionWindows.with(ofMillis(10L)).grace(ofMillis(10L)),
+            SessionWindows.with(ofMillis(10L)).grace(ofMillis(0L)),
             STORE_NAME,
             initializer,
             aggregator,
@@ -343,14 +343,21 @@ public class KStreamSessionWindowAggregateProcessorTest {
         initStore(false);
         processor.init(context);
 
-        // dummy record to advance stream time
-        context.setRecordContext(new ProcessorRecordContext(20, -2, -3, "topic", null));
+        // dummy record to establish stream time = 0
+        context.setRecordContext(new ProcessorRecordContext(0, -2, -3, "topic", null));
         processor.process("dummy", "dummy");
 
+        // record arrives on time, should not be skipped
         context.setRecordContext(new ProcessorRecordContext(0, -2, -3, "topic", null));
-        processor.process("A", "1");
+        processor.process("OnTime1", "1");
+
+        // dummy record to advance stream time = 1
         context.setRecordContext(new ProcessorRecordContext(1, -2, -3, "topic", null));
-        processor.process("A", "1");
+        processor.process("dummy", "dummy");
+
+        // record is late
+        context.setRecordContext(new ProcessorRecordContext(0, -2, -3, "topic", null));
+        processor.process("Late1", "1");
         LogCaptureAppender.unregister(appender);
 
         final MetricName dropMetric = new MetricName(
@@ -364,7 +371,7 @@ public class KStreamSessionWindowAggregateProcessorTest {
             )
         );
 
-        assertThat(metrics.metrics().get(dropMetric).metricValue(), is(2.0));
+        assertThat(metrics.metrics().get(dropMetric).metricValue(), is(1.0));
 
         final MetricName dropRate = new MetricName(
             "late-record-drop-rate",
@@ -382,9 +389,80 @@ public class KStreamSessionWindowAggregateProcessorTest {
             greaterThan(0.0));
         assertThat(
             appender.getMessages(),
-            hasItem("Skipping record for expired window. key=[A] topic=[topic] partition=[-3]
offset=[-2] timestamp=[0] window=[0,0) expiration=[10] streamTime=[20]"));
+            hasItem("Skipping record for expired window. key=[Late1] topic=[topic] partition=[-3]
offset=[-2] timestamp=[0] window=[0,0] expiration=[1] streamTime=[1]"));
+    }
+
+    @Test
+    public void shouldLogAndMeterWhenSkippingLateRecordWithNonzeroGrace() {
+        LogCaptureAppender.setClassLoggerToDebug(KStreamSessionWindowAggregate.class);
+        final LogCaptureAppender appender = LogCaptureAppender.createAndRegister();
+        final Processor<String, String> processor = new KStreamSessionWindowAggregate<>(
+            SessionWindows.with(ofMillis(10L)).grace(ofMillis(1L)),
+            STORE_NAME,
+            initializer,
+            aggregator,
+            sessionMerger
+        ).get();
+
+        initStore(false);
+        processor.init(context);
+
+        // dummy record to establish stream time = 0
+        context.setRecordContext(new ProcessorRecordContext(0, -2, -3, "topic", null));
+        processor.process("dummy", "dummy");
+
+        // record arrives on time, should not be skipped
+        context.setRecordContext(new ProcessorRecordContext(0, -2, -3, "topic", null));
+        processor.process("OnTime1", "1");
+
+        // dummy record to advance stream time = 1
+        context.setRecordContext(new ProcessorRecordContext(1, -2, -3, "topic", null));
+        processor.process("dummy", "dummy");
+
+        // delayed record arrives on time, should not be skipped
+        context.setRecordContext(new ProcessorRecordContext(0, -2, -3, "topic", null));
+        processor.process("OnTime2", "1");
+
+        // dummy record to advance stream time = 2
+        context.setRecordContext(new ProcessorRecordContext(2, -2, -3, "topic", null));
+        processor.process("dummy", "dummy");
+
+        // delayed record arrives late
+        context.setRecordContext(new ProcessorRecordContext(0, -2, -3, "topic", null));
+        processor.process("Late1", "1");
+
+
+        LogCaptureAppender.unregister(appender);
+
+        final MetricName dropMetric = new MetricName(
+            "late-record-drop-total",
+            "stream-processor-node-metrics",
+            "The total number of occurrence of late-record-drop operations.",
+            mkMap(
+                mkEntry("client-id", "test"),
+                mkEntry("task-id", "0_0"),
+                mkEntry("processor-node-id", "TESTING_NODE")
+            )
+        );
+
+        assertThat(metrics.metrics().get(dropMetric).metricValue(), is(1.0));
+
+        final MetricName dropRate = new MetricName(
+            "late-record-drop-rate",
+            "stream-processor-node-metrics",
+            "The average number of occurrence of late-record-drop operations.",
+            mkMap(
+                mkEntry("client-id", "test"),
+                mkEntry("task-id", "0_0"),
+                mkEntry("processor-node-id", "TESTING_NODE")
+            )
+        );
+
+        assertThat(
+            (Double) metrics.metrics().get(dropRate).metricValue(),
+            greaterThan(0.0));
         assertThat(
             appender.getMessages(),
-            hasItem("Skipping record for expired window. key=[A] topic=[topic] partition=[-3]
offset=[-2] timestamp=[1] window=[1,1) expiration=[10] streamTime=[20]"));
+            hasItem("Skipping record for expired window. key=[Late1] topic=[topic] partition=[-3]
offset=[-2] timestamp=[0] window=[0,0] expiration=[1] streamTime=[2]"));
     }
 }
diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/SuppressScenarioTest.java
b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/SuppressScenarioTest.java
index b0b87be..24f222b 100644
--- a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/SuppressScenarioTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/SuppressScenarioTest.java
@@ -47,7 +47,6 @@ import org.apache.kafka.streams.test.OutputVerifier;
 import org.apache.kafka.test.TestUtils;
 import org.junit.Test;
 
-import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Iterator;
 import java.util.LinkedList;
@@ -72,9 +71,9 @@ public class SuppressScenarioTest {
     private static final Serde<String> STRING_SERDE = Serdes.String();
     private static final LongDeserializer LONG_DESERIALIZER = new LongDeserializer();
     private final Properties config = Utils.mkProperties(Utils.mkMap(
-            Utils.mkEntry(StreamsConfig.APPLICATION_ID_CONFIG, getClass().getSimpleName().toLowerCase(Locale.getDefault())),
-            Utils.mkEntry(StreamsConfig.STATE_DIR_CONFIG, TestUtils.tempDirectory().getPath()),
-            Utils.mkEntry(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "bogus")
+        Utils.mkEntry(StreamsConfig.APPLICATION_ID_CONFIG, getClass().getSimpleName().toLowerCase(Locale.getDefault())),
+        Utils.mkEntry(StreamsConfig.STATE_DIR_CONFIG, TestUtils.tempDirectory().getPath()),
+        Utils.mkEntry(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "bogus")
     ));
 
     @Test
@@ -304,7 +303,7 @@ public class SuppressScenarioTest {
             .count();
         valueCounts
             // this is a bit brittle, but I happen to know that the entries are a little
over 100 bytes in size.
-            .suppress(untilTimeLimit(Duration.ofMillis(Long.MAX_VALUE), maxBytes(200L).emitEarlyWhenFull()))
+            .suppress(untilTimeLimit(ofMillis(Long.MAX_VALUE), maxBytes(200L).emitEarlyWhenFull()))
             .toStream()
             .to("output-suppressed", Produced.with(STRING_SERDE, Serdes.Long()));
         valueCounts
@@ -352,7 +351,6 @@ public class SuppressScenarioTest {
         }
     }
 
-    @SuppressWarnings("deprecation")
     @Test
     public void shouldSupportFinalResultsForTimeWindows() {
         final StreamsBuilder builder = new StreamsBuilder();
@@ -403,7 +401,6 @@ public class SuppressScenarioTest {
         }
     }
 
-    @SuppressWarnings("deprecation")
     @Test
     public void shouldSupportFinalResultsForTimeWindowsWithLargeJump() {
         final StreamsBuilder builder = new StreamsBuilder();
@@ -465,7 +462,7 @@ public class SuppressScenarioTest {
         final KTable<Windowed<String>, Long> valueCounts = builder
             .stream("input", Consumed.with(STRING_SERDE, STRING_SERDE))
             .groupBy((String k, String v) -> k, Grouped.with(STRING_SERDE, STRING_SERDE))
-            .windowedBy(SessionWindows.with(ofMillis(5L)).grace(ofMillis(5L)))
+            .windowedBy(SessionWindows.with(ofMillis(5L)).grace(ofMillis(0L)))
             .count(Materialized.<String, Long, SessionStore<Bytes, byte[]>>as("counts").withCachingDisabled());
         valueCounts
             .suppress(untilWindowCloses(unbounded()))
@@ -482,34 +479,38 @@ public class SuppressScenarioTest {
         try (final TopologyTestDriver driver = new TopologyTestDriver(topology, config))
{
             // first window
             driver.pipeInput(recordFactory.create("input", "k1", "v1", 0L));
+            driver.pipeInput(recordFactory.create("input", "k1", "v1", 5L));
+            // arbitrarily disordered records are admitted, because the *window* is not closed
until stream-time > window-end + grace
             driver.pipeInput(recordFactory.create("input", "k1", "v1", 1L));
-            // new window
-            driver.pipeInput(recordFactory.create("input", "k1", "v1", 7L));
+            // any record in the same partition advances stream time (note the key is different)
+            driver.pipeInput(recordFactory.create("input", "k2", "v1", 6L));
             // late event for first window - this should get dropped from all streams, since
the first window is now closed.
-            driver.pipeInput(recordFactory.create("input", "k1", "v1", 1L));
+            driver.pipeInput(recordFactory.create("input", "k1", "v1", 5L));
             // just pushing stream time forward to flush the other events through.
             driver.pipeInput(recordFactory.create("input", "k1", "v1", 30L));
             verify(
                 drainProducerRecords(driver, "output-raw", STRING_DESERIALIZER, LONG_DESERIALIZER),
                 asList(
                     new KeyValueTimestamp<>("[k1@0/0]", 1L, 0L),
-                    new KeyValueTimestamp<>("[k1@0/0]", null, 1L),
-                    new KeyValueTimestamp<>("[k1@0/1]", 2L, 1L),
-                    new KeyValueTimestamp<>("[k1@7/7]", 1L, 7L),
+                    new KeyValueTimestamp<>("[k1@0/0]", null, 5L),
+                    new KeyValueTimestamp<>("[k1@0/5]", 2L, 5L),
+                    new KeyValueTimestamp<>("[k1@0/5]", null, 1L),
+                    new KeyValueTimestamp<>("[k1@0/5]", 3L, 1L),
+                    new KeyValueTimestamp<>("[k2@6/6]", 1L, 6L),
                     new KeyValueTimestamp<>("[k1@30/30]", 1L, 30L)
                 )
             );
             verify(
                 drainProducerRecords(driver, "output-suppressed", STRING_DESERIALIZER, LONG_DESERIALIZER),
                 asList(
-                    new KeyValueTimestamp<>("[k1@0/1]", 2L, 1L),
-                    new KeyValueTimestamp<>("[k1@7/7]", 1L, 7L)
+                    new KeyValueTimestamp<>("[k1@0/5]", 3L, 1L),
+                    new KeyValueTimestamp<>("[k2@6/6]", 1L, 6L)
                 )
             );
         }
     }
 
-    private <K, V> void verify(final List<ProducerRecord<K, V>> results,
final List<KeyValueTimestamp<K, V>> expectedResults) {
+    private static <K, V> void verify(final List<ProducerRecord<K, V>>
results, final List<KeyValueTimestamp<K, V>> expectedResults) {
         if (results.size() != expectedResults.size()) {
             throw new AssertionError(printRecords(results) + " != " + expectedResults);
         }
@@ -524,7 +525,7 @@ public class SuppressScenarioTest {
         }
     }
 
-    private <K, V> List<ProducerRecord<K, V>> drainProducerRecords(final
TopologyTestDriver driver, final String topic, final Deserializer<K> keyDeserializer,
final Deserializer<V> valueDeserializer) {
+    private static <K, V> List<ProducerRecord<K, V>> drainProducerRecords(final
TopologyTestDriver driver, final String topic, final Deserializer<K> keyDeserializer,
final Deserializer<V> valueDeserializer) {
         final List<ProducerRecord<K, V>> result = new LinkedList<>();
         for (ProducerRecord<K, V> next = driver.readOutput(topic, keyDeserializer,
valueDeserializer);
              next != null;
@@ -534,11 +535,11 @@ public class SuppressScenarioTest {
         return new ArrayList<>(result);
     }
 
-    private <K, V> String printRecords(final List<ProducerRecord<K, V>>
result) {
+    private static <K, V> String printRecords(final List<ProducerRecord<K, V>>
result) {
         final StringBuilder resultStr = new StringBuilder();
         resultStr.append("[\n");
         for (final ProducerRecord<?, ?> record : result) {
-            resultStr.append("  ").append(record.toString()).append("\n");
+            resultStr.append("  ").append(record).append("\n");
         }
         resultStr.append("]");
         return resultStr.toString();
diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/graph/GraphGraceSearchUtilTest.java
b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/graph/GraphGraceSearchUtilTest.java
index 5e426d9..45fe845 100644
--- a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/graph/GraphGraceSearchUtilTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/graph/GraphGraceSearchUtilTest.java
@@ -121,11 +121,11 @@ public class GraphGraceSearchUtilTest {
         );
 
         final long extracted = GraphGraceSearchUtil.findAndVerifyWindowGrace(node);
-        assertThat(extracted, is(windows.gracePeriodMs()));
+        assertThat(extracted, is(windows.gracePeriodMs() + windows.inactivityGap()));
     }
 
     @Test
-    public void shouldExtractGraceFromAncestorThroughStatefulParent() {
+    public void shouldExtractGraceFromSessionAncestorThroughStatefulParent() {
         final SessionWindows windows = SessionWindows.with(ofMillis(10L)).grace(ofMillis(1234L));
         final StatefulProcessorNode<String, Long> graceGrandparent = new StatefulProcessorNode<>(
             "asdf",
@@ -160,11 +160,11 @@ public class GraphGraceSearchUtilTest {
         statefulParent.addChild(node);
 
         final long extracted = GraphGraceSearchUtil.findAndVerifyWindowGrace(node);
-        assertThat(extracted, is(windows.gracePeriodMs()));
+        assertThat(extracted, is(windows.gracePeriodMs() + windows.inactivityGap()));
     }
 
     @Test
-    public void shouldExtractGraceFromAncestorThroughStatelessParent() {
+    public void shouldExtractGraceFromSessionAncestorThroughStatelessParent() {
         final SessionWindows windows = SessionWindows.with(ofMillis(10L)).grace(ofMillis(1234L));
         final StatefulProcessorNode<String, Long> graceGrandparent = new StatefulProcessorNode<>(
             "asdf",
@@ -189,7 +189,7 @@ public class GraphGraceSearchUtilTest {
         statelessParent.addChild(node);
 
         final long extracted = GraphGraceSearchUtil.findAndVerifyWindowGrace(node);
-        assertThat(extracted, is(windows.gracePeriodMs()));
+        assertThat(extracted, is(windows.gracePeriodMs() + windows.inactivityGap()));
     }
 
     @Test


Mime
View raw message