HDFS-16660. Improve Code With Lambda in IPCLoggerChannel class (#4561)

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

@@ -23,7 +23,6 @@
 import java.net.URI;
 import java.net.URL;
 import java.security.PrivilegedExceptionAction;
-import java.util.concurrent.Callable;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
@@ -67,6 +66,7 @@
 import org.apache.hadoop.thirdparty.com.google.common.util.concurrent.MoreExecutors;
 import org.apache.hadoop.thirdparty.com.google.common.util.concurrent.ThreadFactoryBuilder;
 import org.apache.hadoop.thirdparty.com.google.common.util.concurrent.UncaughtExceptionHandlers;
+import org.apache.hadoop.util.Time;
 
 /**
  * Channel to a remote JournalNode using Hadoop IPC.
@@ -154,26 +154,15 @@ public class IPCLoggerChannel implements AsyncLogger {
 
   private static final long WARN_JOURNAL_MILLIS_THRESHOLD = 1000;
   
-  static final Factory FACTORY = new AsyncLogger.Factory() {
+  static final Factory FACTORY = IPCLoggerChannel::new;
-    @Override
-    public AsyncLogger createLogger(Configuration conf, NamespaceInfo nsInfo,
-        String journalId, String nameServiceId, InetSocketAddress addr) {
-      return new IPCLoggerChannel(conf, nsInfo, journalId, nameServiceId, addr);
-    }
-  };
 
-  public IPCLoggerChannel(Configuration conf,
+  public IPCLoggerChannel(Configuration conf, NamespaceInfo nsInfo,
-      NamespaceInfo nsInfo,
+      String journalId, InetSocketAddress addr) {
-      String journalId,
-      InetSocketAddress addr) {
     this(conf, nsInfo, journalId, null, addr);
   }
 
-  public IPCLoggerChannel(Configuration conf,
+  public IPCLoggerChannel(Configuration conf, NamespaceInfo nsInfo,
-                          NamespaceInfo nsInfo,
+      String journalId, String nameServiceId, InetSocketAddress addr) {
-                          String journalId,
-                          String nameServiceId,
-                          InetSocketAddress addr) {
     this.conf = conf;
     this.nsInfo = nsInfo;
     this.journalId = journalId;
@@ -202,7 +191,7 @@ public synchronized void setCommittedTxId(long txid) {
         "Trying to move committed txid backwards in client " +
          "old: %s new: %s", committedTxId, txid);
     this.committedTxId = txid;
-    this.lastCommitNanos = System.nanoTime();
+    this.lastCommitNanos = Time.monotonicNowNanos();
   }
   
   @Override
@@ -229,17 +218,12 @@ protected QJournalProtocol createProxy() throws IOException {
     final Configuration confCopy = new Configuration(conf);
     
     // Need to set NODELAY or else batches larger than MTU can trigger 
-    // 40ms nagling delays.
+    // 40ms nailing delays.
-    confCopy.setBoolean(
+    confCopy.setBoolean(CommonConfigurationKeysPublic.IPC_CLIENT_TCPNODELAY_KEY, true);
-        CommonConfigurationKeysPublic.IPC_CLIENT_TCPNODELAY_KEY,
-        true);
-    
     RPC.setProtocolEngine(confCopy,
         QJournalProtocolPB.class, ProtobufRpcEngine2.class);
     return SecurityUtil.doAsLoginUser(
-        new PrivilegedExceptionAction<QJournalProtocol>() {
+        (PrivilegedExceptionAction<QJournalProtocol>) () -> {
-          @Override
-          public QJournalProtocol run() throws IOException {
           RPC.setProtocolEngine(confCopy,
               QJournalProtocolPB.class, ProtobufRpcEngine2.class);
           QJournalProtocolPB pbproxy = RPC.getProxy(
@@ -247,7 +231,6 @@ public QJournalProtocol run() throws IOException {
               RPC.getProtocolVersion(QJournalProtocolPB.class),
               addr, confCopy);
           return new QJournalProtocolTranslatorPB(pbproxy);
-          }
         });
   }
   
@@ -260,10 +243,8 @@ protected ExecutorService createSingleThreadExecutor() {
     return Executors.newSingleThreadExecutor(
         new ThreadFactoryBuilder()
           .setDaemon(true)
-          .setNameFormat("Logger channel (from single-thread executor) to " +
+          .setNameFormat("Logger channel (from single-thread executor) to " + addr)
-              addr)
+          .setUncaughtExceptionHandler(UncaughtExceptionHandlers.systemExit())
-          .setUncaughtExceptionHandler(
-              UncaughtExceptionHandlers.systemExit())
           .build());
   }
 
@@ -308,11 +289,6 @@ private synchronized RequestInfo createReqInfo() {
         epoch, ipcSerial++, committedTxId);
   }
 
-  @VisibleForTesting
-  synchronized long getNextIpcSerial() {
-    return ipcSerial;
-  }
-
   public synchronized int getQueuedEditsSize() {
     return queuedEditsSizeBytes;
   }
@@ -333,11 +309,7 @@ public synchronized boolean isOutOfSync() {
   @VisibleForTesting
   void waitForAllPendingCalls() throws InterruptedException {
     try {
-      singleThreadExecutor.submit(new Runnable() {
+      singleThreadExecutor.submit(() -> {}).get();
-        @Override
-        public void run() {
-        }
-      }).get();
     } catch (ExecutionException e) {
       // This can't happen!
       throw new AssertionError(e);
@@ -346,36 +318,23 @@ public void run() {
 
   @Override
   public ListenableFuture<Boolean> isFormatted() {
-    return singleThreadExecutor.submit(new Callable<Boolean>() {
+    return singleThreadExecutor.submit(() -> getProxy().isFormatted(journalId, nameServiceId));
-      @Override
-      public Boolean call() throws IOException {
-        return getProxy().isFormatted(journalId, nameServiceId);
-      }
-    });
   }
 
   @Override
   public ListenableFuture<GetJournalStateResponseProto> getJournalState() {
-    return singleThreadExecutor.submit(new Callable<GetJournalStateResponseProto>() {
+    return singleThreadExecutor.submit(() -> {
-      @Override
+      GetJournalStateResponseProto ret = getProxy().getJournalState(journalId, nameServiceId);
-      public GetJournalStateResponseProto call() throws IOException {
-        GetJournalStateResponseProto ret =
-            getProxy().getJournalState(journalId, nameServiceId);
       constructHttpServerURI(ret);
       return ret;
-      }
     });
   }
 
   @Override
   public ListenableFuture<NewEpochResponseProto> newEpoch(
       final long epoch) {
-    return singleThreadExecutor.submit(new Callable<NewEpochResponseProto>() {
+    return singleThreadExecutor.submit(
-      @Override
+        () -> getProxy().newEpoch(journalId, nameServiceId, nsInfo, epoch));
-      public NewEpochResponseProto call() throws IOException {
-        return getProxy().newEpoch(journalId, nameServiceId, nsInfo, epoch);
-      }
-    });
   }
   
   @Override
@@ -390,42 +349,36 @@ public ListenableFuture<Void> sendEdits(
     
     // When this batch is acked, we use its submission time in order
     // to calculate how far we are lagging.
-    final long submitNanos = System.nanoTime();
+    final long submitNanos = Time.monotonicNowNanos();
     
     ListenableFuture<Void> ret = null;
     try {
-      ret = singleThreadExecutor.submit(new Callable<Void>() {
+      ret = singleThreadExecutor.submit(() -> {
-        @Override
-        public Void call() throws IOException {
         throwIfOutOfSync();
 
-          long rpcSendTimeNanos = System.nanoTime();
+        final long rpcSendTimeNanos = Time.monotonicNowNanos();
         try {
-            getProxy().journal(createReqInfo(),
+          getProxy().journal(createReqInfo(), segmentTxId, firstTxnId, numTxns, data);
-                segmentTxId, firstTxnId, numTxns, data);
         } catch (IOException e) {
-            QuorumJournalManager.LOG.warn(
+          QuorumJournalManager.LOG.warn("Remote journal {} failed to write txns {}-{}."
-                "Remote journal " + IPCLoggerChannel.this + " failed to " +
+                  + " Will try to write to this JN again after the next log roll.",
-                "write txns " + firstTxnId + "-" + (firstTxnId + numTxns - 1) +
+              IPCLoggerChannel.this, firstTxnId, (firstTxnId + numTxns - 1), e);
-                ". Will try to write to this JN again after the next " +
-                "log roll.", e); 
           synchronized (IPCLoggerChannel.this) {
             outOfSync = true;
           }
           throw e;
         } finally {
-            long now = System.nanoTime();
+          final long nowNanos = Time.monotonicNowNanos();
-            long rpcTime = TimeUnit.MICROSECONDS.convert(
+          final long rpcTimeMicros = TimeUnit.MICROSECONDS.convert(
-                now - rpcSendTimeNanos, TimeUnit.NANOSECONDS);
+              (nowNanos - rpcSendTimeNanos), TimeUnit.NANOSECONDS);
-            long endToEndTime = TimeUnit.MICROSECONDS.convert(
+          final long endToEndTimeMicros = TimeUnit.MICROSECONDS.convert(
-                now - submitNanos, TimeUnit.NANOSECONDS);
+              (nowNanos - submitNanos), TimeUnit.NANOSECONDS);
-            metrics.addWriteEndToEndLatency(endToEndTime);
+          metrics.addWriteEndToEndLatency(endToEndTimeMicros);
-            metrics.addWriteRpcLatency(rpcTime);
+          metrics.addWriteRpcLatency(rpcTimeMicros);
-            if (rpcTime / 1000 > WARN_JOURNAL_MILLIS_THRESHOLD) {
+          if (rpcTimeMicros / 1000 > WARN_JOURNAL_MILLIS_THRESHOLD) {
             QuorumJournalManager.LOG.warn(
-                  "Took " + (rpcTime / 1000) + "ms to send a batch of " +
+                "Took {}ms to send a batch of {} edits ({} bytes) to remote journal {}.",
-                  numTxns + " edits (" + data.length + " bytes) to " +
+                rpcTimeMicros / 1000, numTxns, data.length, IPCLoggerChannel.this);
-                  "remote journal " + IPCLoggerChannel.this);
           }
         }
         synchronized (IPCLoggerChannel.this) {
@@ -433,7 +386,6 @@ public Void call() throws IOException {
           lastAckNanos = submitNanos;
         }
         return null;
-        }
       });
     } finally {
       if (ret == null) {
@@ -460,14 +412,12 @@ public void onSuccess(Void t) {
     return ret;
   }
 
-  private void throwIfOutOfSync()
+  private void throwIfOutOfSync() throws IOException {
-      throws JournalOutOfSyncException, IOException {
     if (isOutOfSync()) {
       // Even if we're out of sync, it's useful to send an RPC
       // to the remote node in order to update its lag metrics, etc.
       heartbeatIfNecessary();
-      throw new JournalOutOfSyncException(
+      throw new JournalOutOfSyncException("Journal disabled until next roll");
-          "Journal disabled until next roll");
     }
   }
 
@@ -497,12 +447,10 @@ private void heartbeatIfNecessary() throws IOException {
   private synchronized void reserveQueueSpace(int size)
       throws LoggerTooFarBehindException {
     Preconditions.checkArgument(size >= 0);
-    if (queuedEditsSizeBytes + size > queueSizeLimitBytes &&
+    if (queuedEditsSizeBytes + size > queueSizeLimitBytes && queuedEditsSizeBytes > 0) {
-        queuedEditsSizeBytes > 0) {
+      QuorumJournalManager.LOG.warn("Pending edits to {} is going to exceed limit size: {}"
-      QuorumJournalManager.LOG.warn("Pending edits to " + IPCLoggerChannel.this
+          + ", current queued edits size: {}, will silently drop {} bytes of edits!",
-          + " is going to exceed limit size: " + queueSizeLimitBytes
+          IPCLoggerChannel.class, queueSizeLimitBytes, queuedEditsSizeBytes, size);
-          + ", current queued edits size: " + queuedEditsSizeBytes
-          + ", will silently drop " + size + " bytes of edits!");
       throw new LoggerTooFarBehindException();
     }
     queuedEditsSizeBytes += size;
@@ -514,203 +462,144 @@ private synchronized void unreserveQueueSpace(int size) {
   }
 
   @Override
-  public ListenableFuture<Void> format(final NamespaceInfo nsInfo,
+  public ListenableFuture<Void> format(final NamespaceInfo nsInfo, final boolean force) {
-      final boolean force) {
+    return singleThreadExecutor.submit(() -> {
-    return singleThreadExecutor.submit(new Callable<Void>() {
-      @Override
-      public Void call() throws Exception {
       getProxy().format(journalId, nameServiceId, nsInfo, force);
       return null;
-      }
     });
   }
   
   @Override
-  public ListenableFuture<Void> startLogSegment(final long txid,
+  public ListenableFuture<Void> startLogSegment(final long txid, final int layoutVersion) {
-      final int layoutVersion) {
+    return singleThreadExecutor.submit(() -> {
-    return singleThreadExecutor.submit(new Callable<Void>() {
-      @Override
-      public Void call() throws IOException {
       getProxy().startLogSegment(createReqInfo(), txid, layoutVersion);
       synchronized (IPCLoggerChannel.this) {
         if (outOfSync) {
           outOfSync = false;
           QuorumJournalManager.LOG.info(
-                "Restarting previously-stopped writes to " +
+              "Restarting previously-stopped writes to {} in segment starting at txid {}.",
-                IPCLoggerChannel.this + " in segment starting at txid " +
+                  IPCLoggerChannel.class, txid);
-                txid);
         }
       }
       return null;
-      }
     });
   }
   
   @Override
-  public ListenableFuture<Void> finalizeLogSegment(
+  public ListenableFuture<Void> finalizeLogSegment(final long startTxId, final long endTxId) {
-      final long startTxId, final long endTxId) {
+    return singleThreadExecutor.submit(() -> {
-    return singleThreadExecutor.submit(new Callable<Void>() {
-      @Override
-      public Void call() throws IOException {
       throwIfOutOfSync();
-        
       getProxy().finalizeLogSegment(createReqInfo(), startTxId, endTxId);
       return null;
-      }
     });
   }
   
   @Override
   public ListenableFuture<Void> purgeLogsOlderThan(final long minTxIdToKeep) {
-    return singleThreadExecutor.submit(new Callable<Void>() {
+    return singleThreadExecutor.submit(() -> {
-      @Override
-      public Void call() throws Exception {
       getProxy().purgeLogsOlderThan(createReqInfo(), minTxIdToKeep);
       return null;
-      }
     });
   }
 
   @Override
   public ListenableFuture<GetJournaledEditsResponseProto> getJournaledEdits(
       long fromTxnId, int maxTransactions) {
-    return parallelExecutor.submit(
+    return parallelExecutor.submit(() -> getProxy().getJournaledEdits(
-        new Callable<GetJournaledEditsResponseProto>() {
+        journalId, nameServiceId, fromTxnId, maxTransactions));
-          @Override
-          public GetJournaledEditsResponseProto call() throws IOException {
-            return getProxy().getJournaledEdits(journalId, nameServiceId,
-                fromTxnId, maxTransactions);
-          }
-        });
   }
 
   @Override
   public ListenableFuture<RemoteEditLogManifest> getEditLogManifest(
       final long fromTxnId, final boolean inProgressOk) {
-    return parallelExecutor.submit(new Callable<RemoteEditLogManifest>() {
+    return parallelExecutor.submit(() -> {
-      @Override
-      public RemoteEditLogManifest call() throws IOException {
       GetEditLogManifestResponseProto ret = getProxy().getEditLogManifest(
           journalId, nameServiceId, fromTxnId, inProgressOk);
       // Update the http port, since we need this to build URLs to any of the
       // returned logs.
       constructHttpServerURI(ret);
       return PBHelper.convert(ret.getManifest());
-      }
     });
   }
 
   @Override
-  public ListenableFuture<PrepareRecoveryResponseProto> prepareRecovery(
+  public ListenableFuture<PrepareRecoveryResponseProto> prepareRecovery(final long segmentTxId) {
-      final long segmentTxId) {
+    return singleThreadExecutor.submit(() -> {
-    return singleThreadExecutor.submit(new Callable<PrepareRecoveryResponseProto>() {
-      @Override
-      public PrepareRecoveryResponseProto call() throws IOException {
       if (!hasHttpServerEndPoint()) {
-          // force an RPC call so we know what the HTTP port should be if it
+        // force an RPC call, so we know what the HTTP port should be if it
-          // haven't done so.
+        // hasn't done so.
         GetJournalStateResponseProto ret = getProxy().getJournalState(
             journalId, nameServiceId);
         constructHttpServerURI(ret);
       }
       return getProxy().prepareRecovery(createReqInfo(), segmentTxId);
-      }
     });
   }
 
   @Override
-  public ListenableFuture<Void> acceptRecovery(
+  public ListenableFuture<Void> acceptRecovery(final SegmentStateProto log, final URL url) {
-      final SegmentStateProto log, final URL url) {
+    return singleThreadExecutor.submit(() -> {
-    return singleThreadExecutor.submit(new Callable<Void>() {
-      @Override
-      public Void call() throws IOException {
       getProxy().acceptRecovery(createReqInfo(), log, url);
       return null;
-      }
     });
   }
   
   @Override
   public ListenableFuture<Void> doPreUpgrade() {
-    return singleThreadExecutor.submit(new Callable<Void>() {
+    return singleThreadExecutor.submit(() -> {
-      @Override
-      public Void call() throws IOException {
       getProxy().doPreUpgrade(journalId);
       return null;
-      }
     });
   }
   
   @Override
   public ListenableFuture<Void> doUpgrade(final StorageInfo sInfo) {
-    return singleThreadExecutor.submit(new Callable<Void>() {
+    return singleThreadExecutor.submit(() -> {
-      @Override
-      public Void call() throws IOException {
       getProxy().doUpgrade(journalId, sInfo);
       return null;
-      }
     });
   }
   
   @Override
   public ListenableFuture<Void> doFinalize() {
-    return singleThreadExecutor.submit(new Callable<Void>() {
+    return singleThreadExecutor.submit(() -> {
-      @Override
-      public Void call() throws IOException {
       getProxy().doFinalize(journalId, nameServiceId);
       return null;
-      }
     });
   }
   
   @Override
   public ListenableFuture<Boolean> canRollBack(final StorageInfo storage,
       final StorageInfo prevStorage, final int targetLayoutVersion) {
-    return singleThreadExecutor.submit(new Callable<Boolean>() {
+    return singleThreadExecutor.submit(
-      @Override
+        () -> getProxy().canRollBack(journalId, nameServiceId,
-      public Boolean call() throws IOException {
+            storage, prevStorage, targetLayoutVersion));
-        return getProxy().canRollBack(journalId, nameServiceId,
-            storage, prevStorage, targetLayoutVersion);
-      }
-    });
   }
 
   @Override
   public ListenableFuture<Void> doRollback() {
-    return singleThreadExecutor.submit(new Callable<Void>() {
+    return singleThreadExecutor.submit(() -> {
-      @Override
-      public Void call() throws IOException {
       getProxy().doRollback(journalId, nameServiceId);
       return null;
-      }
     });
   }
 
   @Override
   public ListenableFuture<Void> discardSegments(final long startTxId) {
-    return singleThreadExecutor.submit(new Callable<Void>() {
+    return singleThreadExecutor.submit(() -> {
-      @Override
-      public Void call() throws IOException {
       getProxy().discardSegments(journalId, nameServiceId, startTxId);
       return null;
-      }
     });
   }
 
   @Override
   public ListenableFuture<Long> getJournalCTime() {
-    return singleThreadExecutor.submit(new Callable<Long>() {
+    return singleThreadExecutor.submit(() -> getProxy().getJournalCTime(journalId, nameServiceId));
-      @Override
-      public Long call() throws IOException {
-        return getProxy().getJournalCTime(journalId, nameServiceId);
-      }
-    });
   }
 
   @Override
   public String toString() {
-    return InetAddresses.toAddrString(addr.getAddress()) + ':' +
+    return InetAddresses.toAddrString(addr.getAddress()) + ':' + addr.getPort();
-        addr.getPort();
   }
 
   @Override
@@ -778,5 +667,4 @@ private URL getHttpServerURI(String scheme, int port) {
   private boolean hasHttpServerEndPoint() {
    return httpServerURL != null;
   }
-
 }