[FLINK-18063][checkpointing][refactoring] Implement default #isBlocke…

…d method in CheckpointBarrierHandler Simplify the implementations of CheckpointBarrierTracker and CheckpointBarrierUnaligner to reuse the parent default implementation. This closes apache#12460.
ruslandanilin · Jun 9, 2020 · 0cbd5e3 · 0cbd5e3
1 parent bb0ebca
commit 0cbd5e3
Show file tree

Hide file tree

Showing 3 changed files with 3 additions and 14 deletions.
diff --git a/...ng-java/src/main/java/org/apache/flink/streaming/runtime/io/CheckpointBarrierHandler.java b/...ng-java/src/main/java/org/apache/flink/streaming/runtime/io/CheckpointBarrierHandler.java
@@ -60,7 +60,9 @@ public CheckpointBarrierHandler(AbstractInvokable toNotifyOnCheckpoint) {
 	 * @param channelIndex The channel index to check.
 	 * @return True if the channel is blocked, false if not.
 	 */
-	public abstract boolean isBlocked(int channelIndex);
+	public boolean isBlocked(int channelIndex) {
+		return false;
+	}
 
 	@Override
 	public void close() throws IOException {

diff --git a/...ng-java/src/main/java/org/apache/flink/streaming/runtime/io/CheckpointBarrierTracker.java b/...ng-java/src/main/java/org/apache/flink/streaming/runtime/io/CheckpointBarrierTracker.java
@@ -79,11 +79,6 @@ public CheckpointBarrierTracker(int totalNumberOfInputChannels, AbstractInvokabl
 	public void releaseBlocksAndResetBarriers() {
 	}
 
-	@Override
-	public boolean isBlocked(int channelIndex) {
-		return false;
-	}
-
 	@Override
 	public void processBarrier(CheckpointBarrier receivedBarrier, int channelIndex) throws Exception {
 		final long barrierId = receivedBarrier.getId();

diff --git a/...-java/src/main/java/org/apache/flink/streaming/runtime/io/CheckpointBarrierUnaligner.java b/...-java/src/main/java/org/apache/flink/streaming/runtime/io/CheckpointBarrierUnaligner.java
@@ -129,14 +129,6 @@ public void releaseBlocksAndResetBarriers() {
 		threadSafeUnaligner.resetReceivedBarriers(currentConsumedCheckpointId);
 	}
 
-	/**
-	 * For unaligned checkpoint, it never blocks processing from the task aspect.
-	 */
-	@Override
-	public boolean isBlocked(int channelIndex) {
-		return false;
-	}
-
 	/**
 	 * We still need to trigger checkpoint via {@link ThreadSafeUnaligner#notifyBarrierReceived(CheckpointBarrier, InputChannelInfo)}
 	 * while reading the first barrier from one channel, because this might happen