Support for Flink's SpeculativeExecution in batch execution mode

[venkata91]

Summary

Add support for Flink's Speculative Execution in batch execution mode

Testing

Added an integration test to verify the expected speculative execution behavior with IcebergSource

[becketqin]

@venkata91 Thanks for the patch. Left a comment. It looks like there is a simpler approach.

[venkata91]

cc @stevenzwu for review. Should this change also be made in other Flink versions like Flink-1.17 and Flink-1.18?

[pvary]

@venkata91: How can we be sure that the tests are exercising the speculative execution code path?

Does any of the tests reads some splits multiple times, and use the result of the faster one?

I think it would be useful to have a test demonstrating that the behavior works, to prevent disabling it by an unrelated change by accident.

[venkata91]

@venkata91: How can we be sure that the tests are exercising the speculative execution code path?

Does any of the tests reads some splits multiple times, and use the result of the faster one?

I think it would be useful to have a test demonstrating that the behavior works, to prevent disabling it by an unrelated change by accident.

Sure sounds good. Will add a test.

[venkata91]

@venkata91: How can we be sure that the tests are exercising the speculative execution code path?

Does any of the tests reads some splits multiple times, and use the result of the faster one?

I think it would be useful to have a test demonstrating that the behavior works, to prevent disabling it by an unrelated change by accident.

@pvary Added an integration test to verify the tasks are speculated and produces the expected output. PTAL. btw, should this change also be made in other Flink versions like Flink-1.17 and Flink-1.18?

[venkata91]

Gentle ping for reviews cc @pvary @stevenzwu Thanks!

[pvary]

In my local tests this passes even without the changes in AbstractIcebergEnumerator.
Does that mean, that the speculative execution is supported even without this change?

[venkata91]

Nice catch!

Looks like I didn't set table.exec.iceberg.use-flip27-source to true. With this, without the changes in AbstractIcebergEnumerator, it will fail with an exception saying:

The split enumerator StaticIcebergEnumerator must implement SupportsHandleExecutionAttemptSourceEvent to be used in concurrent execution attempts scenario (e.g. if speculative execution is enabled"

[venkata91]

@pvary should this change be ported to other flink versions like 1.17 and 1.18?

[pvary]

@pvary should this change be ported to other flink versions like 1.17 and 1.18?

In a follow-up backport pr

[pvary]

Merged to main.
Thanks for the PR @venkata91!

Could you please create the backport PR to the other Flink versions? The PR could be generated like this:

git diff <HASH>^ <HASH> | sed "s/v1.19/v1.18/g" | git apply -3 -p1
git diff <HASH>^ <HASH> | sed "s/v1.19/v1.17/g" | git apply -3 -p1

[venkata91]

Could you please create the backport PR to the other Flink versions?

@pvary Should we have 2 backport PRs one for 1.17 and 1.18 or is it fine to do it in a single PR?

[pvary]

Could you please create the backport PR to the other Flink versions?

@pvary Should we have 2 backport PRs one for 1.17 and 1.18 or is it fine to do it in a single PR?

A single PR would be fine, as we don't expect any serious changes to review