diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/LocalInputChannel.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/LocalInputChannel.java
index 9a7aa4b96dfc15..fd995cc7cbe84a 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/LocalInputChannel.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/LocalInputChannel.java
@@ -24,7 +24,6 @@
 import org.apache.flink.runtime.checkpoint.CheckpointFailureReason;
 import org.apache.flink.runtime.checkpoint.channel.ChannelStateWriter;
 import org.apache.flink.runtime.checkpoint.channel.RecoveryCheckpointBarrier;
-import org.apache.flink.runtime.event.AbstractEvent;
 import org.apache.flink.runtime.event.TaskEvent;
 import org.apache.flink.runtime.execution.CancelTaskException;
 import org.apache.flink.runtime.io.network.TaskEventPublisher;
@@ -349,7 +348,8 @@ private List<Buffer> collectPreRecoveryBarrier(long checkpointId) throws IOExcep
             Iterator<Buffer> it = recoveredBuffers.iterator();
             while (it.hasNext()) {
                 Buffer b = it.next();
-                RecoveryCheckpointBarrier barrier = asRecoveryCheckpointBarrier(b);
+                RecoveryCheckpointBarrier barrier =
+                        RecoveryCheckpointBarrierUtils.asRecoveryCheckpointBarrier(b);
                 if (barrier != null) {
                     long barrierId = barrier.getCheckpointId();
                     if (barrierId == checkpointId) {
@@ -383,10 +383,10 @@ private List<Buffer> collectPreRecoveryBarrier(long checkpointId) throws IOExcep
                 }
             }
         } catch (IOException e) {
-            releaseRetainedBuffers(retained);
+            RecoveryCheckpointBarrierUtils.releaseRetainedBuffers(retained);
             throw e;
         }
-        releaseRetainedBuffers(retained);
+        RecoveryCheckpointBarrierUtils.releaseRetainedBuffers(retained);
         throw new IOException(
                 "Missing RecoveryCheckpointBarrier for checkpoint "
                         + checkpointId
@@ -394,26 +394,6 @@ private List<Buffer> collectPreRecoveryBarrier(long checkpointId) throws IOExcep
                         + getChannelInfo());
     }
 
-    private static void releaseRetainedBuffers(List<Buffer> retained) {
-        for (Buffer buffer : retained) {
-            buffer.recycleBuffer();
-        }
-    }
-
-    @Nullable
-    private static RecoveryCheckpointBarrier asRecoveryCheckpointBarrier(Buffer b)
-            throws IOException {
-        if (b.isBuffer()) {
-            return null;
-        }
-        AbstractEvent event =
-                EventSerializer.fromBuffer(b, RecoveryCheckpointBarrier.class.getClassLoader());
-        b.setReaderIndex(0);
-        return event instanceof RecoveryCheckpointBarrier
-                ? (RecoveryCheckpointBarrier) event
-                : null;
-    }
-
     // ------------------------------------------------------------------------
     // Consume
     // ------------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RecoveryCheckpointBarrierUtils.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RecoveryCheckpointBarrierUtils.java
new file mode 100644
index 00000000000000..31c83f9c9a7415
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RecoveryCheckpointBarrierUtils.java
@@ -0,0 +1,57 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.io.network.partition.consumer;
+
+import org.apache.flink.runtime.checkpoint.channel.RecoveryCheckpointBarrier;
+import org.apache.flink.runtime.event.AbstractEvent;
+import org.apache.flink.runtime.io.network.api.serialization.EventSerializer;
+import org.apache.flink.runtime.io.network.buffer.Buffer;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+import java.util.List;
+
+/**
+ * Shared stateless helpers for the checkpointing-during-recovery sentinel protocol, used by both
+ * {@link LocalInputChannel} and {@link RemoteInputChannel}.
+ */
+final class RecoveryCheckpointBarrierUtils {
+
+    private RecoveryCheckpointBarrierUtils() {}
+
+    static void releaseRetainedBuffers(List<Buffer> retained) {
+        for (Buffer buffer : retained) {
+            buffer.recycleBuffer();
+        }
+    }
+
+    @Nullable
+    static RecoveryCheckpointBarrier asRecoveryCheckpointBarrier(Buffer b) throws IOException {
+        if (b.isBuffer()) {
+            return null;
+        }
+        AbstractEvent event =
+                EventSerializer.fromBuffer(b, RecoveryCheckpointBarrier.class.getClassLoader());
+        b.setReaderIndex(0);
+        return event instanceof RecoveryCheckpointBarrier
+                ? (RecoveryCheckpointBarrier) event
+                : null;
+    }
+}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteInputChannel.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteInputChannel.java
index 8b9241e19da968..c22ba964ae19a7 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteInputChannel.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteInputChannel.java
@@ -986,7 +986,8 @@ private List<Buffer> collectPreRecoveryBarrier(long checkpointId) throws IOExcep
             Iterators.advance(it, receivedBuffers.getNumPriorityElements());
             while (it.hasNext()) {
                 SequenceBuffer sb = it.next();
-                RecoveryCheckpointBarrier barrier = asRecoveryCheckpointBarrier(sb.buffer);
+                RecoveryCheckpointBarrier barrier =
+                        RecoveryCheckpointBarrierUtils.asRecoveryCheckpointBarrier(sb.buffer);
                 if (barrier != null) {
                     long barrierId = barrier.getCheckpointId();
                     if (barrierId == checkpointId) {
@@ -1020,11 +1021,11 @@ private List<Buffer> collectPreRecoveryBarrier(long checkpointId) throws IOExcep
                 }
             }
         } catch (IOException e) {
-            releaseRetainedBuffers(retained);
+            RecoveryCheckpointBarrierUtils.releaseRetainedBuffers(retained);
             throw e;
         }
         if (sentinel == null) {
-            releaseRetainedBuffers(retained);
+            RecoveryCheckpointBarrierUtils.releaseRetainedBuffers(retained);
             throw new IOException(
                     "Missing RecoveryCheckpointBarrier for checkpoint "
                             + checkpointId
@@ -1047,26 +1048,6 @@ private void removeRecoverySentinel(SequenceBuffer sentinel) {
         sentinel.buffer.recycleBuffer();
     }
 
-    private static void releaseRetainedBuffers(List<Buffer> retained) {
-        for (Buffer buffer : retained) {
-            buffer.recycleBuffer();
-        }
-    }
-
-    @Nullable
-    private static RecoveryCheckpointBarrier asRecoveryCheckpointBarrier(Buffer b)
-            throws IOException {
-        if (b.isBuffer()) {
-            return null;
-        }
-        AbstractEvent event =
-                EventSerializer.fromBuffer(b, RecoveryCheckpointBarrier.class.getClassLoader());
-        b.setReaderIndex(0);
-        return event instanceof RecoveryCheckpointBarrier
-                ? (RecoveryCheckpointBarrier) event
-                : null;
-    }
-
     public void checkpointStopped(long checkpointId) {
         synchronized (receivedBuffers) {
             channelStatePersister.stopPersisting(checkpointId);