You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "hanyuzheng7 (via GitHub)" <gi...@apache.org> on 2023/10/10 21:30:54 UTC
[PR] Update test and add test cases in IQv2StoreIntegrationTest [kafka]
hanyuzheng7 opened a new pull request, #14523:
URL: https://github.com/apache/kafka/pull/14523
Update test and add test cases in IQv2StoreIntegrationTest
Change the input key-value pair timestamp and add more test cases.
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15569: test and add test cases in IQv2StoreIntegrationTest [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #14523:
URL: https://github.com/apache/kafka/pull/14523#discussion_r1357200716
##########
streams/src/test/java/org/apache/kafka/streams/integration/IQv2StoreIntegrationTest.java:
##########
@@ -880,25 +888,113 @@ private <T> void shouldHandleWindowKeyQueries(final Function<T, Integer> extract
extractor,
mkSet()
);
+
+ shouldHandleWindowKeyQuery(
+ 0,
+ Instant.ofEpochMilli(WINDOW_START),
+ Instant.ofEpochMilli(WINDOW_START + Duration.ofMinutes(5).toMillis()),
+ extractor,
+ mkSet(0)
+ );
+
+ shouldHandleWindowKeyQuery(
+ 1,
+ Instant.ofEpochMilli(WINDOW_START + Duration.ofMinutes(4).toMillis()),
+ Instant.ofEpochMilli(WINDOW_START + Duration.ofMinutes(11).toMillis()),
+ extractor,
+ mkSet(1)
+ );
+
+ shouldHandleWindowKeyQuery(
+ 2,
+ Instant.ofEpochMilli(WINDOW_START + Duration.ofMinutes(5).toMillis()),
+ Instant.ofEpochMilli(WINDOW_START + Duration.ofMinutes(12).toMillis()),
+ extractor,
+ mkSet(2)
+ );
+
+ shouldHandleWindowKeyQuery(
+ 3,
+ Instant.ofEpochMilli(WINDOW_START + Duration.ofMinutes(13).toMillis()),
+ Instant.ofEpochMilli(WINDOW_START + Duration.ofMinutes(20).toMillis()),
+ extractor,
+ mkSet(3)
+ );
}
private <T> void shouldHandleWindowRangeQueries(final Function<T, Integer> extractor) {
final long windowSize = WINDOW_SIZE.toMillis();
final long windowStart = (RECORD_TIME / windowSize) * windowSize;
+ // miss the window start
+ shouldHandleWindowRangeQuery(
+ Instant.ofEpochMilli(windowStart - 1),
+ Instant.ofEpochMilli(windowStart - 1),
+ extractor,
+ mkSet()
+ );
+
+ shouldHandleWindowRangeQuery(
+ Instant.ofEpochMilli(windowStart),
+ Instant.ofEpochMilli(windowStart),
+ extractor,
+ mkSet(0)
+ );
+
shouldHandleWindowRangeQuery(
Instant.ofEpochMilli(windowStart),
- Instant.ofEpochMilli(windowStart),
+ Instant.ofEpochMilli(windowStart + Duration.ofMinutes(5).toMillis()),
extractor,
- mkSet(0, 1, 2, 3)
+ mkSet(0, 1)
);
- // miss the window start
shouldHandleWindowRangeQuery(
Review Comment:
Sweet -- yes, that is what I was getting at (cf my other comment above).
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15569: test and add test cases in IQv2StoreIntegrationTest [kafka]
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
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 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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15569: test and add test cases in IQv2StoreIntegrationTest [kafka]
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
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 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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15569: test and add test cases in IQv2StoreIntegrationTest [kafka]
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
hanyuzheng7 commented on code in PR #14523:
URL: https://github.com/apache/kafka/pull/14523#discussion_r1356043842
##########
streams/src/test/java/org/apache/kafka/streams/integration/IQv2StoreIntegrationTest.java:
##########
@@ -413,7 +413,7 @@ public static void before()
new ProducerRecord<>(
INPUT_TOPIC_NAME,
i % partitions,
- RECORD_TIME,
+ WINDOW_START + Duration.ofMinutes(6).toMillis() * i,
Review Comment:
This time the first record time is same as WINDOW_START
##########
streams/src/test/java/org/apache/kafka/streams/integration/IQv2StoreIntegrationTest.java:
##########
@@ -413,7 +413,7 @@ public static void before()
new ProducerRecord<>(
INPUT_TOPIC_NAME,
i % partitions,
- RECORD_TIME,
+ WINDOW_START + Duration.ofMinutes(6).toMillis() * i,
Review Comment:
This time the first record time is same as WINDOW_START.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15569: test and add test cases in IQv2StoreIntegrationTest [kafka]
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
hanyuzheng7 commented on code in PR #14523:
URL: https://github.com/apache/kafka/pull/14523#discussion_r1356042346
##########
streams/src/test/java/org/apache/kafka/streams/integration/IQv2StoreIntegrationTest.java:
##########
@@ -880,25 +888,113 @@ private <T> void shouldHandleWindowKeyQueries(final Function<T, Integer> extract
extractor,
mkSet()
);
+
+ shouldHandleWindowKeyQuery(
+ 0,
+ Instant.ofEpochMilli(WINDOW_START),
+ Instant.ofEpochMilli(WINDOW_START + Duration.ofMinutes(5).toMillis()),
+ extractor,
+ mkSet(0)
+ );
+
+ shouldHandleWindowKeyQuery(
+ 1,
+ Instant.ofEpochMilli(WINDOW_START + Duration.ofMinutes(4).toMillis()),
+ Instant.ofEpochMilli(WINDOW_START + Duration.ofMinutes(11).toMillis()),
+ extractor,
+ mkSet(1)
+ );
+
+ shouldHandleWindowKeyQuery(
+ 2,
+ Instant.ofEpochMilli(WINDOW_START + Duration.ofMinutes(5).toMillis()),
+ Instant.ofEpochMilli(WINDOW_START + Duration.ofMinutes(12).toMillis()),
+ extractor,
+ mkSet(2)
+ );
+
+ shouldHandleWindowKeyQuery(
+ 3,
+ Instant.ofEpochMilli(WINDOW_START + Duration.ofMinutes(13).toMillis()),
+ Instant.ofEpochMilli(WINDOW_START + Duration.ofMinutes(20).toMillis()),
+ extractor,
+ mkSet(3)
+ );
}
private <T> void shouldHandleWindowRangeQueries(final Function<T, Integer> extractor) {
final long windowSize = WINDOW_SIZE.toMillis();
final long windowStart = (RECORD_TIME / windowSize) * windowSize;
+ // miss the window start
+ shouldHandleWindowRangeQuery(
+ Instant.ofEpochMilli(windowStart - 1),
+ Instant.ofEpochMilli(windowStart - 1),
+ extractor,
+ mkSet()
+ );
+
+ shouldHandleWindowRangeQuery(
+ Instant.ofEpochMilli(windowStart),
+ Instant.ofEpochMilli(windowStart),
+ extractor,
+ mkSet(0)
+ );
+
shouldHandleWindowRangeQuery(
Instant.ofEpochMilli(windowStart),
- Instant.ofEpochMilli(windowStart),
+ Instant.ofEpochMilli(windowStart + Duration.ofMinutes(5).toMillis()),
extractor,
- mkSet(0, 1, 2, 3)
+ mkSet(0, 1)
);
- // miss the window start
shouldHandleWindowRangeQuery(
Review Comment:
https://github.com/apache/kafka/pull/14523#discussion_r1356041608
I might consider choosing the query bounds more methodically. Additionally, I'll annotate my code with comments to clarify the rationale behind my boundary selections.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15569: test and add test cases in IQv2StoreIntegrationTest [kafka]
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
hanyuzheng7 commented on code in PR #14523:
URL: https://github.com/apache/kafka/pull/14523#discussion_r1356034620
##########
streams/src/test/java/org/apache/kafka/streams/integration/IQv2StoreIntegrationTest.java:
##########
@@ -863,6 +863,14 @@ private <T> void shouldHandleWindowKeyQueries(final Function<T, Integer> extract
mkSet()
);
+ shouldHandleWindowKeyQuery(
+ 2,
Review Comment:
ok, I will fix it. This isn't the new test case I introduced; it's the previous one. However, the outcome has now changed.
The previous like this
```
// tightest possible start range
shouldHandleWindowKeyQuery(
2,
Instant.ofEpochMilli(WINDOW_START),
Instant.ofEpochMilli(WINDOW_START),
extractor,
mkSet(2)
);
```
this time the answer is mkSet()
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15569: test and add test cases in IQv2StoreIntegrationTest [kafka]
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
hanyuzheng7 commented on code in PR #14523:
URL: https://github.com/apache/kafka/pull/14523#discussion_r1357233277
##########
streams/src/test/java/org/apache/kafka/streams/integration/IQv2StoreIntegrationTest.java:
##########
@@ -863,6 +863,14 @@ private <T> void shouldHandleWindowKeyQueries(final Function<T, Integer> extract
mkSet()
);
+ shouldHandleWindowKeyQuery(
+ 2,
Review Comment:
```
// tightest possible start range
shouldHandleWindowKeyQuery(
2,
Instant.ofEpochMilli(WINDOW_START),
Instant.ofEpochMilli(WINDOW_START),
extractor,
mkSet()
);
```
this test comment maybe change to cannot find key 2 at first window
##########
streams/src/test/java/org/apache/kafka/streams/integration/IQv2StoreIntegrationTest.java:
##########
@@ -863,6 +863,14 @@ private <T> void shouldHandleWindowKeyQueries(final Function<T, Integer> extract
mkSet()
);
+ shouldHandleWindowKeyQuery(
+ 2,
Review Comment:
```
// tightest possible start range
shouldHandleWindowKeyQuery(
2,
Instant.ofEpochMilli(WINDOW_START),
Instant.ofEpochMilli(WINDOW_START),
extractor,
mkSet()
);
```
this test comment maybe change to cannot find key 2 at first 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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15569: test and add test cases in IQv2StoreIntegrationTest [kafka]
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
hanyuzheng7 commented on code in PR #14523:
URL: https://github.com/apache/kafka/pull/14523#discussion_r1356034620
##########
streams/src/test/java/org/apache/kafka/streams/integration/IQv2StoreIntegrationTest.java:
##########
@@ -863,6 +863,14 @@ private <T> void shouldHandleWindowKeyQueries(final Function<T, Integer> extract
mkSet()
);
+ shouldHandleWindowKeyQuery(
+ 2,
Review Comment:
ok, I will fix it. This isn't the new test case I introduced; it's the previous one. However, the outcome has now changed.
The previous like this
```
// tightest possible start range
shouldHandleWindowKeyQuery(
2,
Instant.ofEpochMilli(WINDOW_START),
Instant.ofEpochMilli(WINDOW_START),
extractor,
mkSet(2)
);
```
this time the answer is mkSet()
I should I add comment in this one like
```
// tightest possible start range
shouldHandleWindowKeyQuery(
2,
Instant.ofEpochMilli(WINDOW_START),
Instant.ofEpochMilli(WINDOW_START),
extractor,
mkSet()
);
```
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15569: test and add test cases in IQv2StoreIntegrationTest [kafka]
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
hanyuzheng7 commented on code in PR #14523:
URL: https://github.com/apache/kafka/pull/14523#discussion_r1356034620
##########
streams/src/test/java/org/apache/kafka/streams/integration/IQv2StoreIntegrationTest.java:
##########
@@ -863,6 +863,14 @@ private <T> void shouldHandleWindowKeyQueries(final Function<T, Integer> extract
mkSet()
);
+ shouldHandleWindowKeyQuery(
+ 2,
Review Comment:
ok, I will fix it. This isn't the new test case I introduced; it's the previous one. However, the outcome has now changed.
The previous like this
```
// tightest possible start range
shouldHandleWindowKeyQuery(
2,
Instant.ofEpochMilli(WINDOW_START),
Instant.ofEpochMilli(WINDOW_START),
extractor,
mkSet(2)
);
```
this time the answer is mkSet()
I should I add comment in this one like // tightest possible start range
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15569: test and add test cases in IQv2StoreIntegrationTest [kafka]
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
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 their own session windows.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15569: test and add test cases in IQv2StoreIntegrationTest [kafka]
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
hanyuzheng7 commented on PR #14523:
URL: https://github.com/apache/kafka/pull/14523#issuecomment-1758892421
> Thanks for extending this test. This is very helpful.
>
> I am wondering, if it would be beneficial to add more input record, such that a few windows have more than one record in them?
Yes, we can add more records and let them in same 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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15569: test and add test cases in IQv2StoreIntegrationTest [kafka]
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15569: test and add test cases in IQv2StoreIntegrationTest [kafka]
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
hanyuzheng7 commented on code in PR #14523:
URL: https://github.com/apache/kafka/pull/14523#discussion_r1356034620
##########
streams/src/test/java/org/apache/kafka/streams/integration/IQv2StoreIntegrationTest.java:
##########
@@ -863,6 +863,14 @@ private <T> void shouldHandleWindowKeyQueries(final Function<T, Integer> extract
mkSet()
);
+ shouldHandleWindowKeyQuery(
+ 2,
Review Comment:
ok, I will fix it. This isn't the new test case I introduced; it's the previous one. However, the outcome has now changed.
The previous like this
```
// tightest possible start range
shouldHandleWindowKeyQuery(
2,
Instant.ofEpochMilli(WINDOW_START),
Instant.ofEpochMilli(WINDOW_START),
extractor,
mkSet(2)
);
```
this time the answer is mkSet()
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15569: test and add test cases in IQv2StoreIntegrationTest [kafka]
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
hanyuzheng7 commented on code in PR #14523:
URL: https://github.com/apache/kafka/pull/14523#discussion_r1360875255
##########
streams/src/test/java/org/apache/kafka/streams/integration/IQv2StoreIntegrationTest.java:
##########
@@ -1071,16 +1568,51 @@ public <V> void shouldHandleKeyQuery(
assertThrows(IllegalArgumentException.class, queryResult::getFailureReason);
assertThrows(
- IllegalArgumentException.class,
- queryResult::getFailureMessage
+ IllegalArgumentException.class,
+ queryResult::getFailureMessage
);
final V result1 = queryResult.getResult();
final Integer integer = valueExtactor.apply(result1);
assertThat(integer, is(expectedValue));
assertThat(queryResult.getExecutionInfo(), is(empty()));
assertThat(queryResult.getPosition(), is(POSITION_0));
- }
+ }
+
+ public <V> void shouldHandleKeyPAPIQuery(
Review Comment:
At this time, these two methods are the same. We can use one method to handle both topologies. So we can use shouldHandleKeyQuery one to handle two topologies. I will update it.
##########
streams/src/test/java/org/apache/kafka/streams/integration/IQv2StoreIntegrationTest.java:
##########
@@ -655,7 +657,7 @@ public void process(final Record<Integer, Integer> record) {
ValueAndTimestamp.make(
record.value(), record.timestamp()
Review Comment:
ok
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15569: test and add test cases in IQv2StoreIntegrationTest [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #14523:
URL: https://github.com/apache/kafka/pull/14523#discussion_r1360050468
##########
streams/src/test/java/org/apache/kafka/streams/integration/IQv2StoreIntegrationTest.java:
##########
@@ -716,7 +718,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(), record.timestamp())),
Review Comment:
We should add a comment above this line:
```
// we do not re-implement the actual session-window logic from the DSL here to keep the test simple,
// but instead just put each record into it's own session
```
##########
streams/src/test/java/org/apache/kafka/streams/integration/IQv2StoreIntegrationTest.java:
##########
@@ -1071,16 +1568,51 @@ public <V> void shouldHandleKeyQuery(
assertThrows(IllegalArgumentException.class, queryResult::getFailureReason);
assertThrows(
- IllegalArgumentException.class,
- queryResult::getFailureMessage
+ IllegalArgumentException.class,
+ queryResult::getFailureMessage
);
final V result1 = queryResult.getResult();
final Integer integer = valueExtactor.apply(result1);
assertThat(integer, is(expectedValue));
assertThat(queryResult.getExecutionInfo(), is(empty()));
assertThat(queryResult.getPosition(), is(POSITION_0));
- }
+ }
+
+ public <V> void shouldHandleKeyPAPIQuery(
Review Comment:
Maybe I missed it, but how is this method different to `shouldHandleKeyDSLQuery` ? -- Do we really need to versions of exiting `shouldHandleKeyQuery` ?
##########
streams/src/test/java/org/apache/kafka/streams/integration/IQv2StoreIntegrationTest.java:
##########
@@ -671,7 +673,7 @@ public void process(final Record<Integer, Integer> record) {
public void process(final Record<Integer, Integer> record) {
final WindowStore<Integer, Integer> stateStore =
context().getStateStore(windowStoreStoreBuilder.name());
- stateStore.put(record.key(), record.value(), WINDOW_START);
+ stateStore.put(record.key(), record.value(), (record.timestamp() / WINDOW_SIZE.toMillis()) * WINDOW_SIZE.toMillis());
Review Comment:
We should add the same comment as above.
##########
streams/src/test/java/org/apache/kafka/streams/integration/IQv2StoreIntegrationTest.java:
##########
@@ -655,7 +657,7 @@ public void process(final Record<Integer, Integer> record) {
ValueAndTimestamp.make(
record.value(), record.timestamp()
Review Comment:
We should add a comment here:
```
// We don't re-implement the DSL logic (which implements sum) but instead just keep the lasted value per 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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15569: test and add test cases in IQv2StoreIntegrationTest [kafka]
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
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.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15569: test and add test cases in IQv2StoreIntegrationTest [kafka]
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
hanyuzheng7 commented on code in PR #14523:
URL: https://github.com/apache/kafka/pull/14523#discussion_r1356044046
##########
streams/src/test/java/org/apache/kafka/streams/integration/IQv2StoreIntegrationTest.java:
##########
@@ -847,11 +847,11 @@ private <T> void shouldHandleWindowKeyQueries(final Function<T, Integer> extract
// tightest possible start range
shouldHandleWindowKeyQuery(
- 2,
+ 0,
Instant.ofEpochMilli(WINDOW_START),
Instant.ofEpochMilli(WINDOW_START),
extractor,
- mkSet(2)
+ mkSet(0)
);
// miss the window start range
Review Comment:
Yes, the value should be 0. In this case, <0,0> is aligned with the WINDOW_START, making it retrievable.
[[And I set](https://github.com/apache/kafka/pull/14523/files#r1356043842)](https://github.com/apache/kafka/pull/14523/files#r1356043842)
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15569: test and add test cases in IQv2StoreIntegrationTest [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax merged PR #14523:
URL: https://github.com/apache/kafka/pull/14523
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15569: test and add test cases in IQv2StoreIntegrationTest [kafka]
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
hanyuzheng7 commented on code in PR #14523:
URL: https://github.com/apache/kafka/pull/14523#discussion_r1356034620
##########
streams/src/test/java/org/apache/kafka/streams/integration/IQv2StoreIntegrationTest.java:
##########
@@ -863,6 +863,14 @@ private <T> void shouldHandleWindowKeyQueries(final Function<T, Integer> extract
mkSet()
);
+ shouldHandleWindowKeyQuery(
+ 2,
Review Comment:
ok, I will fix it.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15569: test and add test cases in IQv2StoreIntegrationTest [kafka]
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
hanyuzheng7 commented on code in PR #14523:
URL: https://github.com/apache/kafka/pull/14523#discussion_r1356041608
##########
streams/src/test/java/org/apache/kafka/streams/integration/IQv2StoreIntegrationTest.java:
##########
@@ -880,25 +888,113 @@ private <T> void shouldHandleWindowKeyQueries(final Function<T, Integer> extract
extractor,
mkSet()
);
+
+ shouldHandleWindowKeyQuery(
Review Comment:
While adding tests, I observed an intriguing scenario. If we have three windows: 0-5, 5-10, and 10-15, any effective query must encompass the window's start time. For instance, with a key-value pair at time 6, the query range must include 5. We could have ranges like query(5, 6) or query(5,7). If the query's lower bound exceeds 5, we won't retrieve the value for that key-value pair.
When selecting a query range, I ensure the lower bound is less than or equal to the window's start time. For instance, if I want to retrieve the data at timestamp 6 (represented as <1,1>), I would set the lower bound to a value less than or equal to 5. The upper bound, on the other hand, is selected randomly from values greater than or equal to 5.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15569: test and add test cases in IQv2StoreIntegrationTest [kafka]
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
hanyuzheng7 commented on code in PR #14523:
URL: https://github.com/apache/kafka/pull/14523#discussion_r1356044046
##########
streams/src/test/java/org/apache/kafka/streams/integration/IQv2StoreIntegrationTest.java:
##########
@@ -847,11 +847,11 @@ private <T> void shouldHandleWindowKeyQueries(final Function<T, Integer> extract
// tightest possible start range
shouldHandleWindowKeyQuery(
- 2,
+ 0,
Instant.ofEpochMilli(WINDOW_START),
Instant.ofEpochMilli(WINDOW_START),
extractor,
- mkSet(2)
+ mkSet(0)
);
// miss the window start range
Review Comment:
Yes, the value should be 0. In this case, <0,0> is aligned with the WINDOW_START, making it retrievable.
https://github.com/apache/kafka/pull/14523/files#r1356043842
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15569: test and add test cases in IQv2StoreIntegrationTest [kafka]
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
hanyuzheng7 commented on code in PR #14523:
URL: https://github.com/apache/kafka/pull/14523#discussion_r1356041608
##########
streams/src/test/java/org/apache/kafka/streams/integration/IQv2StoreIntegrationTest.java:
##########
@@ -880,25 +888,113 @@ private <T> void shouldHandleWindowKeyQueries(final Function<T, Integer> extract
extractor,
mkSet()
);
+
+ shouldHandleWindowKeyQuery(
Review Comment:
If we have three windows: 0-5, 5-10, and 10-15, any effective query must encompass the window's start time. For instance, with a key-value pair at time 6, the query range must include 5. We could have ranges like query(5, 6) or query(5,7). If the query's lower bound exceeds 5, we won't retrieve the value for that key-value pair.
When selecting a query range, I ensure the lower bound is less than or equal to the window's start time. For instance, if I want to retrieve the data at timestamp 6 (represented as <1,1>), I would set the lower bound to a value less than or equal to 5. The upper bound, on the other hand, is selected randomly from values greater than or equal to 5.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15569: test and add test cases in IQv2StoreIntegrationTest [kafka]
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
hanyuzheng7 commented on code in PR #14523:
URL: https://github.com/apache/kafka/pull/14523#discussion_r1360875657
##########
streams/src/test/java/org/apache/kafka/streams/integration/IQv2StoreIntegrationTest.java:
##########
@@ -716,7 +718,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(), record.timestamp())),
Review Comment:
ok
##########
streams/src/test/java/org/apache/kafka/streams/integration/IQv2StoreIntegrationTest.java:
##########
@@ -671,7 +673,7 @@ public void process(final Record<Integer, Integer> record) {
public void process(final Record<Integer, Integer> record) {
final WindowStore<Integer, Integer> stateStore =
context().getStateStore(windowStoreStoreBuilder.name());
- stateStore.put(record.key(), record.value(), WINDOW_START);
+ stateStore.put(record.key(), record.value(), (record.timestamp() / WINDOW_SIZE.toMillis()) * WINDOW_SIZE.toMillis());
Review Comment:
ok
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15569: test and add test cases in IQv2StoreIntegrationTest [kafka]
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
hanyuzheng7 commented on PR #14523:
URL: https://github.com/apache/kafka/pull/14523#issuecomment-1765551586
@mjsax Thank you for reviewing the code.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15569: test and add test cases in IQv2StoreIntegrationTest [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #14523:
URL: https://github.com/apache/kafka/pull/14523#discussion_r1357188427
##########
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:
I think it should be `new SessionWindow(record.timestamp(), record.timestamp())`
> but their session window size seem to become 0
Yes, the size would become `1` (note, for sessions window, both upper and lower bound are inclusive in contrast to Tumbling/Hopping window). For session windows, there is no fixed size, but the "inactivity gap" parameter (the test only re-uses `WINDOW_SIZE` variable as "gap" -- this might be a little bit confusing). The size of the created windows depend on the data -- in the current test setup, records are apart from each other more than the "gap" and thus each record creates it's own session, and thus each session has size `1` -- only if two (or more) records fall into the same session (ie, their timestamp are close enough to each other, ie, `<= gap` the session size would become larger).
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15569: test and add test cases in IQv2StoreIntegrationTest [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #14523:
URL: https://github.com/apache/kafka/pull/14523#discussion_r1357193113
##########
streams/src/test/java/org/apache/kafka/streams/integration/IQv2StoreIntegrationTest.java:
##########
@@ -863,6 +863,14 @@ private <T> void shouldHandleWindowKeyQueries(final Function<T, Integer> extract
mkSet()
);
+ shouldHandleWindowKeyQuery(
+ 2,
Review Comment:
Well, yes and no. The original test was updated to use key 0 now -- the goal if the test is to verify `miss the window start range` -- This test with key 2 does not test "miss the window start range" -- so what's the purpose of it?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15569: test and add test cases in IQv2StoreIntegrationTest [kafka]
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15569: test and add test cases in IQv2StoreIntegrationTest [kafka]
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
hanyuzheng7 commented on code in PR #14523:
URL: https://github.com/apache/kafka/pull/14523#discussion_r1356041608
##########
streams/src/test/java/org/apache/kafka/streams/integration/IQv2StoreIntegrationTest.java:
##########
@@ -880,25 +888,113 @@ private <T> void shouldHandleWindowKeyQueries(final Function<T, Integer> extract
extractor,
mkSet()
);
+
+ shouldHandleWindowKeyQuery(
Review Comment:
When selecting a query range, I ensure the lower bound is less than or equal to the window's start time. For instance, if I want to retrieve the data at timestamp 6 (represented as <1,1>), I would set the lower bound to a value less than or equal to 5. The upper bound, on the other hand, is selected randomly from values greater than or equal to 5.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15569: test and add test cases in IQv2StoreIntegrationTest [kafka]
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
hanyuzheng7 commented on code in PR #14523:
URL: https://github.com/apache/kafka/pull/14523#discussion_r1356043842
##########
streams/src/test/java/org/apache/kafka/streams/integration/IQv2StoreIntegrationTest.java:
##########
@@ -413,7 +413,7 @@ public static void before()
new ProducerRecord<>(
INPUT_TOPIC_NAME,
i % partitions,
- RECORD_TIME,
+ WINDOW_START + Duration.ofMinutes(6).toMillis() * i,
Review Comment:
Now, the first record time is same as WINDOW_START.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15569: test and add test cases in IQv2StoreIntegrationTest [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #14523:
URL: https://github.com/apache/kafka/pull/14523#discussion_r1355991839
##########
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:
We use a session window gap of `WINDOW_SIZE` -- this means that every record falls into it's own session, and `session-start = session-end = record.ts`.
##########
streams/src/test/java/org/apache/kafka/streams/integration/IQv2StoreIntegrationTest.java:
##########
@@ -880,25 +888,113 @@ private <T> void shouldHandleWindowKeyQueries(final Function<T, Integer> extract
extractor,
mkSet()
);
+
+ shouldHandleWindowKeyQuery(
+ 0,
+ Instant.ofEpochMilli(WINDOW_START),
+ Instant.ofEpochMilli(WINDOW_START + Duration.ofMinutes(5).toMillis()),
+ extractor,
+ mkSet(0)
+ );
+
+ shouldHandleWindowKeyQuery(
+ 1,
+ Instant.ofEpochMilli(WINDOW_START + Duration.ofMinutes(4).toMillis()),
+ Instant.ofEpochMilli(WINDOW_START + Duration.ofMinutes(11).toMillis()),
+ extractor,
+ mkSet(1)
+ );
+
+ shouldHandleWindowKeyQuery(
+ 2,
+ Instant.ofEpochMilli(WINDOW_START + Duration.ofMinutes(5).toMillis()),
+ Instant.ofEpochMilli(WINDOW_START + Duration.ofMinutes(12).toMillis()),
+ extractor,
+ mkSet(2)
+ );
+
+ shouldHandleWindowKeyQuery(
+ 3,
+ Instant.ofEpochMilli(WINDOW_START + Duration.ofMinutes(13).toMillis()),
+ Instant.ofEpochMilli(WINDOW_START + Duration.ofMinutes(20).toMillis()),
+ extractor,
+ mkSet(3)
+ );
}
private <T> void shouldHandleWindowRangeQueries(final Function<T, Integer> extractor) {
final long windowSize = WINDOW_SIZE.toMillis();
final long windowStart = (RECORD_TIME / windowSize) * windowSize;
+ // miss the window start
+ shouldHandleWindowRangeQuery(
+ Instant.ofEpochMilli(windowStart - 1),
+ Instant.ofEpochMilli(windowStart - 1),
+ extractor,
+ mkSet()
+ );
+
+ shouldHandleWindowRangeQuery(
+ Instant.ofEpochMilli(windowStart),
+ Instant.ofEpochMilli(windowStart),
+ extractor,
+ mkSet(0)
+ );
+
shouldHandleWindowRangeQuery(
Instant.ofEpochMilli(windowStart),
- Instant.ofEpochMilli(windowStart),
+ Instant.ofEpochMilli(windowStart + Duration.ofMinutes(5).toMillis()),
extractor,
- mkSet(0, 1, 2, 3)
+ mkSet(0, 1)
);
- // miss the window start
shouldHandleWindowRangeQuery(
Review Comment:
Similar to above -- it's not obvious why you picked the time bounds for the queries. Can you explain?
##########
streams/src/test/java/org/apache/kafka/streams/integration/IQv2StoreIntegrationTest.java:
##########
@@ -847,11 +847,11 @@ private <T> void shouldHandleWindowKeyQueries(final Function<T, Integer> extract
// tightest possible start range
shouldHandleWindowKeyQuery(
- 2,
+ 0,
Instant.ofEpochMilli(WINDOW_START),
Instant.ofEpochMilli(WINDOW_START),
extractor,
- mkSet(2)
+ mkSet(0)
);
// miss the window start range
Review Comment:
The key below should be updated to `0`? Otherwise we don't miss the window of the key be 1 ms as intended by the test?
##########
streams/src/test/java/org/apache/kafka/streams/integration/IQv2StoreIntegrationTest.java:
##########
@@ -880,25 +888,113 @@ private <T> void shouldHandleWindowKeyQueries(final Function<T, Integer> extract
extractor,
mkSet()
);
+
+ shouldHandleWindowKeyQuery(
Review Comment:
For this and the other added queries. How do you pick the query search bounds? It's not obvious to me.
##########
streams/src/test/java/org/apache/kafka/streams/integration/IQv2StoreIntegrationTest.java:
##########
@@ -863,6 +863,14 @@ private <T> void shouldHandleWindowKeyQueries(final Function<T, Integer> extract
mkSet()
);
+ shouldHandleWindowKeyQuery(
+ 2,
Review Comment:
nit: fix indention -- should only be 4 spaces (similar below)
What is the indent of adding this test case?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15569: test and add test cases in IQv2StoreIntegrationTest [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #14523:
URL: https://github.com/apache/kafka/pull/14523#discussion_r1357200115
##########
streams/src/test/java/org/apache/kafka/streams/integration/IQv2StoreIntegrationTest.java:
##########
@@ -880,25 +888,113 @@ private <T> void shouldHandleWindowKeyQueries(final Function<T, Integer> extract
extractor,
mkSet()
);
+
+ shouldHandleWindowKeyQuery(
Review Comment:
Thanks. Maybe add some (short) comment about it similar to the existing test cases.
The "confusing" thing is, that the bounds are not "strict": for example you use range 4 to 11, what gives the same result at 5 to 11, or 4 to 10.... (so why is 4,11 used but not some other bound? if it's random, also ok, but should be clarified). For testing, the interesting part is usually when "bounds" are crossed:
For example, we could have a query range 5,9 that only returns the window starting at 5, plus range 5,10 that returns two windows (thus we test the upper bound from 9 and 10), plus range 6,10 that would only return the window at 10 (and thus verify the lower bound from 5 to 6). Does this make sense?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org