Flink: Backport #8553 to v1.15, v1.16

[pvary]

Flink: Backport #8553 to v1.15, v1.16

• Clean backport of Flink: Emit watermarks from the IcebergSource #8553
• Fix backport differences for Flink 1.17: Use awaitility instead of Thread.sleep() #8852 (Flink 1.15: Use awaitility instead of Thread.sleep() #8877, Flink 1.16: Use awaitility instead of Thread.sleep() #8880)

[pvary]

CC: @nastra, @nk1506 - When backporting changes, please make sure that all of the changes as similar as possible.

After fixing the backport differences caused by previous backports I had a clean backport of #8553 to 1.16.

Flink 1.15 missing a feature, so I had to improvise there. See c09aab0

CC: @stevenzwu

[stevenzwu]

@pvary can you separate this out as a separate PR?

Fix backport differences for #8852 (#8877, #8880)

[pvary]

@pvary can you separate this out as a separate PR?

Fix backport differences for #8852 (#8877, #8880)

TBH, I do not really understand your comment here 😞

To decrease the turnaround time, I have created 3 PRs, and you can decide which one we should use to move forward:

• Flink: Proper backport for #8852 #9146 - Flink 1.17: Use awaitility instead of Thread.sleep() #8852 was the original change, and Flink 1.15: Use awaitility instead of Thread.sleep() #8877 and Flink 1.16: Use awaitility instead of Thread.sleep() #8880 were the backports done. Sadly these backports were not clean backports, resulting in a slightly different code in v1.17/v1.16/v1.15 versions. Flink: Proper backport for #8852 #9146 is a standalone commit which cleans up the differences, and makes sure that v1.16/v1.15 has the same code as v1.17. The code in v1.17 was used as a master of the changes.
• Flink missing backports #9144 - This has 2 commits. The first one (c7b484e) is the same as Flink: Proper backport for #8852 #9146, the second one (21b3eea) is the backport of the Flink part of Core: Enable column statistics filtering after planning #8803. If we accept this one, we can merge the changes without squashing the commits - this way we have a clean history, but we can move forward faster.
• Flink: Backport #8553 to v1.15, v1.16 #9145 - This has 3 commits. The first two are the same as Flink missing backports #9144, the third one (27333e3) is the backport of Flink: Emit watermarks from the IcebergSource #8553. If we accept this one, we can merge the changes without squashing the commits - this way we have a clean history, but we can move forward faster.

If someone wants to understand better the contents of the actual backport commit of #8553, the 27333e3 is a merge of the clean backport (67d5cdc), and the required v1.15 related fixes (c09aab0) which are needed as Flink v1.15 has a missing feature used by the tests

[stevenzwu]

@pvary I meant separate to multiple back port PRs. one PR for one purpose. it is easier to review too.

[pvary]

@pvary I meant separate to multiple back port PRs. one PR for one purpose. it is easier to review too.

Thanks!

[pvary]

Merged with specific PRs for the different parts