accumulo-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From ctubb...@apache.org
Subject [4/6] accumulo git commit: Merge branch '1.6' into 1.7
Date Thu, 03 Dec 2015 21:08:20 GMT
Merge branch '1.6' into 1.7


Project: http://git-wip-us.apache.org/repos/asf/accumulo/repo
Commit: http://git-wip-us.apache.org/repos/asf/accumulo/commit/0e3af1e8
Tree: http://git-wip-us.apache.org/repos/asf/accumulo/tree/0e3af1e8
Diff: http://git-wip-us.apache.org/repos/asf/accumulo/diff/0e3af1e8

Branch: refs/heads/master
Commit: 0e3af1e8560b0a28a4415f8da38a561965d552dc
Parents: e35e189 bd8cf5e
Author: Christopher Tubbs <ctubbsii@apache.org>
Authored: Thu Dec 3 15:36:49 2015 -0500
Committer: Christopher Tubbs <ctubbsii@apache.org>
Committed: Thu Dec 3 15:36:49 2015 -0500

----------------------------------------------------------------------
 .../src/main/java/org/apache/accumulo/proxy/Proxy.java  |  2 +-
 .../java/org/apache/accumulo/server/rpc/RpcWrapper.java | 12 ++++++------
 .../org/apache/accumulo/server/rpc/RpcWrapperTest.java  |  9 +++------
 .../org/apache/accumulo/gc/SimpleGarbageCollector.java  |  2 +-
 .../main/java/org/apache/accumulo/master/Master.java    |  4 ++--
 .../java/org/apache/accumulo/tserver/TabletServer.java  |  4 ++--
 6 files changed, 15 insertions(+), 18 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/accumulo/blob/0e3af1e8/proxy/src/main/java/org/apache/accumulo/proxy/Proxy.java
