You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "mjsax (via GitHub)" <gi...@apache.org> on 2024/11/19 04:46:56 UTC
[PR] MINOR: fix warnings in Kafka Streams state store tests [kafka]
mjsax opened a new pull request, #17855:
URL: https://github.com/apache/kafka/pull/17855
This PR reduced compiler warnings, and WARN logs (by closing iterator expliclity).
--
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] MINOR: fix warnings in Kafka Streams state store tests [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #17855:
URL: https://github.com/apache/kafka/pull/17855#discussion_r1847639466
##########
streams/src/test/java/org/apache/kafka/streams/state/internals/AbstractSessionBytesStoreTest.java:
##########
@@ -747,10 +749,11 @@ public void testIteratorPeek() {
sessionStore.put(new Windowed<>("aa", new SessionWindow(10, 20)), 4L);
try (final KeyValueIterator<Windowed<String>, Long> iterator = sessionStore.findSessions("a", 0L, 20)) {
-
- assertEquals(iterator.peekNextKey(), new Windowed<>("a", new SessionWindow(0L, 0L)));
- assertEquals(iterator.peekNextKey(), iterator.next().key);
- assertEquals(iterator.peekNextKey(), iterator.next().key);
+ assertEquals(new Windowed<>("a", new SessionWindow(0L, 0L)), iterator.peekNextKey());
+ final Windowed<String> k1 = iterator.peekNextKey();
+ assertEquals(iterator.next().key, k1);
+ final Windowed<String> k2 = iterator.peekNextKey();
+ assertEquals(iterator.next().key, k2);
Review Comment:
Swapping parameters, because "expected" should be first, and "actual" should be second. Need to add variable `k1` and `k2` to call `peekNextKey` first to not change test semantics.
(Same below)
--
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] MINOR: fix warnings in Kafka Streams state store tests [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #17855:
URL: https://github.com/apache/kafka/pull/17855#discussion_r1847802852
##########
streams/src/test/java/org/apache/kafka/streams/state/internals/AbstractKeyValueStoreTest.java:
##########
@@ -520,12 +520,13 @@ public void shouldPutReverseAll() {
final List<KeyValue<Integer, String>> allReturned = new ArrayList<>();
final List<KeyValue<Integer, String>> expectedReturned =
Arrays.asList(KeyValue.pair(2, "two"), KeyValue.pair(1, "one"));
- final Iterator<KeyValue<Integer, String>> iterator = store.reverseAll();
- while (iterator.hasNext()) {
- allReturned.add(iterator.next());
+ try (final KeyValueIterator<Integer, String> iterator = store.reverseAll()) {
+ while (iterator.hasNext()) {
Review Comment:
We can reuse `StreamsTestUtils.toList`:
```java
try (final KeyValueIterator<Integer, String> iterator = store.reverseAll()) {
final List<KeyValue<Integer, String>> allReturned = StreamsTestUtils.toList(iterator);
final List<KeyValue<Integer, String>> expectedReturned =
Arrays.asList(KeyValue.pair(2, "two"), KeyValue.pair(1, "one"));
assertThat(allReturned, equalTo(expectedReturned));
}
```
##########
streams/src/test/java/org/apache/kafka/test/StreamsTestUtils.java:
##########
@@ -150,6 +150,17 @@ public static <K, V> Set<V> valuesToSet(final Iterator<KeyValue<K, V>> iterator)
return results;
}
+ public static <K, V> Set<V> valuesToSetAndCloseIterator(final KeyValueIterator<K, V> iterator) {
+ try (iterator) {
Review Comment:
We can reuse `valuesToSet`:
```java
try (iterator) {
return valuesToSet(iterator);
}
```
##########
streams/src/test/java/org/apache/kafka/streams/state/internals/AbstractKeyValueStoreTest.java:
##########
@@ -501,12 +500,13 @@ public void shouldPutAll() {
final List<KeyValue<Integer, String>> allReturned = new ArrayList<>();
final List<KeyValue<Integer, String>> expectedReturned =
Arrays.asList(KeyValue.pair(1, "one"), KeyValue.pair(2, "two"));
- final Iterator<KeyValue<Integer, String>> iterator = store.all();
- while (iterator.hasNext()) {
- allReturned.add(iterator.next());
+ try (final KeyValueIterator<Integer, String> iterator = store.all()) {
+ while (iterator.hasNext()) {
+ allReturned.add(iterator.next());
Review Comment:
We can reuse `StreamsTestUtils.toList`:
```java
try (final KeyValueIterator<Integer, String> iterator = store.all()) {
final List<KeyValue<Integer, String>> allReturned = StreamsTestUtils.toList(iterator);
final List<KeyValue<Integer, String>> expectedReturned =
Arrays.asList(KeyValue.pair(1, "one"), KeyValue.pair(2, "two"));
assertThat(allReturned, equalTo(expectedReturned));
}
```
##########
streams/src/test/java/org/apache/kafka/streams/state/internals/AbstractDualSchemaRocksDBSegmentedBytesStoreTest.java:
##########
@@ -251,7 +250,7 @@ public void shouldPutAndFetch() {
// all records expired as actual from is 59001 and to is 1000
final List<KeyValue<Windowed<String>, Long>> expected = Collections.emptyList();
- assertEquals(expected, toList(values));
+ assertEquals(expected, toListAndCloseIterator(values));
Review Comment:
line#246 already use try-resource to handle the close, so why does `toListAndCloseIterator` need to call close again?
--
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] MINOR: fix warnings in Kafka Streams state store tests [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on PR #17855:
URL: https://github.com/apache/kafka/pull/17855#issuecomment-2495045249
PR updated.
--
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] MINOR: fix warnings in Kafka Streams state store tests [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #17855:
URL: https://github.com/apache/kafka/pull/17855#discussion_r1852778867
##########
streams/src/test/java/org/apache/kafka/streams/state/internals/AbstractDualSchemaRocksDBSegmentedBytesStoreTest.java:
##########
@@ -251,7 +250,7 @@ public void shouldPutAndFetch() {
// all records expired as actual from is 59001 and to is 1000
final List<KeyValue<Windowed<String>, Long>> expected = Collections.emptyList();
- assertEquals(expected, toList(values));
+ assertEquals(expected, toListAndCloseIterator(values));
Review Comment:
You are right that it does not need to close the iterator, but it would be unnecessary code duplication to have both `toList` and `toListAndCloseIterator`...
And it does not "hurt" to close it twice.
It was just easier to refactor `toList` across the broad, instead of add `try-with-resources` to all callers which don't close the passed in iterator.
Thoughts? Also happy to do it differently.
--
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] MINOR: fix warnings in Kafka Streams state store tests [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #17855:
URL: https://github.com/apache/kafka/pull/17855#discussion_r1852783030
##########
streams/src/test/java/org/apache/kafka/test/StreamsTestUtils.java:
##########
@@ -150,6 +150,17 @@ public static <K, V> Set<V> valuesToSet(final Iterator<KeyValue<K, V>> iterator)
return results;
}
+ public static <K, V> Set<V> valuesToSetAndCloseIterator(final KeyValueIterator<K, V> iterator) {
+ try (iterator) {
Review Comment:
Good idea!
--
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] MINOR: fix warnings in Kafka Streams state store tests [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #17855:
URL: https://github.com/apache/kafka/pull/17855#discussion_r1847640621
##########
streams/src/test/java/org/apache/kafka/streams/state/internals/WindowKeySchemaTest.java:
##########
@@ -126,7 +126,7 @@ interface TriFunction<A, B, C, R> {
private final Window window = new TimeWindow(startTime, endTime);
private final Windowed<String> windowedKey = new Windowed<>(key, window);
private KeySchema keySchema;
- private final Serde<Windowed<String>> keySerde = new WindowedSerdes.TimeWindowedSerde<>(serde, Long.MAX_VALUE);
+ private final Serde<Windowed<String>> keySerde = new WindowedSerdes.TimeWindowedSerde<>(serde, endTime - startTime);
Review Comment:
Fixing window size to avoid WARN logs during test run (same below)
##########
streams/src/test/java/org/apache/kafka/streams/state/internals/WindowKeySchemaTest.java:
##########
@@ -126,7 +126,7 @@ interface TriFunction<A, B, C, R> {
private final Window window = new TimeWindow(startTime, endTime);
private final Windowed<String> windowedKey = new Windowed<>(key, window);
private KeySchema keySchema;
- private final Serde<Windowed<String>> keySerde = new WindowedSerdes.TimeWindowedSerde<>(serde, Long.MAX_VALUE);
+ private final Serde<Windowed<String>> keySerde = new WindowedSerdes.TimeWindowedSerde<>(serde, endTime - startTime);
Review Comment:
Fixing window size to avoid WARN logs during test run
--
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] MINOR: fix warnings in Kafka Streams state store tests [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 merged PR #17855:
URL: https://github.com/apache/kafka/pull/17855
--
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] MINOR: fix warnings in Kafka Streams state store tests [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #17855:
URL: https://github.com/apache/kafka/pull/17855#discussion_r1854416691
##########
streams/src/test/java/org/apache/kafka/streams/state/internals/AbstractDualSchemaRocksDBSegmentedBytesStoreTest.java:
##########
@@ -251,7 +250,7 @@ public void shouldPutAndFetch() {
// all records expired as actual from is 59001 and to is 1000
final List<KeyValue<Windowed<String>, Long>> expected = Collections.emptyList();
- assertEquals(expected, toList(values));
+ assertEquals(expected, toListAndCloseIterator(values));
Review Comment:
Maybe we could remove the try-with-resources to match the style used in AbstractRocksDBWindowStoreTest. For example:
```
// For all tests, actualFrom is computed using observedStreamTime - retention + 1.
// so actualFrom = 60000(observedStreamTime) - 1000(retention) + 1 = 59001
// all records expired as actual from is 59001 and to is 1000
assertEquals(Collections.emptyList(), toListAndCloseIterator(bytesStore.fetch(
Bytes.wrap(keyA.getBytes()), 0, windows[2].start())));
// all records expired as actual from is 59001 and to is 1000
assertEquals(Collections.emptyList(), toListAndCloseIterator(bytesStore.fetch(
Bytes.wrap(keyA.getBytes()), Bytes.wrap(keyB.getBytes()), 0, windows[2].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] MINOR: fix warnings in Kafka Streams state store tests [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #17855:
URL: https://github.com/apache/kafka/pull/17855#discussion_r1847636072
##########
streams/src/test/java/org/apache/kafka/streams/state/internals/AbstractDualSchemaRocksDBSegmentedBytesStoreTest.java:
##########
@@ -1354,17 +1356,17 @@ public void shouldCreateWriteBatches() {
@Test
public void shouldRestoreToByteStoreForActiveTask() {
- shouldRestoreToByteStore(TaskType.ACTIVE);
+ shouldRestoreToByteStore();
}
@Test
public void shouldRestoreToByteStoreForStandbyTask() {
context.transitionToStandby(null);
- shouldRestoreToByteStore(TaskType.STANDBY);
+ shouldRestoreToByteStore();
}
- private void shouldRestoreToByteStore(final TaskType taskType) {
Review Comment:
Unused parameter
--
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] MINOR: fix warnings in Kafka Streams state store tests [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #17855:
URL: https://github.com/apache/kafka/pull/17855#discussion_r1847640118
##########
streams/src/test/java/org/apache/kafka/streams/state/internals/AbstractWindowBytesStoreTest.java:
##########
@@ -652,90 +651,90 @@ public void testPutAndFetchAfter() {
assertEquals(
new HashSet<>(Collections.singletonList("zero")),
- valuesToSet(windowStore.fetch(0, ofEpochMilli(defaultStartTime + 0L),
- ofEpochMilli(defaultStartTime + 0L + WINDOW_SIZE))));
+ valuesToSetAndCloseIterator(windowStore.fetch(0, ofEpochMilli(defaultStartTime),
+ ofEpochMilli(defaultStartTime + WINDOW_SIZE))));
assertEquals(
new HashSet<>(Collections.singletonList("one")),
- valuesToSet(windowStore.fetch(1, ofEpochMilli(defaultStartTime + 1L),
+ valuesToSetAndCloseIterator(windowStore.fetch(1, ofEpochMilli(defaultStartTime + 1L),
ofEpochMilli(defaultStartTime + 1L + WINDOW_SIZE))));
assertEquals(
new HashSet<>(Collections.singletonList("two")),
- valuesToSet(windowStore.fetch(2, ofEpochMilli(defaultStartTime + 2L),
+ valuesToSetAndCloseIterator(windowStore.fetch(2, ofEpochMilli(defaultStartTime + 2L),
ofEpochMilli(defaultStartTime + 2L + WINDOW_SIZE))));
assertEquals(
new HashSet<>(Collections.emptyList()),
- valuesToSet(windowStore.fetch(3, ofEpochMilli(defaultStartTime + 3L),
+ valuesToSetAndCloseIterator(windowStore.fetch(3, ofEpochMilli(defaultStartTime + 3L),
ofEpochMilli(defaultStartTime + 3L + WINDOW_SIZE))));
assertEquals(
new HashSet<>(Collections.singletonList("four")),
- valuesToSet(windowStore.fetch(4, ofEpochMilli(defaultStartTime + 4L),
+ valuesToSetAndCloseIterator(windowStore.fetch(4, ofEpochMilli(defaultStartTime + 4L),
ofEpochMilli(defaultStartTime + 4L + WINDOW_SIZE))));
assertEquals(
new HashSet<>(Collections.singletonList("five")),
- valuesToSet(windowStore.fetch(5, ofEpochMilli(defaultStartTime + 5L),
+ valuesToSetAndCloseIterator(windowStore.fetch(5, ofEpochMilli(defaultStartTime + 5L),
ofEpochMilli(defaultStartTime + 5L + WINDOW_SIZE))));
- putSecondBatch(windowStore, defaultStartTime, context);
Review Comment:
removing unused parameter
--
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] MINOR: fix warnings in Kafka Streams state store tests [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #17855:
URL: https://github.com/apache/kafka/pull/17855#discussion_r1854825512
##########
streams/src/test/java/org/apache/kafka/streams/state/internals/AbstractDualSchemaRocksDBSegmentedBytesStoreTest.java:
##########
@@ -251,7 +250,7 @@ public void shouldPutAndFetch() {
// all records expired as actual from is 59001 and to is 1000
final List<KeyValue<Windowed<String>, Long>> expected = Collections.emptyList();
- assertEquals(expected, toList(values));
+ assertEquals(expected, toListAndCloseIterator(values));
Review Comment:
Lots of reformatting, but 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