sync

hadoop-cloud-storage-project/hadoop-huaweicloud/pom.xml

@@ -60,10 +60,9 @@
   <build>
     <plugins>
       <plugin>
-        <groupId>org.codehaus.mojo</groupId>
-        <artifactId>findbugs-maven-plugin</artifactId>
+        <groupId>com.github.spotbugs</groupId>
+        <artifactId>spotbugs-maven-plugin</artifactId>
         <configuration>
-          <findbugsXmlOutput>true</findbugsXmlOutput>
           <xmlOutput>true</xmlOutput>
           <excludeFilterFile>${basedir}/dev-support/findbugs-exclude.xml
           </excludeFilterFile>

hadoop-common-project/hadoop-auth/src/main/java/org/apache/hadoop/security/authentication/util/KerberosUtil.java

@@ -79,11 +79,16 @@ private static Oid getNumericOidInstance(String oidName) {
    *
    * @return Oid instance
    * @param oidName The oid Name
+   * @throws ClassNotFoundException for backward compatibility.
+   * @throws GSSException for backward compatibility.
    * @throws NoSuchFieldException if the input is not supported.
+   * @throws IllegalAccessException for backward compatibility.
+   *
    */
   @Deprecated
   public static Oid getOidInstance(String oidName)
-      throws NoSuchFieldException {
+      throws ClassNotFoundException, GSSException, NoSuchFieldException,
+      IllegalAccessException {
     switch (oidName) {
     case "GSS_SPNEGO_MECH_OID":
       return GSS_SPNEGO_MECH_OID;

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/statistics/StoreStatisticNames.java

@@ -84,6 +84,12 @@ public final class StoreStatisticNames {
   /** {@value}. */
   public static final String OP_IS_FILE = "op_is_file";
 
+  /** {@value}. */
+  public static final String OP_HFLUSH = "op_hflush";
+
+  /** {@value}. */
+  public static final String OP_HSYNC = "op_hsync";
+
   /** {@value}. */
   public static final String OP_IS_DIRECTORY = "op_is_directory";
 

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

@@ -27,6 +27,7 @@
 import org.apache.hadoop.thirdparty.com.google.common.annotations.VisibleForTesting;
 import org.apache.hadoop.util.Options;
 import org.apache.hadoop.fs.*;
+import org.apache.hadoop.fs.StreamCapabilities;
 import org.apache.hadoop.fs.Options.CreateOpts;
 import org.apache.hadoop.io.compress.CodecPool;
 import org.apache.hadoop.io.compress.CompressionCodec;
@@ -834,7 +835,8 @@ public String toString() {
   }
 
   /** Write key/value pairs to a sequence-format file. */
-  public static class Writer implements java.io.Closeable, Syncable {
+  public static class Writer implements java.io.Closeable, Syncable,
+                  Flushable, StreamCapabilities {
     private Configuration conf;
     FSDataOutputStream out;
     boolean ownOutputStream = true;
@@ -1367,6 +1369,21 @@ public void hflush() throws IOException {
         out.hflush();
       }
     }
+
+    @Override
+    public void flush() throws IOException {
+      if (out != null) {
+        out.flush();
+      }
+    }
+
+    @Override
+    public boolean hasCapability(String capability) {
+      if (out !=null && capability != null) {
+        return out.hasCapability(capability);
+      }
+      return false;
+    }
 
     /** Returns the configuration of this file. */
     Configuration getConf() { return conf; }

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

@@ -122,7 +122,7 @@ public ProtocolProxy<ProtocolMetaInfoPB> getProtocolMetaInfoProxy(
                 factory)), false);
   }
 
-  private static class Invoker implements RpcInvocationHandler {
+  protected static class Invoker implements RpcInvocationHandler {
     private final Map<String, Message> returnTypes = 
         new ConcurrentHashMap<String, Message>();
     private boolean isClosed = false;
@@ -133,7 +133,7 @@ private static class Invoker implements RpcInvocationHandler {
     private AtomicBoolean fallbackToSimpleAuth;
     private AlignmentContext alignmentContext;
 
-    private Invoker(Class<?> protocol, InetSocketAddress addr,
+    protected Invoker(Class<?> protocol, InetSocketAddress addr,
         UserGroupInformation ticket, Configuration conf, SocketFactory factory,
         int rpcTimeout, RetryPolicy connectionRetryPolicy,
         AtomicBoolean fallbackToSimpleAuth, AlignmentContext alignmentContext)
@@ -148,7 +148,7 @@ private Invoker(Class<?> protocol, InetSocketAddress addr,
     /**
      * This constructor takes a connectionId, instead of creating a new one.
      */
-    private Invoker(Class<?> protocol, Client.ConnectionId connId,
+    protected Invoker(Class<?> protocol, Client.ConnectionId connId,
         Configuration conf, SocketFactory factory) {
       this.remoteId = connId;
       this.client = CLIENTS.getClient(conf, factory, RpcWritable.Buffer.class);
@@ -225,8 +225,6 @@ public Message invoke(Object proxy, final Method method, Object[] args)
         traceScope = tracer.newScope(RpcClientUtil.methodToTraceString(method));
       }
 
-      RequestHeaderProto rpcRequestHeader = constructRpcRequestHeader(method);
-
       if (LOG.isTraceEnabled()) {
         LOG.trace(Thread.currentThread().getId() + ": Call -> " +
             remoteId + ": " + method.getName() +
@@ -238,7 +236,7 @@ public Message invoke(Object proxy, final Method method, Object[] args)
       final RpcWritable.Buffer val;
       try {
         val = (RpcWritable.Buffer) client.call(RPC.RpcKind.RPC_PROTOCOL_BUFFER,
-            new RpcProtobufRequest(rpcRequestHeader, theRequest), remoteId,
+            constructRpcRequest(method, theRequest), remoteId,
             fallbackToSimpleAuth, alignmentContext);
 
       } catch (Throwable e) {
@@ -283,6 +281,11 @@ public boolean isDone() {
       }
     }
 
+    protected Writable constructRpcRequest(Method method, Message theRequest) {
+      RequestHeaderProto rpcRequestHeader = constructRpcRequestHeader(method);
+      return new RpcProtobufRequest(rpcRequestHeader, theRequest);
+    }
+
     private Message getReturnMessage(final Method method,
         final RpcWritable.Buffer buf) throws ServiceException {
       Message prototype = null;
@@ -332,6 +335,14 @@ private Message getReturnProtoType(Method method) throws Exception {
     public ConnectionId getConnectionId() {
       return remoteId;
     }
+
+    protected long getClientProtocolVersion() {
+      return clientProtocolVersion;
+    }
+
+    protected String getProtocolName() {
+      return protocolName;
+    }
   }
 
   @VisibleForTesting
@@ -518,6 +529,13 @@ public Writable call(RPC.Server server, String connectionProtocolName,
         String declaringClassProtoName = 
             rpcRequest.getDeclaringClassProtocolName();
         long clientVersion = rpcRequest.getClientProtocolVersion();
+        return call(server, connectionProtocolName, request, receiveTime,
+            methodName, declaringClassProtoName, clientVersion);
+      }
+
+      protected Writable call(RPC.Server server, String connectionProtocolName,
+          RpcWritable.Buffer request, long receiveTime, String methodName,
+          String declaringClassProtoName, long clientVersion) throws Exception {
         if (server.verbose)
           LOG.info("Call: connectionProtocolName=" + connectionProtocolName + 
               ", method=" + methodName);

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

@@ -116,7 +116,7 @@ public ProtocolProxy<ProtocolMetaInfoPB> getProtocolMetaInfoProxy(
                 factory)), false);
   }
 
-  private static final class Invoker implements RpcInvocationHandler {
+  protected static class Invoker implements RpcInvocationHandler {
     private final Map<String, Message> returnTypes =
         new ConcurrentHashMap<String, Message>();
     private boolean isClosed = false;
@@ -127,7 +127,7 @@ private static final class Invoker implements RpcInvocationHandler {
     private AtomicBoolean fallbackToSimpleAuth;
     private AlignmentContext alignmentContext;
 
-    private Invoker(Class<?> protocol, InetSocketAddress addr,
+    protected Invoker(Class<?> protocol, InetSocketAddress addr,
         UserGroupInformation ticket, Configuration conf, SocketFactory factory,
         int rpcTimeout, RetryPolicy connectionRetryPolicy,
         AtomicBoolean fallbackToSimpleAuth, AlignmentContext alignmentContext)
@@ -142,7 +142,7 @@ private Invoker(Class<?> protocol, InetSocketAddress addr,
     /**
      * This constructor takes a connectionId, instead of creating a new one.
      */
-    private Invoker(Class<?> protocol, Client.ConnectionId connId,
+    protected Invoker(Class<?> protocol, Client.ConnectionId connId,
         Configuration conf, SocketFactory factory) {
       this.remoteId = connId;
       this.client = CLIENTS.getClient(conf, factory, RpcWritable.Buffer.class);
@@ -219,8 +219,6 @@ public Message invoke(Object proxy, final Method method, Object[] args)
         traceScope = tracer.newScope(RpcClientUtil.methodToTraceString(method));
       }
 
-      RequestHeaderProto rpcRequestHeader = constructRpcRequestHeader(method);
-
       if (LOG.isTraceEnabled()) {
         LOG.trace(Thread.currentThread().getId() + ": Call -> " +
             remoteId + ": " + method.getName() +
@@ -232,7 +230,7 @@ public Message invoke(Object proxy, final Method method, Object[] args)
       final RpcWritable.Buffer val;
       try {
         val = (RpcWritable.Buffer) client.call(RPC.RpcKind.RPC_PROTOCOL_BUFFER,
-            new RpcProtobufRequest(rpcRequestHeader, theRequest), remoteId,
+            constructRpcRequest(method, theRequest), remoteId,
             fallbackToSimpleAuth, alignmentContext);
 
       } catch (Throwable e) {
@@ -279,6 +277,11 @@ public boolean isDone() {
       }
     }
 
+    protected Writable constructRpcRequest(Method method, Message theRequest) {
+      RequestHeaderProto rpcRequestHeader = constructRpcRequestHeader(method);
+      return new RpcProtobufRequest(rpcRequestHeader, theRequest);
+    }
+
     private Message getReturnMessage(final Method method,
         final RpcWritable.Buffer buf) throws ServiceException {
       Message prototype = null;
@@ -328,6 +331,14 @@ private Message getReturnProtoType(Method method) throws Exception {
     public ConnectionId getConnectionId() {
       return remoteId;
     }
+
+    protected long getClientProtocolVersion() {
+      return clientProtocolVersion;
+    }
+
+    protected String getProtocolName() {
+      return protocolName;
+    }
   }
 
   @VisibleForTesting
@@ -509,6 +520,13 @@ public Writable call(RPC.Server server, String connectionProtocolName,
         String declaringClassProtoName =
             rpcRequest.getDeclaringClassProtocolName();
         long clientVersion = rpcRequest.getClientProtocolVersion();
+        return call(server, connectionProtocolName, request, receiveTime,
+            methodName, declaringClassProtoName, clientVersion);
+      }
+
+      protected Writable call(RPC.Server server, String connectionProtocolName,
+          RpcWritable.Buffer request, long receiveTime, String methodName,
+          String declaringClassProtoName, long clientVersion) throws Exception {
         if (server.verbose) {
           LOG.info("Call: connectionProtocolName=" + connectionProtocolName +
               ", method=" + methodName);

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/security/LdapGroupsMapping.java

@@ -643,7 +643,29 @@ private DirContext getDirContext() throws NamingException {
       env.put("com.sun.jndi.ldap.read.timeout", conf.get(READ_TIMEOUT,
           String.valueOf(READ_TIMEOUT_DEFAULT)));
 
-      ctx = new InitialDirContext(env);
+      // See HADOOP-17675 for details TLDR:
+      // From a native thread the thread's context classloader is null.
+      // jndi internally in the InitialDirContext specifies the context
+      // classloader for Class.forName, and as it is null, jndi will use the
+      // bootstrap classloader in this case to laod the socket factory
+      // implementation.
+      // BUT
+      // Bootstrap classloader does not have it in its classpath, so throws a
+      // ClassNotFoundException.
+      // This affects Impala for example when it uses LdapGroupsMapping.
+      ClassLoader currentContextLoader =
+          Thread.currentThread().getContextClassLoader();
+      if (currentContextLoader == null) {
+        try {
+          Thread.currentThread().setContextClassLoader(
+              this.getClass().getClassLoader());
+          ctx = new InitialDirContext(env);
+        } finally {
+          Thread.currentThread().setContextClassLoader(null);
+        }
+      } else {
+        ctx = new InitialDirContext(env);
+      }
     }
     return ctx;
   }