Skip to content

MINOR: clarify why suppress can sometimes drop tombstones #6195

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
guozhangwang merged 2 commits into from
Jan 25, 2019
Merged

MINOR: clarify why suppress can sometimes drop tombstones #6195

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
dceee75
MINOR: clarify why suppress can sometimes drop tombstones
vvcephei Jan 24, 2019
9e8d56d
fix missed method name
vvcephei Jan 25, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 3 additions & 3 deletions ...ain/java/org/apache/kafka/streams/kstream/internals/suppress/KTableSuppressProcessor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,7 +42,7 @@ public class KTableSuppressProcessor<K, V> implements Processor<K, Change<V>> {
private final long suppressDurationMillis;
private final TimeDefinition<K> bufferTimeDefinition;
private final BufferFullStrategy bufferFullStrategy;
private final boolean shouldSuppressTombstones;
private final boolean safeToDropTombstones;
private final String storeName;

private TimeOrderedKeyValueBuffer buffer;
Expand All @@ -64,7 +64,7 @@ public KTableSuppressProcessor(final SuppressedInternal<K> suppress,
suppressDurationMillis = suppress.timeToWaitForMoreEvents().toMillis();
bufferTimeDefinition = suppress.timeDefinition();
bufferFullStrategy = suppress.bufferConfig().bufferFullStrategy();
shouldSuppressTombstones = suppress.shouldSuppressTombstones();
safeToDropTombstones = suppress.safeToDropTombstones();
}

@SuppressWarnings("unchecked")
Expand Down Expand Up @@ -136,7 +136,7 @@ private void emit(final KeyValue<Bytes, ContextualRecord> toEmit) {
}

private boolean shouldForward(final Change<V> value) {
return !(value.newValue == null && shouldSuppressTombstones);
return value.newValue != null || !safeToDropTombstones;
}

@Override
Expand Down
31 changes: 22 additions & 9 deletions ...src/main/java/org/apache/kafka/streams/kstream/internals/suppress/SuppressedInternal.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,23 +30,36 @@ public class SuppressedInternal<K> implements Suppressed<K> {
private final BufferConfigInternal bufferConfig;
private final Duration timeToWaitForMoreEvents;
private final TimeDefinition<K> 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

@guozhangwang guozhangwang Jan 25, 2019

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Sounds good.

Actually I've seen the same situation for our current caching layer flushing logic as well: e.g. put(A, v) -> delete(A) and both only hit the cache layer. When flushing we tried to read the old value and found its null bytes, so we know nothing was flushed for A and nothing written to downstream before so we can skip putting a tombstone to underlying store as well as downstream.

For suppression buffer though, it is harder since you do not have an underlying store to fetch the old value, and of course reading the whole changelog to see if there's any updates on this key A costs you everything. But suppose we always have a persistent buffer, this may be an easier task.

@vvcephei vvcephei Jan 25, 2019

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks @guozhangwang ,

Yes, I think that's why the record cache keeps a set of "dirty keys", just like the suppression buffer. If you have a dirty-key, but don't find any value for it, then you know it was a delete and you can send a tombstone.

It is indeed harder for the suppression buffer, since we try not to completely duplicate all the data. I guess it's not all the data, but it is all the keys at least. For example, let's say we store the data as the heterogeneous type [nonTombstoneValueHasBeenEmittedFlag] | [nonTombstoneValueHasBeenEmittedFlag, valueToBeEmitted]. Then, when we get a tombstone, if the pre-existing value has nonTombstoneValueHasBeenEmittedFlag == true, we know we must emit the tombstone. If there is no value, or if nonTombstoneValueHasBeenEmittedFlag == false, we know that we don't need to send the tombstone. Upon sending the tombstone, we would simply delete the record from the store. Upon emitting a non-tombstone value, we would drop the value from the store and only store [nonTombstoneValueHasBeenEmittedFlag := true].

Not sure I would do this in memory, but as you point out, it could be an option for the persistent version. I guess if you think the whole "live" key space would fit in memory (plus one bit per key for the flag), then in-memory would work too.

@guozhangwang guozhangwang Jan 25, 2019

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Right. But note that the current dirty-key in cache is not enough determining if we have, ever, write for a key to the underlying store which is not deleted yet: dirty-key only contains the dirty-keys since last flush, i.e. the key not in the dirty-key is only a necessary, not sufficient condition. And that's why we can only consider not writing the tombstone if the read on this key returns null-bytes, indicating nothing was there.

@vvcephei vvcephei Jan 25, 2019

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Oh, right, I guess we would need to do something like what I described above, or some equivalent solution...

* 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<K> 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<K> withName(final String name) {
return new SuppressedInternal<>(name, timeToWaitForMoreEvents, bufferConfig, timeDefinition, suppressTombstones);
return new SuppressedInternal<>(name, timeToWaitForMoreEvents, bufferConfig, timeDefinition, safeToDropTombstones);
}

public String name() {
Expand All @@ -65,8 +78,8 @@ Duration timeToWaitForMoreEvents() {
return timeToWaitForMoreEvents == null ? Duration.ZERO : timeToWaitForMoreEvents;
}

boolean shouldSuppressTombstones() {
return suppressTombstones;
boolean safeToDropTombstones() {
return safeToDropTombstones;
}

@Override
Expand All @@ -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) &&
Expand All @@ -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
Expand All @@ -96,7 +109,7 @@ public String toString() {
", bufferConfig=" + bufferConfig +
", timeToWaitForMoreEvents=" + timeToWaitForMoreEvents +
", timeDefinition=" + timeDefinition +
", suppressTombstones=" + suppressTombstones +
", safeToDropTombstones=" + safeToDropTombstones +
'}';
}
}
33 changes: 28 additions & 5 deletions ...java/org/apache/kafka/streams/kstream/internals/suppress/KTableSuppressProcessorTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<Windowed<String>, Long> harness =
new Harness<>(finalResults(ofMillis(0L)), timeWindowedSerdeFrom(String.class, 100L), Long());
final MockInternalProcessorContext context = harness.context;
Expand All @@ -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<Windowed<String>, Long> harness =
new Harness<>(finalResults(ofMillis(0L)), sessionWindowedSerdeFrom(String.class), Long());
final MockInternalProcessorContext context = harness.context;
Expand All @@ -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<Windowed<String>, Long> harness =
new Harness<>(untilTimeLimit(ofMillis(0), maxRecords(0)), timeWindowedSerdeFrom(String.class, 100L), Long());
final MockInternalProcessorContext context = harness.context;
Expand All @@ -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<Windowed<String>, Long> harness =
new Harness<>(untilTimeLimit(ofMillis(0), maxRecords(0)), sessionWindowedSerdeFrom(String.class), Long());
final MockInternalProcessorContext context = harness.context;
Expand All @@ -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<String, Long> harness =
new Harness<>(untilTimeLimit(ofMillis(0), maxRecords(0)), String(), Long());
final MockInternalProcessorContext context = harness.context;
Expand Down