StephanEwen commented on a change in pull request #15557:
URL: https://github.com/apache/flink/pull/15557#discussion_r613167120
##########
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) {
Review comment:
I am not sure we can do this here, because when the valve is shut, the
event and future are stored in a list, and the completion sending (and future
completion) is separately triggered later when the valve is opened again.
I would prefer to look at beautifying the future passing/returning after the
last PR of the sequence is done, because this is one part where things build
upon each other and tightly rely on each other. To change this part, we'd also
need to change other parts and I would rather do that jointly in one commit at
the end.
--
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]
