HDFS-17223. Add journalnode maintenance node list

Author: 顾鹏

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSConfigKeys.java

@@ -1466,6 +1466,10 @@ public class DFSConfigKeys extends CommonConfigurationKeys {
           "dfs.journalnode.edit-cache-size.fraction";
   public static final float DFS_JOURNALNODE_EDIT_CACHE_SIZE_FRACTION_DEFAULT = 0.5f;
 
+  public static final String  DFS_JOURNALNODE_MAINTENANCE_NODES_KEY =
+      "dfs.journalnode.maintenance.nodes";
+  public static final String[]  DFS_JOURNALNODE_MAINTENANCE_NODES_DEFAULT = {};
+
   // Journal-node related configs for the client side.
   public static final String  DFS_QJOURNAL_QUEUE_SIZE_LIMIT_KEY = "dfs.qjournal.queued-edits.limit.mb";
   public static final int     DFS_QJOURNAL_QUEUE_SIZE_LIMIT_DEFAULT = 10;

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSUtil.java

@@ -70,6 +70,7 @@
 import org.apache.hadoop.classification.InterfaceStability;
 import org.apache.hadoop.fs.ParentNotDirectoryException;
 import org.apache.hadoop.fs.UnresolvedLinkException;
+import org.apache.hadoop.hdfs.server.blockmanagement.HostSet;
 import org.apache.hadoop.hdfs.server.datanode.metrics.DataNodeMetrics;
 import org.apache.hadoop.hdfs.server.namenode.FSDirectory;
 import org.apache.hadoop.hdfs.server.namenode.INodesInPath;
@@ -1982,4 +1983,27 @@ public static void addTransferRateMetric(final DataNodeMetrics metrics, final lo
       LOG.warn("Unexpected value for data transfer bytes={} duration={}", read, duration);
     }
   }
+
+  /**
+   * Retrieve InetSocketAddress array by ip port string array.
+   * @param nodesHostPort ip port string array
+   */
+  public static HostSet getInetSocketAddress(String[] nodesHostPort) {
+    HostSet retSet = new HostSet();
+    for (String hostPort : nodesHostPort) {
+      try {
+        URI uri = new URI("dummy", hostPort, null, null, null);
+        int port = uri.getPort() == -1 ? 0 : uri.getPort();
+        InetSocketAddress inetSocketAddress = new InetSocketAddress(uri.getHost(), port);
+        if (inetSocketAddress.isUnresolved()) {
+          LOG.warn(String.format("Failed to resolve address `%s`", hostPort));
+          continue;
+        }
+        retSet.add(inetSocketAddress);
+      } catch (URISyntaxException e) {
+        LOG.warn(String.format("Failed to parse `%s`", hostPort));
+      }
+    }
+    return retSet;
+  }
 }

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/qjournal/client/AsyncLoggerSet.java

