Re: [PR] Flink: Support watermark alignment of source splits [iceberg]

pvary commented on code in PR #9308: URL: https://github.com/apache/iceberg/pull/9308#discussion_r1431841694 ## flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/source/reader/IcebergSourceSplitReader.java: ## @@ -80,6 +88,35 @@ public RecordsWithSplitIds> fetch() throws

Re: [PR] Flink: Support watermark alignment of source splits [iceberg]

pvary commented on code in PR #9308: URL: https://github.com/apache/iceberg/pull/9308#discussion_r1431830538 ## flink/v1.18/flink/src/test/java/org/apache/iceberg/flink/source/reader/TestIcebergSourceSplitReader.java: ## @@ -0,0 +1,167 @@ +/* + * Licensed to the Apache Software

Re: [PR] Flink: Support watermark alignment of source splits [iceberg]

stevenzwu commented on code in PR #9308: URL: https://github.com/apache/iceberg/pull/9308#discussion_r1431802868 ## flink/v1.18/flink/src/test/java/org/apache/iceberg/flink/source/reader/TestIcebergSourceSplitReader.java: ## @@ -0,0 +1,167 @@ +/* + * Licensed to the Apache Softw

Re: [PR] Flink: Support watermark alignment of source splits [iceberg]

stevenzwu commented on code in PR #9308: URL: https://github.com/apache/iceberg/pull/9308#discussion_r1431758647 ## flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/source/reader/IcebergSourceSplitReader.java: ## @@ -80,6 +88,35 @@ public RecordsWithSplitIds> fetch() th

Re: [PR] Flink: Support watermark alignment of source splits [iceberg]

stevenzwu commented on code in PR #9308: URL: https://github.com/apache/iceberg/pull/9308#discussion_r1431754664 ## flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/source/reader/IcebergSourceSplitReader.java: ## @@ -46,6 +51,8 @@ class IcebergSourceSplitReader implement

Re: [PR] Flink: Support watermark alignment of source splits [iceberg]

stevenzwu commented on code in PR #9308: URL: https://github.com/apache/iceberg/pull/9308#discussion_r1431726956 ## flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/source/reader/IcebergSourceSplitReader.java: ## @@ -80,6 +88,35 @@ public RecordsWithSplitIds> fetch() th

Re: [PR] Flink: Support watermark alignment of source splits [iceberg]

pvary commented on code in PR #9308: URL: https://github.com/apache/iceberg/pull/9308#discussion_r1431743938 ## flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/source/reader/IcebergSourceSplitReader.java: ## @@ -80,6 +88,35 @@ public RecordsWithSplitIds> fetch() throws

Re: [PR] Flink: Support watermark alignment of source splits [iceberg]

pvary commented on code in PR #9308: URL: https://github.com/apache/iceberg/pull/9308#discussion_r1431742741 ## flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/source/reader/IcebergSourceSplitReader.java: ## @@ -46,6 +51,8 @@ class IcebergSourceSplitReader implements S

Re: [PR] Flink: Support watermark alignment of source splits [iceberg]

stevenzwu commented on code in PR #9308: URL: https://github.com/apache/iceberg/pull/9308#discussion_r1431740597 ## flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/source/reader/IcebergSourceSplitReader.java: ## @@ -46,6 +51,8 @@ class IcebergSourceSplitReader implement

Re: [PR] Flink: Support watermark alignment of source splits [iceberg]

pvary commented on code in PR #9308: URL: https://github.com/apache/iceberg/pull/9308#discussion_r1431735881 ## flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/source/reader/IcebergSourceSplitReader.java: ## @@ -80,6 +88,35 @@ public RecordsWithSplitIds> fetch() throws

Re: [PR] Flink: Support watermark alignment of source splits [iceberg]

stevenzwu commented on code in PR #9308: URL: https://github.com/apache/iceberg/pull/9308#discussion_r1431736659 ## flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/source/reader/IcebergSourceSplitReader.java: ## @@ -80,6 +88,35 @@ public RecordsWithSplitIds> fetch() th

Re: [PR] Flink: Support watermark alignment of source splits [iceberg]

pvary commented on code in PR #9308: URL: https://github.com/apache/iceberg/pull/9308#discussion_r1431731166 ## flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/source/reader/IcebergSourceSplitReader.java: ## @@ -80,6 +88,35 @@ public RecordsWithSplitIds> fetch() throws

Re: [PR] Flink: Support watermark alignment of source splits [iceberg]

