HBASE-26120 New replication gets stuck or data loss when multiwal groups more than 10 (#3528)

Signed-off-by: Andrew Purtell <[email protected]>
Signed-off-by: Michael Stack <[email protected]>

Conflicts:
	hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceManager.java
	hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestReplicationSourceManager.java

Author: Apache9

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

@@ -19,14 +19,13 @@
 
 package org.apache.hadoop.hbase.replication.regionserver;
 
+import com.google.common.collect.Sets;
 import com.google.common.util.concurrent.ThreadFactoryBuilder;
 
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.HashMap;
-import java.util.HashSet;
-import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 import java.util.Random;
@@ -108,7 +107,7 @@ public class ReplicationSourceManager implements ReplicationListener {
   private final Configuration conf;
   private final FileSystem fs;
   // The paths to the latest log of each wal group, for new coming peers
-  private Set<Path> latestPaths;
+  private final Map<String, Path> latestPaths;
   // Path to the wals directories
   private final Path logDir;
   // Path to the wal archive
@@ -171,7 +170,7 @@ public ReplicationSourceManager(final ReplicationQueues replicationQueues,
     tfb.setDaemon(true);
     this.executor.setThreadFactory(tfb.build());
     this.rand = new Random();
-    this.latestPaths = Collections.synchronizedSet(new HashSet<Path>());
+    this.latestPaths = new HashMap<>();
     replicationForBulkLoadDataEnabled =
         conf.getBoolean(HConstants.REPLICATION_BULKLOAD_ENABLE_KEY,
           HConstants.REPLICATION_BULKLOAD_ENABLE_DEFAULT);
@@ -305,23 +304,22 @@ protected ReplicationSourceInterface addSource(String id) throws IOException,
       this.walsById.put(id, walsByGroup);
       // Add the latest wal to that source's queue
       synchronized (latestPaths) {
-        if (this.latestPaths.size() > 0) {
-          for (Path logPath : latestPaths) {
-            String name = logPath.getName();
-            String walPrefix = DefaultWALProvider.getWALPrefixFromWALName(name);
-            SortedSet<String> logs = new TreeSet<String>();
-            logs.add(name);
-            walsByGroup.put(walPrefix, logs);
+        if (!latestPaths.isEmpty()) {
+          for (Map.Entry<String, Path> walPrefixAndPath : latestPaths.entrySet()) {
+            Path walPath = walPrefixAndPath.getValue();
+            SortedSet<String> wals = new TreeSet<>();
+            wals.add(walPath.getName());
+            walsByGroup.put(walPrefixAndPath.getKey(), wals);
             try {
-              this.replicationQueues.addLog(id, name);
+              this.replicationQueues.addLog(id, walPath.getName());
             } catch (ReplicationException e) {
               String message =
                   "Cannot add log to queue when creating a new source, queueId=" + id
-                      + ", filename=" + name;
+                      + ", filename=" + walPath.getName();
               server.stop(message);
               throw e;
             }
-            src.enqueueLog(logPath);
+            src.enqueueLog(walPath);
           }
         }
       }
@@ -409,15 +407,7 @@ void preLogRoll(Path newLog) throws IOException {
     String logName = newLog.getName();
     String logPrefix = DefaultWALProvider.getWALPrefixFromWALName(logName);
     synchronized (latestPaths) {
-      Iterator<Path> iterator = latestPaths.iterator();
-      while (iterator.hasNext()) {
-        Path path = iterator.next();
-        if (path.getName().contains(logPrefix)) {
-          iterator.remove();
-          break;
-        }
-      }
-      this.latestPaths.add(newLog);
+      latestPaths.put(logPrefix, newLog);
     }
   }
 
@@ -693,6 +683,12 @@ public void peerListChanged(List<String> peerIds) {
     }
   }
 
+  Set<Path> getLastestPath() {
+    synchronized (latestPaths) {
+      return Sets.newHashSet(latestPaths.values());
+    }
+  }
+
   /**
    * Class responsible to setup new ReplicationSources to take care of the
    * queues from dead region servers.

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

@@ -34,6 +34,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.NavigableMap;
+import java.util.Set;
 import java.util.SortedMap;
 import java.util.SortedSet;
 import java.util.TreeMap;
@@ -509,6 +510,20 @@ private void removePeerAndWait(final String peerId) throws Exception {
     });
   }
 
+  @Test
+  public void testSameWALPrefix() throws IOException {
+    Set<Path> latestWalsBefore = manager.getLastestPath();
+    Path walName1 = new Path("localhost,8080,12345-45678-Peer.34567");
+    Path walName2 = new Path("localhost,8080,12345.56789");
+    manager.preLogRoll(walName1);
+    manager.preLogRoll(walName2);
+    Set<Path> latestWals = manager.getLastestPath();
+    latestWals.removeAll(latestWalsBefore);
+    assertEquals(2, latestWals.size());
+    assertTrue(latestWals.contains(walName1));
+    assertTrue(latestWals.contains(walName2));
+  }
+
   private WALEdit getBulkLoadWALEdit() {
     // 1. Create store files for the families
     Map<byte[], List<Path>> storeFiles = new HashMap<>(1);