pnowojski commented on a change in pull request #11507: [FLINK-16587] Add basic
CheckpointBarrierHandler for unaligned checkpoint
URL: https://github.com/apache/flink/pull/11507#discussion_r399244373
##########
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteInputChannel.java
##########
@@ -514,20 +532,29 @@ public void onBuffer(Buffer buffer, int sequenceNumber,
int backlog) throws IOEx
try {
- final boolean wasEmpty;
- synchronized (receivedBuffers) {
- // Similar to notifyBufferAvailable(), make
sure that we never add a buffer
- // after releaseAllResources() released all
buffers from receivedBuffers
- // (see above for details).
- if (isReleased.get()) {
- return;
- }
+ // Similar to notifyBufferAvailable(), make sure that
we never add a buffer
+ // after releaseAllResources() released all buffers
from receivedBuffers
+ // (see above for details).
+ if (isReleased.get()) {
+ return;
+ }
- if (expectedSequenceNumber != sequenceNumber) {
- onError(new
BufferReorderingException(expectedSequenceNumber, sequenceNumber));
- return;
+ if (expectedSequenceNumber != sequenceNumber) {
+ onError(new
BufferReorderingException(expectedSequenceNumber, sequenceNumber));
+ return;
+ }
+
+ if (inputGate.bufferReceivedListener != null) {
+ CheckpointBarrier checkpointBarrier =
parseCheckpointBarrier(buffer);
+ if (checkpointBarrier == null) {
+
inputGate.bufferReceivedListener.notifyBufferReceived(buffer, channelInfo);
+ } else {
+
inputGate.bufferReceivedListener.notifyBarrierReceived(checkpointBarrier,
channelInfo);
}
+ }
+ final boolean wasEmpty;
+ synchronized (receivedBuffers) {
Review comment:
Do we even need this commit? What is it's purpose (missing detailed commit
message)? I don't see how could it affect performance, as I would expect:
- `isReleased.get()` check should be as fast inside the synchronized
section as outside of it. Actually I would hope that JIT would optimise both to
just one single memory barrier
- `onError` is not executed on the hot path
and it makes code a bit more "fragile"?
Or is there another motivation for it? Like avoiding some deadlocks ?
----------------------------------------------------------------
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:
us...@infra.apache.org
With regards,
Apache Git Services
