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

• 7e3700a 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 7e3700a, and the changes are the same.

[davidradl]

can we have a unit test?

[davidradl]

should we be eagerly grabbing the lock in all these methods so all processing is done under the lock? for example all references to batch should be under the lock as the value could change under us. I was thinking we could synchronize all the methods referencing batch, so we do not need the lock.

[raulpardo]

This is a good idea that would solve all data races. However, I noticed that the current implementation seems to be acquiring locks depending on what thread is executing the method by invoking checkIsMailboxThread(). It might be that this was introduced for performance reasons; as acquiring the lock is a rather slow operation. But this is not entirely clear to me, as there are other methods like createBatch that acquire the lock even though the first instruction is to check is to ensure that only the mailbox thread is executing this.

This is why I reported only the data race on the drain method, as there is no instruction indicating that only the mailbox thread can execute that method.

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