feat: call KStreams API grace() when GRACE PERIOD is used in joins

[spena]

Description

Follow up on #7662. This PR makes the connection from the GRACE PERIOD syntax in stream-stream joins and the call to the KStreams grace API, which enables the use of grace in join windows.

This is the workflow that happens internally to help review the code:

---

When a join happens. The parsing creates a WithinExpression which has the windows and optional grace (updated in #7662).

This window + optional grace are obtained from the JoinNode and passed to the SchemaKStream class, which accepts
an optional grace. The SchemaKStream then passes the window + optional grace to the ExecutionStepFactory which
gets the before/after and grace values in milliseconds, and then passes them to the StreamStreamJoin.

The StreamStreamJoin accepts graceMillis as optional because seems it is needed for compatibility with an old plan written
in JSON?. Anyway, the grace must be optional to continue with the call to the KSPlanBuilder.visitStreamStreamJoin(),
which later ends up calling the StreamStreamJoinBuilder.build().

The StreamStreamJoinBuilder is the final class that builds the windows + grace and calls the Kafka streams API. The
grace is optional in this class in order to know whether to call the new KStreams API ofSizeAndGrace() or just call
the normal of/after calls.

The ofSizeAndGrace() is called is not done in this PR. A follow-up PR will be done after the Kafka PR that contains this new method is merged.

---

Testing done

Describe the testing strategy. Unit and integration tests are expected for any behavior changes.
Updated unit tests
Add QTT tests

Reviewer checklist

• Ensure docs are updated if necessary. (eg. if a user visible feature is being added or changed).
• Ensure relevant issues are linked (description should include text like "Fixes #")

[guozhangwang]

Thank you so much for the great PR description!

I'm half way there and would try to complete by eow.

[guozhangwang]

I have a major question about the added test: with the fix in place, the output seems still spurious? Otherwise, just minor questions.

[guozhangwang]

Why can't we just retrieve the grace period from the JoinWindows? I.e. we just set the grace time in JoinWindows if GRACE specified, and then pass around that object only.

[guozhangwang]

EDIT: I knew this is only used for WITH and the GRACE PERIOD is out of the scope; I guess my meta question is: why can't we have the GRACE PERIOD captured as optional in WithinExpression as well?

[spena]

The JoinWindows has a default grace (24h) if it is not passed one. I cannot know if the grace was optional in the ksql syntax with that default. Say a user set GRACE PERIOD 24 HOURS, then JoinWindows will be set to 24h, which it could be the default (no fix enabled) or manually set (fix enabled).

The grace is optional in the WithinExpression. I could pass the WithinExpression all down to the StreamStreamJoinBuilder but I would need to add the ksqldb-parser dependency into the ksqldb-streams module.

[guozhangwang]

Let's say in the future if we do not need to rely on GRACE to enable the fix or not, then would this code be changed again to only pass in the JoinWindows object still?

If yes, I'd suggest we just keep doing that right away: we could, e.g., a wild idea is to have an extended JoinWindows that adds a flag indicating if user specified grace, and then pass around that object from WithinExpression -> ExecutionStepFactory -> StreamStreamJoin, and let it to return a graceSpecified rather than join.getGraceMillis(), used by the StreamStreamJoinBuilder. In the future we can just remove the extended class and the graceSpecified from StreamStreamJoin. Does that work?

[spena]

I could do that. But it would only be used to pass the window+graceSpecified from SchemaKStream#leftJoin() -> ExecutionStepFactory#streamStreamJoin(). After that, the rest should be passed as millis to the StreamStreamJoin because of compatibility. Look at the StreamStreamJoin class. It has a constructor that is used by a JSON mapper when reads a plan. That already accepts before/after millis. I cannot change that constructor to pass the JoinWindows.

I'm not sure if it is worth creating a class around the JoinWindows for one single change and then remove it. I think we can remove the gracePeriod later easier than removing a class.

[guozhangwang]

Okay, I think I'm convinced. Let's just keep it as is then.

[guozhangwang]

Hmm.. is the output expected? With our fix shouldn't we only output one record for key 2, with C-c, and one record for key 3, with D-d?

[spena]

The fix is not enabled yet. These results are working with the spurious results bug.

[guozhangwang]

Ditto above.

[spena]

The fix is not enabled yet. These results are working with the spurious results bug.

[guozhangwang]

I see. You mean that we need to wait until the dependency ccs-kafka has @mjsax 's merged code, and then update this PR here and merge?

[spena]

We need to wait for the fix in kafka. But it should not block this PR 'cause enabling the fix will involve other changes in QTT. A follow-up PR for making changes in the QTT framework is here (which is required to enable the fix and update the tests) - #7695

[guozhangwang]

Sounds good.

[spena]

Thanks @guozhangwang. I just merged the PR.

[mjsax]

This is calling the old/exiting grace() and thus it does not enable this fix. I think we need to change this (starting in L88):

final JoinWindows joinWindows;
if (join.getGraceMillis().isPresent()) {
    joinWindows = JoinsWindow.ofTimeDifferenceAndGrace(
        join.getBeforeMillis(),
        join.getGraceMillis().get()
    ).after(join.getAfterMillis();
} else {
    joinWindows  = JoinWindows.of(join.getBeforeMillis()).after(join.getAfterMillis())
}

[guozhangwang]

I think #7695 would update this.

[spena]

Right. I need #7695 merged before I make the changes for the new ofTimeDifferenceAndGrace.