stevenzwu commented on code in PR #10393:
URL: https://github.com/apache/iceberg/pull/10393#discussion_r1620956381
##########
flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/source/TestIcebergSourceFailover.java:
##########
@@ -333,28 +336,28 @@ private static void restartTaskManager(Runnable
afterFailAction, MiniCluster min
private static class RecordCounterToFail {
private static AtomicInteger records;
- private static CompletableFuture<Void> fail;
+ private static CountDownLatch countDownLatch;
private static CompletableFuture<Void> continueProcessing;
private static <T> DataStream<T> wrapWithFailureAfter(DataStream<T>
stream, int failAfter) {
records = new AtomicInteger();
- fail = new CompletableFuture<>();
continueProcessing = new CompletableFuture<>();
+ countDownLatch = new CountDownLatch(stream.getParallelism());
return stream.map(
record -> {
boolean reachedFailPoint = records.incrementAndGet() > failAfter;
- boolean notFailedYet = !fail.isDone();
+ boolean notFailedYet = countDownLatch.getCount() != 0;
if (notFailedYet && reachedFailPoint) {
- fail.complete(null);
+ countDownLatch.countDown();
continueProcessing.get();
}
return record;
});
}
- private static void waitToFail() throws ExecutionException,
InterruptedException {
- fail.get();
+ private static void waitToFail() throws InterruptedException {
Review Comment:
I didn't mean we have to do it in this PR. it is definitely reasonable to
separate it out.
while we are at it, I just noticed the intention and coverage of this class
has changed. hence want to point it out.
--
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: issues-unsubscr...@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscr...@iceberg.apache.org
For additional commands, e-mail: issues-h...@iceberg.apache.org
