Support for Flink's SpeculativeExecution in batch execution mode

[venkata91]

Summary

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

Testing

Existing tests should take care of it

[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.