make only one fbr task is executing

Author: lgh

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

@@ -972,8 +972,8 @@ public class DFSConfigKeys extends CommonConfigurationKeys {
   public static final boolean DFS_DATANODE_TRANSFERTO_ALLOWED_DEFAULT = true;
   public static final String  DFS_HEARTBEAT_INTERVAL_KEY = "dfs.heartbeat.interval";
   public static final long    DFS_HEARTBEAT_INTERVAL_DEFAULT = 3;
-  public static final String  DFS_HEARTBEAT_REREGISTER_INTERVAL_KEY = "dfs.heartbeat.reregister,interval";
-  public static final long    DFS_HEARTBEAT_REREGISTER_INTERVAL__DEFAULT = 9;
+  public static final String  DFS_HEARTBEAT_REREGISTER_INTERVAL_KEY = "dfs.heartbeat.reregister.interval";
+  public static final long DFS_HEARTBEAT_REREGISTER_INTERVAL_DEFAULT = 6;
   public static final String  DFS_DATANODE_LIFELINE_INTERVAL_SECONDS_KEY =
       "dfs.datanode.lifeline.interval.seconds";
   public static final String  DFS_NAMENODE_PATH_BASED_CACHE_RETRY_INTERVAL_MS = "dfs.namenode.path.based.cache.retry.interval.ms";

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/BPServiceActor.java

@@ -123,6 +123,7 @@ enum RunningState {
   private final DNConf dnConf;
   private long prevBlockReportId;
   private volatile long fullBlockReportLeaseId;
+  private volatile boolean shouldSendFBR = true;
   private final SortedSet<Integer> blockReportSizes =
       Collections.synchronizedSortedSet(new TreeSet<>());
   private final int maxDataLength;
@@ -149,6 +150,7 @@ enum RunningState {
         dn.getMetrics());
     prevBlockReportId = ThreadLocalRandom.current().nextLong();
     fullBlockReportLeaseId = 0;
+    shouldSendFBR = true;
     scheduler = new Scheduler(dnConf.heartBeatInterval,
         dnConf.getLifelineIntervalMs(), dnConf.blockReportInterval,
         dnConf.outliersReportIntervalMs, dnConf.heartBeatReRegisterInterval);
@@ -771,7 +773,8 @@ private void offerService() throws Exception {
         if (forceFullBr) {
           LOG.info("Forcing a full block report to " + nnAddr);
         }
-        if ((fullBlockReportLeaseId != 0) || forceFullBr) {
+        if ((fullBlockReportLeaseId != 0 && shouldSendFBR) || forceFullBr) {
+          shouldSendFBR = false;
           fbrExecutorService.submit(new FBRTaskHandler());
         }
 
@@ -799,6 +802,7 @@ private void offerService() throws Exception {
         }
         if (InvalidBlockReportLeaseException.class.getName().equals(reClass)) {
           fullBlockReportLeaseId = 0;
+          shouldSendFBR = true;
         }
         LOG.warn("RemoteException in offerService", re);
         sleepAfterException();
@@ -873,6 +877,7 @@ void register(NamespaceInfo nsInfo) throws IOException {
     // reset lease id whenever registered to NN.
     // ask for a new lease id at the next heartbeat.
     fullBlockReportLeaseId = 0;
+    shouldSendFBR = true;
 
     // random short delay - helps scatter the BR from all DNs
     scheduler.scheduleBlockReport(dnConf.initialBlockReportDelayMs, true);
@@ -1212,8 +1217,10 @@ public void run() {
         }
         fullBlockReportLeaseId = 0;
         commandProcessingThread.enqueue(cmds);
+        shouldSendFBR = true;
       } catch (Throwable t) {
         fullBlockReportLeaseId = 0;
+        shouldSendFBR = true;
         LOG.warn("InterruptedException in FBR Task Handler.", t);
         sleepAndLogInterrupts(5000, "offering FBR service");
         synchronized(ibrManager) {

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DNConf.java

@@ -40,7 +40,7 @@
 import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_ENCRYPT_DATA_OVERWRITE_DOWNSTREAM_DERIVED_QOP_DEFAULT;
 import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_ENCRYPT_DATA_OVERWRITE_DOWNSTREAM_DERIVED_QOP_KEY;
 import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_HEARTBEAT_REREGISTER_INTERVAL_KEY;
-import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_HEARTBEAT_REREGISTER_INTERVAL__DEFAULT;
+import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_HEARTBEAT_REREGISTER_INTERVAL_DEFAULT;
 import static org.apache.hadoop.hdfs.client.HdfsClientConfigKeys.DFS_CLIENT_SOCKET_TIMEOUT_KEY;
 import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_DATANODE_MAX_LOCKED_MEMORY_DEFAULT;
 import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_DATANODE_MAX_LOCKED_MEMORY_KEY;
@@ -223,7 +223,7 @@ public DNConf(final Configurable dn) {
         DFS_HEARTBEAT_INTERVAL_DEFAULT, TimeUnit.SECONDS,
         TimeUnit.MILLISECONDS);
     heartBeatReRegisterInterval = getConf().getTimeDuration(DFS_HEARTBEAT_REREGISTER_INTERVAL_KEY,
-        DFS_HEARTBEAT_REREGISTER_INTERVAL__DEFAULT, TimeUnit.SECONDS,
+        DFS_HEARTBEAT_REREGISTER_INTERVAL_DEFAULT, TimeUnit.SECONDS,
         TimeUnit.MILLISECONDS);
     long confLifelineIntervalMs =
         getConf().getLong(DFS_DATANODE_LIFELINE_INTERVAL_SECONDS_KEY,