You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "pvary (via GitHub)" <gi...@apache.org> on 2024/05/29 07:39:46 UTC
[PR] Flink 1.19: Fix flaky TestIcebergSourceFailover > testBoundedWithSavepoint [iceberg]
pvary opened a new pull request, #10393:
URL: https://github.com/apache/iceberg/pull/10393
This attempts to fix https://github.com/apache/iceberg/issues/10356
Also related: https://github.com/apache/iceberg/pull/10370
--
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-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink 1.19: Fix flaky TestIcebergSourceFailover > testBoundedWithSavepoint [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #10393:
URL: https://github.com/apache/iceberg/pull/10393#discussion_r1619213639
##########
flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/source/TestIcebergSourceFailoverWithWatermarkExtractor.java:
##########
@@ -48,10 +52,17 @@ public class TestIcebergSourceFailoverWithWatermarkExtractor extends TestIceberg
@Override
protected IcebergSource.Builder<RowData> sourceBuilder() {
- return IcebergSource.<RowData>builder()
+ Configuration config = new Configuration();
+ config.setInteger(FlinkConfigOptions.SOURCE_READER_FETCH_BATCH_RECORD_COUNT, 128);
Review Comment:
Copied from the `TestIcebergSourceFailover`.
Do you happen to remember why they were set in the first place?
The tests are running without 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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink 1.19: Fix flaky TestIcebergSourceFailover > testBoundedWithSavepoint [iceberg]
Posted by "stevenzwu (via GitHub)" <gi...@apache.org>.
stevenzwu commented on code in PR #10393:
URL: https://github.com/apache/iceberg/pull/10393#discussion_r1619547088
##########
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());
Review Comment:
thanks for confirming. assume you meant breakpoints (not checkpoints)
--
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-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink 1.19: Fix flaky TestIcebergSourceFailover > testBoundedWithSavepoint [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #10393:
URL: https://github.com/apache/iceberg/pull/10393#discussion_r1619214564
##########
flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/source/TestIcebergSourceFailoverWithWatermarkExtractor.java:
##########
@@ -48,10 +52,17 @@ public class TestIcebergSourceFailoverWithWatermarkExtractor extends TestIceberg
@Override
protected IcebergSource.Builder<RowData> sourceBuilder() {
- return IcebergSource.<RowData>builder()
+ Configuration config = new Configuration();
+ config.setInteger(FlinkConfigOptions.SOURCE_READER_FETCH_BATCH_RECORD_COUNT, 128);
+ return IcebergSource.forRowData()
.tableLoader(sourceTableResource.tableLoader())
.watermarkColumn("ts")
- .project(TestFixtures.TS_SCHEMA);
+ .project(TestFixtures.TS_SCHEMA)
+ // Prevent combining splits
+ .set(
Review Comment:
Yeah, but the error was not with this one 😄.
I have just find it when copied the other configs to here.
--
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-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink 1.19: Fix flaky TestIcebergSourceFailover > testBoundedWithSavepoint [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #10393:
URL: https://github.com/apache/iceberg/pull/10393#discussion_r1619334747
##########
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());
Review Comment:
Tested with checkpoints. The static variables are "global" in the sense, that they are shared between every instance in the JVM. So the latch is used for both operator instances.
--
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-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink 1.19: Fix flaky TestIcebergSourceFailover > testBoundedWithSavepoint [iceberg]
Posted by "stevenzwu (via GitHub)" <gi...@apache.org>.
stevenzwu commented on code in PR #10393:
URL: https://github.com/apache/iceberg/pull/10393#discussion_r1619548308
##########
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:
sorry for the typo. I meant `TestIcebergSourceBounded` is probably a better home for `testBoundedWithSavepoint` if we can figure out the code reuse from `TestIcebergSourceFailover`. Otherwise, we need to rename `TestIcebergSourceFailover` to accurately reflect that it doesn't just test failover. It also tests savepoint trigger and no duplication.
--
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-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink 1.19: Fix flaky TestIcebergSourceFailover > testBoundedWithSavepoint [iceberg]
Posted by "stevenzwu (via GitHub)" <gi...@apache.org>.
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-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink 1.19: Fix flaky TestIcebergSourceFailover > testBoundedWithSavepoint [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #10393:
URL: https://github.com/apache/iceberg/pull/10393#discussion_r1619348996
##########
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 don't get this last comment 😢
--
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-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink 1.19: Fix flaky TestIcebergSourceFailover > testBoundedWithSavepoint [iceberg]
Posted by "stevenzwu (via GitHub)" <gi...@apache.org>.
stevenzwu commented on code in PR #10393:
URL: https://github.com/apache/iceberg/pull/10393#discussion_r1619230274
##########
flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/source/TestIcebergSourceFailoverWithWatermarkExtractor.java:
##########
@@ -48,10 +52,17 @@ public class TestIcebergSourceFailoverWithWatermarkExtractor extends TestIceberg
@Override
protected IcebergSource.Builder<RowData> sourceBuilder() {
- return IcebergSource.<RowData>builder()
+ Configuration config = new Configuration();
+ config.setInteger(FlinkConfigOptions.SOURCE_READER_FETCH_BATCH_RECORD_COUNT, 128);
Review Comment:
I see. this config is probably not required for `TestIcebergSourceFailover`, because only small number of records are generated per data file. This config is probably copied over unnecessarily from some specific tests like `TestArrayPoolDataIteratorBatcherRowData`.
either way is fine with me (copying it here or not).
--
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-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink 1.19: Fix flaky TestIcebergSourceFailover > testBoundedWithSavepoint [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #10393:
URL: https://github.com/apache/iceberg/pull/10393#discussion_r1619876999
##########
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());
Review Comment:
Yes, brakepoints 😄
--
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-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink 1.19: Fix flaky TestIcebergSourceFailover > testBoundedWithSavepoint [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on PR #10393:
URL: https://github.com/apache/iceberg/pull/10393#issuecomment-2139996501
Thanks for the report @manuzhang and @stevenzwu, @manuzhang for the reviews!
--
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-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink 1.19: Fix flaky TestIcebergSourceFailover > testBoundedWithSavepoint [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on PR #10393:
URL: https://github.com/apache/iceberg/pull/10393#issuecomment-2137479894
Seems, like it is working now.
Run 3k test locally, and ~3*6 tests here.
Needed to add the `SPLIT_FILE_OPEN_COST` to the `TestIcebergSourceFailoverWithWatermarkExtractor` as it was missing there.
--
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-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink 1.19: Fix flaky TestIcebergSourceFailover > testBoundedWithSavepoint [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on PR #10393:
URL: https://github.com/apache/iceberg/pull/10393#issuecomment-2136745703
CC: @manuzhang
--
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-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink 1.19: Fix flaky TestIcebergSourceFailover > testBoundedWithSavepoint [iceberg]
Posted by "stevenzwu (via GitHub)" <gi...@apache.org>.
stevenzwu commented on code in PR #10393:
URL: https://github.com/apache/iceberg/pull/10393#discussion_r1619139980
##########
flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/source/TestIcebergSourceFailoverWithWatermarkExtractor.java:
##########
@@ -48,10 +52,17 @@ public class TestIcebergSourceFailoverWithWatermarkExtractor extends TestIceberg
@Override
protected IcebergSource.Builder<RowData> sourceBuilder() {
- return IcebergSource.<RowData>builder()
+ Configuration config = new Configuration();
+ config.setInteger(FlinkConfigOptions.SOURCE_READER_FETCH_BATCH_RECORD_COUNT, 128);
+ return IcebergSource.forRowData()
.tableLoader(sourceTableResource.tableLoader())
.watermarkColumn("ts")
- .project(TestFixtures.TS_SCHEMA);
+ .project(TestFixtures.TS_SCHEMA)
+ // Prevent combining splits
+ .set(
Review Comment:
ah. we missed the bundling earlier.
##########
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());
Review Comment:
curious how is the countdown latch used by the map operator. will it be serialized and a new cloned object is deserialized on the operator/subtask side? if yes, I guess there is no difference with completable future. If no, this ensures all subtasks are running and processed some records before failure is triggered.
could this be verified by break point in IDE?
##########
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 also just realized that this method is probably not accurate anymore with the `testBoundedWithSavepoint`. Maybe it should be called `waitForCondition`? also the inner class name `RecordCounterToFail` could be renamed to `RecordCounterToWait`.
##########
flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/source/TestIcebergSourceFailoverWithWatermarkExtractor.java:
##########
@@ -48,10 +52,17 @@ public class TestIcebergSourceFailoverWithWatermarkExtractor extends TestIceberg
@Override
protected IcebergSource.Builder<RowData> sourceBuilder() {
- return IcebergSource.<RowData>builder()
+ Configuration config = new Configuration();
+ config.setInteger(FlinkConfigOptions.SOURCE_READER_FETCH_BATCH_RECORD_COUNT, 128);
Review Comment:
is this config necessary?
--
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-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink 1.19: Fix flaky TestIcebergSourceFailover > testBoundedWithSavepoint [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary merged PR #10393:
URL: https://github.com/apache/iceberg/pull/10393
--
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-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink 1.19: Fix flaky TestIcebergSourceFailover > testBoundedWithSavepoint [iceberg]
Posted by "stevenzwu (via GitHub)" <gi...@apache.org>.
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` if we can figure out the code reuse.
--
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-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink 1.19: Fix flaky TestIcebergSourceFailover > testBoundedWithSavepoint [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #10393:
URL: https://github.com/apache/iceberg/pull/10393#discussion_r1619909229
##########
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:
The issue was discovered when there was a job restart, because of a temporary issue in a downstream service.
This is a failure recovery scenario, in many ways very similar to the Failover cases tested here. Maybe we can rename the class to `TestIcebergSourceFailureRecovery`?
That said, I see the following issues mixing refactoring and actual fixes in the same PR:
- Makes it hard in retrospect to understand what was the fix, and what was the refactor. If we need to revisit the fix, because it's not working, or we have a similar issue, then we will have hard time separating those.
- If you have a stable branch for internal releases, you might not want to accept the risks inherent to the refactor, but you might need the fix. Mixing fix and refactor code makes this hard.
- If you refactor the code that you are fixing, and move it to another class, then it is very hard to spot small mistakes or changes done in the same part of the code. Makes the reviewer life hard, and the process error prone.
- Blocking flaky test fix on refactoring could waste serious developer hours, if other developers bumps into the same issue and sink time into trying to find and fix the bug.
I am all for cleaning up code, and happy to do it whenever I have time, but I think that if it's not related to the actual fix/PR, then it should be done only with a very limited scope. If we spot something which needs to be fixed, we should create a specific, refactoring PR for it. I would be happy to review it, whenever you make one.
--
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-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink 1.19: Fix flaky TestIcebergSourceFailover > testBoundedWithSavepoint [iceberg]
Posted by "stevenzwu (via GitHub)" <gi...@apache.org>.
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-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org