KAFKA-14936: Add Grace period logic to Stream Table Join (2/N) #13855

wcarlson5 · 2023-06-14T20:30:11Z

This Pr adds the interface for grace period to the Joined object as well as uses the buffer. The majority of it is tests and moving some of the existing join logic.

Committer Checklist (excluded from commit message)

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

cadonna

Thanks for the PR, @wcarlson5!

Do you plan to add the changelogging in on of the next PRs?

Here my feedback on this PR.

I think you should move the eviction logic completely to the buffer. That would make the code easier to maintain.

cadonna · 2023-06-15T07:08:14Z

streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamImpl.java

+
+        Optional<TimeOrderedKeyValueBuffer<K, V, V>> buffer = Optional.empty();
+
+        if (joined.gracePeriod() != null) {


Shouldn't this be

Suggested change

if (joined.gracePeriod() != null) {

if (joined.gracePeriod() != null && joined.gracePeriod() > 0) {

In the KIP discussion we said that a grace period of zero is equal to no grace period.

Note, some conditions later on need to be changed if you agree with this change, like for example gracePeriod.isPresent() ^ buffer.isPresent().

I see what you mean but we actually were going to create the store and leave it empty for zero duration. The point was to make it easier to change the grace period if desired so the store isn't getting created and destroyed. Something about making it more backward compatible.

So we are going to make a store even when the grace period is zero. This should make changing the grace period to and from zero smoother. So if someone set the grace period to zero then we can evict everything from the buffer even after the grace period is changed.

So if someone set the grace period to zero then we can evict everything from the buffer even after the grace period is changed.

Are you saying that if I set the grace period > 0 and I run my application, then I stop my application and set the grace period == 0, and restart my application, my application will process records from my first run?
Why should removing the grace period (grace period == null) behave differently?

Another thought here:
What happens if I specify a grace period > 0 and run my application for a while. Then, I remove the grace period and again run my application for a while. Finally, I specify a grace period > 0 or even == 0 and again run my application for a while. Does my application in the last run process records that the first run wrote into the buffer? (Assuming the task is always assigned to the same node).

What I am trying to say is that we should try to put simplicity over partial support for topology updates.

This seems to be the main sticking point of the PR.

The thought was that this that we would need some way to stop using the grace period without loosing records, but you bring up a good point.

...ams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamKTableJoinProcessor.java

cadonna · 2023-06-15T07:54:22Z

...ams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamKTableJoinProcessor.java

+            doJoin(record);
+        } else {
+            final long deadline = observedStreamTime - gracePeriod.get().toMillis();
+            if (record.timestamp() <= deadline) {


Why here record.timestamp() <= deadline and in RocksDBTimeOrderedKeyValueBuffer at line 127 wrapped().observedStreamTime - gracePeriod > record.timestamp()?

I am wondering, if it would be simpler to remove the condition in RocksDBTimeOrderedKeyValueBuffer at line 127 and only check late records here. In the end, you do not get any return value from the put() in RocksDBTimeOrderedKeyValueBuffer. So it would silently not add the record to the buffer. Such a behavior might become hard to debug in case of a bug and it would also be quite error-prone because you have the condition for late records in two places instead of just one.

I am even in favor of let RocksDBTimeOrderedKeyValueBuffer#put() return whether the record was added to the buffer or not. In such a way, you can move the logic for eviction and late records to the buffer instead of having it half here and half there making maintaining the code harder.

That is a good idea. I'm also in favor of having put decide.

cadonna · 2023-06-15T08:39:56Z

...ams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamKTableJoinProcessor.java

+            } else {
+                buffer.get().put(observedStreamTime, record, internalProcessorContext.recordContext());
+            }
+            buffer.get().evictWhile(() -> buffer.get().minTimestamp() <= deadline, this::emit);


I am wondering if it might not be simpler to use as predicate just () -> true instead of () -> buffer.get().minTimestamp() <= deadline and leave the conditions for evicting to the buffer. Otherwise, we have half of the logic for evicting here and the other half in the buffer. That seems error-prone and hard to maintain and debug to me.

it should be the same. So sure I can go with the simple approach

cadonna · 2023-06-15T09:01:34Z

.../test/java/org/apache/kafka/streams/integration/StreamTableJoinWithGraceIntegrationTest.java

+        super.prepareEnvironment();
+        appID = "stream-table-join-integration-test";
+        builder = new StreamsBuilder();
+        joined = Joined.with(Serdes.Long(), Serdes.String(), Serdes.String(), "Grace", Duration.ZERO);


Why do you have only tests with grace period set to zero. I thought a grace period of zero is equivalent to no grace period. Basically, you are not testing the new code if you look at it from the outside.

I'll add some, I think I need to rework what is in the integration tests and the unit tests. Should be fixed.

cadonna · 2023-06-15T09:02:53Z

...src/test/java/org/apache/kafka/streams/kstream/internals/KStreamKTableJoinWithGraceTest.java

+    }
+
+    @Test
+    public void shouldFailIfGracePeriodIsLongerThanHistoryRetention() {


I am missing assertions in this test. You should call the init() and process() methods directly.

Ah this should be removed. It will be added back in the next PR (or the one after)

cadonna · 2023-06-15T09:05:22Z

...src/test/java/org/apache/kafka/streams/kstream/internals/KStreamKTableJoinWithGraceTest.java

+        processor.checkAndClearProcessResult(new KeyValueTimestamp<>(0, "X0+Y0", 0),
+            new KeyValueTimestamp<>(1, "X1+Y1", 1));


nit:

Suggested change

processor.checkAndClearProcessResult(new KeyValueTimestamp<>(0, "X0+Y0", 0),

new KeyValueTimestamp<>(1, "X1+Y1", 1));

processor.checkAndClearProcessResult(

new KeyValueTimestamp<>(0, "X0+Y0", 0),

new KeyValueTimestamp<>(1, "X1+Y1", 1));

cadonna · 2023-06-15T09:06:45Z

...src/test/java/org/apache/kafka/streams/kstream/internals/KStreamKTableJoinWithGraceTest.java

+        processor.checkAndClearProcessResult(new KeyValueTimestamp<>(0, "X0+YY0", 0),
+            new KeyValueTimestamp<>(1, "X1+YY1", 1),
+            new KeyValueTimestamp<>(2, "X2+YY2", 2),
+            new KeyValueTimestamp<>(3, "X3+YY3", 3));


nit:

Suggested change

processor.checkAndClearProcessResult(new KeyValueTimestamp<>(0, "X0+YY0", 0),

new KeyValueTimestamp<>(1, "X1+YY1", 1),

new KeyValueTimestamp<>(2, "X2+YY2", 2),

new KeyValueTimestamp<>(3, "X3+YY3", 3));

processor.checkAndClearProcessResult(

new KeyValueTimestamp<>(0, "X0+YY0", 0),

new KeyValueTimestamp<>(1, "X1+YY1", 1),

new KeyValueTimestamp<>(2, "X2+YY2", 2),

new KeyValueTimestamp<>(3, "X3+YY3", 3));

cadonna · 2023-06-15T09:10:08Z

...src/test/java/org/apache/kafka/streams/kstream/internals/KStreamKTableJoinWithGraceTest.java

+    private final int[] expectedKeys = {0, 1, 2, 3};
+
+    private MockApiProcessor<Integer, String, Void, Void> processor;
+    private TopologyTestDriver driver;


Why do you need the topology test driver for unit tests?

So I was following the normal KStreamKTableJoinTest. I think its best if they aline with each other.

I think its best if they aline with each other

Why is that?

I find it a bit sad that we do not have real unit tests for the processor. But it's not your fault. I just wanted to document it somewhere.

Another related question: What is the difference between KStreamKTableJoinWithGraceTest and StreamTableJoinWithGraceIntegrationTest? They both use the TopologyTestDriver so it seems they both test the stream table join on the same level. I would rather expect to have unit tests that test directly the processor and an intergration test that tests the integration of the processor.

BTW, using unit tests would make the tests much simpler. You just need to mock the value getter (and maybe some other parts) instead of building up a whole topology.

Okay so I actually tried to set this up and it was NOT simpler. The mocking got way out of hand. Normally I 100 precent agree with out, mocking is way better.

Due to you other comment I just ended up moving this whole class to the unit testing class for the normal Join anyways, so this is a bit of a moot point

wcarlson5

I made updates to the code, still working on rejiggering the tests to see if I can get Junit5 working. Hopfully will get that soon

wcarlson5 · 2023-06-15T17:19:08Z

streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamImpl.java

+
+        Optional<TimeOrderedKeyValueBuffer<K, V, V>> buffer = Optional.empty();
+
+        if (joined.gracePeriod() != null) {


I see what you mean but we actually were going to create the store and leave it empty for zero duration. The point was to make it easier to change the grace period if desired so the store isn't getting created and destroyed. Something about making it more backward compatible.

wcarlson5 · 2023-06-16T21:57:52Z

streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamImpl.java

+
+        Optional<TimeOrderedKeyValueBuffer<K, V, V>> buffer = Optional.empty();
+
+        if (joined.gracePeriod() != null) {


So we are going to make a store even when the grace period is zero. This should make changing the grace period to and from zero smoother. So if someone set the grace period to zero then we can evict everything from the buffer even after the grace period is changed.

...ams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamKTableJoinProcessor.java

wcarlson5 · 2023-06-16T22:05:59Z

...ams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamKTableJoinProcessor.java

+            } else {
+                buffer.get().put(observedStreamTime, record, internalProcessorContext.recordContext());
+            }
+            buffer.get().evictWhile(() -> buffer.get().minTimestamp() <= deadline, this::emit);


it should be the same. So sure I can go with the simple approach

wcarlson5 · 2023-06-16T22:11:09Z

...ams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamKTableJoinProcessor.java

+            doJoin(record);
+        } else {
+            final long deadline = observedStreamTime - gracePeriod.get().toMillis();
+            if (record.timestamp() <= deadline) {


That is a good idea. I'm also in favor of having put decide.

.../test/java/org/apache/kafka/streams/integration/StreamTableJoinWithGraceIntegrationTest.java

wcarlson5 · 2023-06-16T22:17:49Z

...src/test/java/org/apache/kafka/streams/kstream/internals/KStreamKTableJoinWithGraceTest.java

+    private final int[] expectedKeys = {0, 1, 2, 3};
+
+    private MockApiProcessor<Integer, String, Void, Void> processor;
+    private TopologyTestDriver driver;


So I was following the normal KStreamKTableJoinTest. I think its best if they aline with each other.

wcarlson5 · 2023-06-16T22:18:20Z

...src/test/java/org/apache/kafka/streams/kstream/internals/KStreamKTableJoinWithGraceTest.java

+    }
+
+    @Test
+    public void shouldFailIfGracePeriodIsLongerThanHistoryRetention() {


Ah this should be removed. It will be added back in the next PR (or the one after)

wcarlson5 · 2023-06-16T22:30:35Z

.../test/java/org/apache/kafka/streams/integration/StreamTableJoinWithGraceIntegrationTest.java

+        super.prepareEnvironment();
+        appID = "stream-table-join-integration-test";
+        builder = new StreamsBuilder();
+        joined = Joined.with(Serdes.Long(), Serdes.String(), Serdes.String(), "Grace", Duration.ZERO);


I'll add some, I think I need to rework what is in the integration tests and the unit tests. Should be fixed.

wcarlson5 · 2023-06-20T02:54:50Z

@cadonna I got around to updating the tests as well like you asked

wcarlson5

I made updates to the code, still working on rejiggering the tests to see if I can get Junit5 working. Hopfully will get that soon

.../test/java/org/apache/kafka/streams/integration/StreamTableJoinWithGraceIntegrationTest.java

cadonna

Thanks for the updates, @wcarlson5 !

Here my feedback.

...ams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamKTableJoinProcessor.java

cadonna · 2023-06-20T14:43:18Z

...ams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamKTableJoinProcessor.java

+            }
+
+            buffer.get().setSerdesIfNull(new SerdeGetter(context));
+            //cast doesn't matter, it is just because the processor is deprecated. The context gets converted back with StoreToProcessorContextAdapter.adapt(context)


Is this comment so important or can you remove it?

Yeah we can remove

...src/main/java/org/apache/kafka/streams/state/internals/RocksDBTimeOrderedKeyValueBuffer.java

cadonna · 2023-06-22T14:58:07Z

...src/test/java/org/apache/kafka/streams/kstream/internals/KStreamKTableJoinWithGraceTest.java

+    private final int[] expectedKeys = {0, 1, 2, 3};
+
+    private MockApiProcessor<Integer, String, Void, Void> processor;
+    private TopologyTestDriver driver;


I think its best if they aline with each other

Why is that?

I find it a bit sad that we do not have real unit tests for the processor. But it's not your fault. I just wanted to document it somewhere.

cadonna · 2023-06-22T15:08:52Z

...src/test/java/org/apache/kafka/streams/kstream/internals/KStreamKTableJoinWithGraceTest.java

+    private final int[] expectedKeys = {0, 1, 2, 3};
+
+    private MockApiProcessor<Integer, String, Void, Void> processor;
+    private TopologyTestDriver driver;


Another related question: What is the difference between KStreamKTableJoinWithGraceTest and StreamTableJoinWithGraceIntegrationTest? They both use the TopologyTestDriver so it seems they both test the stream table join on the same level. I would rather expect to have unit tests that test directly the processor and an intergration test that tests the integration of the processor.

cadonna · 2023-06-22T16:38:35Z

...src/test/java/org/apache/kafka/streams/kstream/internals/KStreamKTableJoinWithGraceTest.java

+        final KStream<String, String> streamA = builder.stream("topic", Consumed.with(Serdes.String(), Serdes.String()));
+        final KTable<String, String> tableB = builder.table("topic2", Consumed.with(Serdes.String(), Serdes.String()));
+
+        assertThrows(IllegalArgumentException.class,


Please do also verify the exception message. We had cases in the past where the test did not fail although a different exception was thrown.

sure, good call.

...src/test/java/org/apache/kafka/streams/kstream/internals/KStreamKTableJoinWithGraceTest.java

cadonna · 2023-06-22T16:46:05Z

...src/test/java/org/apache/kafka/streams/kstream/internals/KStreamKTableJoinWithGraceTest.java

+    }
+
+    @Test
+    public void shouldNotJoinOnTableUpdates() {


Same here and for the rest of the test class. I think you should just move the first two tests (i.e., the new ones) to KStreamKTableJoinTest.

That is a good idea, massively simplified the code. I resolved the other comments that were pretty much the same as this one

wcarlson5

Thanks for the review @cadonna ! It looks like we have one thing to figure out.

wcarlson5 · 2023-06-22T16:57:23Z

...ams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamKTableJoinProcessor.java

+            }
+
+            buffer.get().setSerdesIfNull(new SerdeGetter(context));
+            //cast doesn't matter, it is just because the processor is deprecated. The context gets converted back with StoreToProcessorContextAdapter.adapt(context)


Yeah we can remove

...src/main/java/org/apache/kafka/streams/state/internals/RocksDBTimeOrderedKeyValueBuffer.java

wcarlson5 · 2023-06-22T17:07:00Z

...src/test/java/org/apache/kafka/streams/kstream/internals/KStreamKTableJoinWithGraceTest.java

+        final KStream<String, String> streamA = builder.stream("topic", Consumed.with(Serdes.String(), Serdes.String()));
+        final KTable<String, String> tableB = builder.table("topic2", Consumed.with(Serdes.String(), Serdes.String()));
+
+        assertThrows(IllegalArgumentException.class,


sure, good call.

wcarlson5 · 2023-06-22T17:11:16Z

...src/test/java/org/apache/kafka/streams/kstream/internals/KStreamKTableJoinWithGraceTest.java

+    private final int[] expectedKeys = {0, 1, 2, 3};
+
+    private MockApiProcessor<Integer, String, Void, Void> processor;
+    private TopologyTestDriver driver;


Okay so I actually tried to set this up and it was NOT simpler. The mocking got way out of hand. Normally I 100 precent agree with out, mocking is way better.

Due to you other comment I just ended up moving this whole class to the unit testing class for the normal Join anyways, so this is a bit of a moot point

wcarlson5 · 2023-06-22T17:27:13Z

...src/test/java/org/apache/kafka/streams/kstream/internals/KStreamKTableJoinWithGraceTest.java

+    }
+
+    @Test
+    public void shouldNotJoinOnTableUpdates() {


That is a good idea, massively simplified the code. I resolved the other comments that were pretty much the same as this one

wcarlson5 · 2023-06-22T17:30:16Z

streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamImpl.java

+
+        Optional<TimeOrderedKeyValueBuffer<K, V, V>> buffer = Optional.empty();
+
+        if (joined.gracePeriod() != null) {


This seems to be the main sticking point of the PR.

The thought was that this that we would need some way to stop using the grace period without loosing records, but you bring up a good point.

wcarlson5 · 2023-06-22T21:04:05Z

The next PR is the restore logic https://github.com/wcarlson5/kafka/pull/2

I have it targeted to this branch for easy reading and will retarget it when this is merged.

@vcrfxia @cadonna

Also a tangental PR for checking the history retention is not less than the grace period can be reviewed and merged at the same time https://github.com/wcarlson5/kafka/pull/3

vcrfxia

Thanks @wcarlson5 . Some suggestions inline, but nothing blocking.

vcrfxia · 2023-06-23T00:12:05Z

streams/src/main/java/org/apache/kafka/streams/kstream/Joined.java

+    }
+
+    /**
+     * Set the grace period on the stream side of the join. Records will enter a buffer before being processed. Out of order records in the grace period will be processed in timestamp order. Late records, out of the grace period, will be executed right as they come in, if it is past the table history retention this could result in joins on the wrong version or a null join. Long gaps in stream side arriving records will cause records to be delayed in processing, even resulting in be processed out of the grace period window.


if it is past the table history retention this could result in joins on the wrong version or a null join

I think this should be "if it is past the table history retention this could result in a null join" instead? I don't think it's possible to join with the wrong version, the only possible issue is that it joins with a null instead.

even resulting in be processed out of the grace period window

What does this mean?

Could you please also add a couple of line breaks to the java docs as you did for the javadocs of the with() method?

updated and clarified

vcrfxia · 2023-06-23T00:13:00Z

...ams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamKTableJoinProcessor.java

+    private void emit(final TimeOrderedKeyValueBuffer.Eviction<K1, V1> toEmit) {
+        final Record<K1, V1> record = new Record<>(toEmit.key(), toEmit.value(), toEmit.recordContext().timestamp())
+            .withHeaders(toEmit.recordContext().headers());
+        internalProcessorContext.setRecordContext(toEmit.recordContext());


Can you unset the record context (i.e., set it back to the original context) after finishing this step, in order to avoid confusion about what context is being used downstream? That's what we do with the suppress buffer, for example:

kafka/streams/src/main/java/org/apache/kafka/streams/kstream/internals/suppress/KTableSuppressProcessorSupplier.java

Lines 207 to 215 in 1dbcb7d

final ProcessorRecordContext prevRecordContext = internalProcessorContext.recordContext();

internalProcessorContext.setRecordContext(toEmit.recordContext());

try {

internalProcessorContext.forward(toEmit.record()

.withTimestamp(toEmit.recordContext().timestamp())

.withHeaders(toEmit.recordContext().headers()));

suppressionEmitSensor.record(1.0d, internalProcessorContext.currentSystemTimeMs());

} finally {

internalProcessorContext.setRecordContext(prevRecordContext);

vcrfxia · 2023-06-23T00:13:13Z

streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamKTableJoinTest.java

+
+        // push all four items to the primary stream. this should produce two items.
+        pushToStream(4, "X");
+        processor.checkAndClearProcessResult(


The reason this produces two output records is because the max timestamp seen so far is still 3, which means only records with timestamp 0 and 1 are emitted (timestamps 2 and 3 are still in the buffer). Can you add another step to this test which now produces a record with a larger timestamp and verifies that the records with timestamps 2 and 3 are emitted? There should be four of them, and they should be emitted in timestamp order which is different from the offset order that they arrived in.

I added something to flush the buffer at the end of this test. Works just fine :)

cadonna

Thanks for the updates!

I have one question about the need to evict when a late record comes in.

Please also consider @vcrfxia 's suggestions about testing and resetting the record context.

cadonna · 2023-06-27T09:51:22Z

streams/src/main/java/org/apache/kafka/streams/kstream/Joined.java

@@ -203,4 +250,4 @@ public Serde<V> valueSerde() {
    public Serde<VO> otherValueSerde() {
        return otherValueSerde;
    }
-}
+}


nit: Could remove this change?

cadonna · 2023-06-27T09:54:08Z

streams/src/main/java/org/apache/kafka/streams/kstream/Joined.java

+    }
+
+    /**
+     * Set the grace period on the stream side of the join. Records will enter a buffer before being processed. Out of order records in the grace period will be processed in timestamp order. Late records, out of the grace period, will be executed right as they come in, if it is past the table history retention this could result in joins on the wrong version or a null join. Long gaps in stream side arriving records will cause records to be delayed in processing, even resulting in be processed out of the grace period window.


Could you please also add a couple of line breaks to the java docs as you did for the javadocs of the with() method?

cadonna · 2023-06-27T11:25:10Z

streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamKTableJoinTest.java

+
+        // push all four items to the primary stream. this should produce two items.
+        pushToStream(4, "X");
+        processor.checkAndClearProcessResult(


cadonna · 2023-06-27T11:37:18Z

...ams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamKTableJoinProcessor.java

+            if (!buffer.get().put(observedStreamTime, record, internalProcessorContext.recordContext())) {
+                doJoin(record);
+            }
+            buffer.get().evictWhile(() -> true, this::emit);


I think that should be the following way to avoid a unnecessary range query:

Suggested change

if (!buffer.get().put(observedStreamTime, record, internalProcessorContext.recordContext())) {

doJoin(record);

}

buffer.get().evictWhile(() -> true, this::emit);

if (!buffer.get().put(observedStreamTime, record, internalProcessorContext.recordContext())) {

doJoin(record);

} else {

buffer.get().evictWhile(() -> true, this::emit);

}

If the record is a late record it will not update the observed stream time. If the observed stream time is not updated, the range query will not return records that need to be evicted, since they have been already evicted the last time evictWhile() was called.
Does this make sense?

If you agree could also please add a test for this?

Sure that is just fine with me. I'll add a test adding records out of the grace period to

wcarlson5

Thanks @cadonna and @vcrfxia

It didn't seem like any blockers are left, is it good to go?

wcarlson5 · 2023-06-28T06:19:49Z

streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamKTableJoinTest.java

+
+        // push all four items to the primary stream. this should produce two items.
+        pushToStream(4, "X");
+        processor.checkAndClearProcessResult(


I added something to flush the buffer at the end of this test. Works just fine :)

wcarlson5 · 2023-06-28T06:24:21Z

...ams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamKTableJoinProcessor.java

+            if (!buffer.get().put(observedStreamTime, record, internalProcessorContext.recordContext())) {
+                doJoin(record);
+            }
+            buffer.get().evictWhile(() -> true, this::emit);


Sure that is just fine with me. I'll add a test adding records out of the grace period to

vcrfxia · 2023-06-28T18:29:11Z

Hey @wcarlson5 did you mean to push a new commit to this PR since @cadonna and I last reviewed? Some of your replies mention having made changes but I don't see any.

Nothing blocking from my side, though I do think that the latest round of suggestions (including test coverage improvements and @cadonna 's suggestion for avoiding an unnecessary range query) would be good to incorporate either in this PR or the next one.

cadonna

Thanks for the updates, @wcarlson5 !

LGTM!

cadonna · 2023-06-29T12:12:15Z

Build failures are unrelated:

    Build / JDK 17 and Scala 2.13 / org.apache.kafka.connect.mirror.integration.IdentityReplicationIntegrationTest.testOneWayReplicationWithFrequentOffsetSyncs()
    Build / JDK 17 and Scala 2.13 / org.apache.kafka.connect.mirror.integration.MirrorConnectorsIntegrationBaseTest.testReplicateSourceDefault()
    Build / JDK 17 and Scala 2.13 / org.apache.kafka.connect.mirror.integration.MirrorConnectorsIntegrationBaseTest.testRestartReplication()
    Build / JDK 17 and Scala 2.13 / org.apache.kafka.connect.mirror.integration.MirrorConnectorsIntegrationTransactionsTest.testNoCheckpointsIfNoRecordsAreMirrored()
    Build / JDK 17 and Scala 2.13 / org.apache.kafka.connect.mirror.integration.MirrorConnectorsIntegrationTransactionsTest.testOneWayReplicationWithFrequentOffsetSyncs()
    Build / JDK 17 and Scala 2.13 / org.apache.kafka.connect.mirror.integration.MirrorConnectorsIntegrationTransactionsTest.testNoCheckpointsIfNoRecordsAreMirrored()
    Build / JDK 17 and Scala 2.13 / kafka.admin.ReassignPartitionsIntegrationTest.testAlterReassignmentThrottle(String).quorum=kraft
    Build / JDK 17 and Scala 2.13 / kafka.zk.ZkMigrationIntegrationTest.[1] Type=ZK, Name=testDualWrite, MetadataVersion=3.4-IV0, Security=PLAINTEXT
    Build / JDK 17 and Scala 2.13 / org.apache.kafka.controller.QuorumControllerTest.testBalancePartitionLeaders()
    Build / JDK 11 and Scala 2.13 / org.apache.kafka.connect.mirror.integration.MirrorConnectorsIntegrationExactlyOnceTest.testOffsetTranslationBehindReplicationFlow()
    Build / JDK 11 and Scala 2.13 / org.apache.kafka.connect.mirror.integration.MirrorConnectorsIntegrationTransactionsTest.testOneWayReplicationWithAutoOffsetSync()
    Build / JDK 11 and Scala 2.13 / org.apache.kafka.connect.mirror.integration.MirrorConnectorsIntegrationTransactionsTest.testOneWayReplicationWithAutoOffsetSync()
    Build / JDK 11 and Scala 2.13 / org.apache.kafka.connect.integration.ExactlyOnceSourceIntegrationTest.testSeparateOffsetsTopic
    Build / JDK 11 and Scala 2.13 / kafka.server.DynamicBrokerReconfigurationTest.testTrustStoreAlter(String).quorum=kraft
    Build / JDK 11 and Scala 2.13 / kafka.server.RaftClusterSnapshotTest.testSnapshotsGenerated()
    Build / JDK 11 and Scala 2.13 / org.apache.kafka.controller.QuorumControllerTest.testBalancePartitionLeaders()
    Build / JDK 11 and Scala 2.13 / org.apache.kafka.streams.integration.KTableKTableForeignKeyInnerJoinCustomPartitionerIntegrationTest.shouldThrowIllegalArgumentExceptionWhenCustomPartitionerReturnsMultiplePartitions()
    Build / JDK 11 and Scala 2.13 / org.apache.kafka.streams.tests.RelationalSmokeTestTest.verifySmokeTestLogic

The looked into the two Kafka Streams related tests. org.apache.kafka.streams.tests.RelationalSmokeTestTest.verifySmokeTestLogic fails because of a file issue thrown by the JVM. I ran org.apache.kafka.streams.integration.KTableKTableForeignKeyInnerJoinCustomPartitionerIntegrationTest.shouldThrowIllegalArgumentExceptionWhenCustomPartitionerReturnsMultiplePartitions() locally but could not reproduce the issue after multiple runs.

cadonna · 2023-06-29T12:14:47Z

Thanks @wcarlson5 !

wcarlson5 added 26 commits June 10, 2023 21:42

KAFKA-14936: Add On Disk Time Ordered Buffer

0d70af9

update min time correctly

dcf426d

Address most comments

1814684

Address most recent comments version two

16e87d0

update try

58c7058

checkstyle

78c20d1

Address Bruno's comments

64402e5

Remove change

d167a7a

checkstyle

18495fe

checkstyle

952231b

checkstyle

7dfe721

checkstyle

f7108eb

comments

65a873f

Abstract away the Segments

7472f8f

checkstyle

b63bb91

add more checks to the tests

a4fde3c

Add Grace period logic to Stream Table Join

e91e515

checkstyle

07ae99c

checkstyle

99328f7

Address comments

6e6d2db

add grace period table retention check

2ee5c59

remove change records

8c8a161

remove change records

e98d360

remove extra description

572c491

remove extra description

3f8e214

adress comments, need to add more tests

5ba7fac

wcarlson5 changed the title ~~Grace period added~~ KAFKA-14936: Add Grace period logic to Stream Table Join (2/N) Jun 14, 2023

wcarlson5 and others added 3 commits June 14, 2023 15:31

Merge branch 'trunk' into GracePeriodAdded

4be7fb4

Remove change

494a476

quick update

93baf4b

cadonna reviewed Jun 15, 2023

View reviewed changes

divijvaidya added the streams label Jun 15, 2023

wcarlson5 added the kip Requires or implements a KIP label Jun 15, 2023

wcarlson5 commented Jun 16, 2023

View reviewed changes

wcarlson5 added 6 commits June 16, 2023 16:11

address comments

959b797

address comments

1ee826a

first test fixed

41b3c6d

integration test fixed

e5322f1

update unit tests

7bc9408

update unit tests

b05e350

wcarlson5 requested a review from cadonna June 20, 2023 02:54

wcarlson5 commented Jun 20, 2023

View reviewed changes

.../test/java/org/apache/kafka/streams/integration/StreamTableJoinWithGraceIntegrationTest.java Show resolved Hide resolved

cadonna reviewed Jun 22, 2023

View reviewed changes

address round 2 comments

3e5b486

wcarlson5 commented Jun 22, 2023

View reviewed changes

removed extra import

2f7a006

vcrfxia reviewed Jun 23, 2023

View reviewed changes

cadonna reviewed Jun 27, 2023

View reviewed changes

wcarlson5 commented Jun 28, 2023

View reviewed changes

Comments

48e9d02

cadonna approved these changes Jun 29, 2023

View reviewed changes

cadonna merged commit 12be344 into apache:trunk Jun 29, 2023
1 check failed


		Optional<TimeOrderedKeyValueBuffer<K, V, V>> buffer = Optional.empty();

		if (joined.gracePeriod() != null) {

	if (joined.gracePeriod() != null) {
	if (joined.gracePeriod() != null && joined.gracePeriod() > 0) {

		processor.checkAndClearProcessResult(new KeyValueTimestamp<>(0, "X0+Y0", 0),
		new KeyValueTimestamp<>(1, "X1+Y1", 1));

	final ProcessorRecordContext prevRecordContext = internalProcessorContext.recordContext();
	internalProcessorContext.setRecordContext(toEmit.recordContext());
	try {
	internalProcessorContext.forward(toEmit.record()
	.withTimestamp(toEmit.recordContext().timestamp())
	.withHeaders(toEmit.recordContext().headers()));
	suppressionEmitSensor.record(1.0d, internalProcessorContext.currentSystemTimeMs());
	} finally {
	internalProcessorContext.setRecordContext(prevRecordContext);

KAFKA-14936: Add Grace period logic to Stream Table Join (2/N) #13855

KAFKA-14936: Add Grace period logic to Stream Table Join (2/N) #13855

Conversation

wcarlson5 commented Jun 14, 2023 • edited

Committer Checklist (excluded from commit message)

cadonna left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

cadonna Jun 20, 2023 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

wcarlson5 Jun 16, 2023 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

wcarlson5 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

wcarlson5 Jun 16, 2023 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

wcarlson5 commented Jun 20, 2023

wcarlson5 left a comment

Choose a reason for hiding this comment

cadonna left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

wcarlson5 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

wcarlson5 commented Jun 22, 2023 • edited

vcrfxia left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

vcrfxia Jun 23, 2023 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

cadonna left a comment • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

wcarlson5 commented Jun 14, 2023 •

edited

cadonna Jun 20, 2023 •

edited

wcarlson5 Jun 16, 2023 •

edited

wcarlson5 Jun 16, 2023 •

edited

wcarlson5 commented Jun 22, 2023 •

edited

vcrfxia Jun 23, 2023 •

edited

cadonna left a comment •

edited