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() + WINDOW_SIZE.toMillis())`?
but they are same the 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 they interval is window_size,
and in the end each record just falls in their session window.
--
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
