[FLINK-37955] Don't load all records into main memory to avoid OOM

[rkhachatryan]

Replace List of matched records with Iterator to avoid OOM errors in StreamingJoinOperator and the like.
Currently, PR targets only the sync operator version - see the discussion below.

[flinkbot]

CI report:

• fa4b8a2 Azure: PENDING

Bot commands

The @flinkbot bot supports the following commands:

• @flinkbot run azure re-run the last Azure build

[rkhachatryan]

Originally, I fixed the issue in Flink 1.19 by replacing List with Iterator.

While porting the fix to 2.0, I discovered that most of the code was extracted to be re-used in async versions of the operators.

However, without some deeper rework, I don't see how my change can be used in async way; because async iterator requires callbacks to be passed in.

So I'm inclined to have two different versions for sync (iterator) and async (list) for now.

WDYT @xuyangzhong, @pnowojski ?

[pnowojski]

LGTM assuming green build

[rkhachatryan]

@flinkbot run azure

[rkhachatryan]

@flinkbot run azure

[rkhachatryan]

This failure seems to be unrelated:

Sep 17 17:50:34 17:50:34.902 [ERROR] Failures: 
Sep 17 17:50:34 17:50:34.902 [ERROR]   ChangelogNormalizeRestoreTest>RestoreTestBase.testRestore:397 
Sep 17 17:50:34 Expecting actual:
Sep 17 17:50:34   ["+I[two, 2, b]",
Sep 17 17:50:34     "+I[one, 1, a]",
Sep 17 17:50:34     "+I[three, 3, c]",
Sep 17 17:50:34     "-U[one, 1, a]",
Sep 17 17:50:34     "+U[one, 1, aa]",
Sep 17 17:50:34     "-U[three, 3, c]",
Sep 17 17:50:34     "+U[three, 3, cc]",
Sep 17 17:50:34     "-D[two, 2, b]",
Sep 17 17:50:34     "+I[four, 4, d]",
Sep 17 17:50:34     "+I[five, 5, e]",
Sep 17 17:50:34     "-U[four, 4, d]",
Sep 17 17:50:34     "+U[four, 4, dd]"]
Sep 17 17:50:34 to contain exactly in any order:
Sep 17 17:50:34   ["+I[one, 1, a]",
Sep 17 17:50:34     "+I[two, 2, b]",
Sep 17 17:50:34     "-U[one, 1, a]",
Sep 17 17:50:34     "+U[one, 1, aa]",
Sep 17 17:50:34     "+I[three, 3, c]",
Sep 17 17:50:34     "-D[two, 2, b]",
Sep 17 17:50:34     "-U[three, 3, c]",
Sep 17 17:50:34     "+U[three, 3, cc]",
Sep 17 17:50:34     "+I[four, 4, d]",
Sep 17 17:50:34     "+I[five, 5, e]",
Sep 17 17:50:34     "-U[four, 4, d]",
Sep 17 17:50:34     "+U[four, 4, dd]",
Sep 17 17:50:34     "+I[six, 6, f]",
Sep 17 17:50:34     "-D[six, 6, f]"]
Sep 17 17:50:34 but could not find the following elements:
Sep 17 17:50:34   ["+I[six, 6, f]", "-D[six, 6, f]"]

Re-running the CI.

[rkhachatryan]

The failure is unrelated: https://issues.apache.org/jira/browse/FLINK-35012 .
Merging

[rkhachatryan]

Originally, I fixed the issue in Flink 1.19 by replacing List with Iterator.
While porting the fix to 2.0, I discovered that most of the code was extracted to be re-used in async versions of the operators.
However, without some deeper rework, I don't see how my change can be used in async way; because async iterator requires callbacks to be passed in.
So I'm inclined to have two different versions for sync (iterator) and async (list) for now.
WDYT @xuyangzhong, @pnowojski ?

We've discussed this offline with @pnowojski and @Zakelly and decided to have separate versions of the code for now
but create a ticket for the async version: https://issues.apache.org/jira/browse/FLINK-38376