@@ -53,9 +53,16 @@ class AsyncLoggerSet {
 
   private static final long INVALID_EPOCH = -1;
   private long myEpoch = INVALID_EPOCH;
+  private final int majoritySize;
 
   public AsyncLoggerSet(List<AsyncLogger> loggers) {
     this.loggers = ImmutableList.copyOf(loggers);
+    this.majoritySize = loggers.size() / 2 + 1;
+  }
+
+  public AsyncLoggerSet(List<AsyncLogger> loggers, int quorumJournalCount) {
+    this.loggers = ImmutableList.copyOf(loggers);
+    this.majoritySize = quorumJournalCount / 2 + 1;
   }
 
   void setEpoch(long e) {
@@ -151,7 +158,7 @@ <V> Map<AsyncLogger, V> waitForWriteQuorum(QuorumCall<AsyncLogger, V> q,
    * @return the number of nodes which are required to obtain a quorum.
    */
   int getMajoritySize() {
-    return loggers.size() / 2 + 1;
+    return this.majoritySize;
   }
 
   /**

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/qjournal/client/QuorumJournalManager.java

@@ -17,6 +17,9 @@
  */
 package org.apache.hadoop.hdfs.qjournal.client;
 
+import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_JOURNALNODE_MAINTENANCE_NODES_DEFAULT;
+import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_JOURNALNODE_MAINTENANCE_NODES_KEY;
+
 import java.io.IOException;
 import java.net.InetSocketAddress;
 import java.net.URI;
@@ -31,6 +34,7 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 
+import org.apache.hadoop.hdfs.server.blockmanagement.HostSet;
 import org.apache.hadoop.util.Lists;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -62,6 +66,7 @@
 import org.apache.hadoop.classification.VisibleForTesting;
 import org.apache.hadoop.thirdparty.com.google.common.base.Joiner;
 import org.apache.hadoop.util.Preconditions;
+
 import org.apache.hadoop.thirdparty.protobuf.TextFormat;
 
 /**
@@ -108,6 +113,7 @@ public class QuorumJournalManager implements JournalManager {
   private static final int OUTPUT_BUFFER_CAPACITY_DEFAULT = 512 * 1024;
   private int outputBufferCapacity;
   private final URLConnectionFactory connectionFactory;
+  private int quorumJournalCount;
 
   /** Limit logging about input stream selection to every 5 seconds max. */
   private static final long SELECT_INPUT_STREAM_LOG_INTERVAL_MS = 5000;
@@ -144,7 +150,14 @@ public QuorumJournalManager(Configuration conf,
     this.uri = uri;
     this.nsInfo = nsInfo;
     this.nameServiceId = nameServiceId;
-    this.loggers = new AsyncLoggerSet(createLoggers(loggerFactory));
+    this.loggers = new AsyncLoggerSet(createLoggers(loggerFactory), this.quorumJournalCount);
+
+    // Check whether the number of jn maintenance lists is valid
+    int quorumThreshold = quorumJournalCount / 2 + 1;
+    Preconditions.checkArgument(
+        this.loggers.size() >= quorumThreshold,
+        "The total journalnode minus %s the number of blacklists must be greater than or equal to"
+            + " %s!", DFS_JOURNALNODE_MAINTENANCE_NODES_KEY, quorumThreshold);
 
     this.maxTxnsPerRpc =
         conf.getInt(QJM_RPC_MAX_TXNS_KEY, QJM_RPC_MAX_TXNS_DEFAULT);
@@ -250,6 +263,9 @@ Map<AsyncLogger, NewEpochResponseProto> createNewUniqueEpoch()
 
   @Override
   public void format(NamespaceInfo nsInfo, boolean force) throws IOException {
+    if (isEnableJnMaintenance()) {
+      throw new IOException("format() does not support enabling jn maintenance mode");
+    }
     QuorumCall<AsyncLogger, Void> call = loggers.format(nsInfo, force);
     try {
       call.waitFor(loggers.size(), loggers.size(), 0, timeoutMs,
@@ -406,21 +422,39 @@ private void recoverUnclosedSegment(long segmentTxId) throws IOException {
             logToSync.getStartTxId(),
             logToSync.getEndTxId()));
   }
-  
-  static List<AsyncLogger> createLoggers(Configuration conf,
+
+  List<AsyncLogger> createLoggers(Configuration conf,
+                                  URI uri,
+                                  NamespaceInfo nsInfo,
+                                  AsyncLogger.Factory factory,
+                                  String nameServiceId)
+      throws IOException {
+    String[] skipNodesHostPort = conf.getTrimmedStrings(
+        DFS_JOURNALNODE_MAINTENANCE_NODES_KEY, DFS_JOURNALNODE_MAINTENANCE_NODES_DEFAULT);
+    return createLoggers(conf, uri, nsInfo, factory, nameServiceId, skipNodesHostPort);
+  }
+
+  private List<AsyncLogger> createLoggers(Configuration conf,
                                          URI uri,
                                          NamespaceInfo nsInfo,
                                          AsyncLogger.Factory factory,
-                                         String nameServiceId)
+                                         String nameServiceId,
+                                         String[] skipNodesHostPort)
       throws IOException {
     List<AsyncLogger> ret = Lists.newArrayList();
     List<InetSocketAddress> addrs = Util.getAddressesList(uri, conf);
     if (addrs.size() % 2 == 0) {
       LOG.warn("Quorum journal URI '" + uri + "' has an even number " +
           "of Journal Nodes specified. This is not recommended!");
     }
+    this.quorumJournalCount = addrs.size();
+    HostSet skipSet = DFSUtil.getInetSocketAddress(skipNodesHostPort);
     String jid = parseJournalId(uri);
     for (InetSocketAddress addr : addrs) {
+      if(skipSet.match(addr)) {
+        LOG.info("The node {} is a maintenance node and will skip initialization.", addr);
+        continue;
+      }
       ret.add(factory.createLogger(conf, nsInfo, jid, nameServiceId, addr));
     }
     return ret;
@@ -667,6 +701,9 @@ AsyncLoggerSet getLoggerSetForTests() {
 
   @Override
   public void doPreUpgrade() throws IOException {
+    if (isEnableJnMaintenance()) {
+      throw new IOException("doPreUpgrade() does not support enabling jn maintenance mode");
+    }
     QuorumCall<AsyncLogger, Void> call = loggers.doPreUpgrade();
     try {
       call.waitFor(loggers.size(), loggers.size(), 0, timeoutMs,
@@ -684,6 +721,9 @@ public void doPreUpgrade() throws IOException {
 
   @Override
   public void doUpgrade(Storage storage) throws IOException {
+    if (isEnableJnMaintenance()) {
+      throw new IOException("doUpgrade() does not support enabling jn maintenance mode");
+    }
     QuorumCall<AsyncLogger, Void> call = loggers.doUpgrade(storage);
     try {
       call.waitFor(loggers.size(), loggers.size(), 0, timeoutMs,
@@ -701,6 +741,9 @@ public void doUpgrade(Storage storage) throws IOException {
 
   @Override
   public void doFinalize() throws IOException {
+    if (isEnableJnMaintenance()) {
+      throw new IOException("doFinalize() does not support enabling jn maintenance mode");
+    }
     QuorumCall<AsyncLogger, Void> call = loggers.doFinalize();
     try {
       call.waitFor(loggers.size(), loggers.size(), 0, timeoutMs,
@@ -719,6 +762,9 @@ public void doFinalize() throws IOException {
   @Override
   public boolean canRollBack(StorageInfo storage, StorageInfo prevStorage,
       int targetLayoutVersion) throws IOException {
+    if (isEnableJnMaintenance()) {
+      throw new IOException("canRollBack() does not support enabling jn maintenance mode");
+    }
     QuorumCall<AsyncLogger, Boolean> call = loggers.canRollBack(storage,
         prevStorage, targetLayoutVersion);
     try {
@@ -753,6 +799,9 @@ public boolean canRollBack(StorageInfo storage, StorageInfo prevStorage,
 
   @Override
   public void doRollback() throws IOException {
+    if (isEnableJnMaintenance()) {
+      throw new IOException("doRollback() does not support enabling jn maintenance mode");
+    }
     QuorumCall<AsyncLogger, Void> call = loggers.doRollback();
     try {
       call.waitFor(loggers.size(), loggers.size(), 0, timeoutMs,
@@ -770,6 +819,9 @@ public void doRollback() throws IOException {
 
   @Override
   public void discardSegments(long startTxId) throws IOException {
+    if (isEnableJnMaintenance()) {
+      throw new IOException("discardSegments() does not support enabling jn maintenance mode");
+    }
     QuorumCall<AsyncLogger, Void> call = loggers.discardSegments(startTxId);
     try {
       call.waitFor(loggers.size(), loggers.size(), 0,
@@ -789,6 +841,9 @@ public void discardSegments(long startTxId) throws IOException {
 
   @Override
   public long getJournalCTime() throws IOException {
+    if (isEnableJnMaintenance()) {
+      throw new IOException("getJournalCTime() does not support enabling jn maintenance mode");
+    }
     QuorumCall<AsyncLogger, Long> call = loggers.getJournalCTime();
     try {
       call.waitFor(loggers.size(), loggers.size(), 0,
@@ -819,4 +874,8 @@ public long getJournalCTime() throws IOException {
 
     throw new AssertionError("Unreachable code.");
   }
+
+  private boolean isEnableJnMaintenance() {
+    return this.loggers.size() < quorumJournalCount;
+  }
 }

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/HostSet.java

@@ -45,7 +45,7 @@ public class HostSet implements Iterable<InetSocketAddress> {
    * The function that checks whether there exists an entry foo in the set
    * so that foo &lt;= addr.
    */
-  boolean matchedBy(InetSocketAddress addr) {
+  public boolean matchedBy(InetSocketAddress addr) {
     Collection<Integer> ports = addrs.get(addr.getAddress());
     return addr.getPort() == 0 ? !ports.isEmpty() : ports.contains(addr
         .getPort());
@@ -55,23 +55,23 @@ boolean matchedBy(InetSocketAddress addr) {
    * The function that checks whether there exists an entry foo in the set
    * so that addr &lt;= foo.
    */
-  boolean match(InetSocketAddress addr) {
+  public boolean match(InetSocketAddress addr) {
     int port = addr.getPort();
     Collection<Integer> ports = addrs.get(addr.getAddress());
     boolean exactMatch = ports.contains(port);
     boolean genericMatch = ports.contains(0);
     return exactMatch || genericMatch;
   }
 
-  boolean isEmpty() {
+  public boolean isEmpty() {
     return addrs.isEmpty();
   }
 
-  int size() {
+  public int size() {
     return addrs.size();
   }
 
-  void add(InetSocketAddress addr) {
+  public void add(InetSocketAddress addr) {
     Preconditions.checkArgument(!addr.isUnresolved());
     addrs.put(addr.getAddress(), addr.getPort());
   }

hadoop-hdfs-project/hadoop-hdfs/src/main/resources/hdfs-default.xml

@@ -6333,6 +6333,20 @@
     </description>
   </property>
 
+  <property>
+    <name>dfs.journalnode.maintenance.nodes</name>
+    <value></value>
+    <description>
+      In the case of one out of three journal nodes being down, theoretically the service can still
+      continue. However, in reality, the downed node may not recover quickly. If the Namenode needs
+      to be restarted, it will try the downed journal node through the lengthy RPC retry mechanism,
+      resulting in a long initialization time for the Namenode to provide services. By adding the
+      downed journal node to the maintenance nodes, the initialization time of the Namenode in such
+      scenarios can be accelerated.
+    </description>
+  </property>
+
+
   <property>
     <name>dfs.namenode.lease-hard-limit-sec</name>
     <value>1200</value>

hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/qjournal/QJMTestUtil.java

@@ -47,6 +47,7 @@ public abstract class QJMTestUtil {
   public static final NamespaceInfo FAKE_NSINFO = new NamespaceInfo(
       12345, "mycluster", "my-bp", 0L);
   public static final String JID = "test-journal";
+  public static final String FAKE_HOSTNAME = "jn";
 
   public static byte[] createTxnData(int startTxn, int numTxns) throws Exception {
     DataOutputBuffer buf = new DataOutputBuffer();

hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/qjournal/client/TestIPCLoggerChannel.java

@@ -17,6 +17,7 @@
  */
 package org.apache.hadoop.hdfs.qjournal.client;
 
+import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_JOURNALNODE_MAINTENANCE_NODES_KEY;
 import static org.junit.Assert.*;
 
 import java.io.IOException;

hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/qjournal/client/TestQuorumJournalManager.java

@@ -17,6 +17,9 @@
  */
 package org.apache.hadoop.hdfs.qjournal.client;
 
+import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_JOURNALNODE_MAINTENANCE_NODES_KEY;
+import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_JOURNALNODE_RPC_PORT_DEFAULT;
+import static org.apache.hadoop.hdfs.qjournal.QJMTestUtil.FAKE_HOSTNAME;
 import static org.apache.hadoop.hdfs.qjournal.QJMTestUtil.FAKE_NSINFO;
 import static org.apache.hadoop.hdfs.qjournal.QJMTestUtil.JID;
 import static org.apache.hadoop.hdfs.qjournal.QJMTestUtil.verifyEdits;
@@ -27,6 +30,7 @@
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertThrows;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
 import static org.mockito.ArgumentMatchers.eq;
@@ -39,6 +43,7 @@
 import java.net.UnknownHostException;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.StringJoiner;
 import java.util.concurrent.Semaphore;
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicInteger;
@@ -1171,6 +1176,21 @@ public void testSelectViaRpcAfterJNRestart() throws Exception {
     }
   }
 
+  /**
+   * Tests to throw an exception if the jn maintenance nodes exceeds half of the journalnode number.
+   */
+  @Test
+  public void testJNBlackListViaRpcTwoJNsError() throws Exception {
+    StringJoiner blackListBuff = new StringJoiner(",");
+    for (int i = 0; i < 2; i++) {
+      blackListBuff.add(String.format("%s%d:%d",
+          FAKE_HOSTNAME, i, DFS_JOURNALNODE_RPC_PORT_DEFAULT));
+    }
+    this.conf.set(DFS_JOURNALNODE_MAINTENANCE_NODES_KEY, blackListBuff.toString());
+    assertThrows(IllegalArgumentException.class, () -> createSpyingQJM());
+  }
+
+
   @Test
   public void testGetJournalAddressListWithResolution() throws Exception {
     Configuration configuration = new Configuration();