KAFKA-7223: add tests in preparation for suppression by vvcephei · Pull Request #5687 · apache/kafka

vvcephei · 2018-09-24T20:53:39Z

This is Part 2 of suppression.
Part 1 was #5567

In an effort to control the scope of the review, this PR is just the tests for buffered suppression.

Committer Checklist (excluded from commit message)

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

vvcephei · 2018-09-24T20:57:49Z

When you have a chance please take a look at this PR.

As the description says, it contains only the unit tests that we will use for buffered suppressions.
Clearly, since it's not implemented yet, the tests can't all pass, so you'll notice that many have been configured to expect exceptions. This expectation will be removed when I add the implementation.

I'm hoping that this progression is more reviewable than one large chunk, and also that it makes sense in its resemblence to TDD. I'd like to think of this as writing/reviewing the tests first, so we agree on the expected behavior. With that agreement in place, we'll be able to concentrate on the implementation in a follow-on PR.

WDYT?

vvcephei · 2018-09-24T21:03:47Z

+
+import static java.util.Objects.requireNonNull;
+
+public class FullChangeSerde<T> implements Serde<Change<T>> {


Unlike the source/sink nodes, the suppress processor needs to be able to handle Changes in which both new and old are non-null.

vvcephei · 2018-09-24T21:04:23Z

    public KTable<K, V> suppress(final Suppressed<K> suppressed) {
        final String name = builder.newProcessorName(SUPPRESS_NAME);

+        // TODO: follow-up pr to forward the k/v serdes


This will be fixed in part 3

vvcephei · 2018-09-24T21:05:07Z

    }

-    static class NotImplementedException extends RuntimeException {
+    public static class NotImplementedException extends RuntimeException {


Just so we can assert that it happens in the scenario tests (which are in a different package)

vvcephei · 2018-09-24T21:07:08Z

    private static final int SCALE_FACTOR = COMMIT_INTERVAL * 2;
+    private static final long TIMEOUT_MS = 30_000L;
+
+    @Ignore


Note these tests are ignored because they can't run until Part 3.

vvcephei · 2018-09-24T21:08:21Z

        }
    }

+    @Test(expected = ProcessorStateException.class)


Note that we're expecting exceptions for now. This will be removed in Part 3.

vvcephei · 2018-09-24T21:13:16Z

+                drainProducerRecords(driver, "output-suppressed", STRING_DESERIALIZER, LONG_DESERIALIZER),
+                asList(
+                    // TODO: it's not strictly necessary to emit these in final mode, but it's also not harmful... maybe?
+                    new KeyValueTimestamp<>("[k1@0/0]", null, 1L),


The todo refers to emitting this null result, which is a result of session windows merging.

I suppose the suppression operator could drop results with null values if (and only if) it's in "final results" mode. But on the other hand, I worry that it's too much magic.

WDYT?

Here's my thinking: for session windows, when we merge the window we will send negating old values downstream, but with final output we should not send any data yet before merging and sending the final result, so there is no point sending null, and hence in this case we should remove it in the suppressed result as well?

Ok, it's a little more complex to maintain as a special case, but it does seem better not to burden our users with mysteries for the sake of convenience.

I agree with dropping null values

guozhangwang · 2018-09-25T00:45:31Z

+                if (newBytes != null) {
+                    buffer.put(newBytes);
+                }
+                return buffer.array();


I think we cannot always return buffer.array() since it just returns the backing byte array of this ByteBuffer, and since multiple ByteBuffer may potentially sits on the same byte array, this array() may actually be much larger than the ByteBuffer's capacity. To be safer we need to do sth. in the ByteBufferSerializer:

if (data.hasArray()) { byte[] arr = data.array(); if (data.arrayOffset() == 0 && arr.length == data.remaining()) { return arr; } } byte[] ret = new byte[data.remaining()]; data.get(ret, 0, ret.length); data.rewind(); return ret;

Ok, thanks. I'm (clearly) not too familiar with the API. What you say makes sense.

guozhangwang · 2018-09-25T00:49:51Z

+                                   final Serde<K> keySerde,
+                                   final Serde<Change<V>> valueSerde) {
+        this.suppress = requireNonNull(suppress);
+        this.keySerde = keySerde;


Where are these two serdes used in this PR?

They aren't. They will be in part 3, so the tests need to call this constructor. I added this constructor and the dummy fields just so the (ignored) tests can compile.

It's a bit awkward, I know.

guozhangwang · 2018-09-25T00:56:22Z

    private static final LongDeserializer LONG_DESERIALIZER = new LongDeserializer();
    private static final int COMMIT_INTERVAL = 100;
    private static final int SCALE_FACTOR = COMMIT_INTERVAL * 2;
+    private static final long TIMEOUT_MS = 30_000L;


Why SCALE_FACTOR is COMMIT_INTERVAL * 2, is it just to make sure at least one commit is called during this interval of time?

Yes. It doesn't actually matter whether or not a commit in called in between the various times used in tests (I've tested it both ways). But it seemed like a reasonable concern that the commit might mess up the suppression in some way, so I went ahead and made sure the tests included commits.

Sounds good.

guozhangwang · 2018-09-25T01:07:06Z

+            verifyOutput(
+                outputSuppressed,
+                asList(
+                    new KeyValueTimestamp<>("v2", 1L, scaledTime(1L)),


Why the supressed result will not have tick without filtering?

Only because it isn't time to emit it yet. The tick is the newest event, at time 5, so it's still stream-time 5. Based on the suppression config, we won't emit it until after time 7.

The only reason I filtered it out of the "raw" output is that I felt it made the tests more confusing to read. Do you think I should also filter it from the suppressed output (even though it doesn't matter), just to avoid distracting readers with the question you raised?

The reasoning makes sense.

I think not explicitly filtering tick and expect in the output raw is actually fine, and to me it is better exposing this behavior to make sure the raw data did include the tick value while the suppressed one does not.

Ok, it's easy enough to put it back in.

I had the same question about tick I also think it makes sense to have it in the raw data.

…to array

vvcephei · 2018-09-25T02:23:42Z

+                if (newBytes != null) {
+                    buffer.put(newBytes);
+                }
+                return byteBufferSerializer.serialize(null, buffer);


@guozhangwang , rather than copy/paste the non-trivial transformation logic, what do you think about just delegating to the actual ByteBufferSerializer?

The deserializer is trivial, but I went ahead and used it in my deserializer, too, for symmetry.

I'm thinking that as long as the ByteBufferSerializer is stable, and it doesn't do anything weird in the future, this should be fine. And that should be the case, since future versions of the code are going to have to be compatible with data written by older versions.

Sounds good.

guozhangwang

The tests LGTM, most of the comments are minor.

One meta comment though: from the unit tests in seems we are going to register one scheduled punctuator for each key in time-based intermediate suppression, and one punctuator for each window in final result suppression in window store, is that right? I'm a bit concerned about the punctuator overhead and how we can maintain it during failover, but I understand it is for part 3 not this PR.

guozhangwang · 2018-09-25T04:37:04Z

+        cleanStateBeforeTest(input, outputRaw, outputSuppressed);
+
+        final StreamsBuilder builder = new StreamsBuilder();
+        final KTable<String, Long> valueCounts = builder


Can we share the logic of creating the valueCounts as well?

guozhangwang · 2018-09-25T04:42:28Z

+                    new KeyValueTimestamp<>("k1", "v1", scaledTime(0L)),
+                    new KeyValueTimestamp<>("k1", "v2", scaledTime(1L)),
+                    new KeyValueTimestamp<>("k2", "v1", scaledTime(2L)),
+                    new KeyValueTimestamp<>("tick", "tick", scaledTime(5L))


I'd suggest 1) adding another record at 7L and 2) add a filter before line 111 to filter this record out, so that effectively this record will not be processed at all but only moves the stream time. And we can then verify that the final result does not have any outputs from this input, but the tick record did emit because of the stream time advances.

I think I might have misunderstood you. As I read it, the role proposed for this extra record is exactly what tick is there for. It exists only to advance stream time, and it gets filtered from the output.

fwiw, I think your earlier suggestion to not filter it is fine. I'm also adding a couple of comments explaining what's happening with it.

guozhangwang · 2018-09-25T04:48:55Z

+                new KeyValueTimestamp<>("k1", "v1", scaledTime(2L)),
+                new KeyValueTimestamp<>("k1", "v1", scaledTime(1L)),
+                new KeyValueTimestamp<>("k1", "v1", scaledTime(0L)),
+                new KeyValueTimestamp<>("k1", "v1", scaledTime(4L))


Add another late record after this to be falling out of the grace period, and verify that it will not be reflected in the final result and be skipped?

guozhangwang · 2018-09-25T05:02:11Z

+            driver.pipeInput(recordFactory.create("input", "k1", "v1", 2L));
+            driver.pipeInput(recordFactory.create("input", "k1", "v1", 1L));
+            driver.pipeInput(recordFactory.create("input", "k1", "v1", 0L));
+            driver.pipeInput(recordFactory.create("input", "k1", "v1", 5L));


Ditto as in the integration test.

guozhangwang · 2018-09-25T05:14:52Z

+                drainProducerRecords(driver, "output-suppressed", STRING_DESERIALIZER, LONG_DESERIALIZER),
+                asList(
+                    // TODO: it's not strictly necessary to emit these in final mode, but it's also not harmful... maybe?
+                    new KeyValueTimestamp<>("[k1@0/0]", null, 1L),


Here's my thinking: for session windows, when we merge the window we will send negating old values downstream, but with final output we should not send any data yet before merging and sending the final result, so there is no point sending null, and hence in this case we should remove it in the suppressed result as well?

guozhangwang · 2018-09-25T05:17:02Z


 @SuppressWarnings("PointlessArithmeticExpression")
 public class KTableSuppressProcessorTest {
-    /**


Why remove the comments here?

They just seemed unnecessary, as the variable names should make it obvious.

guozhangwang · 2018-09-25T05:44:29Z

-        final KTableSuppressProcessor<Windowed<String>, Long> processor =
-            new KTableSuppressProcessor<>(finalResults(ofMillis(1)));
+    @Test(expected = KTableSuppressProcessor.NotImplementedException.class)
+    public void finalResultsSuppressionShouldBufferAndEmitLater() {


What's the difference between this test and the finalResultsWith0GraceBeforeWindowEndShouldBufferAndEmitLater below? Besides grace they seems exactly the same.

In that case, could we just merge these two, to randomly choose one as the grace period, and with a significant amount of likelihood that the random number is 0?

They are similar, but targeting different behaviors:

finalResultsSuppressionShouldBufferAndEmitLater is the "main" unit test: verifying that the buffer saves up all the data and emits at the grace period expiration.

finalResultsWith0GraceBeforeWindowEndShouldBufferAndEmitLater is somewhat of a "regression" test. I added it during implementation because I initially failed to account for this behavior: even when the grace period is 0, the final results buffer should wait until the end of the window before emitting (instead of immediately emitting). I think I can name the method better and add a comment to make this clear.

As far as randomized testing goes, I'd like to experiment with a fuzzing framework in the future, rather than hacking out a half-baked one right now. There are some important features, such as a mechanism to reproduce failing tests, that are necessary to make it a "randomized test", rather than a "flaky test".

For now, I'd rather just stick with hand-written scenarios.

Makes sense.

vvcephei · 2018-09-25T15:21:20Z

@guozhangwang , thanks for the review. I think I've addressed all your comments.

About the punctuation overhead, I share your concern, and I think we can optimize it, but I agree we should table this discussion until part 3.

bbejeck

Thanks @vvcephei, I made a pass and looks good overall, just a few minor comments.

One additional comment should we add unit tests to KTableSuppressProcessorTest with records falling inside the grace period as well as outside of it?

bbejeck · 2018-09-25T14:18:29Z

+public class FullChangeSerde<T> implements Serde<Change<T>> {
+    private final Serde<T> inner;
+
+    public FullChangeSerde(final Serde<T> inner) {


Can we get a unit test for FullChangeSerde?

Oh! good catch.

Thanks for this. I added one, and it turns out I caught a couple of bugs!

bbejeck · 2018-09-25T14:51:34Z

+            verifyOutput(
+                outputSuppressed,
+                asList(
+                    new KeyValueTimestamp<>("v2", 1L, scaledTime(1L)),


I had the same question about tick I also think it makes sense to have it in the raw data.

bbejeck · 2018-09-25T15:03:14Z

-        final KTableSuppressProcessor<Windowed<String>, Long> processor =
-            new KTableSuppressProcessor<>(finalResults(ofMillis(0)));
+    @Test(expected = KTableSuppressProcessor.NotImplementedException.class)
+    public void finalResultsWith0GraceBeforeWindowEndShouldBufferAndEmitLater() {


nit: I realize this was here in a previous commit, but in this test and the next spell out Zero vs using the number, it's hard to read.

bbejeck · 2018-09-25T15:18:59Z

+                    new KeyValueTimestamp<>("v1", 1L, 2L)
+                )
+            );
+            driver.pipeInput(recordFactory.create("input", "tick", "tick", 5L));


same here as with the integration test, maybe don't filter tick from raw to show that suppression is keeping it from being emitted.

Ok, I've removed the filter, and added expectations as well as explanatory comments for "tick"

bbejeck · 2018-09-25T15:25:05Z

+                drainProducerRecords(driver, "output-suppressed", STRING_DESERIALIZER, LONG_DESERIALIZER),
+                asList(
+                    // TODO: it's not strictly necessary to emit these in final mode, but it's also not harmful... maybe?
+                    new KeyValueTimestamp<>("[k1@0/0]", null, 1L),


I agree with dropping null values

guozhangwang · 2018-09-25T16:56:24Z

@vvcephei Thanks for your replies. I do not have further comments. And once @bbejeck is happy with the current patch we can merge it.

vvcephei · 2018-09-25T17:34:21Z

+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.core.Is.is;
+
+public class FullChangeSerdeTest {


@bbejeck Here's the test you asked for.

vvcephei · 2018-09-25T17:36:52Z

Ok @bbejeck, I believe I've addressed all your and @guozhangwang 's comments.
Do you mind making a final pass?

bbejeck

@vvcephei thanks for the updates, LGTM

vvcephei · 2018-09-25T20:42:04Z

Once again, my sincere thanks for the reviews, @guozhangwang and @bbejeck

This is Part 4 of suppression (durability) Part 1 was #5567 (the API) Part 2 was #5687 (the tests) Part 3 was #5693 (in-memory buffering) Implement a changelog for the suppression buffer so that the buffer state may be recovered on restart or recovery. As of this PR, suppression is suitable for general usage. Reviewers: Bill Bejeck <bill@confluent.io>, Guozhang Wang <guozhang@confluent.io>, Matthias J. Sax <matthias@confluent.io>

This is Part 2 of suppression. Part 1 was apache#5567 In an effort to control the scope of the review, this PR is just the tests for buffered suppression. Reviewers: Bill Bejeck <bill@confluent.io>, Guozhang Wang <wangguoz@gmail.com>

This is Part 4 of suppression (durability) Part 1 was apache#5567 (the API) Part 2 was apache#5687 (the tests) Part 3 was apache#5693 (in-memory buffering) Implement a changelog for the suppression buffer so that the buffer state may be recovered on restart or recovery. As of this PR, suppression is suitable for general usage. Reviewers: Bill Bejeck <bill@confluent.io>, Guozhang Wang <guozhang@confluent.io>, Matthias J. Sax <matthias@confluent.io>

KAFKA-7223: add unit tests in preparation for suppression

3b74c63

add integration tests

70d74df

vvcephei commented Sep 24, 2018

View reviewed changes

vvcephei changed the title ~~KAFKA-7223: add unit tests in preparation for suppression~~ KAFKA-7223: add tests in preparation for suppression Sep 24, 2018

vvcephei mentioned this pull request Sep 24, 2018

KAFKA-7223: Part 3 preview #5688

Closed

3 tasks

guozhangwang reviewed Sep 25, 2018

View reviewed changes

defer to ByteBufferSerDes for non-trivial transformation from buffer …

f7b8ea3

…to array

vvcephei commented Sep 25, 2018

View reviewed changes

guozhangwang reviewed Sep 25, 2018

View reviewed changes

vvcephei added 3 commits September 25, 2018 10:15

cr comments

2a78450

cr comments

189c83b

cr comments

4c2a4d7

bbejeck reviewed Sep 25, 2018

View reviewed changes

update tests

7c000b2

vvcephei commented Sep 25, 2018

View reviewed changes

fix style

b40354c

bbejeck approved these changes Sep 25, 2018

View reviewed changes

guozhangwang merged commit f712ce6 into apache:trunk Sep 25, 2018

vvcephei deleted the KAFKA-7223-in-mem-suppress-unit-tests branch September 25, 2018 20:42

vvcephei mentioned this pull request Sep 25, 2018

KAFKA-7223: In-Memory Suppression Buffering #5693

Merged

3 tasks

mjsax added the streams label Sep 25, 2018

vvcephei mentioned this pull request Oct 2, 2018

KAFKA-7223: Make suppression buffer durable #5724

Merged

3 tasks


		import static java.util.Objects.requireNonNull;

		public class FullChangeSerde<T> implements Serde<Change<T>> {

Uh oh!

Conversation

vvcephei commented Sep 24, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Committer Checklist (excluded from commit message)

Uh oh!

vvcephei commented Sep 24, 2018

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

guozhangwang left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

vvcephei commented Sep 24, 2018 •

edited

Loading