diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceShipper.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceShipper.java
index ee68c38fcb22..b99b6451c85b 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceShipper.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceShipper.java
@@ -213,6 +213,7 @@ void shipEdits(WALEntryBatch entryBatch) throws IOException {
       persistLogPosition();
       return;
     }
+    boolean bufferReleased = false;
     while (isActive()) {
       try {
         try {
@@ -253,6 +254,7 @@ void shipEdits(WALEntryBatch entryBatch) throws IOException {
         // acquireBufferQuota() in ReplicationSourceWALReader because they maintain
         // same variable: totalBufferUsed
         source.postShipEdits(entries, entryBatch.getUsedBufferSize());
+        bufferReleased = true;
         // FIXME check relationship between wal group and overall
         source.getSourceMetrics().shipBatch(entryBatch.getNbOperations(), currentSize,
           entryBatch.getNbHFiles());
@@ -277,12 +279,14 @@ void shipEdits(WALEntryBatch entryBatch) throws IOException {
         }
       }
     }
-
     accumulatedSizeSinceLastUpdate += currentSize;
     lastShippedBatch = entryBatch;
     if (shouldPersistLogPosition()) {
       persistLogPosition();
     }
+    if (!bufferReleased) {
+      source.getSourceManager().releaseWALEntryBatchBufferQuota(entryBatch);
+    }
   }
 
   private boolean shouldPersistLogPosition() {
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestReplicationSource.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestReplicationSource.java
index 468ff3940fcf..e01836d4626e 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestReplicationSource.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestReplicationSource.java
@@ -311,6 +311,46 @@ public void testTerminateClearsBuffer() throws Exception {
     assertEquals(0, source.getSourceManager().getTotalBufferUsed());
   }
 
+  @Test
+  public void testStoppedShipperClearsInflightBuffer() throws Exception {
+    ReplicationSource source = new ReplicationSource();
+    ReplicationSourceManager sourceManager =
+      new ReplicationSourceManager(null, null, conf, null, null, null, null, null, null, null,
+        mock(MetricsReplicationGlobalSourceSource.class));
+    ReplicationPeer mockPeer = mock(ReplicationPeer.class);
+    Mockito.when(mockPeer.getPeerBandwidth()).thenReturn(0L);
+    ReplicationQueueId queueId =
+      new ReplicationQueueId(ServerName.valueOf("test,123,123"), "testPeer");
+    source.init(HBaseConfiguration.create(), null, sourceManager, null, mockPeer,
+      Mockito.mock(Server.class), new ReplicationQueueData(queueId, ImmutableMap.of()), null,
+      p -> OptionalLong.empty(),
+      mock(MetricsSource.class));
+    ReplicationSourceWALReader reader = new ReplicationSourceWALReader(null,
+      conf, null, 0, null, source, null);
+    ReplicationSourceShipper shipper = new ReplicationSourceShipper(conf, null, source, reader);
+    WALEntryBatch batch = new WALEntryBatch(10, logDir);
+    WAL.Entry mockEntry = mock(WAL.Entry.class);
+    WALEdit mockEdit = mock(WALEdit.class);
+    WALKeyImpl mockKey = mock(WALKeyImpl.class);
+    when(mockEntry.getEdit()).thenReturn(mockEdit);
+    when(mockEdit.isEmpty()).thenReturn(false);
+    when(mockEntry.getKey()).thenReturn(mockKey);
+    when(mockKey.estimatedSerializedSizeOf()).thenReturn(1000L);
+    when(mockEdit.heapSize()).thenReturn(10000L);
+    when(mockEdit.size()).thenReturn(0);
+    ArrayList<Cell> cells = new ArrayList<>();
+    KeyValue kv = new KeyValue(Bytes.toBytes("0001"), Bytes.toBytes("f"),
+      Bytes.toBytes("1"), Bytes.toBytes("v1"));
+    cells.add(kv);
+    when(mockEdit.getCells()).thenReturn(cells);
+    reader.addEntryToBatch(batch, mockEntry);
+    assertEquals(11000, source.getSourceManager().getTotalBufferUsed());
+    shipper.setWorkerState(ReplicationSourceShipper.WorkerState.RUNNING);
+    shipper.stopWorker();
+    shipper.shipEdits(batch);
+    assertEquals(0, source.getSourceManager().getTotalBufferUsed());
+  }
+
   /**
    * Tests that recovered queues are preserved on a regionserver shutdown. See HBASE-18192
    */