Merge remote-tracking branch 'origin/YARN-11212' into YARN-11212

Author: slfan1989

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/io/retry/RetryInvocationHandler.java

@@ -46,6 +46,10 @@ public class RetryInvocationHandler<T> implements RpcInvocationHandler {
   public static final Logger LOG = LoggerFactory.getLogger(
       RetryInvocationHandler.class);
 
+  @VisibleForTesting
+  public static final ThreadLocal<Boolean> SET_CALL_ID_FOR_TEST =
+      ThreadLocal.withInitial(() -> true);
+
   static class Call {
     private final Method method;
     private final Object[] args;
@@ -159,7 +163,7 @@ CallReturn invoke() throws Throwable {
     }
 
     Object invokeMethod() throws Throwable {
-      if (isRpc) {
+      if (isRpc && SET_CALL_ID_FOR_TEST.get()) {
         Client.setCallIdAndRetryCount(callId, counters.retries,
             retryInvocationHandler.asyncCallHandler);
       }

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/CallerContext.java

@@ -47,6 +47,8 @@ public final class CallerContext {
   // field names
   public static final String CLIENT_IP_STR = "clientIp";
   public static final String CLIENT_PORT_STR = "clientPort";
+  public static final String CLIENT_ID_STR = "clientId";
+  public static final String CLIENT_CALL_ID_STR = "clientCallId";
 
   /** The caller context.
    *

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/RetryCache.java

@@ -55,14 +55,14 @@ public static class CacheEntry implements LightWeightCache.Entry {
     /**
      * Processing state of the requests.
      */
-    private static byte INPROGRESS = 0;
-    private static byte SUCCESS = 1;
-    private static byte FAILED = 2;
+    private static final byte INPROGRESS = 0;
+    private static final byte SUCCESS = 1;
+    private static final byte FAILED = 2;
 
     private byte state = INPROGRESS;
 
     // Store uuid as two long for better memory utilization
-    private final long clientIdMsb; // Most signficant bytes
+    private final long clientIdMsb; // Most significant bytes
     private final long clientIdLsb; // Least significant bytes
 
     private final int callId;
@@ -140,8 +140,8 @@ public long getExpirationTime() {
 
     @Override
     public String toString() {
-      return (new UUID(this.clientIdMsb, this.clientIdLsb)).toString() + ":"
-          + this.callId + ":" + this.state;
+      return String.format("%s:%s:%s", new UUID(this.clientIdMsb, this.clientIdLsb),
+          this.callId, this.state);
     }
   }
 
@@ -183,7 +183,7 @@ public Object getPayload() {
 
   private final LightWeightGSet<CacheEntry, CacheEntry> set;
   private final long expirationTime;
-  private String cacheName;
+  private final String cacheName;
 
   private final ReentrantLock lock = new ReentrantLock();
 
@@ -195,19 +195,19 @@ public Object getPayload() {
    */
   public RetryCache(String cacheName, double percentage, long expirationTime) {
     int capacity = LightWeightGSet.computeCapacity(percentage, cacheName);
-    capacity = capacity > MAX_CAPACITY ? capacity : MAX_CAPACITY;
+    capacity = Math.max(capacity, MAX_CAPACITY);
     this.set = new LightWeightCache<CacheEntry, CacheEntry>(capacity, capacity,
         expirationTime, 0);
     this.expirationTime = expirationTime;
     this.cacheName = cacheName;
     this.retryCacheMetrics =  RetryCacheMetrics.create(this);
   }
 
-  private static boolean skipRetryCache() {
+  private static boolean skipRetryCache(byte[] clientId, int callId) {
     // Do not track non RPC invocation or RPC requests with
     // invalid callId or clientId in retry cache
-    return !Server.isRpcInvocation() || Server.getCallId() < 0
-        || Arrays.equals(Server.getClientId(), RpcConstants.DUMMY_CLIENT_ID);
+    return !Server.isRpcInvocation() || callId < 0
+        || Arrays.equals(clientId, RpcConstants.DUMMY_CLIENT_ID);
   }
 
   public void lock() {
@@ -332,43 +332,51 @@ public void addCacheEntryWithPayload(byte[] clientId, int callId,
     retryCacheMetrics.incrCacheUpdated();
   }
 
-  private static CacheEntry newEntry(long expirationTime) {
-    return new CacheEntry(Server.getClientId(), Server.getCallId(),
+  private static CacheEntry newEntry(long expirationTime,
+      byte[] clientId, int callId) {
+    return new CacheEntry(clientId, callId,
         System.nanoTime() + expirationTime);
   }
 
   private static CacheEntryWithPayload newEntry(Object payload,
-      long expirationTime) {
-    return new CacheEntryWithPayload(Server.getClientId(), Server.getCallId(),
+      long expirationTime, byte[] clientId, int callId) {
+    return new CacheEntryWithPayload(clientId, callId,
         payload, System.nanoTime() + expirationTime);
   }
 
   /**
    * Static method that provides null check for retryCache.
    * @param cache input Cache.
+   * @param clientId client id of this request
+   * @param callId client call id of this request
    * @return CacheEntry.
    */
-  public static CacheEntry waitForCompletion(RetryCache cache) {
-    if (skipRetryCache()) {
+  public static CacheEntry waitForCompletion(RetryCache cache,
+      byte[] clientId, int callId) {
+    if (skipRetryCache(clientId, callId)) {
       return null;
     }
     return cache != null ? cache
-        .waitForCompletion(newEntry(cache.expirationTime)) : null;
+        .waitForCompletion(newEntry(cache.expirationTime,
+            clientId, callId)) : null;
   }
 
   /**
    * Static method that provides null check for retryCache.
    * @param cache input cache.
    * @param payload input payload.
+   * @param clientId client id of this request
+   * @param callId client call id of this request
    * @return CacheEntryWithPayload.
    */
   public static CacheEntryWithPayload waitForCompletion(RetryCache cache,
-      Object payload) {
-    if (skipRetryCache()) {
+      Object payload, byte[] clientId, int callId) {
+    if (skipRetryCache(clientId, callId)) {
       return null;
     }
     return (CacheEntryWithPayload) (cache != null ? cache
-        .waitForCompletion(newEntry(payload, cache.expirationTime)) : null);
+        .waitForCompletion(newEntry(payload, cache.expirationTime,
+            clientId, callId)) : null);
   }
 
   public static void setState(CacheEntry e, boolean success) {

hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ipc/TestRetryCache.java

@@ -50,14 +50,14 @@ public void setup() {
   static class TestServer {
     AtomicInteger retryCount = new AtomicInteger();
     AtomicInteger operationCount = new AtomicInteger();
-    private RetryCache retryCache = new RetryCache("TestRetryCache", 1,
-        100 * 1000 * 1000 * 1000L);
+    private final RetryCache retryCache = new RetryCache(
+        "TestRetryCache", 1, 100 * 1000 * 1000 * 1000L);
 
     /**
      * A server method implemented using {@link RetryCache}.
      * 
      * @param input is returned back in echo, if {@code success} is true.
-     * @param failureOuput returned on failure, if {@code success} is false.
+     * @param failureOutput returned on failure, if {@code success} is false.
      * @param methodTime time taken by the operation. By passing smaller/larger
      *          value one can simulate an operation that takes short/long time.
      * @param success whether this operation completes successfully or not
@@ -67,7 +67,7 @@ static class TestServer {
     int echo(int input, int failureOutput, long methodTime, boolean success)
         throws InterruptedException {
       CacheEntryWithPayload entry = RetryCache.waitForCompletion(retryCache,
-          null);
+          null, Server.getClientId(), Server.getCallId());
       if (entry != null && entry.isSuccess()) {
         System.out.println("retryCount incremented " + retryCount.get());
         retryCount.incrementAndGet();
@@ -173,16 +173,13 @@ public void testOperations(final int input, final int numberOfThreads,
     final int failureOutput = input + 1;
     ExecutorService executorService = Executors
         .newFixedThreadPool(numberOfThreads);
-    List<Future<Integer>> list = new ArrayList<Future<Integer>>();
+    List<Future<Integer>> list = new ArrayList<>();
     for (int i = 0; i < numberOfThreads; i++) {
-      Callable<Integer> worker = new Callable<Integer>() {
-        @Override
-        public Integer call() throws Exception {
-          Server.getCurCall().set(call);
-          Assert.assertEquals(Server.getCurCall().get(), call);
-          int randomPause = pause == 0 ? pause : r.nextInt(pause);
-          return testServer.echo(input, failureOutput, randomPause, success);
-        }
+      Callable<Integer> worker = () -> {
+        Server.getCurCall().set(call);
+        Assert.assertEquals(Server.getCurCall().get(), call);
+        int randomPause = pause == 0 ? pause : r.nextInt(pause);
+        return testServer.echo(input, failureOutput, randomPause, success);
       };
       Future<Integer> submit = executorService.submit(worker);
       list.add(submit);

hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfs-tests/test_libhdfs_mini_stress.c

@@ -18,24 +18,27 @@
 
 #include "common/util_c.h"
 #include "expect.h"
-#include "hdfs/hdfs.h"
 #include "hdfspp/hdfs_ext.h"
 #include "native_mini_dfs.h"
 #include "os/thread.h"
 #include "x-platform/c-api/syscall.h"
+#include "hdfs/hdfs.h"
 
 #include <errno.h>
 #include <inttypes.h>
-#include <pwd.h>
 #include <stdint.h>
 #include <stdio.h>
 #include <stdlib.h>
 #include <string.h>
-#include <sys/socket.h>
 #include <sys/types.h>
-#include <sys/wait.h>
 #include <unistd.h>
 
+#ifndef WIN32
+#include <sys/socket.h>
+#include <sys/wait.h>
+#include <pwd.h>
+#endif
+
 #define TO_STR_HELPER(X) #X
 #define TO_STR(X) TO_STR_HELPER(X)
 
@@ -197,7 +200,7 @@ static int fileEventCallback1(const char * event, const char * cluster, const ch
   if (randomErrRatioStr) randomErrRatio = (int64_t)atoi(randomErrRatioStr);
   if (randomErrRatio == 0) return DEBUG_SIMULATE_ERROR;
   else if (randomErrRatio < 0) return LIBHDFSPP_EVENT_OK;
-  return random() % randomErrRatio == 0 ? DEBUG_SIMULATE_ERROR : LIBHDFSPP_EVENT_OK;
+  return rand() % randomErrRatio == 0 ? DEBUG_SIMULATE_ERROR : LIBHDFSPP_EVENT_OK;
 }
 
 static int fileEventCallback2(const char * event, const char * cluster, const char * file, int64_t value, int64_t cookie)
@@ -235,7 +238,7 @@ static int doTestHdfsMiniStress(struct tlhThreadInfo *ti, int randomErr)
     EXPECT_ZERO(hdfsCloseFile(ti->hdfs, file));
     file = hdfsOpenFile(ti->hdfs, ti->fileNm, O_RDONLY, 0, 0, 0);
     EXPECT_NONNULL(file);
-    seekPos = (((double)random()) / RAND_MAX) * (fileInfo->mSize - expected);
+    seekPos = (((double)rand()) / RAND_MAX) * (fileInfo->mSize - expected);
     seekPos = (seekPos / expected) * expected;
     ret = hdfsSeek(ti->hdfs, file, seekPos);
     if (ret < 0) {

hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfs/os/windows/unistd.h

@@ -22,7 +22,7 @@
 /* On Windows, unistd.h does not exist, so manually define what we need. */
 
 #include <process.h> /* Declares getpid(). */
-#include <windows.h>
+#include <Windows.h>
 
 /* Re-route sleep to Sleep, converting units from seconds to milliseconds. */
 #define sleep(seconds) Sleep((seconds) * 1000)

hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/bindings/c/hdfs.cc

@@ -16,14 +16,15 @@
  * limitations under the License.
  */
 
+#include "fs/filehandle.h"
+
 #include "hdfspp/hdfspp.h"
 #include "hdfspp/hdfs_ext.h"
 
 #include "common/hdfs_configuration.h"
 #include "common/configuration_loader.h"
 #include "common/logging.h"
 #include "fs/filesystem.h"
-#include "fs/filehandle.h"
 #include "x-platform/utils.h"
 #include "x-platform/syscall.h"
 

hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/fsinfo.cc

@@ -21,6 +21,7 @@
 #include <algorithm>
 #include <sstream>
 #include <iomanip>
+#include <algorithm>
 
 namespace hdfs {
 

hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/logging.h

@@ -19,10 +19,10 @@
 #ifndef LIB_COMMON_LOGGING_H_
 #define LIB_COMMON_LOGGING_H_
 
-#include <boost/asio/ip/tcp.hpp>
-
 #include "hdfspp/log.h"
 
+#include <boost/asio/ip/tcp.hpp>
+
 #include <sstream>
 #include <mutex>
 #include <memory>

hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/namenode_info.cc

@@ -26,6 +26,7 @@
 #include <utility>
 #include <future>
 
+#include <boost/system/error_code.hpp>
 
 namespace hdfs {