[GitHub] [flink] AHeise commented on a change in pull request #11507: [FLINK-16587] Add basic CheckpointBarrierHandler for unaligned checkpoint

[GitBox <gi...@apache.org>]

AHeise 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_r402256150
 
 

 ##########
 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:
   Removed the commit. It had some purpose but it got negated in the meantime. 

----------------------------------------------------------------
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:
users@infra.apache.org


With regards,
Apache Git Services