stevenzwu commented on code in PR #9308: URL: https://github.com/apache/iceberg/pull/9308#discussion_r1431726956 ## flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/source/reader/IcebergSourceSplitReader.java: ## @@ -80,6 +88,35 @@ public RecordsWithSplitIds> fetch() th

Re: [PR] Flink: Support watermark alignment of source splits [iceberg]

stevenzwu commented on code in PR #9308: URL: https://github.com/apache/iceberg/pull/9308#discussion_r1431716141 ## flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/source/reader/IcebergSourceSplitReader.java: ## @@ -80,6 +88,35 @@ public RecordsWithSplitIds> fetch() th

Re: [PR] Flink: Support watermark alignment of source splits [iceberg]

stevenzwu commented on code in PR #9308: URL: https://github.com/apache/iceberg/pull/9308#discussion_r1431716141 ## flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/source/reader/IcebergSourceSplitReader.java: ## @@ -80,6 +88,35 @@ public RecordsWithSplitIds> fetch() th

Re: [PR] Flink: Support watermark alignment of source splits [iceberg]

stevenzwu commented on code in PR #9308: URL: https://github.com/apache/iceberg/pull/9308#discussion_r1431702572 ## flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/source/reader/IcebergSourceSplitReader.java: ## @@ -80,6 +88,35 @@ public RecordsWithSplitIds> fetch() th

Re: [PR] Flink: Support watermark alignment of source splits [iceberg]

mxm commented on code in PR #9308: URL: https://github.com/apache/iceberg/pull/9308#discussion_r1431664838 ## flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/source/reader/IcebergSourceSplitReader.java: ## @@ -80,6 +88,35 @@ public RecordsWithSplitIds> fetch() throws I

Re: [PR] Flink: Support watermark alignment of source splits [iceberg]

pvary commented on code in PR #9308: URL: https://github.com/apache/iceberg/pull/9308#discussion_r1431277065 ## flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/source/reader/IcebergSourceSplitReader.java: ## @@ -113,7 +150,13 @@ public void handleSplitsChanges(SplitsCh

Re: [PR] Flink: Support watermark alignment of source splits [iceberg]

pvary commented on code in PR #9308: URL: https://github.com/apache/iceberg/pull/9308#discussion_r1431272586 ## flink/v1.18/flink/src/test/java/org/apache/iceberg/flink/source/reader/TestIcebergSourceSplitReader.java: ## @@ -0,0 +1,167 @@ +/* + * Licensed to the Apache Software

Re: [PR] Flink: Support watermark alignment of source splits [iceberg]

pvary commented on code in PR #9308: URL: https://github.com/apache/iceberg/pull/9308#discussion_r1431258224 ## flink/v1.18/flink/src/test/java/org/apache/iceberg/flink/source/TestIcebergSourceWithWatermarkExtractor.java: ## @@ -383,8 +381,10 @@ protected IcebergSource source()

Re: [PR] Flink: Support watermark alignment of source splits [iceberg]

pvary commented on code in PR #9308: URL: https://github.com/apache/iceberg/pull/9308#discussion_r1431255938 ## flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/source/reader/IcebergSourceSplitReader.java: ## @@ -80,6 +88,35 @@ public RecordsWithSplitIds> fetch() throws

Re: [PR] Flink: Support watermark alignment of source splits [iceberg]

pvary commented on code in PR #9308: URL: https://github.com/apache/iceberg/pull/9308#discussion_r1430987041 ## flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/source/reader/IcebergSourceSplitReader.java: ## @@ -80,6 +88,35 @@ public RecordsWithSplitIds> fetch() throws

Re: [PR] Flink: Support watermark alignment of source splits [iceberg]

pvary commented on code in PR #9308: URL: https://github.com/apache/iceberg/pull/9308#discussion_r1430985248 ## flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/source/reader/IcebergSourceSplitReader.java: ## @@ -80,6 +88,35 @@ public RecordsWithSplitIds> fetch() throws

Re: [PR] Flink: Support watermark alignment of source splits [iceberg]

pvary commented on code in PR #9308: URL: https://github.com/apache/iceberg/pull/9308#discussion_r1430978486 ## flink/v1.18/flink/src/test/java/org/apache/iceberg/flink/source/TestIcebergSourceWithWatermarkExtractor.java: ## @@ -383,8 +381,10 @@ protected IcebergSource source()

