From dceee75707614b353cbabcee5d55c6c835b84dcf Mon Sep 17 00:00:00 2001 From: John Roesler Date: Thu, 24 Jan 2019 08:54:50 -0600 Subject: [PATCH 1/2] MINOR: clarify why suppress can sometimes drop tombstones --- .../suppress/KTableSuppressProcessor.java | 6 ++-- .../suppress/SuppressedInternal.java | 29 +++++++++++----- .../suppress/KTableSuppressProcessorTest.java | 33 ++++++++++++++++--- 3 files changed, 52 insertions(+), 16 deletions(-) diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/suppress/KTableSuppressProcessor.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/suppress/KTableSuppressProcessor.java index 06d5004f65fe1..b5f1bd1f8c4cc 100644 --- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/suppress/KTableSuppressProcessor.java +++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/suppress/KTableSuppressProcessor.java @@ -42,7 +42,7 @@ public class KTableSuppressProcessor implements Processor> { private final long suppressDurationMillis; private final TimeDefinition bufferTimeDefinition; private final BufferFullStrategy bufferFullStrategy; - private final boolean shouldSuppressTombstones; + private final boolean safeToDropTombstones; private final String storeName; private TimeOrderedKeyValueBuffer buffer; @@ -64,7 +64,7 @@ public KTableSuppressProcessor(final SuppressedInternal suppress, suppressDurationMillis = suppress.timeToWaitForMoreEvents().toMillis(); bufferTimeDefinition = suppress.timeDefinition(); bufferFullStrategy = suppress.bufferConfig().bufferFullStrategy(); - shouldSuppressTombstones = suppress.shouldSuppressTombstones(); + safeToDropTombstones = suppress.shouldSuppressTombstones(); } @SuppressWarnings("unchecked") @@ -136,7 +136,7 @@ private void emit(final KeyValue toEmit) { } private boolean shouldForward(final Change value) { - return !(value.newValue == null && shouldSuppressTombstones); + return value.newValue != null || !safeToDropTombstones; } @Override diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/suppress/SuppressedInternal.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/suppress/SuppressedInternal.java index 042a81aa2fe11..78a972c34b711 100644 --- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/suppress/SuppressedInternal.java +++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/suppress/SuppressedInternal.java @@ -30,23 +30,36 @@ public class SuppressedInternal implements Suppressed { private final BufferConfigInternal bufferConfig; private final Duration timeToWaitForMoreEvents; private final TimeDefinition timeDefinition; - private final boolean suppressTombstones; + private final boolean safeToDropTombstones; + /** + * @param safeToDropTombstones Note: it's *only* safe to drop tombstones for windowed KTables in "final results" mode. + * In that case, we have a priori knowledge that we have never before emitted any + * results for a given key, and therefore the tombstone is unnecessary (albeit + * idempotent and correct). We decided that the unnecessary tombstones would not be + * desirable in the output stream, though, hence the ability to drop them. + * + * A alternative is to remember whether a result has previously been emitted + * for a key and drop tombstones in that case, but it would be a little complicated to + * figure out when to forget the fact that we have emitted some result (currently, the + * buffer immediately forgets all about a key when we emit, which helps to keep it + * compact). + */ public SuppressedInternal(final String name, final Duration suppressionTime, final BufferConfig bufferConfig, final TimeDefinition timeDefinition, - final boolean suppressTombstones) { + final boolean safeToDropTombstones) { this.name = name; this.timeToWaitForMoreEvents = suppressionTime == null ? DEFAULT_SUPPRESSION_TIME : suppressionTime; this.timeDefinition = timeDefinition == null ? TimeDefinitions.RecordTimeDefintion.instance() : timeDefinition; this.bufferConfig = bufferConfig == null ? DEFAULT_BUFFER_CONFIG : (BufferConfigInternal) bufferConfig; - this.suppressTombstones = suppressTombstones; + this.safeToDropTombstones = safeToDropTombstones; } @Override public Suppressed withName(final String name) { - return new SuppressedInternal<>(name, timeToWaitForMoreEvents, bufferConfig, timeDefinition, suppressTombstones); + return new SuppressedInternal<>(name, timeToWaitForMoreEvents, bufferConfig, timeDefinition, safeToDropTombstones); } public String name() { @@ -66,7 +79,7 @@ Duration timeToWaitForMoreEvents() { } boolean shouldSuppressTombstones() { - return suppressTombstones; + return safeToDropTombstones; } @Override @@ -78,7 +91,7 @@ public boolean equals(final Object o) { return false; } final SuppressedInternal that = (SuppressedInternal) o; - return suppressTombstones == that.suppressTombstones && + return safeToDropTombstones == that.safeToDropTombstones && Objects.equals(name, that.name) && Objects.equals(bufferConfig, that.bufferConfig) && Objects.equals(timeToWaitForMoreEvents, that.timeToWaitForMoreEvents) && @@ -87,7 +100,7 @@ public boolean equals(final Object o) { @Override public int hashCode() { - return Objects.hash(name, bufferConfig, timeToWaitForMoreEvents, timeDefinition, suppressTombstones); + return Objects.hash(name, bufferConfig, timeToWaitForMoreEvents, timeDefinition, safeToDropTombstones); } @Override @@ -96,7 +109,7 @@ public String toString() { ", bufferConfig=" + bufferConfig + ", timeToWaitForMoreEvents=" + timeToWaitForMoreEvents + ", timeDefinition=" + timeDefinition + - ", suppressTombstones=" + suppressTombstones + + ", safeToDropTombstones=" + safeToDropTombstones + '}'; } } diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/suppress/KTableSuppressProcessorTest.java b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/suppress/KTableSuppressProcessorTest.java index 335fae1875c4e..0f2b36b605aad 100644 --- a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/suppress/KTableSuppressProcessorTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/suppress/KTableSuppressProcessorTest.java @@ -255,8 +255,12 @@ public void finalResultsWithZeroGraceAtWindowEndShouldImmediatelyEmit() { assertThat(capturedForward.timestamp(), is(timestamp)); } + /** + * It's desirable to drop tombstones for final-results windowed streams, since (as described in the + * {@link SuppressedInternal} javadoc), they are unnecessary to emit. + */ @Test - public void finalResultsShouldSuppressTombstonesForTimeWindows() { + public void finalResultsShouldDropTombstonesForTimeWindows() { final Harness, Long> harness = new Harness<>(finalResults(ofMillis(0L)), timeWindowedSerdeFrom(String.class, 100L), Long()); final MockInternalProcessorContext context = harness.context; @@ -272,8 +276,13 @@ public void finalResultsShouldSuppressTombstonesForTimeWindows() { assertThat(context.forwarded(), hasSize(0)); } + + /** + * It's desirable to drop tombstones for final-results windowed streams, since (as described in the + * {@link SuppressedInternal} javadoc), they are unnecessary to emit. + */ @Test - public void finalResultsShouldSuppressTombstonesForSessionWindows() { + public void finalResultsShouldDropTombstonesForSessionWindows() { final Harness, Long> harness = new Harness<>(finalResults(ofMillis(0L)), sessionWindowedSerdeFrom(String.class), Long()); final MockInternalProcessorContext context = harness.context; @@ -289,8 +298,12 @@ public void finalResultsShouldSuppressTombstonesForSessionWindows() { assertThat(context.forwarded(), hasSize(0)); } + /** + * It's NOT OK to drop tombstones for non-final-results windowed streams, since we may have emitted some results for + * the window before getting the tombstone (see the {@link SuppressedInternal} javadoc). + */ @Test - public void suppressShouldNotSuppressTombstonesForTimeWindows() { + public void suppressShouldNotDropTombstonesForTimeWindows() { final Harness, Long> harness = new Harness<>(untilTimeLimit(ofMillis(0), maxRecords(0)), timeWindowedSerdeFrom(String.class, 100L), Long()); final MockInternalProcessorContext context = harness.context; @@ -309,8 +322,13 @@ public void suppressShouldNotSuppressTombstonesForTimeWindows() { assertThat(capturedForward.timestamp(), is(timestamp)); } + + /** + * It's NOT OK to drop tombstones for non-final-results windowed streams, since we may have emitted some results for + * the window before getting the tombstone (see the {@link SuppressedInternal} javadoc). + */ @Test - public void suppressShouldNotSuppressTombstonesForSessionWindows() { + public void suppressShouldNotDropTombstonesForSessionWindows() { final Harness, Long> harness = new Harness<>(untilTimeLimit(ofMillis(0), maxRecords(0)), sessionWindowedSerdeFrom(String.class), Long()); final MockInternalProcessorContext context = harness.context; @@ -329,8 +347,13 @@ public void suppressShouldNotSuppressTombstonesForSessionWindows() { assertThat(capturedForward.timestamp(), is(timestamp)); } + + /** + * It's SUPER NOT OK to drop tombstones for non-windowed streams, since we may have emitted some results for + * the key before getting the tombstone (see the {@link SuppressedInternal} javadoc). + */ @Test - public void suppressShouldNotSuppressTombstonesForKTable() { + public void suppressShouldNotDropTombstonesForKTable() { final Harness harness = new Harness<>(untilTimeLimit(ofMillis(0), maxRecords(0)), String(), Long()); final MockInternalProcessorContext context = harness.context; From 9e8d56d77b4ffd00fdbe87b99dc60223f0b11efc Mon Sep 17 00:00:00 2001 From: John Roesler Date: Fri, 25 Jan 2019 07:49:02 -0600 Subject: [PATCH 2/2] fix missed method name --- .../kstream/internals/suppress/KTableSuppressProcessor.java | 2 +- .../streams/kstream/internals/suppress/SuppressedInternal.java | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/suppress/KTableSuppressProcessor.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/suppress/KTableSuppressProcessor.java index b5f1bd1f8c4cc..813c5581a1d69 100644 --- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/suppress/KTableSuppressProcessor.java +++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/suppress/KTableSuppressProcessor.java @@ -64,7 +64,7 @@ public KTableSuppressProcessor(final SuppressedInternal suppress, suppressDurationMillis = suppress.timeToWaitForMoreEvents().toMillis(); bufferTimeDefinition = suppress.timeDefinition(); bufferFullStrategy = suppress.bufferConfig().bufferFullStrategy(); - safeToDropTombstones = suppress.shouldSuppressTombstones(); + safeToDropTombstones = suppress.safeToDropTombstones(); } @SuppressWarnings("unchecked") diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/suppress/SuppressedInternal.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/suppress/SuppressedInternal.java index 78a972c34b711..c3877007c5a42 100644 --- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/suppress/SuppressedInternal.java +++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/suppress/SuppressedInternal.java @@ -78,7 +78,7 @@ Duration timeToWaitForMoreEvents() { return timeToWaitForMoreEvents == null ? Duration.ZERO : timeToWaitForMoreEvents; } - boolean shouldSuppressTombstones() { + boolean safeToDropTombstones() { return safeToDropTombstones; }