----------------------------------------------------------------------
diff --cc proxy/src/main/java/org/apache/accumulo/proxy/Proxy.java
index ca39786,53221e0..87e2c58
--- a/proxy/src/main/java/org/apache/accumulo/proxy/Proxy.java
+++ b/proxy/src/main/java/org/apache/accumulo/proxy/Proxy.java
@@@ -161,113 -117,38 +161,113 @@@ public class Proxy implements KeywordEx
  
      Class<? extends TProtocolFactory> protoFactoryClass = Class.forName(opts.prop.getProperty("protocolFactory",
TCompactProtocol.Factory.class.getName()))
          .asSubclass(TProtocolFactory.class);
 +    TProtocolFactory protoFactory = protoFactoryClass.newInstance();
      int port = Integer.parseInt(opts.prop.getProperty("port"));
 -    TServer server = createProxyServer(AccumuloProxy.class, ProxyServer.class, port, protoFactoryClass,
opts.prop);
 -    server.serve();
 +    String hostname = opts.prop.getProperty(THRIFT_SERVER_HOSTNAME, THRIFT_SERVER_HOSTNAME_DEFAULT);
 +    HostAndPort address = HostAndPort.fromParts(hostname, port);
 +    ServerAddress server = createProxyServer(address, protoFactory, opts.prop);
 +    // Wait for the server to come up
 +    while (!server.server.isServing()) {
 +      Thread.sleep(100);
 +    }
 +    log.info("Proxy server started on " + server.getAddress());
 +    while (server.server.isServing()) {
 +      Thread.sleep(1000);
 +    }
 +  }
 +
 +  public static void main(String[] args) throws Exception {
 +    new Proxy().execute(args);
    }
  
 -  public static TServer createProxyServer(Class<?> api, Class<?> implementor,
final int port, Class<? extends TProtocolFactory> protoClass,
 -      Properties properties) throws Exception {
 -    Class<?> proxyIfaceClass = Class.forName(api.getName() + "$Iface");
 -    return createProxyServer(proxyIfaceClass, api, implementor, port, protoClass, properties);
 +  public static ServerAddress createProxyServer(HostAndPort address, TProtocolFactory protocolFactory,
Properties properties) throws Exception {
 +    return createProxyServer(address, protocolFactory, properties, ClientConfiguration.loadDefault());
    }
  
 -  private static <I,P extends TBaseProcessor<I>> TServer createProxyServer(final
Class<I> proxyIfaceClass, Class<?> api, final Class<?> implementor,
 -      final int port, final Class<? extends TProtocolFactory> protoClass, final Properties
properties) throws Exception {
 -    @SuppressWarnings("unchecked")
 -    Class<P> proxyProcClass = (Class<P>) Class.forName(api.getName() + "$Processor");
 -
 -    // create the implementor
 -    @SuppressWarnings("unchecked")
 -    I impl = (I) implementor.getConstructor(Properties.class).newInstance(properties);
 -
 -    Constructor<P> proxyProcConstructor = proxyProcClass.getConstructor(proxyIfaceClass);
 -    P processor = proxyProcConstructor.newInstance(RpcWrapper.service(impl, proxyProcConstructor.newInstance(impl)));
 -
 -    TNonblockingServerSocket socket = new TNonblockingServerSocket(port);
 -    THsHaServer.Args args = new THsHaServer.Args(socket);
 -    args.processor(processor);
 -    final long maxFrameSize = AccumuloConfiguration.getMemoryInBytes(properties.getProperty("maxFrameSize",
"16M"));
 -    if (maxFrameSize > Integer.MAX_VALUE)
 -      throw new RuntimeException(maxFrameSize + " is larger than MAX_INT");
 -    args.transportFactory(new TFramedTransport.Factory((int) maxFrameSize));
 -    args.protocolFactory(protoClass.newInstance());
 -    args.maxReadBufferBytes = maxFrameSize;
 -    return new THsHaServer(args);
 +  public static ServerAddress createProxyServer(HostAndPort address, TProtocolFactory protocolFactory,
Properties properties, ClientConfiguration clientConf)
 +      throws Exception {
 +    final int numThreads = Integer.parseInt(properties.getProperty(THRIFT_THREAD_POOL_SIZE_KEY,
THRIFT_THREAD_POOL_SIZE_DEFAULT));
 +    final long maxFrameSize = AccumuloConfiguration.getMemoryInBytes(properties.getProperty(THRIFT_MAX_FRAME_SIZE_KEY,
THRIFT_MAX_FRAME_SIZE_DEFAULT));
 +    final int simpleTimerThreadpoolSize = Integer.parseInt(Property.GENERAL_SIMPLETIMER_THREADPOOL_SIZE.getDefaultValue());
 +    // How frequently to try to resize the thread pool
 +    final long threadpoolResizeInterval = 1000l * 5;
 +    // No timeout
 +    final long serverSocketTimeout = 0l;
 +    // Use the new hadoop metrics2 support
 +    final MetricsFactory metricsFactory = new MetricsFactory(false);
 +    final String serverName = "Proxy", threadName = "Accumulo Thrift Proxy";
 +
 +    // create the implementation of the proxy interface
 +    ProxyServer impl = new ProxyServer(properties);
 +
 +    // Wrap the implementation -- translate some exceptions
-     AccumuloProxy.Iface wrappedImpl = RpcWrapper.service(impl, new AccumuloProxy.Processor<AccumuloProxy.Iface>(impl).getProcessMapView());
++    AccumuloProxy.Iface wrappedImpl = RpcWrapper.service(impl, new AccumuloProxy.Processor<AccumuloProxy.Iface>(impl));
 +
 +    // Create the processor from the implementation
 +    TProcessor processor = new AccumuloProxy.Processor<AccumuloProxy.Iface>(wrappedImpl);
 +
 +    // Get the type of thrift server to instantiate
 +    final String serverTypeStr = properties.getProperty(THRIFT_SERVER_TYPE, THRIFT_SERVER_TYPE_DEFAULT);
 +    ThriftServerType serverType = DEFAULT_SERVER_TYPE;
 +    if (!THRIFT_SERVER_TYPE_DEFAULT.equals(serverTypeStr)) {
 +      serverType = ThriftServerType.get(serverTypeStr);
 +    }
 +
 +    SslConnectionParams sslParams = null;
 +    SaslServerConnectionParams saslParams = null;
 +    switch (serverType) {
 +      case SSL:
 +        sslParams = SslConnectionParams.forClient(ClientContext.convertClientConfig(clientConf));
 +        break;
 +      case SASL:
 +        if (!clientConf.getBoolean(ClientProperty.INSTANCE_RPC_SASL_ENABLED.getKey(), false))
{
 +          // ACCUMULO-3651 Changed level to error and added FATAL to message for slf4j capability
 +          log.error("FATAL: SASL thrift server was requested but it is disabled in client
configuration");
 +          throw new RuntimeException("SASL is not enabled in configuration");
 +        }
 +
 +        // Kerberos needs to be enabled to use it
 +        if (!UserGroupInformation.isSecurityEnabled()) {
 +          // ACCUMULO-3651 Changed level to error and added FATAL to message for slf4j capability
 +          log.error("FATAL: Hadoop security is not enabled");
 +          throw new RuntimeException();
 +        }
 +
 +        // Login via principal and keytab
 +        final String kerberosPrincipal = properties.getProperty(KERBEROS_PRINCIPAL, ""),
 +        kerberosKeytab = properties.getProperty(KERBEROS_KEYTAB, "");
 +        if (StringUtils.isBlank(kerberosPrincipal) || StringUtils.isBlank(kerberosKeytab))
{
 +          // ACCUMULO-3651 Changed level to error and added FATAL to message for slf4j capability
 +          log.error("FATAL: Kerberos principal and keytab must be provided");
 +          throw new RuntimeException();
 +        }
 +        UserGroupInformation.loginUserFromKeytab(kerberosPrincipal, kerberosKeytab);
 +        UserGroupInformation ugi = UserGroupInformation.getCurrentUser();
 +        log.info("Logged in as " + ugi.getUserName());
 +
 +        // The kerberosPrimary set in the SASL server needs to match the principal we're
logged in as.
 +        final String shortName = ugi.getShortUserName();
 +        log.info("Setting server primary to {}", shortName);
 +        clientConf.setProperty(ClientProperty.KERBEROS_SERVER_PRIMARY, shortName);
 +
 +        KerberosToken token = new KerberosToken();
 +        saslParams = new SaslServerConnectionParams(clientConf, token, null);
 +
 +        processor = new UGIAssumingProcessor(processor);
 +
 +        break;
 +      default:
 +        // nothing to do -- no extra configuration necessary
 +        break;
 +    }
 +
 +    // Hook up support for tracing for thrift calls
 +    TimedProcessor timedProcessor = new TimedProcessor(metricsFactory, processor, serverName,
threadName);
 +
 +    // Create the thrift server with our processor and properties
 +    ServerAddress serverAddr = TServerUtils.startTServer(address, serverType, timedProcessor,
protocolFactory, serverName, threadName, numThreads,
 +        simpleTimerThreadpoolSize, threadpoolResizeInterval, maxFrameSize, sslParams, saslParams,
serverSocketTimeout);
 +
 +    return serverAddr;
    }
  }

http://git-wip-us.apache.org/repos/asf/accumulo/blob/0e3af1e8/server/base/src/main/java/org/apache/accumulo/server/rpc/RpcWrapper.java
----------------------------------------------------------------------
diff --cc server/base/src/main/java/org/apache/accumulo/server/rpc/RpcWrapper.java
index 585eb27,0000000..ec68166
mode 100644,000000..100644
--- a/server/base/src/main/java/org/apache/accumulo/server/rpc/RpcWrapper.java
+++ b/server/base/src/main/java/org/apache/accumulo/server/rpc/RpcWrapper.java
@@@ -1,128 -1,0 +1,128 @@@
 +/*
 + * 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
 + *
 + *     http://www.apache.org/licenses/LICENSE-2.0
 + *
 + * 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.accumulo.server.rpc;
 +
 +import java.lang.reflect.InvocationHandler;
 +import java.lang.reflect.Method;
 +import java.lang.reflect.Proxy;
 +import java.util.HashSet;
 +import java.util.Map;
 +import java.util.Map.Entry;
 +import java.util.Set;
 +
 +import org.apache.accumulo.core.trace.wrappers.RpcServerInvocationHandler;
 +import org.apache.accumulo.core.trace.wrappers.TraceWrap;
 +import org.apache.thrift.ProcessFunction;
 +import org.apache.thrift.TApplicationException;
- import org.apache.thrift.TBase;
++import org.apache.thrift.TBaseProcessor;
 +import org.apache.thrift.TException;
 +import org.slf4j.Logger;
 +import org.slf4j.LoggerFactory;
 +
 +/**
 + * This class accommodates the changes in THRIFT-1805, which appeared in Thrift 0.9.1 and
restricts client-side notification of server-side errors to
 + * {@link TException} only, by wrapping {@link RuntimeException} and {@link Error} as {@link
TException}, so it doesn't just close the connection and look like
 + * a network issue, but informs the client that a {@link TApplicationException} had occurred,
as it did in Thrift 0.9.0. This performs similar functions as
 + * {@link TraceWrap}, but with the additional action of translating exceptions. See also
ACCUMULO-1691 and ACCUMULO-2950.
 + *
 + * ACCUMULO-4065 found that the above exception-wrapping is not appropriate for Thrift's
implementation of oneway methods. Oneway methods are defined as a
 + * method which the client does not wait for it to return. Normally, this is acceptable
as these methods are void. Therefore, if another client reuses the
 + * connection to send a new RPC, there is no "extra" data sitting on the InputStream from
the Socket (that the server sent). However, the implementation of a
 + * oneway method <em>does</em> send a response to the client when the implementation
throws a {@link TException}. This message showing up on the client's
 + * InputStream causes future use of the Thrift Connection to become unusable. As long as
the Thrift implementation sends a message back when oneway methods
 + * throw a {@link TException}, we much make sure that we don't re-wrap-and-throw any exceptions
as {@link TException}s.
 + *
 + * @since 1.6.1
 + */
 +public class RpcWrapper {
 +  private static final Logger log = LoggerFactory.getLogger(RpcWrapper.class);
 +
-   public static <T> T service(final T instance, @SuppressWarnings("rawtypes") final
Map<String,ProcessFunction<T,? extends TBase>> processorView) {
++  public static <I> I service(final I instance, final TBaseProcessor<I> processor)
{
++    final Map<String,ProcessFunction<I,?>> processorView = processor.getProcessMapView();
 +    final Set<String> onewayMethods = getOnewayMethods(processorView);
 +    log.debug("Found oneway Thrift methods: " + onewayMethods);
 +
 +    InvocationHandler handler = getInvocationHandler(instance, onewayMethods);
 +
 +    @SuppressWarnings("unchecked")
-     T proxiedInstance = (T) Proxy.newProxyInstance(instance.getClass().getClassLoader(),
instance.getClass().getInterfaces(), handler);
++    I proxiedInstance = (I) Proxy.newProxyInstance(instance.getClass().getClassLoader(),
instance.getClass().getInterfaces(), handler);
 +    return proxiedInstance;
 +  }
 +
 +  protected static <T> RpcServerInvocationHandler<T> getInvocationHandler(final
T instance, final Set<String> onewayMethods) {
 +    return new RpcServerInvocationHandler<T>(instance) {
 +      private final Logger log = LoggerFactory.getLogger(instance.getClass());
 +
 +      @Override
 +      public Object invoke(Object obj, Method method, Object[] args) throws Throwable {
 +        // e.g. ThriftClientHandler.flush(TInfo, TCredentials, ...)
 +        try {
 +          return super.invoke(obj, method, args);
 +        } catch (RuntimeException e) {
 +          String msg = e.getMessage();
 +          log.error(msg, e);
 +          if (onewayMethods.contains(method.getName())) {
 +            throw e;
 +          }
 +          throw new TException(msg);
 +        } catch (Error e) {
 +          String msg = e.getMessage();
 +          log.error(msg, e);
 +          if (onewayMethods.contains(method.getName())) {
 +            throw e;
 +          }
 +          throw new TException(msg);
 +        }
 +      }
 +    };
 +  }
 +
-   protected static <T> Set<String> getOnewayMethods(@SuppressWarnings("rawtypes")
Map<String,ProcessFunction<T,? extends TBase>> processorView) {
++  protected static Set<String> getOnewayMethods(Map<String,?> processorView)
{
 +    // Get a handle on the isOnewayMethod and make it accessible
 +    final Method isOnewayMethod;
 +    try {
 +      isOnewayMethod = ProcessFunction.class.getDeclaredMethod("isOneway");
 +    } catch (NoSuchMethodException e) {
 +      throw new RuntimeException("Could not access isOneway method", e);
 +    } catch (SecurityException e) {
 +      throw new RuntimeException("Could not access isOneway method", e);
 +    }
 +    // In java7, this appears to be copying the method, but it's trivial for us to return
the object to how it was before.
 +    final boolean accessible = isOnewayMethod.isAccessible();
 +    isOnewayMethod.setAccessible(true);
 +
 +    try {
 +      final Set<String> onewayMethods = new HashSet<String>();
-       for (@SuppressWarnings("rawtypes")
-       Entry<String,ProcessFunction<T,? extends TBase>> entry : processorView.entrySet())
{
++      for (Entry<String,?> entry : processorView.entrySet()) {
 +        try {
 +          if ((Boolean) isOnewayMethod.invoke(entry.getValue())) {
 +            onewayMethods.add(entry.getKey());
 +          }
 +        } catch (RuntimeException e) {
 +          throw e;
 +        } catch (Exception e) {
 +          throw new RuntimeException(e);
 +        }
 +      }
 +
 +      return onewayMethods;
 +    } finally {
 +      // Reset it back to how it was.
 +      isOnewayMethod.setAccessible(accessible);
 +    }
 +  }
 +}

http://git-wip-us.apache.org/repos/asf/accumulo/blob/0e3af1e8/server/base/src/test/java/org/apache/accumulo/server/rpc/RpcWrapperTest.java
----------------------------------------------------------------------
diff --cc server/base/src/test/java/org/apache/accumulo/server/rpc/RpcWrapperTest.java
index e4e9402,0000000..39d3705
mode 100644,000000..100644
--- a/server/base/src/test/java/org/apache/accumulo/server/rpc/RpcWrapperTest.java
+++ b/server/base/src/test/java/org/apache/accumulo/server/rpc/RpcWrapperTest.java
@@@ -1,304 -1,0 +1,301 @@@
 +/*
 + * 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
 + *
 + * http://www.apache.org/licenses/LICENSE-2.0
 + *
 + * 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.accumulo.server.rpc;
 +
 +import java.util.Collections;
 +import java.util.HashMap;
 +import java.util.Map;
 +import java.util.Set;
 +
 +import org.apache.accumulo.core.trace.wrappers.RpcServerInvocationHandler;
 +import org.apache.accumulo.server.rpc.RpcWrapper;
 +import org.apache.thrift.ProcessFunction;
 +import org.apache.thrift.TBase;
 +import org.apache.thrift.TException;
 +import org.apache.thrift.protocol.TProtocol;
 +import org.junit.Assert;
 +import org.junit.Test;
 +
 +import com.google.common.collect.Sets;
 +
 +/**
 + * Verification that RpcWrapper correctly mangles Exceptions to work around Thrift.
 + */
 +public class RpcWrapperTest {
 +
 +  private static final String RTE_MESSAGE = "RpcWrapperTest's RuntimeException Message";
 +
 +  /**
 +   * Given a method name and whether or not the method is oneway, construct a ProcessFunction.
 +   *
 +   * @param methodName
 +   *          The service method name.
 +   * @param isOneway
 +   *          Is the method oneway.
 +   * @return A ProcessFunction.
 +   */
 +  private fake_proc<FakeService> createProcessFunction(String methodName, boolean
isOneway) {
 +    return new fake_proc<FakeService>(methodName, isOneway);
 +  }
 +
 +  @Test
 +  public void testSomeOnewayMethods() {
-     @SuppressWarnings("rawtypes")
-     Map<String,ProcessFunction<FakeService,? extends TBase>> procs = new HashMap<String,ProcessFunction<FakeService,?
extends TBase>>();
++    Map<String,ProcessFunction<FakeService,?>> procs = new HashMap<String,ProcessFunction<FakeService,?>>();
 +    procs.put("foo", createProcessFunction("foo", true));
 +    procs.put("foobar", createProcessFunction("foobar", false));
 +    procs.put("bar", createProcessFunction("bar", true));
 +    procs.put("barfoo", createProcessFunction("barfoo", false));
 +
 +    Set<String> onewayMethods = RpcWrapper.getOnewayMethods(procs);
 +    Assert.assertEquals(Sets.newHashSet("foo", "bar"), onewayMethods);
 +  }
 +
 +  @Test
 +  public void testNoOnewayMethods() {
-     @SuppressWarnings("rawtypes")
-     Map<String,ProcessFunction<FakeService,? extends TBase>> procs = new HashMap<String,ProcessFunction<FakeService,?
extends TBase>>();
++    Map<String,ProcessFunction<FakeService,?>> procs = new HashMap<String,ProcessFunction<FakeService,?>>();
 +    procs.put("foo", createProcessFunction("foo", false));
 +    procs.put("foobar", createProcessFunction("foobar", false));
 +    procs.put("bar", createProcessFunction("bar", false));
 +    procs.put("barfoo", createProcessFunction("barfoo", false));
 +
 +    Set<String> onewayMethods = RpcWrapper.getOnewayMethods(procs);
 +    Assert.assertEquals(Collections.<String> emptySet(), onewayMethods);
 +  }
 +
 +  @Test
 +  public void testAllOnewayMethods() {
-     @SuppressWarnings("rawtypes")
-     Map<String,ProcessFunction<FakeService,? extends TBase>> procs = new HashMap<String,ProcessFunction<FakeService,?
extends TBase>>();
++    Map<String,ProcessFunction<FakeService,?>> procs = new HashMap<String,ProcessFunction<FakeService,?>>();
 +    procs.put("foo", createProcessFunction("foo", true));
 +    procs.put("foobar", createProcessFunction("foobar", true));
 +    procs.put("bar", createProcessFunction("bar", true));
 +    procs.put("barfoo", createProcessFunction("barfoo", true));
 +
 +    Set<String> onewayMethods = RpcWrapper.getOnewayMethods(procs);
 +    Assert.assertEquals(Sets.newHashSet("foo", "foobar", "bar", "barfoo"), onewayMethods);
 +  }
 +
 +  @Test
 +  public void testNoExceptionWrappingForOneway() throws Throwable {
 +    final Object[] args = new Object[0];
 +
 +    final FakeService impl = new FakeServiceImpl();
 +
 +    // "short" names throw RTEs and are oneway, while long names do not throw exceptions
and are not oneway.
 +    RpcServerInvocationHandler<FakeService> handler = RpcWrapper.getInvocationHandler(impl,
Sets.newHashSet("foo", "bar"));
 +
 +    // Should throw an exception, but not be wrapped because the method is oneway
 +    try {
 +      handler.invoke(impl, FakeServiceImpl.class.getMethod("foo"), args);
 +      Assert.fail("Expected an exception");
 +    } catch (RuntimeException e) {
 +      Assert.assertEquals(RTE_MESSAGE, e.getMessage());
 +    }
 +
 +    // Should not throw an exception
 +    handler.invoke(impl, FakeServiceImpl.class.getMethod("foobar"), args);
 +  }
 +
 +  @Test
 +  public void testExceptionWrappingForNonOneway() throws Throwable {
 +    final Object[] args = new Object[0];
 +
 +    final FakeService impl = new FakeServiceImpl();
 +
 +    // "short" names throw RTEs and are not oneway, while long names do not throw exceptions
and are oneway.
 +    RpcServerInvocationHandler<FakeService> handler = RpcWrapper.getInvocationHandler(impl,
Sets.newHashSet("foobar", "barfoo"));
 +
 +    // Should throw an exception, but not be wrapped because the method is oneway
 +    try {
 +      handler.invoke(impl, FakeServiceImpl.class.getMethod("foo"), args);
 +      Assert.fail("Expected an exception");
 +    } catch (TException e) {
 +      // The InvocationHandler should take the exception from the RTE and make it a TException
 +      Assert.assertEquals(RTE_MESSAGE, e.getMessage());
 +    }
 +
 +    // Should not throw an exception
 +    handler.invoke(impl, FakeServiceImpl.class.getMethod("foobar"), args);
 +  }
 +
 +  //
 +  // Some hacked together classes/interfaces that mimic what Thrift is doing.
 +  //
 +
 +  /**
 +   * Some fake fields for our fake arguments.
 +   */
 +  private static class fake_fields implements org.apache.thrift.TFieldIdEnum {
 +    @Override
 +    public short getThriftFieldId() {
 +      throw new UnsupportedOperationException();
 +    }
 +
 +    @Override
 +    public String getFieldName() {
 +      throw new UnsupportedOperationException();
 +    }
 +
 +    @Override
 +    public boolean equals(Object o) {
 +      throw new UnsupportedOperationException();
 +    }
 +
 +    @Override
 +    public int hashCode() {
 +      throw new UnsupportedOperationException();
 +    }
 +  }
 +
 +  /**
 +   * A fake thrift service
 +   */
 +  interface FakeService {
 +    void foo();
 +
 +    String foobar();
 +
 +    int bar();
 +
 +    long barfoo();
 +  }
 +
 +  /**
 +   * An implementation of the fake thrift service. The "short" names throw RTEs, while long
names do not.
 +   */
 +  public static class FakeServiceImpl implements FakeService {
 +    @Override
 +    public void foo() {
 +      throw new RuntimeException(RTE_MESSAGE);
 +    }
 +
 +    @Override
 +    public String foobar() {
 +      return "";
 +    }
 +
 +    @Override
 +    public int bar() {
 +      throw new RuntimeException(RTE_MESSAGE);
 +    }
 +
 +    @Override
 +    public long barfoo() {
 +      return 0;
 +    }
 +  };
 +
 +  /**
 +   * A fake ProcessFunction implementation for testing that allows injection of method name
and oneway.
 +   */
 +  private static class fake_proc<I extends FakeService> extends org.apache.thrift.ProcessFunction<I,foo_args>
{
 +    final private boolean isOneway;
 +
 +    public fake_proc(String methodName, boolean isOneway) {
 +      super(methodName);
 +      this.isOneway = isOneway;
 +    }
 +
 +    @Override
 +    protected boolean isOneway() {
 +      return isOneway;
 +    }
 +
 +    @SuppressWarnings("rawtypes")
 +    @Override
 +    public TBase getResult(I iface, foo_args args) throws TException {
 +      throw new UnsupportedOperationException();
 +    }
 +
 +    @Override
 +    public foo_args getEmptyArgsInstance() {
 +      throw new UnsupportedOperationException();
 +    }
 +
 +    @Override
 +    public boolean equals(Object o) {
 +      throw new UnsupportedOperationException();
 +    }
 +
 +    @Override
 +    public int hashCode() {
 +      throw new UnsupportedOperationException();
 +    }
 +  }
 +
 +  /**
 +   * Fake arguments for our fake service.
 +   */
 +  private static class foo_args implements org.apache.thrift.TBase<foo_args,fake_fields>
{
 +
 +    private static final long serialVersionUID = 1L;
 +
 +    @Override
 +    public boolean equals(Object o) {
 +      throw new UnsupportedOperationException();
 +    }
 +
 +    @Override
 +    public int hashCode() {
 +      throw new UnsupportedOperationException();
 +    }
 +
 +    @Override
 +    public int compareTo(foo_args o) {
 +      throw new UnsupportedOperationException();
 +    }
 +
 +    @Override
 +    public void read(TProtocol iprot) throws TException {
 +      throw new UnsupportedOperationException();
 +    }
 +
 +    @Override
 +    public void write(TProtocol oprot) throws TException {
 +      throw new UnsupportedOperationException();
 +    }
 +
 +    @Override
 +    public fake_fields fieldForId(int fieldId) {
 +      throw new UnsupportedOperationException();
 +    }
 +
 +    @Override
 +    public boolean isSet(fake_fields field) {
 +      throw new UnsupportedOperationException();
 +    }
 +
 +    @Override
 +    public Object getFieldValue(fake_fields field) {
 +      throw new UnsupportedOperationException();
 +    }
 +
 +    @Override
 +    public void setFieldValue(fake_fields field, Object value) {
 +      throw new UnsupportedOperationException();
 +    }
 +
 +    @Override
 +    public TBase<foo_args,fake_fields> deepCopy() {
 +      throw new UnsupportedOperationException();
 +    }
 +
 +    @Override
 +    public void clear() {
 +      throw new UnsupportedOperationException();
 +    }
 +  }
 +}

http://git-wip-us.apache.org/repos/asf/accumulo/blob/0e3af1e8/server/gc/src/main/java/org/apache/accumulo/gc/SimpleGarbageCollector.java
----------------------------------------------------------------------
diff --cc server/gc/src/main/java/org/apache/accumulo/gc/SimpleGarbageCollector.java
index 4d55461,1c6836b..6bb8f9a
--- a/server/gc/src/main/java/org/apache/accumulo/gc/SimpleGarbageCollector.java
+++ b/server/gc/src/main/java/org/apache/accumulo/gc/SimpleGarbageCollector.java
@@@ -709,16 -702,9 +709,16 @@@ public class SimpleGarbageCollector ext
    }
  
    private HostAndPort startStatsService() throws UnknownHostException {
-     Iface rpcProxy = RpcWrapper.service(this, new Processor<Iface>(this).getProcessMapView());
 -    Processor<Iface> processor = new Processor<Iface>(RpcWrapper.service(this,
new Processor<Iface>(this)));
 -    int port = config.getPort(Property.GC_PORT);
 -    long maxMessageSize = config.getMemoryInBytes(Property.GENERAL_MAX_MESSAGE_SIZE);
++    Iface rpcProxy = RpcWrapper.service(this, new Processor<Iface>(this));
 +    final Processor<Iface> processor;
 +    if (ThriftServerType.SASL == getThriftServerType()) {
 +      Iface tcProxy = TCredentialsUpdatingWrapper.service(rpcProxy, getClass(), getConfiguration());
 +      processor = new Processor<Iface>(tcProxy);
 +    } else {
 +      processor = new Processor<Iface>(rpcProxy);
 +    }
 +    int port = getConfiguration().getPort(Property.GC_PORT);
 +    long maxMessageSize = getConfiguration().getMemoryInBytes(Property.GENERAL_MAX_MESSAGE_SIZE);
      HostAndPort result = HostAndPort.fromParts(opts.getAddress(), port);
      log.debug("Starting garbage collector listening on " + result);
      try {

http://git-wip-us.apache.org/repos/asf/accumulo/blob/0e3af1e8/server/master/src/main/java/org/apache/accumulo/master/Master.java
----------------------------------------------------------------------
diff --cc server/master/src/main/java/org/apache/accumulo/master/Master.java
index 02e1132,0434714..1370989
--- a/server/master/src/main/java/org/apache/accumulo/master/Master.java
+++ b/server/master/src/main/java/org/apache/accumulo/master/Master.java
@@@ -1147,38 -1032,10 +1147,38 @@@ public class Master extends AccumuloSer
        throw new IOException(e);
      }
  
 -    Processor<Iface> processor = new Processor<Iface>(
 -        RpcWrapper.service(new MasterClientServiceHandler(this), new Processor<Iface>(new
MasterClientServiceHandler(this))));
 -    ServerAddress sa = TServerUtils.startServer(getSystemConfiguration(), hostname, Property.MASTER_CLIENTPORT,
processor, "Master",
 -        "Master Client Service Handler", null, Property.MASTER_MINTHREADS, Property.MASTER_THREADCHECK,
Property.GENERAL_MAX_MESSAGE_SIZE);
 +    ZooKeeperInitialization.ensureZooKeeperInitialized(zReaderWriter, zroot);
 +
 +    // Make sure that we have a secret key (either a new one or an old one from ZK) before
we start
 +    // the master client service.
 +    if (null != authenticationTokenKeyManager && null != keyDistributor) {
 +      log.info("Starting delegation-token key manager");
 +      keyDistributor.initialize();
 +      authenticationTokenKeyManager.start();
 +      boolean logged = false;
 +      while (!authenticationTokenKeyManager.isInitialized()) {
 +        // Print out a status message when we start waiting for the key manager to get initialized
 +        if (!logged) {
 +          log.info("Waiting for AuthenticationTokenKeyManager to be initialized");
 +          logged = true;
 +        }
 +        UtilWaitThread.sleep(200);
 +      }
 +      // And log when we are initialized
 +      log.info("AuthenticationTokenSecretManager is initialized");
 +    }
 +
 +    clientHandler = new MasterClientServiceHandler(this);
-     Iface rpcProxy = RpcWrapper.service(clientHandler, new Processor<Iface>(clientHandler).getProcessMapView());
++    Iface rpcProxy = RpcWrapper.service(clientHandler, new Processor<Iface>(clientHandler));
 +    final Processor<Iface> processor;
 +    if (ThriftServerType.SASL == getThriftServerType()) {
 +      Iface tcredsProxy = TCredentialsUpdatingWrapper.service(rpcProxy, clientHandler.getClass(),
getConfiguration());
 +      processor = new Processor<Iface>(tcredsProxy);
 +    } else {
 +      processor = new Processor<Iface>(rpcProxy);
 +    }
 +    ServerAddress sa = TServerUtils.startServer(this, hostname, Property.MASTER_CLIENTPORT,
processor, "Master", "Master Client Service Handler", null,
 +        Property.MASTER_MINTHREADS, Property.MASTER_THREADCHECK, Property.GENERAL_MAX_MESSAGE_SIZE);
      clientService = sa.server;
      String address = sa.address.toString();
      log.info("Setting master lock data to " + address);
@@@ -1187,43 -1044,6 +1187,43 @@@
      while (!clientService.isServing()) {
        UtilWaitThread.sleep(100);
      }
 +
 +    // Start the daemon to scan the replication table and make units of work
 +    replicationWorkDriver = new ReplicationDriver(this);
 +    replicationWorkDriver.start();
 +
 +    // Start the daemon to assign work to tservers to replicate to our peers
 +    try {
 +      replicationWorkAssigner = new WorkDriver(this);
 +    } catch (AccumuloException | AccumuloSecurityException e) {
 +      log.error("Caught exception trying to initialize replication WorkDriver", e);
 +      throw new RuntimeException(e);
 +    }
 +    replicationWorkAssigner.start();
 +
 +    // Start the replication coordinator which assigns tservers to service replication requests
 +    MasterReplicationCoordinator impl = new MasterReplicationCoordinator(this);
 +    ReplicationCoordinator.Processor<ReplicationCoordinator.Iface> replicationCoordinatorProcessor
= new ReplicationCoordinator.Processor<ReplicationCoordinator.Iface>(
-         RpcWrapper.service(impl, new ReplicationCoordinator.Processor<ReplicationCoordinator.Iface>(impl).getProcessMapView()));
++        RpcWrapper.service(impl, new ReplicationCoordinator.Processor<ReplicationCoordinator.Iface>(impl)));
 +    ServerAddress replAddress = TServerUtils.startServer(this, hostname, Property.MASTER_REPLICATION_COORDINATOR_PORT,
replicationCoordinatorProcessor,
 +        "Master Replication Coordinator", "Replication Coordinator", null, Property.MASTER_REPLICATION_COORDINATOR_MINTHREADS,
 +        Property.MASTER_REPLICATION_COORDINATOR_THREADCHECK, Property.GENERAL_MAX_MESSAGE_SIZE);
 +
 +    log.info("Started replication coordinator service at " + replAddress.address);
 +
 +    // Advertise that port we used so peers don't have to be told what it is
 +    ZooReaderWriter.getInstance().putPersistentData(ZooUtil.getRoot(getInstance()) + Constants.ZMASTER_REPLICATION_COORDINATOR_ADDR,
 +        replAddress.address.toString().getBytes(UTF_8), NodeExistsPolicy.OVERWRITE);
 +
 +    // Register replication metrics
 +    MasterMetricsFactory factory = new MasterMetricsFactory(getConfiguration(), this);
 +    Metrics replicationMetrics = factory.createReplicationMetrics();
 +    try {
 +      replicationMetrics.register();
 +    } catch (Exception e) {
 +      log.error("Failed to register replication metrics", e);
 +    }
 +
      while (clientService.isServing()) {
        UtilWaitThread.sleep(500);
      }

http://git-wip-us.apache.org/repos/asf/accumulo/blob/0e3af1e8/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java
----------------------------------------------------------------------
diff --cc server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java
index 3022a76,dce66b0..1080d8d
--- a/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java
+++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java
@@@ -2332,46 -3160,14 +2332,46 @@@ public class TabletServer extends Accum
    private HostAndPort startTabletClientService() throws UnknownHostException {
      // start listening for client connection last
      ThriftClientHandler handler = new ThriftClientHandler();
-     Iface rpcProxy = RpcWrapper.service(handler, new Processor<Iface>(handler).getProcessMapView());
 -    Iface tch = RpcWrapper.service(handler, new Processor<Iface>(handler));
 -    Processor<Iface> processor = new Processor<Iface>(tch);
 -    HostAndPort address = startServer(getSystemConfiguration(), clientAddress.getHostText(),
Property.TSERV_CLIENTPORT, processor, "Thrift Client Server");
++    Iface rpcProxy = RpcWrapper.service(handler, new Processor<Iface>(handler));
 +    final Processor<Iface> processor;
 +    if (ThriftServerType.SASL == getThriftServerType()) {
 +      Iface tcredProxy = TCredentialsUpdatingWrapper.service(rpcProxy, ThriftClientHandler.class,
getConfiguration());
 +      processor = new Processor<Iface>(tcredProxy);
 +    } else {
 +      processor = new Processor<Iface>(rpcProxy);
 +    }
 +    HostAndPort address = startServer(getServerConfigurationFactory().getConfiguration(),
clientAddress.getHostText(), Property.TSERV_CLIENTPORT, processor,
 +        "Thrift Client Server");
      log.info("address = " + address);
      return address;
    }
  
 -  ZooLock getLock() {
 +  private HostAndPort startReplicationService() throws UnknownHostException {
 +    final ReplicationServicerHandler handler = new ReplicationServicerHandler(this);
-     ReplicationServicer.Iface rpcProxy = RpcWrapper.service(handler, new ReplicationServicer.Processor<ReplicationServicer.Iface>(handler).getProcessMapView());
++    ReplicationServicer.Iface rpcProxy = RpcWrapper.service(handler, new ReplicationServicer.Processor<ReplicationServicer.Iface>(handler));
 +    ReplicationServicer.Iface repl = TCredentialsUpdatingWrapper.service(rpcProxy, handler.getClass(),
getConfiguration());
 +    ReplicationServicer.Processor<ReplicationServicer.Iface> processor = new ReplicationServicer.Processor<ReplicationServicer.Iface>(repl);
 +    AccumuloConfiguration conf = getServerConfigurationFactory().getConfiguration();
 +    Property maxMessageSizeProperty = (conf.get(Property.TSERV_MAX_MESSAGE_SIZE) != null
? Property.TSERV_MAX_MESSAGE_SIZE : Property.GENERAL_MAX_MESSAGE_SIZE);
 +    ServerAddress sp = TServerUtils.startServer(this, clientAddress.getHostText(), Property.REPLICATION_RECEIPT_SERVICE_PORT,
processor,
 +        "ReplicationServicerHandler", "Replication Servicer", null, Property.REPLICATION_MIN_THREADS,
Property.REPLICATION_THREADCHECK, maxMessageSizeProperty);
 +    this.replServer = sp.server;
 +    log.info("Started replication service on " + sp.address);
 +
 +    try {
 +      // The replication service is unique to the thrift service for a tserver, not just
a host.
 +      // Advertise the host and port for replication service given the host and port for
the tserver.
 +      ZooReaderWriter.getInstance().putPersistentData(ZooUtil.getRoot(getInstance()) + ReplicationConstants.ZOO_TSERVERS
+ "/" + clientAddress.toString(),
 +          sp.address.toString().getBytes(UTF_8), NodeExistsPolicy.OVERWRITE);
 +    } catch (Exception e) {
 +      log.error("Could not advertise replication service port", e);
 +      throw new RuntimeException(e);
 +    }
 +
 +    return sp.address;
 +  }
 +
 +  public ZooLock getLock() {
      return tabletServerLock;
    }
  


Mime
View raw message