Re: [PR] Flink: Support watermark alignment of source splits [iceberg]

pvary commented on code in PR #9308: URL: https://github.com/apache/iceberg/pull/9308#discussion_r1430978486 ## flink/v1.18/flink/src/test/java/org/apache/iceberg/flink/source/TestIcebergSourceWithWatermarkExtractor.java: ## @@ -383,8 +381,10 @@ protected IcebergSource source()

Re: [PR] Flink: Support watermark alignment of source splits [iceberg]

stevenzwu commented on code in PR #9308: URL: https://github.com/apache/iceberg/pull/9308#discussion_r1430727887 ## flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/source/reader/IcebergSourceSplitReader.java: ## @@ -80,6 +88,35 @@ public RecordsWithSplitIds> fetch() th

Re: [PR] Flink: Support watermark alignment of source splits [iceberg]

stevenzwu commented on code in PR #9308: URL: https://github.com/apache/iceberg/pull/9308#discussion_r1430722369 ## flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/source/reader/IcebergSourceSplitReader.java: ## @@ -123,6 +155,19 @@ public void close() throws Exception

Re: [PR] Flink: Support watermark alignment of source splits [iceberg]

pvary commented on code in PR #9308: URL: https://github.com/apache/iceberg/pull/9308#discussion_r1430553479 ## flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/source/reader/IcebergSourceSplitReader.java: ## @@ -80,6 +88,35 @@ public RecordsWithSplitIds> fetch() throws

Re: [PR] Flink: Support watermark alignment of source splits [iceberg]

pvary commented on code in PR #9308: URL: https://github.com/apache/iceberg/pull/9308#discussion_r1430544840 ## flink/v1.18/flink/src/test/java/org/apache/iceberg/flink/source/TestIcebergSourceWithWatermarkExtractor.java: ## @@ -383,8 +381,10 @@ protected IcebergSource source()

Re: [PR] Flink: Support watermark alignment of source splits [iceberg]

pvary commented on PR #9308: URL: https://github.com/apache/iceberg/pull/9308#issuecomment-1861356815 > BTW, should we add an integration test? That reminds me, to remove the `set(PipelineOptions.ALLOW_UNALIGNED_SOURCE_SPLITS, true)` from the `TestIcebergSourceWithWatermarkExtractor`

Re: [PR] Flink: Support watermark alignment of source splits [iceberg]

pvary commented on code in PR #9308: URL: https://github.com/apache/iceberg/pull/9308#discussion_r1430539001 ## flink/v1.18/flink/src/test/java/org/apache/iceberg/flink/source/reader/TestIcebergSourceSplitReader.java: ## @@ -0,0 +1,167 @@ +/* + * Licensed to the Apache Software

Re: [PR] Flink: Support watermark alignment of source splits [iceberg]

pvary commented on code in PR #9308: URL: https://github.com/apache/iceberg/pull/9308#discussion_r1430516299 ## flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/source/reader/IcebergSourceSplitReader.java: ## @@ -123,6 +155,19 @@ public void close() throws Exception {

Re: [PR] Flink: Support watermark alignment of source splits [iceberg]

pvary commented on code in PR #9308: URL: https://github.com/apache/iceberg/pull/9308#discussion_r1430536963 ## flink/v1.18/flink/src/test/java/org/apache/iceberg/flink/source/reader/TestIcebergSourceSplitReader.java: ## @@ -0,0 +1,167 @@ +/* + * Licensed to the Apache Software

Re: [PR] Flink: Support watermark alignment of source splits [iceberg]

pvary commented on code in PR #9308: URL: https://github.com/apache/iceberg/pull/9308#discussion_r1430531849 ## flink/v1.18/flink/src/test/java/org/apache/iceberg/flink/source/reader/TestIcebergSourceSplitReader.java: ## @@ -0,0 +1,167 @@ +/* + * Licensed to the Apache Software

Re: [PR] Flink: Support watermark alignment of source splits [iceberg]

pvary commented on code in PR #9308: URL: https://github.com/apache/iceberg/pull/9308#discussion_r1430516299 ## flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/source/reader/IcebergSourceSplitReader.java: ## @@ -123,6 +155,19 @@ public void close() throws Exception {

Re: [PR] Flink: Support watermark alignment of source splits [iceberg]

mas-chen commented on code in PR #9308: URL: https://github.com/apache/iceberg/pull/9308#discussion_r1430484861 ## flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/source/reader/IcebergSourceSplitReader.java: ## @@ -80,6 +88,35 @@ public RecordsWithSplitIds> fetch() thr

