[hotfix] fix data race in method `drain()` in `TaskMailboxImpl.java`

[raulpardo]

What is the purpose of the change

This is a hotfix fixing a data race in drain() method in TaskMailboxImpl.java. This can lead to undesired executions. For instance, consider the that mailbox thread is adding a mail by executing putFirst, and concurrently another thread is executing drain(). This can produce the following execution:

1. The read of batch here is executed before addFirst(mail) (in this line) adds the element,
2. Then, mail is added to batch (consequentely, mail is not in drainedMails)
3. Finally, batch.clear() is executed.

This execution results in missing mail (as it is neither included in batch nor drainedMails).

This hotfix PR resolves the data race by including the read to batch and its clearing within the critical section in the drain() method.

Brief change log

• Reading and clearing variable batch in the method drain() in TaskMailboxImpl.java are moved into the critical section within the method.

Verifying this change

This change is a trivial rework / code cleanup without any test coverage.

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: don't know
• The runtime per-record code paths (performance sensitive): don't know
• Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: don't know
• The S3 file system connector: don't know

Documentation

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

[flinkbot]

CI report:

• 7e3700a061b28be16b1661b4573b24400461ab2d Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

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

[raulpardo]

CI report:

* [9400657](https://github.com/apache/flink/commit/9400657bdbc244a1eb03d157b13de9ca5f7f7555) UNKNOWN

Bot commands

The link to the commit is unkown because I amended the commit message to clarify the subject of the PR. The new commit link is https://github.com/apache/flink/commit/7e3700a061b28be16b1661b4573b24400461ab2d, and the changes are the same.

[raulpardo]

can we have a unit test?

Sure, I can write a test. I am looking at this page to read about how to write unit tests for flink. I would like to add a java test, but there is no information on how to run Java tests. Is there some documentation on this? (it would be better, if it is possible to write a junit5 test. The test must be executed repeatedly to increase the chance of triggering the bug, and junit5 has a @RepeatedTest functionality which would be handy in this case.

[github-actions[bot]]

This PR is being marked as stale since it has not had any activity in the last 90 days. If you would like to keep this PR alive, please leave a comment asking for a review. If the PR has merge conflicts, update it with the latest from the base branch.

If you are having difficulty finding a reviewer, please reach out to the community, contact details can be found here: https://flink.apache.org/what-is-flink/community/

If this PR is no longer valid or desired, please feel free to close it. If no activity occurs in the next 30 days, it will be automatically closed.

[github-actions[bot]]

This PR is being marked as stale since it has not had any activity in the last 90 days. If you would like to keep this PR alive, please leave a comment asking for a review. If the PR has merge conflicts, update it with the latest from the base branch.

If you are having difficulty finding a reviewer, please reach out to the community, contact details can be found here: https://flink.apache.org/what-is-flink/community/

If this PR is no longer valid or desired, please feel free to close it. If no activity occurs in the next 30 days, it will be automatically closed.