tillrohrmann commented on a change in pull request #15557:
URL: https://github.com/apache/flink/pull/15557#discussion_r613211976
##########
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/operators/coordination/OperatorEventValve.java
##########
@@ -96,42 +94,44 @@ public boolean isShut() {
* (because it gets dropped through a call to {@link #reset()} or {@link
#resetForTask(int)},
* then the returned future till be completed exceptionally.
*/
- public CompletableFuture<Acknowledge> sendEvent(
- SerializedValue<OperatorEvent> event, int subtask) {
- synchronized (lock) {
- if (!shut) {
- return eventSender.apply(event, subtask);
- }
-
- final List<BlockedEvent> eventsForTask =
- blockedEvents.computeIfAbsent(subtask, (key) -> new
ArrayList<>());
- final CompletableFuture<Acknowledge> future = new
CompletableFuture<>();
- eventsForTask.add(new BlockedEvent(event, subtask, future));
- return future;
+ public void sendEvent(
+ SerializedValue<OperatorEvent> event,
+ int subtask,
+ CompletableFuture<Acknowledge> result) {
+ checkRunsInMainThread();
+
+ if (!shut) {
+ final CompletableFuture<Acknowledge> ack =
eventSender.apply(event, subtask);
Review comment:
I am not entirely sure about this. The problem in my opinion is that
`Execution` is not thread-safe. `sendOperatorEvent` accesses internal state of
the `Execution` which is set by another thread (`assignedResource` and
`getState()`). Since there is no synchronization it is not guaranteed that all
changes are visible.
Moreover, it breaks a bit with the existing threading model of the
`ExecutionGraph` where we said that there is only a single thread accessing it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