Re: [PR] Flink: Support watermark alignment of source splits [iceberg]

mas-chen commented on PR #9308: URL: https://github.com/apache/iceberg/pull/9308#issuecomment-1861255463 BTW, should we add an integration test? -- 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 th

Re: [PR] Flink: Support watermark alignment of source splits [iceberg]

pvary commented on PR #9308: URL: https://github.com/apache/iceberg/pull/9308#issuecomment-1860397381 > These messages are executed in the same thread as the fetch method, so in this case we have to return from the fetch loop, even with empty results. This is somewhat concerning to me consi

Re: [PR] Flink: Support watermark alignment of source splits [iceberg]

pvary commented on PR #9308: URL: https://github.com/apache/iceberg/pull/9308#issuecomment-1859147694 @stevenzwu: Here is what I have found: `wakeUp` is called whenever there is a new message concerning the `SplitReader`: - new split added/removed - split paused/resumed These

Re: [PR] Flink: Support watermark alignment of source splits [iceberg]

pvary commented on code in PR #9308: URL: https://github.com/apache/iceberg/pull/9308#discussion_r1429133489 ## flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/source/reader/IcebergSourceSplitReader.java: ## @@ -80,6 +87,25 @@ public RecordsWithSplitIds> fetch() throws

Re: [PR] Flink: Support watermark alignment of source splits [iceberg]

pvary commented on code in PR #9308: URL: https://github.com/apache/iceberg/pull/9308#discussion_r1429133074 ## flink/v1.18/flink/src/test/java/org/apache/iceberg/flink/source/TestIcebergSourceWithWatermarkExtractor.java: ## @@ -273,23 +271,40 @@ public void apply( public voi

Re: [PR] Flink: Support watermark alignment of source splits [iceberg]

pvary commented on code in PR #9308: URL: https://github.com/apache/iceberg/pull/9308#discussion_r1429132495 ## flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/source/reader/IcebergSourceSplitReader.java: ## @@ -113,7 +139,13 @@ public void handleSplitsChanges(SplitsCh

Re: [PR] Flink: Support watermark alignment of source splits [iceberg]

pvary commented on code in PR #9308: URL: https://github.com/apache/iceberg/pull/9308#discussion_r1429132559 ## flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/source/reader/IcebergSourceSplitReader.java: ## @@ -113,7 +139,13 @@ public void handleSplitsChanges(SplitsCh

Re: [PR] Flink: Support watermark alignment of source splits [iceberg]

pvary commented on code in PR #9308: URL: https://github.com/apache/iceberg/pull/9308#discussion_r1429040047 ## flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/source/reader/IcebergSourceSplitReader.java: ## @@ -80,6 +87,25 @@ public RecordsWithSplitIds> fetch() throws

Re: [PR] Flink: Support watermark alignment of source splits [iceberg]

pvary commented on code in PR #9308: URL: https://github.com/apache/iceberg/pull/9308#discussion_r1429023418 ## flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/source/reader/IcebergSourceSplitReader.java: ## @@ -123,6 +155,19 @@ public void close() throws Exception {

Re: [PR] Flink: Support watermark alignment of source splits [iceberg]

pvary commented on PR #9308: URL: https://github.com/apache/iceberg/pull/9308#issuecomment-1859040607 > @pvary we are changing some critical code here. I would prefer we take a little more time here and get more people to review. > > Hence I would prefer if we disable/ignore the test

Re: [PR] Flink: Support watermark alignment of source splits [iceberg]

stevenzwu commented on code in PR #9308: URL: https://github.com/apache/iceberg/pull/9308#discussion_r1428932405 ## flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/source/reader/IcebergSourceSplitReader.java: ## @@ -80,6 +87,25 @@ public RecordsWithSplitIds> fetch() th

Re: [PR] Flink: Support watermark alignment of source splits [iceberg]

stevenzwu commented on code in PR #9308: URL: https://github.com/apache/iceberg/pull/9308#discussion_r1428899523 ## flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/source/reader/IcebergSourceSplitReader.java: ## @@ -113,7 +139,13 @@ public void handleSplitsChanges(Spli

Re: [PR] Flink: Support watermark alignment of source splits [iceberg]

pvary commented on code in PR #9308: URL: https://github.com/apache/iceberg/pull/9308#discussion_r1428804100 ## flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/source/reader/IcebergSourceSplitReader.java: ## @@ -80,6 +87,18 @@ public RecordsWithSplitIds> fetch() throws

