pnowojski commented on a change in pull request #10345: [FLINK-12484][runtime]
synchronize all mailbox actions
URL: https://github.com/apache/flink/pull/10345#discussion_r352522237
##########
File path:
flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java
##########
@@ -262,7 +262,7 @@ protected StreamTask(
this.accumulatorMap =
getEnvironment().getAccumulatorRegistry().getUserMap();
this.recordWriter = createRecordWriterDelegate(configuration,
environment);
this.executionDecorator =
Preconditions.checkNotNull(executionDecorator);
- this.mailboxProcessor = new
MailboxProcessor(this::processInput);
+ this.mailboxProcessor = new
MailboxProcessor(this::processInput, this.executionDecorator);
Review comment:
I think `invokeProcessingTimeCallback` was not addressed.
Performance for `performCheckpoint` will not matter, however double
acquiring the lock there I think is some kind of code smell?
`performCheckpoint` is used from two contexts:
1. `triggerCheckpointOnBarrier`
2. `triggerCheckpointAsync`
The latter is used as a mailbox action so whole code is now being executed
under the checkpoint lock anyway - for this case `performCheckpoint` can safely
drop the lock altogether.
For the first one, it's being executed from within the default action and
it's also executing `runSynchronousSavepointMailboxLoop`, which maybe shouldn't
be done under the `checkpointLock`.
Maybe it would be cleaner If we had two versions of `performCheckpoint`, one
safe acquiring the `checkpointLock` on it's own (used for
`triggerCheckpointOnBarrier`). Second unsafe (with `Unsafe` suffix), that
doesn't acquire the lock on its own (used in `triggerCheckpointAsync`)?
----------------------------------------------------------------
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]
With regards,
Apache Git Services
