HBASE-22784 OldWALs not cleared in a replication slave cluster (cyclic replication bw 2 clusters)

Signed-off-by: Andrew Purtell <apurtell@apache.org>

Author: wchevreuil

hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSource.java

@@ -783,6 +783,7 @@ protected void shipEdits(WALEntryBatch entryBatch) {
     }
 
     private void updateLogPosition(long lastReadPosition) {
+      manager.setPendingShipment(false);
       manager.logPositionAndCleanOldLogs(currentPath, peerClusterZnode, lastReadPosition,
         this.replicationQueueInfo.isQueueRecovered(), false);
       lastLoggedPosition = lastReadPosition;

hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceManager.java

@@ -121,9 +121,10 @@ public class ReplicationSourceManager implements ReplicationListener {
   private final Random rand;
   private final boolean replicationForBulkLoadDataEnabled;
 
-
   private AtomicLong totalBufferUsed = new AtomicLong();
 
+  private boolean pendingShipment;
+
   /**
    * Creates a replication manager and sets the watch on all the other registered region servers
    * @param replicationQueues the interface for manipulating replication queues
@@ -189,14 +190,20 @@ public ReplicationSourceManager(final ReplicationQueues replicationQueues,
    * @param queueRecovered indicates if this queue comes from another region server
    * @param holdLogInZK if true then the log is retained in ZK
    */
-  public void logPositionAndCleanOldLogs(Path log, String id, long position,
+  public synchronized void logPositionAndCleanOldLogs(Path log, String id, long position,
       boolean queueRecovered, boolean holdLogInZK) {
-    String fileName = log.getName();
-    this.replicationQueues.setLogPosition(id, fileName, position);
-    if (holdLogInZK) {
-     return;
+    if (!this.pendingShipment) {
+      String fileName = log.getName();
+      this.replicationQueues.setLogPosition(id, fileName, position);
+      if (holdLogInZK) {
+        return;
+      }
+      cleanOldLogs(fileName, id, queueRecovered);
     }
-    cleanOldLogs(fileName, id, queueRecovered);
+  }
+
+  public synchronized void setPendingShipment(boolean pendingShipment) {
+    this.pendingShipment = pendingShipment;
   }
 
   /**
@@ -209,9 +216,12 @@ public void logPositionAndCleanOldLogs(Path log, String id, long position,
   public void cleanOldLogs(String key, String id, boolean queueRecovered) {
     String logPrefix = DefaultWALProvider.getWALPrefixFromWALName(key);
     if (queueRecovered) {
-      SortedSet<String> wals = walsByIdRecoveredQueues.get(id).get(logPrefix);
-      if (wals != null && !wals.first().equals(key)) {
-        cleanOldLogs(wals, key, id);
+      Map<String, SortedSet<String>> walsForPeer = walsByIdRecoveredQueues.get(id);
+      if(walsForPeer != null) {
+        SortedSet<String> wals = walsForPeer.get(logPrefix);
+        if (wals != null && !wals.first().equals(key)) {
+          cleanOldLogs(wals, key, id);
+        }
       }
     } else {
       synchronized (this.walsById) {

hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceWALReaderThread.java

@@ -81,6 +81,8 @@ public class ReplicationSourceWALReaderThread extends Thread {
   private AtomicLong totalBufferUsed;
   private long totalBufferQuota;
 
+  private ReplicationSourceManager replicationSourceManager;
+
   /**
    * Creates a reader worker for a given WAL queue. Reads WAL entries off a given queue, batches the
    * entries, and puts them on a batch queue.
@@ -109,6 +111,7 @@ public ReplicationSourceWALReaderThread(ReplicationSourceManager manager,
     // memory used will be batchSizeCapacity * (nb.batches + 1)
     // the +1 is for the current thread reading before placing onto the queue
     int batchCount = conf.getInt("replication.source.nb.batches", 1);
+    this.replicationSourceManager = manager;
     this.totalBufferUsed = manager.getTotalBufferUsed();
     this.totalBufferQuota = conf.getLong(HConstants.REPLICATION_SOURCE_TOTAL_BUFFER_KEY,
       HConstants.REPLICATION_SOURCE_TOTAL_BUFFER_DFAULT);
@@ -148,6 +151,7 @@ public void run() {
                 long entrySize = getEntrySizeIncludeBulkLoad(entry);
                 long entrySizeExlucdeBulkLoad = getEntrySizeExcludeBulkLoad(entry);
                 batch.addEntry(entry);
+                replicationSourceManager.setPendingShipment(true);
                 updateBatchStats(batch, entry, entryStream.getPosition(), entrySize);
                 boolean totalBufferTooLarge = acquireBufferQuota(entrySizeExlucdeBulkLoad);
                 // Stop if too many entries or too big
@@ -156,6 +160,11 @@ public void run() {
                   break;
                 }
               }
+            } else {
+              replicationSourceManager.logPositionAndCleanOldLogs(entryStream.getCurrentPath(),
+                this.replicationQueueInfo.getPeerClusterZnode(),
+                entryStream.getPosition(),
+                this.replicationQueueInfo.isQueueRecovered(), false);
             }
           }
           if (batch != null && (!batch.getLastSeqIds().isEmpty() || batch.getNbEntries() > 0)) {

hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestWALEntryStream.java

@@ -25,6 +25,11 @@
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
+import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.anyBoolean;
+import static org.mockito.Matchers.anyString;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
 import java.io.IOException;
@@ -72,6 +77,7 @@
 import org.junit.experimental.categories.Category;
 import org.junit.rules.TestName;
 import org.junit.runner.RunWith;
+import org.mockito.ArgumentCaptor;
 import org.mockito.Mockito;
 import org.mockito.runners.MockitoJUnitRunner;
 
@@ -371,6 +377,40 @@ public void testReplicationSourceWALReaderThread() throws Exception {
     assertEquals(getRow(entryBatch.getWalEntries().get(0)), "foo");
   }
 
+  @Test
+  public void testReplicationSourceUpdatesLogPositionOnFilteredEntries() throws Exception {
+    appendEntriesToLog(3);
+    // get ending position
+    long position;
+    try (WALEntryStream entryStream =
+      new WALEntryStream(walQueue, fs, conf, new MetricsSource("1"))) {
+      entryStream.next();
+      entryStream.next();
+      entryStream.next();
+      position = entryStream.getPosition();
+    }
+    // start up a readerThread with a WALEntryFilter that always filter the entries
+    ReplicationSourceManager mockSourceManager = Mockito.mock(ReplicationSourceManager.class);
+    when(mockSourceManager.getTotalBufferUsed()).thenReturn(new AtomicLong(0));
+    ReplicationSourceWALReaderThread readerThread = new ReplicationSourceWALReaderThread(
+      mockSourceManager, getQueueInfo(), walQueue, 0, fs, conf, new WALEntryFilter() {
+        @Override
+        public Entry filter(Entry entry) {
+          return null;
+        }
+      }, new MetricsSource("1"));
+    readerThread.start();
+    Thread.sleep(100);
+    ArgumentCaptor<Long> positionCaptor = ArgumentCaptor.forClass(Long.class);
+    verify(mockSourceManager, times(3))
+      .logPositionAndCleanOldLogs(any(Path.class),
+        anyString(),
+        positionCaptor.capture(),
+        anyBoolean(),
+        anyBoolean());
+    assertEquals(position, positionCaptor.getValue().longValue());
+  }
+
   @Test
   public void testWALKeySerialization() throws Exception {
     Map<String, byte[]> attributes = new HashMap<String, byte[]>();