Re: [PR] Flink: Support watermark alignment of source splits [iceberg]

pvary commented on code in PR #9308: URL: https://github.com/apache/iceberg/pull/9308#discussion_r1428755761 ## flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/source/reader/IcebergSourceSplitReader.java: ## @@ -61,6 +67,7 @@ class IcebergSourceSplitReader implements S

Re: [PR] Flink: Support watermark alignment of source splits [iceberg]

stevenzwu commented on code in PR #9308: URL: https://github.com/apache/iceberg/pull/9308#discussion_r1428500261 ## flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/source/reader/IcebergSourceSplitReader.java: ## @@ -80,6 +87,18 @@ public RecordsWithSplitIds> fetch() th

Re: [PR] Flink: Support watermark alignment of source splits [iceberg]

pvary commented on code in PR #9308: URL: https://github.com/apache/iceberg/pull/9308#discussion_r1428447497 ## flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/source/reader/IcebergSourceSplitReader.java: ## @@ -80,6 +87,18 @@ public RecordsWithSplitIds> fetch() throws

Re: [PR] Flink: Support watermark alignment of source splits [iceberg]

pvary commented on code in PR #9308: URL: https://github.com/apache/iceberg/pull/9308#discussion_r1428446262 ## flink/v1.18/build.gradle: ## @@ -63,6 +63,7 @@ project(":iceberg-flink:iceberg-flink-${flinkMajorVersion}") { // These artifacts are shaded and included in the

Re: [PR] Flink: Support watermark alignment of source splits [iceberg]

pvary commented on PR #9308: URL: https://github.com/apache/iceberg/pull/9308#issuecomment-1858414638 > @pvary is the whole PR addressing the flaky test? if not, please separate out the flaky test part (hopefully small) into a separate PR. I would like to merge the flaky test fix ASAP. And

Re: [PR] Flink: Support watermark alignment of source splits [iceberg]

stevenzwu commented on PR #9308: URL: https://github.com/apache/iceberg/pull/9308#issuecomment-1858352676 @pvary is the whole PR addressing the flaky test? if not, please separate out the flaky test part (hopefully very small) into a separate PR. I would like to merge the flaky test fix ASA

Re: [PR] Flink: Support watermark alignment of source splits [iceberg]

rodmeneses commented on code in PR #9308: URL: https://github.com/apache/iceberg/pull/9308#discussion_r1428246189 ## flink/v1.18/flink/src/test/java/org/apache/iceberg/flink/source/reader/TestIcebergSourceSplitReader.java: ## @@ -0,0 +1,104 @@ +/* + * Licensed to the Apache Soft

Re: [PR] Flink: Support watermark alignment of source splits [iceberg]

rodmeneses commented on code in PR #9308: URL: https://github.com/apache/iceberg/pull/9308#discussion_r1428244482 ## flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/source/reader/IcebergSourceSplitReader.java: ## @@ -80,6 +87,18 @@ public RecordsWithSplitIds> fetch() t

Re: [PR] Flink: Support watermark alignment of source splits [iceberg]

rodmeneses commented on code in PR #9308: URL: https://github.com/apache/iceberg/pull/9308#discussion_r1428240323 ## flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/source/reader/IcebergSourceSplitReader.java: ## @@ -61,6 +67,7 @@ class IcebergSourceSplitReader implemen

Re: [PR] Flink: Support watermark alignment of source splits [iceberg]

rodmeneses commented on code in PR #9308: URL: https://github.com/apache/iceberg/pull/9308#discussion_r1428236699 ## flink/v1.18/build.gradle: ## @@ -63,6 +63,7 @@ project(":iceberg-flink:iceberg-flink-${flinkMajorVersion}") { // These artifacts are shaded and included in

Re: [PR] Flink: Support watermark alignment of source splits [iceberg]

pvary commented on PR #9308: URL: https://github.com/apache/iceberg/pull/9308#issuecomment-1858055733 CC: @ajantha-bhat - further fixes on the test -- 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

[PR] Flink: Support watermark alignment of source splits [iceberg]

pvary opened a new pull request, #9308: URL: https://github.com/apache/iceberg/pull/9308 Fix unit test failures found here: #9216 Before the changes it was not possible to run the tests multiple times using IntelliJ. Removed the `static` from `InMemoryReporter` so the tests could b