[GitHub] [flink] AHeise commented on a change in pull request #13228: [FLINK-19026][network] Improve threading model of CheckpointBarrierUnaligner

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

AHeise commented on a change in pull request #13228:
URL: https://github.com/apache/flink/pull/13228#discussion_r476657231



##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/PipelinedSubpartition.java
##########
@@ -171,19 +175,42 @@ private void handleAddingBarrier(BufferConsumer bufferConsumer, boolean insertAs
 			checkState(inflightBufferSnapshot.isEmpty(), "Supporting only one concurrent checkpoint in unaligned " +
 				"checkpoints");
 
-			// Meanwhile prepare the collection of in-flight buffers which would be fetched in the next step later.
-			for (BufferConsumer buffer : buffers) {
-				try (BufferConsumer bc = buffer.copy()) {
-					if (bc.isBuffer()) {
-						inflightBufferSnapshot.add(bc.build());
+			final int pos = buffers.getNumPriorityElements();
+			buffers.addPriorityElement(bufferConsumer);
+
+			boolean unalignedCheckpoint = isUnalignedCheckpoint(bufferConsumer);
+			if (unalignedCheckpoint) {
+				final Iterator<BufferConsumer> iterator = buffers.iterator();
+				Iterators.advance(iterator, pos + 1);
+				while (iterator.hasNext()) {
+					BufferConsumer buffer = iterator.next();
+
+					if (buffer.isBuffer()) {
+						try (BufferConsumer bc = buffer.copy()) {
+							inflightBufferSnapshot.add(bc.build());
+						}
 					}
 				}
 			}
+			return;
+		}
+		buffers.add(bufferConsumer);

Review comment:
       Yes, good idea. In general that change looks a bit odd, because it's isolated from the upcoming changes (I had to split somewhere and probably didn't hit the sweet spot everywhere).




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