mjsax commented on code in PR #13927:
URL: https://github.com/apache/kafka/pull/13927#discussion_r1253708580
##########
streams/src/test/java/org/apache/kafka/streams/integration/LagFetchIntegrationTest.java:
##########
@@ -277,6 +277,7 @@ public void shouldFetchLagsDuringRestoration() throws
Exception {
t1.toStream().to(outputTopicName);
final KafkaStreams streams = new KafkaStreams(builder.build(), props);
Review Comment:
We need to `close()` a `KafkaStreams` client even if we never `.start()` it,
and thus should use `try-with-resource` or split this up and do:
```
final KafkaStreams streams;
try {
streams = new KafkaStreams(...);
} finally {
...
}
```
##########
streams/src/test/java/org/apache/kafka/streams/integration/LagFetchIntegrationTest.java:
##########
@@ -313,9 +313,14 @@ public void shouldFetchLagsDuringRestoration() throws
Exception {
Files.walk(stateDir.toPath()).sorted(Comparator.reverseOrder())
.map(Path::toFile)
.forEach(f -> assertTrue(f.delete(), "Some state " + f + "
could not be deleted"));
+ } finally {
+ streams.close();
+ streams.cleanUp();
+ }
- // wait till the lag goes down to 0
- final KafkaStreams restartedStreams = new
KafkaStreams(builder.build(), props);
+ // wait till the lag goes down to 0
+ final KafkaStreams restartedStreams = new
KafkaStreams(builder.build(), props);
Review Comment:
as 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-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
