What is the purpose of the change

This pull request addresses the issue detailed inFLINK-38349 which describes scenarios in which certain edge-case values formaxFanIn andchannelIDs.size() could result in floating-point errors that resulted in thrownArithmeticException exceptions (specifically division by zero). This adds the appropriate safeguards to prevent these issues.

The originating JIRA ticket provides detailed examples illustrating how the floating-point errors were reproduced. In summary,the issue occurs whenever the channel count (channelIDs.size()) is greater than or equal to a power ofmaxFanIn:

channelIDs.size() ≥ maxFanInⁿ

The overall change not only addresses the floating-point issues but alsoimproves scaling calculation performance by 3-4x on average (see chart below).

Brief change log

• Updated the scaling logic that occurs withinSpillingThread.mergeChannelList() by replacing the previousMath.log()-based calculations, which could result in floating point errors, in favor of an iterative, exponential search.
  • Added a privatecomputeMergeScaleForChannelSize() function to encapsulate and document the updated scaling calculations.
• Added a newSpillingThreadTest file with an associated, parameterizedtestMergeChannelListDivideByZeroSafety test that was used to reproduce the original issue and was later adjusted to confirm the fix.

Verifying this change

This change added tests and can be verified as follows:

• Added newSpillingThreadTest.testMergeChannelListDivideByZeroSafety which originally reproduced the issue by creating a series of parameterized tests founds within the originating JIRA ticket.
• Ensured all previous test suites within the related areas were passing as expected.

Verifying Performance

In addition to these tests related to reproduction, I performed a series of manual tests to verify performance between the previousMath.log()-based approach (prone to floating-point error) and the newer iterative, exponential approach.

You cansee the full gist here with all of the necessary code comparing the older and newer approaches (elected not to commit as it was purely for demonstration purposes).

Verifying Correctness

In addition to the above performance tests, I also performed a similar equivalence test comparing the generated results between the previous and new approaches to ensure parity (sans floating-point errors) for many of the common and edge case values presented earlier:

You can finda related gist containing these tests as well.

Does this pull request potentially affect one of the following parts:

• Dependencies (does it add or upgrade a dependency):no
• The public API, i.e., is any changed class annotated with@Public(Evolving):no
• The serializers:no
• The runtime per-record code paths (performance sensitive):yes / don't know
• Anything that affects deployment or recovery:no
• The S3 file system connector:no

Documentation

• Does this pull request introduce a new feature?no
• If yes, how is the feature documented?not applicable