hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/server/namenode/ha/RequestHedgingProxyProvider.java - hadoop - Git at Google

 /**
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
  * distributed with this work for additional information
  * regarding copyright ownership.  The ASF licenses this file
  * to you under the Apache License, Version 2.0 (the
  * "License"); you may not use this file except in compliance
  * with the License.  You may obtain a copy of the License at
  * <p>
  * http://www.apache.org/licenses/LICENSE-2.0
  * <p>
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
 package org.apache.hadoop.hdfs.server.namenode.ha;

 import java.io.IOException;
 import java.lang.reflect.InvocationTargetException;
 import java.lang.reflect.Method;
 import java.lang.reflect.Proxy;
 import java.net.URI;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.concurrent.Callable;
 import java.util.concurrent.CompletionService;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorCompletionService;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.Future;

 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.io.retry.MultiException;
 import org.apache.hadoop.ipc.Client;
 import org.apache.hadoop.ipc.Client.ConnectionId;
 import org.apache.hadoop.ipc.RPC;
 import org.apache.hadoop.ipc.RemoteException;
 import org.apache.hadoop.ipc.RpcInvocationHandler;
 import org.apache.hadoop.ipc.StandbyException;

 /**
  * A FailoverProxyProvider implementation that technically does not "failover"
  * per-se. It constructs a wrapper proxy that sends the request to ALL
  * underlying proxies simultaneously. It assumes the in an HA setup, there will
  * be only one Active, and the active should respond faster than any configured
  * standbys. Once it receive a response from any one of the configured proxies,
  * outstanding requests to other proxies are immediately cancelled.
  */
 public class RequestHedgingProxyProvider<T> extends
         ConfiguredFailoverProxyProvider<T> {

   public static final Logger LOG =
       LoggerFactory.getLogger(RequestHedgingProxyProvider.class);

   class RequestHedgingInvocationHandler implements RpcInvocationHandler {

     final Map<String, ProxyInfo<T>> targetProxies;
     // Proxy of the active nn
     private volatile ProxyInfo<T> currentUsedProxy = null;

     public RequestHedgingInvocationHandler(
             Map<String, ProxyInfo<T>> targetProxies) {
       this.targetProxies = new HashMap<>(targetProxies);
     }

     /**
      * Creates a Executor and invokes all proxies concurrently. This
      * implementation assumes that Clients have configured proper socket
      * timeouts, else the call can block forever.
      *
      * @param proxy
      * @param method
      * @param args
      * @return
      * @throws Throwable
      */
     @Override
     public Object
     invoke(Object proxy, final Method method, final Object[] args)
             throws Throwable {
       // Need double check locking to guarantee thread-safe since
       // currentUsedProxy is lazily initialized.
       if (currentUsedProxy == null) {
         synchronized (this) {
           if (currentUsedProxy == null) {
             Map<Future<Object>, ProxyInfo<T>> proxyMap = new HashMap<>();
             int numAttempts = 0;

             ExecutorService executor = null;
             CompletionService<Object> completionService;
             try {
               // Optimization : if only 2 proxies are configured and one had
               // failed
               // over, then we dont need to create a threadpool etc.
               targetProxies.remove(toIgnore);
               if (targetProxies.size() == 0) {
                 LOG.trace("No valid proxies left");
                 throw new RemoteException(IOException.class.getName(),
                     "No valid proxies left. "
                         + "All NameNode proxies have failed over.");
               }
               if (targetProxies.size() == 1) {
                 ProxyInfo<T> proxyInfo =
                     targetProxies.values().iterator().next();
                 try {
                   currentUsedProxy = proxyInfo;
                   Object retVal = method.invoke(proxyInfo.proxy, args);
                   LOG.debug("Invocation successful on [{}]",
                       currentUsedProxy.proxyInfo);
                   return retVal;
                 } catch (InvocationTargetException ex) {
                   Exception unwrappedException =
                       unwrapInvocationTargetException(ex);
                   logProxyException(unwrappedException,
                       currentUsedProxy.proxyInfo);
                   LOG.trace("Unsuccessful invocation on [{}]",
                       currentUsedProxy.proxyInfo);
                   throw unwrappedException;
                 }
               }
               executor = Executors.newFixedThreadPool(proxies.size());
               completionService = new ExecutorCompletionService<>(executor);
               // Set the callId and other informations from current thread.
               final int callId = Client.getCallId();
               final int retryCount = Client.getRetryCount();
               final Object externalHandler = Client.getExternalHandler();
               for (final Map.Entry<String, ProxyInfo<T>> pEntry : targetProxies
                   .entrySet()) {
                 Callable<Object> c = new Callable<Object>() {
                   @Override
                   public Object call() throws Exception {
                     // Call Id and other informations from parent thread.
                     Client.setCallIdAndRetryCount(callId, retryCount,
                         externalHandler);
                     LOG.trace("Invoking method {} on proxy {}", method,
                         pEntry.getValue().proxyInfo);
                     return method.invoke(pEntry.getValue().proxy, args);
                   }
                 };
                 proxyMap.put(completionService.submit(c), pEntry.getValue());
                 numAttempts++;
               }
               // Current thread's callId will not be cleared as RPC happens in
               // separate threads. Reset the CallId information Forcefully.
               Client.setCallIdAndRetryCountUnprotected(null, 0, null);
               Map<String, Exception> badResults = new HashMap<>();
               while (numAttempts > 0) {
                 Future<Object> callResultFuture = completionService.take();
                 Object retVal;
                 try {
                   currentUsedProxy = proxyMap.get(callResultFuture);
                   retVal = callResultFuture.get();
                   LOG.debug("Invocation successful on [{}]",
                       currentUsedProxy.proxyInfo);
                   return retVal;
                 } catch (ExecutionException ex) {
                   Exception unwrappedException = unwrapExecutionException(ex);
                   ProxyInfo<T> tProxyInfo = proxyMap.get(callResultFuture);
                   logProxyException(unwrappedException, tProxyInfo.proxyInfo);
                   badResults.put(tProxyInfo.proxyInfo, unwrappedException);
                   LOG.trace("Unsuccessful invocation on [{}]",
                       tProxyInfo.proxyInfo);
                   numAttempts--;
                 }
               }

               // At this point we should have All bad results (Exceptions)
               // Or should have returned with successful result.
               if (badResults.size() == 1) {
                 throw badResults.values().iterator().next();
               } else {
                 throw new MultiException(badResults);
               }
             } finally {
               if (executor != null) {
                 LOG.trace("Shutting down threadpool executor");
                 executor.shutdownNow();
               }
             }
           }
         }
       }
       // Because the above synchronized block will return or throw an exception,
       // so we don't need to do any check to prevent the first initialized
       // thread from stepping to following codes.
       try {
         Object retVal = method.invoke(currentUsedProxy.proxy, args);
         LOG.debug("Invocation successful on [{}]", currentUsedProxy.proxyInfo);
         return retVal;
       } catch (InvocationTargetException ex) {
         Exception unwrappedException = unwrapInvocationTargetException(ex);
         logProxyException(unwrappedException, currentUsedProxy.proxyInfo);
         LOG.trace("Unsuccessful invocation on [{}]",
             currentUsedProxy.proxyInfo);
         throw unwrappedException;
       }
     }

     @Override
     public void close() throws IOException {
     }

     @Override
     public ConnectionId getConnectionId() {
       if (currentUsedProxy == null) {
         return null;
       }
       return RPC.getConnectionIdForProxy(currentUsedProxy.proxy);
     }
   }

   /** A proxy wrapping {@link RequestHedgingInvocationHandler}. */
   private ProxyInfo<T> currentUsedHandler = null;
   private volatile String toIgnore = null;

   public RequestHedgingProxyProvider(Configuration conf, URI uri,
       Class<T> xface, HAProxyFactory<T> proxyFactory) {
     super(conf, uri, xface, proxyFactory);
   }

   @SuppressWarnings("unchecked")
   @Override
   public synchronized ProxyInfo<T> getProxy() {
     if (currentUsedHandler != null) {
       return currentUsedHandler;
     }
     Map<String, ProxyInfo<T>> targetProxyInfos = new HashMap<>();
     StringBuilder combinedInfo = new StringBuilder("[");
     for (int i = 0; i < proxies.size(); i++) {
       ProxyInfo<T> pInfo = super.getProxy();
       incrementProxyIndex();
       targetProxyInfos.put(pInfo.proxyInfo, pInfo);
       combinedInfo.append(pInfo.proxyInfo).append(',');
     }
     combinedInfo.append(']');
     T wrappedProxy = (T) Proxy.newProxyInstance(
             RequestHedgingInvocationHandler.class.getClassLoader(),
             new Class<?>[]{xface},
             new RequestHedgingInvocationHandler(targetProxyInfos));
     currentUsedHandler =
         new ProxyInfo<T>(wrappedProxy, combinedInfo.toString());
     return currentUsedHandler;
   }

   @Override
   public synchronized void performFailover(T currentProxy) {
     toIgnore = ((RequestHedgingInvocationHandler) Proxy.getInvocationHandler(
         currentUsedHandler.proxy)).currentUsedProxy.proxyInfo;
     this.currentUsedHandler = null;
   }

   /**
    * Check the exception returned by the proxy log a warning message if it's
    * not a StandbyException (expected exception).
    * @param ex Exception to evaluate.
    * @param proxyInfo Information of the proxy reporting the exception.
    */
   private void logProxyException(Exception ex, String proxyInfo) {
     if (isStandbyException(ex)) {
       LOG.debug("Invocation returned standby exception on [{}]", proxyInfo, ex);
     } else {
       LOG.warn("Invocation returned exception on [{}]", proxyInfo, ex);
     }
   }

   /**
    * Check if the returned exception is caused by an standby namenode.
    * @param exception Exception to check.
    * @return If the exception is caused by an standby namenode.
    */
   private boolean isStandbyException(Exception exception) {
     if (exception instanceof RemoteException) {
       return ((RemoteException) exception).unwrapRemoteException()
           instanceof StandbyException;
     }
     return false;
   }

   /**
    * Unwraps the ExecutionException. <p>
    * Example:
    * <blockquote><pre>
    * if ex is
    * ExecutionException(InvocationTargetException(SomeException))
    * returns SomeException
    * </pre></blockquote>
    *
    * @return unwrapped exception
    */
   private Exception unwrapExecutionException(ExecutionException ex) {
     if (ex != null) {
       Throwable cause = ex.getCause();
       if (cause instanceof InvocationTargetException) {
         return
             unwrapInvocationTargetException((InvocationTargetException)cause);
       }
     }
     return ex;

   }

   /**
    * Unwraps the InvocationTargetException. <p>
    * Example:
    * <blockquote><pre>
    * if ex is InvocationTargetException(SomeException)
    * returns SomeException
    * </pre></blockquote>
    *
    * @return unwrapped exception
    */
   private Exception unwrapInvocationTargetException(
       InvocationTargetException ex) {
     if (ex != null) {
       Throwable cause = ex.getCause();
       if (cause instanceof Exception) {
         return (Exception) cause;
       }
     }
     return ex;
   }
 }
	/**
	* Licensed to the Apache Software Foundation (ASF) under one
	* or more contributor license agreements. See the NOTICE file
	* distributed with this work for additional information
	* regarding copyright ownership. The ASF licenses this file
	* to you under the Apache License, Version 2.0 (the
	* "License"); you may not use this file except in compliance
	* with the License. You may obtain a copy of the License at
	* <p>
	* http://www.apache.org/licenses/LICENSE-2.0
	* <p>
	* Unless required by applicable law or agreed to in writing, software
	* distributed under the License is distributed on an "AS IS" BASIS,
	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
	* See the License for the specific language governing permissions and
	* limitations under the License.
	*/
	package org.apache.hadoop.hdfs.server.namenode.ha;

	import java.io.IOException;
	import java.lang.reflect.InvocationTargetException;
	import java.lang.reflect.Method;
	import java.lang.reflect.Proxy;
	import java.net.URI;
	import java.util.HashMap;
	import java.util.Map;
	import java.util.concurrent.Callable;
	import java.util.concurrent.CompletionService;
	import java.util.concurrent.ExecutionException;
	import java.util.concurrent.ExecutorCompletionService;
	import java.util.concurrent.ExecutorService;
	import java.util.concurrent.Executors;
	import java.util.concurrent.Future;

	import org.slf4j.Logger;
	import org.slf4j.LoggerFactory;

	import org.apache.hadoop.conf.Configuration;
	import org.apache.hadoop.io.retry.MultiException;
	import org.apache.hadoop.ipc.Client;
	import org.apache.hadoop.ipc.Client.ConnectionId;
	import org.apache.hadoop.ipc.RPC;
	import org.apache.hadoop.ipc.RemoteException;
	import org.apache.hadoop.ipc.RpcInvocationHandler;
	import org.apache.hadoop.ipc.StandbyException;

	/**
	* A FailoverProxyProvider implementation that technically does not "failover"
	* per-se. It constructs a wrapper proxy that sends the request to ALL
	* underlying proxies simultaneously. It assumes the in an HA setup, there will
	* be only one Active, and the active should respond faster than any configured
	* standbys. Once it receive a response from any one of the configured proxies,
	* outstanding requests to other proxies are immediately cancelled.
	*/
	public class RequestHedgingProxyProvider<T> extends
	ConfiguredFailoverProxyProvider<T> {

	public static final Logger LOG =
	LoggerFactory.getLogger(RequestHedgingProxyProvider.class);

	class RequestHedgingInvocationHandler implements RpcInvocationHandler {

	final Map<String, ProxyInfo<T>> targetProxies;
	// Proxy of the active nn
	private volatile ProxyInfo<T> currentUsedProxy = null;

	public RequestHedgingInvocationHandler(
	Map<String, ProxyInfo<T>> targetProxies) {
	this.targetProxies = new HashMap<>(targetProxies);
	}

	/**
	* Creates a Executor and invokes all proxies concurrently. This
	* implementation assumes that Clients have configured proper socket
	* timeouts, else the call can block forever.
	*
	* @param proxy
	* @param method
	* @param args
	* @return
	* @throws Throwable
	*/
	@Override
	public Object
	invoke(Object proxy, final Method method, final Object[] args)
	throws Throwable {
	// Need double check locking to guarantee thread-safe since
	// currentUsedProxy is lazily initialized.
	if (currentUsedProxy == null) {
	synchronized (this) {
	if (currentUsedProxy == null) {
	Map<Future<Object>, ProxyInfo<T>> proxyMap = new HashMap<>();
	int numAttempts = 0;

	ExecutorService executor = null;
	CompletionService<Object> completionService;
	try {
	// Optimization : if only 2 proxies are configured and one had
	// failed
	// over, then we dont need to create a threadpool etc.
	targetProxies.remove(toIgnore);
	if (targetProxies.size() == 0) {
	LOG.trace("No valid proxies left");
	throw new RemoteException(IOException.class.getName(),
	"No valid proxies left. "
	+ "All NameNode proxies have failed over.");
	}
	if (targetProxies.size() == 1) {
	ProxyInfo<T> proxyInfo =
	targetProxies.values().iterator().next();
	try {
	currentUsedProxy = proxyInfo;
	Object retVal = method.invoke(proxyInfo.proxy, args);
	LOG.debug("Invocation successful on [{}]",
	currentUsedProxy.proxyInfo);
	return retVal;
	} catch (InvocationTargetException ex) {
	Exception unwrappedException =
	unwrapInvocationTargetException(ex);
	logProxyException(unwrappedException,
	currentUsedProxy.proxyInfo);
	LOG.trace("Unsuccessful invocation on [{}]",
	currentUsedProxy.proxyInfo);
	throw unwrappedException;
	}
	}
	executor = Executors.newFixedThreadPool(proxies.size());
	completionService = new ExecutorCompletionService<>(executor);
	// Set the callId and other informations from current thread.
	final int callId = Client.getCallId();
	final int retryCount = Client.getRetryCount();
	final Object externalHandler = Client.getExternalHandler();
	for (final Map.Entry<String, ProxyInfo<T>> pEntry : targetProxies
	.entrySet()) {
	Callable<Object> c = new Callable<Object>() {
	@Override
	public Object call() throws Exception {
	// Call Id and other informations from parent thread.
	Client.setCallIdAndRetryCount(callId, retryCount,
	externalHandler);
	LOG.trace("Invoking method {} on proxy {}", method,
	pEntry.getValue().proxyInfo);
	return method.invoke(pEntry.getValue().proxy, args);
	}
	};
	proxyMap.put(completionService.submit(c), pEntry.getValue());
	numAttempts++;
	}
	// Current thread's callId will not be cleared as RPC happens in
	// separate threads. Reset the CallId information Forcefully.
	Client.setCallIdAndRetryCountUnprotected(null, 0, null);
	Map<String, Exception> badResults = new HashMap<>();
	while (numAttempts > 0) {
	Future<Object> callResultFuture = completionService.take();
	Object retVal;
	try {
	currentUsedProxy = proxyMap.get(callResultFuture);
	retVal = callResultFuture.get();
	LOG.debug("Invocation successful on [{}]",
	currentUsedProxy.proxyInfo);
	return retVal;
	} catch (ExecutionException ex) {
	Exception unwrappedException = unwrapExecutionException(ex);
	ProxyInfo<T> tProxyInfo = proxyMap.get(callResultFuture);
	logProxyException(unwrappedException, tProxyInfo.proxyInfo);
	badResults.put(tProxyInfo.proxyInfo, unwrappedException);
	LOG.trace("Unsuccessful invocation on [{}]",
	tProxyInfo.proxyInfo);
	numAttempts--;
	}
	}

	// At this point we should have All bad results (Exceptions)
	// Or should have returned with successful result.
	if (badResults.size() == 1) {
	throw badResults.values().iterator().next();
	} else {
	throw new MultiException(badResults);
	}
	} finally {
	if (executor != null) {
	LOG.trace("Shutting down threadpool executor");
	executor.shutdownNow();
	}
	}
	}
	}
	}
	// Because the above synchronized block will return or throw an exception,
	// so we don't need to do any check to prevent the first initialized
	// thread from stepping to following codes.
	try {
	Object retVal = method.invoke(currentUsedProxy.proxy, args);
	LOG.debug("Invocation successful on [{}]", currentUsedProxy.proxyInfo);
	return retVal;
	} catch (InvocationTargetException ex) {
	Exception unwrappedException = unwrapInvocationTargetException(ex);
	logProxyException(unwrappedException, currentUsedProxy.proxyInfo);
	LOG.trace("Unsuccessful invocation on [{}]",
	currentUsedProxy.proxyInfo);
	throw unwrappedException;
	}
	}

	@Override
	public void close() throws IOException {
	}

	@Override
	public ConnectionId getConnectionId() {
	if (currentUsedProxy == null) {
	return null;
	}
	return RPC.getConnectionIdForProxy(currentUsedProxy.proxy);
	}
	}

	/** A proxy wrapping {@link RequestHedgingInvocationHandler}. */
	private ProxyInfo<T> currentUsedHandler = null;
	private volatile String toIgnore = null;

	public RequestHedgingProxyProvider(Configuration conf, URI uri,
	Class<T> xface, HAProxyFactory<T> proxyFactory) {
	super(conf, uri, xface, proxyFactory);
	}

	@SuppressWarnings("unchecked")
	@Override
	public synchronized ProxyInfo<T> getProxy() {
	if (currentUsedHandler != null) {
	return currentUsedHandler;
	}
	Map<String, ProxyInfo<T>> targetProxyInfos = new HashMap<>();
	StringBuilder combinedInfo = new StringBuilder("[");
	for (int i = 0; i < proxies.size(); i++) {
	ProxyInfo<T> pInfo = super.getProxy();
	incrementProxyIndex();
	targetProxyInfos.put(pInfo.proxyInfo, pInfo);
	combinedInfo.append(pInfo.proxyInfo).append(',');
	}
	combinedInfo.append(']');
	T wrappedProxy = (T) Proxy.newProxyInstance(
	RequestHedgingInvocationHandler.class.getClassLoader(),
	new Class<?>[]{xface},
	new RequestHedgingInvocationHandler(targetProxyInfos));
	currentUsedHandler =
	new ProxyInfo<T>(wrappedProxy, combinedInfo.toString());
	return currentUsedHandler;
	}

	@Override
	public synchronized void performFailover(T currentProxy) {
	toIgnore = ((RequestHedgingInvocationHandler) Proxy.getInvocationHandler(
	currentUsedHandler.proxy)).currentUsedProxy.proxyInfo;
	this.currentUsedHandler = null;
	}

	/**
	* Check the exception returned by the proxy log a warning message if it's
	* not a StandbyException (expected exception).
	* @param ex Exception to evaluate.
	* @param proxyInfo Information of the proxy reporting the exception.
	*/
	private void logProxyException(Exception ex, String proxyInfo) {
	if (isStandbyException(ex)) {
	LOG.debug("Invocation returned standby exception on [{}]", proxyInfo, ex);
	} else {
	LOG.warn("Invocation returned exception on [{}]", proxyInfo, ex);
	}
	}

	/**
	* Check if the returned exception is caused by an standby namenode.
	* @param exception Exception to check.
	* @return If the exception is caused by an standby namenode.
	*/
	private boolean isStandbyException(Exception exception) {
	if (exception instanceof RemoteException) {
	return ((RemoteException) exception).unwrapRemoteException()
	instanceof StandbyException;
	}
	return false;
	}

	/**
	* Unwraps the ExecutionException. <p>
	* Example:
	* <blockquote><pre>
	* if ex is
	* ExecutionException(InvocationTargetException(SomeException))
	* returns SomeException
	* </pre></blockquote>
	*
	* @return unwrapped exception
	*/
	private Exception unwrapExecutionException(ExecutionException ex) {
	if (ex != null) {
	Throwable cause = ex.getCause();
	if (cause instanceof InvocationTargetException) {
	return
	unwrapInvocationTargetException((InvocationTargetException)cause);
	}
	}
	return ex;

	}

	/**
	* Unwraps the InvocationTargetException. <p>
	* Example:
	* <blockquote><pre>
	* if ex is InvocationTargetException(SomeException)
	* returns SomeException
	* </pre></blockquote>
	*
	* @return unwrapped exception
	*/
	private Exception unwrapInvocationTargetException(
	InvocationTargetException ex) {
	if (ex != null) {
	Throwable cause = ex.getCause();
	if (cause instanceof Exception) {
	return (Exception) cause;
	}
	}
	return ex;
	}
	}