stevenzwu commented on code in PR #10393:
URL: https://github.com/apache/iceberg/pull/10393#discussion_r1619233709
##########
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:
this will also invalidate the class name `TestIcebergSourceFailover`. It was
added here probably for code reuse. otherwise, the better place for
`testBoundedWithSavepoint ` is probably `TestIcebergSourceFailover`
--
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
