hanyuzheng7 commented on code in PR #14523:
URL: https://github.com/apache/kafka/pull/14523#discussion_r1356033429
##########
streams/src/test/java/org/apache/kafka/streams/integration/IQv2StoreIntegrationTest.java:
##########
@@ -716,7 +716,7 @@ public void process(final Record<Integer, Integer> record) {
final SessionStore<Integer, Integer> stateStore =
context().getStateStore(sessionStoreStoreBuilder.name());
stateStore.put(
- new Windowed<>(record.key(), new
SessionWindow(WINDOW_START, WINDOW_START)),
+ new Windowed<>(record.key(), new
SessionWindow((record.timestamp() / WINDOW_SIZE.toMillis()) *
WINDOW_SIZE.toMillis(), (record.timestamp() / WINDOW_SIZE.toMillis()) *
WINDOW_SIZE.toMillis() + WINDOW_SIZE.toMillis())),
Review Comment:
So it should be
`new SessionWindow((record.timestamp() / WINDOW_SIZE.toMillis()) *
WINDOW_SIZE.toMillis(), (record.timestamp() / WINDOW_SIZE.toMillis()) *
WINDOW_SIZE.toMillis())`
but their session window size seem to become 0
```
/**
* Create a new window for the given start time and end time (both
inclusive).
*
* @param startMs the start timestamp of the window
* @param endMs the end timestamp of the window
* @throws IllegalArgumentException if {@code startMs} is negative or if
{@code endMs} is smaller than {@code startMs}
*/
public SessionWindow(final long startMs, final long endMs) throws
IllegalArgumentException {
super(startMs, endMs);
}
```
so you mean, each session window size is 0 and their interval is
window_size, and in the end each record just falls in it's own session window.
Ok I got that, it should be like this picture each record stand for each
session.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
