You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@accumulo.apache.org by el...@apache.org on 2014/12/09 06:15:38 UTC
[1/3] accumulo git commit: ACCUMULO-3394 Change thrift package to rpc
and make an rpc package in core too
Repository: accumulo
Updated Branches:
refs/heads/master b7927906f -> de1d3ee38
http://git-wip-us.apache.org/repos/asf/accumulo/blob/de1d3ee3/server/base/src/main/java/org/apache/accumulo/server/thrift/TServerUtils.java
----------------------------------------------------------------------
diff --git a/server/base/src/main/java/org/apache/accumulo/server/thrift/TServerUtils.java b/server/base/src/main/java/org/apache/accumulo/server/thrift/TServerUtils.java
deleted file mode 100644
index 123caab..0000000
--- a/server/base/src/main/java/org/apache/accumulo/server/thrift/TServerUtils.java
+++ /dev/null
@@ -1,255 +0,0 @@
-/*
- * 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.thrift;
-
-import java.lang.reflect.Field;
-import java.net.BindException;
-import java.net.InetAddress;
-import java.net.InetSocketAddress;
-import java.net.UnknownHostException;
-import java.util.Random;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.ThreadPoolExecutor;
-
-import org.apache.accumulo.core.conf.AccumuloConfiguration;
-import org.apache.accumulo.core.conf.Property;
-import org.apache.accumulo.core.util.Daemon;
-import org.apache.accumulo.core.util.LoggingRunnable;
-import org.apache.accumulo.core.util.SimpleThreadPool;
-import org.apache.accumulo.core.util.SslConnectionParams;
-import org.apache.accumulo.core.util.ThriftUtil;
-import org.apache.accumulo.core.util.UtilWaitThread;
-import org.apache.accumulo.server.AccumuloServerContext;
-import org.apache.accumulo.server.util.Halt;
-import org.apache.accumulo.server.util.time.SimpleTimer;
-import org.apache.log4j.Logger;
-import org.apache.thrift.TProcessor;
-import org.apache.thrift.TProcessorFactory;
-import org.apache.thrift.server.TServer;
-import org.apache.thrift.server.TThreadPoolServer;
-import org.apache.thrift.transport.TServerTransport;
-import org.apache.thrift.transport.TTransportException;
-
-import com.google.common.net.HostAndPort;
-
-public class TServerUtils {
- private static final Logger log = Logger.getLogger(TServerUtils.class);
-
- public static final ThreadLocal<String> clientAddress = new ThreadLocal<String>();
-
- /**
- * Start a server, at the given port, or higher, if that port is not available.
- *
- * @param portHintProperty
- * the port to attempt to open, can be zero, meaning "any available port"
- * @param processor
- * the service to be started
- * @param serverName
- * the name of the class that is providing the service
- * @param threadName
- * name this service's thread for better debugging
- * @return the server object created, and the port actually used
- * @throws UnknownHostException
- * when we don't know our own address
- */
- public static ServerAddress startServer(AccumuloServerContext service, String address, Property portHintProperty, TProcessor processor, String serverName,
- String threadName, Property portSearchProperty, Property minThreadProperty, Property timeBetweenThreadChecksProperty, Property maxMessageSizeProperty)
- throws UnknownHostException {
- int portHint = service.getConfiguration().getPort(portHintProperty);
- int minThreads = 2;
- if (minThreadProperty != null)
- minThreads = service.getConfiguration().getCount(minThreadProperty);
- long timeBetweenThreadChecks = 1000;
- if (timeBetweenThreadChecksProperty != null)
- timeBetweenThreadChecks = service.getConfiguration().getTimeInMillis(timeBetweenThreadChecksProperty);
- long maxMessageSize = 10 * 1000 * 1000;
- if (maxMessageSizeProperty != null)
- maxMessageSize = service.getConfiguration().getMemoryInBytes(maxMessageSizeProperty);
- boolean portSearch = false;
- if (portSearchProperty != null)
- portSearch = service.getConfiguration().getBoolean(portSearchProperty);
- // create the TimedProcessor outside the port search loop so we don't try to register the same metrics mbean more than once
- TimedProcessor timedProcessor = new TimedProcessor(service.getConfiguration(), processor, serverName, threadName);
- Random random = new Random();
- for (int j = 0; j < 100; j++) {
-
- // Are we going to slide around, looking for an open port?
- int portsToSearch = 1;
- if (portSearch)
- portsToSearch = 1000;
-
- for (int i = 0; i < portsToSearch; i++) {
- int port = portHint + i;
- if (portHint != 0 && i > 0)
- port = 1024 + random.nextInt(65535 - 1024);
- if (port > 65535)
- port = 1024 + port % (65535 - 1024);
- try {
- HostAndPort addr = HostAndPort.fromParts(address, port);
- return TServerUtils.startTServer(addr, timedProcessor, serverName, threadName, minThreads,
- service.getConfiguration().getCount(Property.GENERAL_SIMPLETIMER_THREADPOOL_SIZE), timeBetweenThreadChecks, maxMessageSize,
- service.getServerSslParams(), service.getClientTimeoutInMillis());
- } catch (TTransportException ex) {
- log.error("Unable to start TServer", ex);
- if (ex.getCause() == null || ex.getCause().getClass() == BindException.class) {
- // Note: with a TNonblockingServerSocket a "port taken" exception is a cause-less
- // TTransportException, and with a TSocket created by TSSLTransportFactory, it
- // comes through as caused by a BindException.
- log.info("Unable to use port " + port + ", retrying. (Thread Name = " + threadName + ")");
- UtilWaitThread.sleep(250);
- } else {
- // thrift is passing up a nested exception that isn't a BindException,
- // so no reason to believe retrying on a different port would help.
- log.error("Unable to start TServer", ex);
- break;
- }
- }
- }
- }
- throw new UnknownHostException("Unable to find a listen port");
- }
-
- /**
- * Create a NonBlockingServer with a custom thread pool that can dynamically resize itself.
- */
- public static ServerAddress createNonBlockingServer(HostAndPort address, TProcessor processor, final String serverName, String threadName,
- final int numThreads, final int numSTThreads, long timeBetweenThreadChecks, long maxMessageSize) throws TTransportException {
- TNonblockingServerSocket transport = new TNonblockingServerSocket(new InetSocketAddress(address.getHostText(), address.getPort()));
- CustomNonBlockingServer.Args options = new CustomNonBlockingServer.Args(transport);
- options.protocolFactory(ThriftUtil.protocolFactory());
- options.transportFactory(ThriftUtil.transportFactory(maxMessageSize));
- options.maxReadBufferBytes = maxMessageSize;
- options.stopTimeoutVal(5);
- /*
- * Create our own very special thread pool.
- */
- final ThreadPoolExecutor pool = new SimpleThreadPool(numThreads, "ClientPool");
- // periodically adjust the number of threads we need by checking how busy our threads are
- SimpleTimer.getInstance(numSTThreads).schedule(new Runnable() {
- @Override
- public void run() {
- if (pool.getCorePoolSize() <= pool.getActiveCount()) {
- int larger = pool.getCorePoolSize() + Math.min(pool.getQueue().size(), 2);
- log.info("Increasing server thread pool size on " + serverName + " to " + larger);
- pool.setMaximumPoolSize(larger);
- pool.setCorePoolSize(larger);
- } else {
- if (pool.getCorePoolSize() > pool.getActiveCount() + 3) {
- int smaller = Math.max(numThreads, pool.getCorePoolSize() - 1);
- if (smaller != pool.getCorePoolSize()) {
- // ACCUMULO-2997 there is a race condition here... the active count could be higher by the time
- // we decrease the core pool size... so the active count could end up higher than
- // the core pool size, in which case everything will be queued... the increase case
- // should handle this and prevent deadlock
- log.info("Decreasing server thread pool size on " + serverName + " to " + smaller);
- pool.setCorePoolSize(smaller);
- }
- }
- }
- }
- }, timeBetweenThreadChecks, timeBetweenThreadChecks);
- options.executorService(pool);
- options.processorFactory(new TProcessorFactory(processor));
- if (address.getPort() == 0) {
- address = HostAndPort.fromParts(address.getHostText(), transport.getPort());
- }
- return new ServerAddress(new CustomNonBlockingServer(options), address);
- }
-
- public static TServer createThreadPoolServer(TServerTransport transport, TProcessor processor) {
- TThreadPoolServer.Args options = new TThreadPoolServer.Args(transport);
- options.protocolFactory(ThriftUtil.protocolFactory());
- options.transportFactory(ThriftUtil.transportFactory());
- options.processorFactory(new ClientInfoProcessorFactory(clientAddress, processor));
- return new TThreadPoolServer(options);
- }
-
- public static ServerAddress createSslThreadPoolServer(HostAndPort address, TProcessor processor, long socketTimeout, SslConnectionParams sslParams)
- throws TTransportException {
- org.apache.thrift.transport.TServerSocket transport;
- try {
- transport = ThriftUtil.getServerSocket(address.getPort(), (int) socketTimeout, InetAddress.getByName(address.getHostText()), sslParams);
- } catch (UnknownHostException e) {
- throw new TTransportException(e);
- }
- if (address.getPort() == 0) {
- address = HostAndPort.fromParts(address.getHostText(), transport.getServerSocket().getLocalPort());
- }
- return new ServerAddress(createThreadPoolServer(transport, processor), address);
- }
-
- public static ServerAddress startTServer(AccumuloConfiguration conf, HostAndPort address, TProcessor processor, String serverName, String threadName, int numThreads, int numSTThreads,
- long timeBetweenThreadChecks, long maxMessageSize, SslConnectionParams sslParams, long sslSocketTimeout) throws TTransportException {
- return startTServer(address, new TimedProcessor(conf, processor, serverName, threadName), serverName, threadName, numThreads, numSTThreads,
- timeBetweenThreadChecks, maxMessageSize, sslParams, sslSocketTimeout);
- }
-
- /**
- * Start the appropriate Thrift server (SSL or non-blocking server) for the given parameters. Non-null SSL parameters will cause an SSL server to be started.
- *
- * @return A ServerAddress encapsulating the Thrift server created and the host/port which it is bound to.
- */
- public static ServerAddress startTServer(HostAndPort address, TimedProcessor processor, String serverName, String threadName, int numThreads,
- int numSTThreads, long timeBetweenThreadChecks, long maxMessageSize, SslConnectionParams sslParams, long sslSocketTimeout) throws TTransportException {
-
- ServerAddress serverAddress;
- if (sslParams != null) {
- serverAddress = createSslThreadPoolServer(address, processor, sslSocketTimeout, sslParams);
- } else {
- serverAddress = createNonBlockingServer(address, processor, serverName, threadName, numThreads, numSTThreads, timeBetweenThreadChecks, maxMessageSize);
- }
- final TServer finalServer = serverAddress.server;
- Runnable serveTask = new Runnable() {
- @Override
- public void run() {
- try {
- finalServer.serve();
- } catch (Error e) {
- Halt.halt("Unexpected error in TThreadPoolServer " + e + ", halting.");
- }
- }
- };
- serveTask = new LoggingRunnable(TServerUtils.log, serveTask);
- Thread thread = new Daemon(serveTask, threadName);
- thread.start();
- // check for the special "bind to everything address"
- if (serverAddress.address.getHostText().equals("0.0.0.0")) {
- // can't get the address from the bind, so we'll do our best to invent our hostname
- try {
- serverAddress = new ServerAddress(finalServer, HostAndPort.fromParts(InetAddress.getLocalHost().getHostName(), serverAddress.address.getPort()));
- } catch (UnknownHostException e) {
- throw new TTransportException(e);
- }
- }
- return serverAddress;
- }
-
- // Existing connections will keep our thread running: reach in with reflection and insist that they shutdown.
- public static void stopTServer(TServer s) {
- if (s == null)
- return;
- s.stop();
- try {
- Field f = s.getClass().getDeclaredField("executorService_");
- f.setAccessible(true);
- ExecutorService es = (ExecutorService) f.get(s);
- es.shutdownNow();
- } catch (Exception e) {
- TServerUtils.log.error("Unable to call shutdownNow", e);
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/de1d3ee3/server/base/src/main/java/org/apache/accumulo/server/thrift/TimedProcessor.java
----------------------------------------------------------------------
diff --git a/server/base/src/main/java/org/apache/accumulo/server/thrift/TimedProcessor.java b/server/base/src/main/java/org/apache/accumulo/server/thrift/TimedProcessor.java
deleted file mode 100644
index 56c235c..0000000
--- a/server/base/src/main/java/org/apache/accumulo/server/thrift/TimedProcessor.java
+++ /dev/null
@@ -1,69 +0,0 @@
-/*
- * 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.thrift;
-
-import org.apache.accumulo.core.conf.AccumuloConfiguration;
-import org.apache.accumulo.server.metrics.Metrics;
-import org.apache.accumulo.server.metrics.MetricsFactory;
-import org.apache.accumulo.server.metrics.ThriftMetrics;
-import org.apache.thrift.TException;
-import org.apache.thrift.TProcessor;
-import org.apache.thrift.protocol.TProtocol;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * A {@link TProcessor} which tracks the duration of an RPC and adds it to the metrics subsystem.
- */
-public class TimedProcessor implements TProcessor {
- private static final Logger log = LoggerFactory.getLogger(TimedProcessor.class);
-
- private final TProcessor other;
- private final Metrics metrics;
- private long idleStart = 0;
-
- public TimedProcessor(AccumuloConfiguration conf, TProcessor next, String serverName, String threadName) {
- this.other = next;
- // Register the metrics MBean
- MetricsFactory factory = new MetricsFactory(conf);
- metrics = factory.createThriftMetrics(serverName, threadName);
- try {
- metrics.register();
- } catch (Exception e) {
- log.error("Exception registering MBean with MBean Server", e);
- }
- idleStart = System.currentTimeMillis();
- }
-
- @Override
- public boolean process(TProtocol in, TProtocol out) throws TException {
- long now = 0;
- final boolean metricsEnabled = metrics.isEnabled();
- if (metricsEnabled) {
- now = System.currentTimeMillis();
- metrics.add(ThriftMetrics.idle, (now - idleStart));
- }
- try {
- return other.process(in, out);
- } finally {
- if (metricsEnabled) {
- idleStart = System.currentTimeMillis();
- metrics.add(ThriftMetrics.execute, idleStart - now);
- }
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/de1d3ee3/server/base/src/main/java/org/apache/accumulo/server/util/VerifyTabletAssignments.java
----------------------------------------------------------------------
diff --git a/server/base/src/main/java/org/apache/accumulo/server/util/VerifyTabletAssignments.java b/server/base/src/main/java/org/apache/accumulo/server/util/VerifyTabletAssignments.java
index 9646991..1e25e44 100644
--- a/server/base/src/main/java/org/apache/accumulo/server/util/VerifyTabletAssignments.java
+++ b/server/base/src/main/java/org/apache/accumulo/server/util/VerifyTabletAssignments.java
@@ -43,13 +43,13 @@ import org.apache.accumulo.core.data.thrift.TColumn;
import org.apache.accumulo.core.data.thrift.TKeyExtent;
import org.apache.accumulo.core.data.thrift.TRange;
import org.apache.accumulo.core.metadata.MetadataServicer;
+import org.apache.accumulo.core.rpc.ThriftUtil;
import org.apache.accumulo.core.security.Authorizations;
import org.apache.accumulo.core.security.Credentials;
import org.apache.accumulo.core.tabletserver.thrift.NoSuchScanIDException;
import org.apache.accumulo.core.tabletserver.thrift.TabletClientService;
import org.apache.accumulo.core.trace.Tracer;
import org.apache.accumulo.core.trace.thrift.TInfo;
-import org.apache.accumulo.core.util.ThriftUtil;
import org.apache.accumulo.server.cli.ClientOpts;
import org.apache.hadoop.io.Text;
import org.apache.log4j.Logger;
http://git-wip-us.apache.org/repos/asf/accumulo/blob/de1d3ee3/server/base/src/test/java/org/apache/accumulo/server/util/TServerUtilsTest.java
----------------------------------------------------------------------
diff --git a/server/base/src/test/java/org/apache/accumulo/server/util/TServerUtilsTest.java b/server/base/src/test/java/org/apache/accumulo/server/util/TServerUtilsTest.java
index 337c055..75bf953 100644
--- a/server/base/src/test/java/org/apache/accumulo/server/util/TServerUtilsTest.java
+++ b/server/base/src/test/java/org/apache/accumulo/server/util/TServerUtilsTest.java
@@ -18,7 +18,7 @@ package org.apache.accumulo.server.util;
import java.util.concurrent.ExecutorService;
-import org.apache.accumulo.server.thrift.TServerUtils;
+import org.apache.accumulo.server.rpc.TServerUtils;
import org.apache.thrift.server.TServer;
import org.apache.thrift.transport.TServerSocket;
import org.junit.Test;
http://git-wip-us.apache.org/repos/asf/accumulo/blob/de1d3ee3/server/gc/src/main/java/org/apache/accumulo/gc/GarbageCollectWriteAheadLogs.java
----------------------------------------------------------------------
diff --git a/server/gc/src/main/java/org/apache/accumulo/gc/GarbageCollectWriteAheadLogs.java b/server/gc/src/main/java/org/apache/accumulo/gc/GarbageCollectWriteAheadLogs.java
index 774548e..5aee0c5 100644
--- a/server/gc/src/main/java/org/apache/accumulo/gc/GarbageCollectWriteAheadLogs.java
+++ b/server/gc/src/main/java/org/apache/accumulo/gc/GarbageCollectWriteAheadLogs.java
@@ -48,6 +48,7 @@ import org.apache.accumulo.core.replication.ReplicationTable;
import org.apache.accumulo.core.replication.ReplicationTableOfflineException;
import org.apache.accumulo.core.replication.StatusUtil;
import org.apache.accumulo.core.replication.proto.Replication.Status;
+import org.apache.accumulo.core.rpc.ThriftUtil;
import org.apache.accumulo.core.security.Authorizations;
import org.apache.accumulo.core.tabletserver.log.LogEntry;
import org.apache.accumulo.core.tabletserver.thrift.TabletClientService;
@@ -56,7 +57,6 @@ import org.apache.accumulo.core.trace.Span;
import org.apache.accumulo.core.trace.Trace;
import org.apache.accumulo.core.trace.Tracer;
import org.apache.accumulo.core.util.AddressUtil;
-import org.apache.accumulo.core.util.ThriftUtil;
import org.apache.accumulo.core.zookeeper.ZooUtil;
import org.apache.accumulo.server.AccumuloServerContext;
import org.apache.accumulo.server.ServerConstants;
http://git-wip-us.apache.org/repos/asf/accumulo/blob/de1d3ee3/server/gc/src/main/java/org/apache/accumulo/gc/SimpleGarbageCollector.java
----------------------------------------------------------------------
diff --git a/server/gc/src/main/java/org/apache/accumulo/gc/SimpleGarbageCollector.java b/server/gc/src/main/java/org/apache/accumulo/gc/SimpleGarbageCollector.java
index bbb6306..206b5db 100644
--- a/server/gc/src/main/java/org/apache/accumulo/gc/SimpleGarbageCollector.java
+++ b/server/gc/src/main/java/org/apache/accumulo/gc/SimpleGarbageCollector.java
@@ -91,9 +91,9 @@ import org.apache.accumulo.server.fs.VolumeManager;
import org.apache.accumulo.server.fs.VolumeManager.FileType;
import org.apache.accumulo.server.fs.VolumeManagerImpl;
import org.apache.accumulo.server.fs.VolumeUtil;
+import org.apache.accumulo.server.rpc.RpcWrapper;
+import org.apache.accumulo.server.rpc.TServerUtils;
import org.apache.accumulo.server.tables.TableManager;
-import org.apache.accumulo.server.thrift.RpcWrapper;
-import org.apache.accumulo.server.thrift.TServerUtils;
import org.apache.accumulo.server.util.Halt;
import org.apache.accumulo.server.util.TabletIterator;
import org.apache.accumulo.server.zookeeper.ZooLock;
http://git-wip-us.apache.org/repos/asf/accumulo/blob/de1d3ee3/server/gc/src/main/java/org/apache/accumulo/gc/replication/CloseWriteAheadLogReferences.java
----------------------------------------------------------------------
diff --git a/server/gc/src/main/java/org/apache/accumulo/gc/replication/CloseWriteAheadLogReferences.java b/server/gc/src/main/java/org/apache/accumulo/gc/replication/CloseWriteAheadLogReferences.java
index d99da6a..83b1d54 100644
--- a/server/gc/src/main/java/org/apache/accumulo/gc/replication/CloseWriteAheadLogReferences.java
+++ b/server/gc/src/main/java/org/apache/accumulo/gc/replication/CloseWriteAheadLogReferences.java
@@ -43,6 +43,7 @@ import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.Lo
import org.apache.accumulo.core.replication.ReplicationTable;
import org.apache.accumulo.core.replication.StatusUtil;
import org.apache.accumulo.core.replication.proto.Replication.Status;
+import org.apache.accumulo.core.rpc.ThriftUtil;
import org.apache.accumulo.core.security.Authorizations;
import org.apache.accumulo.core.tabletserver.log.LogEntry;
import org.apache.accumulo.core.tabletserver.thrift.TabletClientService;
@@ -50,7 +51,6 @@ import org.apache.accumulo.core.trace.Span;
import org.apache.accumulo.core.trace.Trace;
import org.apache.accumulo.core.trace.Tracer;
import org.apache.accumulo.core.trace.thrift.TInfo;
-import org.apache.accumulo.core.util.ThriftUtil;
import org.apache.accumulo.server.AccumuloServerContext;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
http://git-wip-us.apache.org/repos/asf/accumulo/blob/de1d3ee3/server/master/src/main/java/org/apache/accumulo/master/Master.java
----------------------------------------------------------------------
diff --git a/server/master/src/main/java/org/apache/accumulo/master/Master.java b/server/master/src/main/java/org/apache/accumulo/master/Master.java
index 9f6a2c0..977bb1d 100644
--- a/server/master/src/main/java/org/apache/accumulo/master/Master.java
+++ b/server/master/src/main/java/org/apache/accumulo/master/Master.java
@@ -117,14 +117,14 @@ import org.apache.accumulo.server.master.state.ZooStore;
import org.apache.accumulo.server.master.state.ZooTabletStateStore;
import org.apache.accumulo.server.metrics.Metrics;
import org.apache.accumulo.server.replication.ZooKeeperInitialization;
+import org.apache.accumulo.server.rpc.RpcWrapper;
+import org.apache.accumulo.server.rpc.ServerAddress;
+import org.apache.accumulo.server.rpc.TServerUtils;
import org.apache.accumulo.server.security.AuditedSecurityOperation;
import org.apache.accumulo.server.security.SecurityOperation;
import org.apache.accumulo.server.security.handler.ZKPermHandler;
import org.apache.accumulo.server.tables.TableManager;
import org.apache.accumulo.server.tables.TableObserver;
-import org.apache.accumulo.server.thrift.RpcWrapper;
-import org.apache.accumulo.server.thrift.ServerAddress;
-import org.apache.accumulo.server.thrift.TServerUtils;
import org.apache.accumulo.server.util.DefaultMap;
import org.apache.accumulo.server.util.Halt;
import org.apache.accumulo.server.util.MetadataTableUtil;
http://git-wip-us.apache.org/repos/asf/accumulo/blob/de1d3ee3/server/monitor/src/main/java/org/apache/accumulo/monitor/Monitor.java
----------------------------------------------------------------------
diff --git a/server/monitor/src/main/java/org/apache/accumulo/monitor/Monitor.java b/server/monitor/src/main/java/org/apache/accumulo/monitor/Monitor.java
index c761081..dd33a98 100644
--- a/server/monitor/src/main/java/org/apache/accumulo/monitor/Monitor.java
+++ b/server/monitor/src/main/java/org/apache/accumulo/monitor/Monitor.java
@@ -43,6 +43,7 @@ import org.apache.accumulo.core.master.thrift.MasterClientService;
import org.apache.accumulo.core.master.thrift.MasterMonitorInfo;
import org.apache.accumulo.core.master.thrift.TableInfo;
import org.apache.accumulo.core.master.thrift.TabletServerStatus;
+import org.apache.accumulo.core.rpc.ThriftUtil;
import org.apache.accumulo.core.security.SecurityUtil;
import org.apache.accumulo.core.tabletserver.thrift.ActiveScan;
import org.apache.accumulo.core.tabletserver.thrift.TabletClientService.Client;
@@ -53,7 +54,6 @@ import org.apache.accumulo.core.util.LoggingRunnable;
import org.apache.accumulo.core.util.Pair;
import org.apache.accumulo.core.util.ServerServices;
import org.apache.accumulo.core.util.ServerServices.Service;
-import org.apache.accumulo.core.util.ThriftUtil;
import org.apache.accumulo.core.util.UtilWaitThread;
import org.apache.accumulo.core.zookeeper.ZooUtil;
import org.apache.accumulo.fate.zookeeper.ZooLock.LockLossReason;
http://git-wip-us.apache.org/repos/asf/accumulo/blob/de1d3ee3/server/monitor/src/main/java/org/apache/accumulo/monitor/ZooKeeperStatus.java
----------------------------------------------------------------------
diff --git a/server/monitor/src/main/java/org/apache/accumulo/monitor/ZooKeeperStatus.java b/server/monitor/src/main/java/org/apache/accumulo/monitor/ZooKeeperStatus.java
index 7aefa1d..907d1f1 100644
--- a/server/monitor/src/main/java/org/apache/accumulo/monitor/ZooKeeperStatus.java
+++ b/server/monitor/src/main/java/org/apache/accumulo/monitor/ZooKeeperStatus.java
@@ -24,7 +24,7 @@ import java.util.TreeSet;
import org.apache.accumulo.core.conf.Property;
import org.apache.accumulo.core.conf.SiteConfiguration;
-import org.apache.accumulo.core.util.TTimeoutTransport;
+import org.apache.accumulo.core.rpc.TTimeoutTransport;
import org.apache.accumulo.core.util.UtilWaitThread;
import org.apache.log4j.Logger;
import org.apache.thrift.transport.TTransport;
http://git-wip-us.apache.org/repos/asf/accumulo/blob/de1d3ee3/server/monitor/src/main/java/org/apache/accumulo/monitor/servlets/TServersServlet.java
----------------------------------------------------------------------
diff --git a/server/monitor/src/main/java/org/apache/accumulo/monitor/servlets/TServersServlet.java b/server/monitor/src/main/java/org/apache/accumulo/monitor/servlets/TServersServlet.java
index ecb4ff7..e7ab43d 100644
--- a/server/monitor/src/main/java/org/apache/accumulo/monitor/servlets/TServersServlet.java
+++ b/server/monitor/src/main/java/org/apache/accumulo/monitor/servlets/TServersServlet.java
@@ -32,13 +32,13 @@ import org.apache.accumulo.core.master.thrift.DeadServer;
import org.apache.accumulo.core.master.thrift.MasterMonitorInfo;
import org.apache.accumulo.core.master.thrift.TableInfo;
import org.apache.accumulo.core.master.thrift.TabletServerStatus;
+import org.apache.accumulo.core.rpc.ThriftUtil;
import org.apache.accumulo.core.tabletserver.thrift.ActionStats;
import org.apache.accumulo.core.tabletserver.thrift.TabletClientService;
import org.apache.accumulo.core.tabletserver.thrift.TabletStats;
import org.apache.accumulo.core.trace.Tracer;
import org.apache.accumulo.core.util.Base64;
import org.apache.accumulo.core.util.Duration;
-import org.apache.accumulo.core.util.ThriftUtil;
import org.apache.accumulo.monitor.Monitor;
import org.apache.accumulo.monitor.util.Table;
import org.apache.accumulo.monitor.util.TableRow;
http://git-wip-us.apache.org/repos/asf/accumulo/blob/de1d3ee3/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java
----------------------------------------------------------------------
diff --git a/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java b/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java
index d05f8d1..19d101e 100644
--- a/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java
+++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java
@@ -107,6 +107,7 @@ import org.apache.accumulo.core.metadata.RootTable;
import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection;
import org.apache.accumulo.core.replication.ReplicationConstants;
import org.apache.accumulo.core.replication.thrift.ReplicationServicer;
+import org.apache.accumulo.core.rpc.ThriftUtil;
import org.apache.accumulo.core.security.Authorizations;
import org.apache.accumulo.core.security.SecurityUtil;
import org.apache.accumulo.core.security.thrift.TCredentials;
@@ -135,7 +136,6 @@ import org.apache.accumulo.core.util.Pair;
import org.apache.accumulo.core.util.ServerServices;
import org.apache.accumulo.core.util.ServerServices.Service;
import org.apache.accumulo.core.util.SimpleThreadPool;
-import org.apache.accumulo.core.util.ThriftUtil;
import org.apache.accumulo.core.util.UtilWaitThread;
import org.apache.accumulo.core.zookeeper.ZooUtil;
import org.apache.accumulo.fate.zookeeper.IZooReaderWriter;
@@ -171,11 +171,11 @@ import org.apache.accumulo.server.metrics.Metrics;
import org.apache.accumulo.server.problems.ProblemReport;
import org.apache.accumulo.server.problems.ProblemReports;
import org.apache.accumulo.server.replication.ZooKeeperInitialization;
+import org.apache.accumulo.server.rpc.RpcWrapper;
+import org.apache.accumulo.server.rpc.ServerAddress;
+import org.apache.accumulo.server.rpc.TServerUtils;
import org.apache.accumulo.server.security.AuditedSecurityOperation;
import org.apache.accumulo.server.security.SecurityOperation;
-import org.apache.accumulo.server.thrift.RpcWrapper;
-import org.apache.accumulo.server.thrift.ServerAddress;
-import org.apache.accumulo.server.thrift.TServerUtils;
import org.apache.accumulo.server.util.FileSystemMonitor;
import org.apache.accumulo.server.util.Halt;
import org.apache.accumulo.server.util.MasterMetadataUtil;
http://git-wip-us.apache.org/repos/asf/accumulo/blob/de1d3ee3/server/tserver/src/main/java/org/apache/accumulo/tserver/session/Session.java
----------------------------------------------------------------------
diff --git a/server/tserver/src/main/java/org/apache/accumulo/tserver/session/Session.java b/server/tserver/src/main/java/org/apache/accumulo/tserver/session/Session.java
index 3762a83..dfe38e4 100644
--- a/server/tserver/src/main/java/org/apache/accumulo/tserver/session/Session.java
+++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/session/Session.java
@@ -17,7 +17,7 @@
package org.apache.accumulo.tserver.session;
import org.apache.accumulo.core.security.thrift.TCredentials;
-import org.apache.accumulo.server.thrift.TServerUtils;
+import org.apache.accumulo.server.rpc.TServerUtils;
public class Session {
public final String client;
http://git-wip-us.apache.org/repos/asf/accumulo/blob/de1d3ee3/test/src/main/java/org/apache/accumulo/test/WrongTabletTest.java
----------------------------------------------------------------------
diff --git a/test/src/main/java/org/apache/accumulo/test/WrongTabletTest.java b/test/src/main/java/org/apache/accumulo/test/WrongTabletTest.java
index 3cf4098..f6f2571 100644
--- a/test/src/main/java/org/apache/accumulo/test/WrongTabletTest.java
+++ b/test/src/main/java/org/apache/accumulo/test/WrongTabletTest.java
@@ -20,11 +20,11 @@ import org.apache.accumulo.core.client.Instance;
import org.apache.accumulo.core.client.impl.ClientContext;
import org.apache.accumulo.core.data.KeyExtent;
import org.apache.accumulo.core.data.Mutation;
+import org.apache.accumulo.core.rpc.ThriftUtil;
import org.apache.accumulo.core.security.Credentials;
import org.apache.accumulo.core.tabletserver.thrift.TDurability;
import org.apache.accumulo.core.tabletserver.thrift.TabletClientService;
import org.apache.accumulo.core.trace.Tracer;
-import org.apache.accumulo.core.util.ThriftUtil;
import org.apache.accumulo.server.AccumuloServerContext;
import org.apache.accumulo.server.cli.ClientOpts;
import org.apache.accumulo.server.conf.ServerConfigurationFactory;
http://git-wip-us.apache.org/repos/asf/accumulo/blob/de1d3ee3/test/src/main/java/org/apache/accumulo/test/functional/ZombieTServer.java
----------------------------------------------------------------------
diff --git a/test/src/main/java/org/apache/accumulo/test/functional/ZombieTServer.java b/test/src/main/java/org/apache/accumulo/test/functional/ZombieTServer.java
index 8e14bf2..eb84533 100644
--- a/test/src/main/java/org/apache/accumulo/test/functional/ZombieTServer.java
+++ b/test/src/main/java/org/apache/accumulo/test/functional/ZombieTServer.java
@@ -40,8 +40,8 @@ import org.apache.accumulo.fate.zookeeper.ZooUtil.NodeExistsPolicy;
import org.apache.accumulo.server.AccumuloServerContext;
import org.apache.accumulo.server.client.HdfsZooInstance;
import org.apache.accumulo.server.conf.ServerConfigurationFactory;
-import org.apache.accumulo.server.thrift.ServerAddress;
-import org.apache.accumulo.server.thrift.TServerUtils;
+import org.apache.accumulo.server.rpc.ServerAddress;
+import org.apache.accumulo.server.rpc.TServerUtils;
import org.apache.accumulo.server.zookeeper.TransactionWatcher;
import org.apache.accumulo.server.zookeeper.ZooLock;
import org.apache.accumulo.server.zookeeper.ZooReaderWriter;
http://git-wip-us.apache.org/repos/asf/accumulo/blob/de1d3ee3/test/src/main/java/org/apache/accumulo/test/performance/thrift/NullTserver.java
----------------------------------------------------------------------
diff --git a/test/src/main/java/org/apache/accumulo/test/performance/thrift/NullTserver.java b/test/src/main/java/org/apache/accumulo/test/performance/thrift/NullTserver.java
index b311358..2ebc2e3 100644
--- a/test/src/main/java/org/apache/accumulo/test/performance/thrift/NullTserver.java
+++ b/test/src/main/java/org/apache/accumulo/test/performance/thrift/NullTserver.java
@@ -69,7 +69,7 @@ import org.apache.accumulo.server.master.state.MetaDataStateStore;
import org.apache.accumulo.server.master.state.MetaDataTableScanner;
import org.apache.accumulo.server.master.state.TServerInstance;
import org.apache.accumulo.server.master.state.TabletLocationState;
-import org.apache.accumulo.server.thrift.TServerUtils;
+import org.apache.accumulo.server.rpc.TServerUtils;
import org.apache.accumulo.server.zookeeper.TransactionWatcher;
import org.apache.hadoop.io.Text;
import org.apache.thrift.TException;
http://git-wip-us.apache.org/repos/asf/accumulo/blob/de1d3ee3/test/src/test/java/org/apache/accumulo/test/TotalQueuedIT.java
----------------------------------------------------------------------
diff --git a/test/src/test/java/org/apache/accumulo/test/TotalQueuedIT.java b/test/src/test/java/org/apache/accumulo/test/TotalQueuedIT.java
index 8cb0290..586ab5e 100644
--- a/test/src/test/java/org/apache/accumulo/test/TotalQueuedIT.java
+++ b/test/src/test/java/org/apache/accumulo/test/TotalQueuedIT.java
@@ -27,8 +27,8 @@ import org.apache.accumulo.core.client.Connector;
import org.apache.accumulo.core.conf.Property;
import org.apache.accumulo.core.data.Mutation;
import org.apache.accumulo.core.master.thrift.TabletServerStatus;
+import org.apache.accumulo.core.rpc.ThriftUtil;
import org.apache.accumulo.core.tabletserver.thrift.TabletClientService;
-import org.apache.accumulo.core.util.ThriftUtil;
import org.apache.accumulo.core.util.UtilWaitThread;
import org.apache.accumulo.minicluster.MemoryUnit;
import org.apache.accumulo.minicluster.impl.MiniAccumuloConfigImpl;
http://git-wip-us.apache.org/repos/asf/accumulo/blob/de1d3ee3/test/src/test/java/org/apache/accumulo/test/replication/GarbageCollectorCommunicatesWithTServersIT.java
----------------------------------------------------------------------
diff --git a/test/src/test/java/org/apache/accumulo/test/replication/GarbageCollectorCommunicatesWithTServersIT.java b/test/src/test/java/org/apache/accumulo/test/replication/GarbageCollectorCommunicatesWithTServersIT.java
index f2733ce..7f42aa0 100644
--- a/test/src/test/java/org/apache/accumulo/test/replication/GarbageCollectorCommunicatesWithTServersIT.java
+++ b/test/src/test/java/org/apache/accumulo/test/replication/GarbageCollectorCommunicatesWithTServersIT.java
@@ -41,11 +41,11 @@ import org.apache.accumulo.core.metadata.schema.MetadataSchema;
import org.apache.accumulo.core.protobuf.ProtobufUtil;
import org.apache.accumulo.core.replication.ReplicationTable;
import org.apache.accumulo.core.replication.proto.Replication.Status;
+import org.apache.accumulo.core.rpc.ThriftUtil;
import org.apache.accumulo.core.security.Authorizations;
import org.apache.accumulo.core.security.Credentials;
import org.apache.accumulo.core.tabletserver.thrift.TabletClientService.Client;
import org.apache.accumulo.core.trace.Tracer;
-import org.apache.accumulo.core.util.ThriftUtil;
import org.apache.accumulo.minicluster.impl.MiniAccumuloConfigImpl;
import org.apache.accumulo.test.functional.AbstractMacIT;
import org.apache.accumulo.test.functional.ConfigurableMacIT;
[3/3] accumulo git commit: ACCUMULO-3394 Change thrift package to rpc
and make an rpc package in core too
Posted by el...@apache.org.
ACCUMULO-3394 Change thrift package to rpc and make an rpc package in core too
Project: http://git-wip-us.apache.org/repos/asf/accumulo/repo
Commit: http://git-wip-us.apache.org/repos/asf/accumulo/commit/de1d3ee3
Tree: http://git-wip-us.apache.org/repos/asf/accumulo/tree/de1d3ee3
Diff: http://git-wip-us.apache.org/repos/asf/accumulo/diff/de1d3ee3
Branch: refs/heads/master
Commit: de1d3ee381d308c9e4d9d9f41532bde4cc1e6706
Parents: b792790
Author: Josh Elser <el...@apache.org>
Authored: Tue Dec 9 00:05:19 2014 -0500
Committer: Josh Elser <el...@apache.org>
Committed: Tue Dec 9 00:05:19 2014 -0500
----------------------------------------------------------------------
.../core/client/impl/ClientContext.java | 2 +-
.../core/client/impl/ConditionalWriterImpl.java | 2 +-
.../client/impl/InstanceOperationsImpl.java | 2 +-
.../accumulo/core/client/impl/MasterClient.java | 2 +-
.../core/client/impl/ReplicationClient.java | 2 +-
.../accumulo/core/client/impl/ServerClient.java | 2 +-
.../core/client/impl/TableOperationsImpl.java | 4 +-
.../impl/TabletServerBatchReaderIterator.java | 2 +-
.../client/impl/TabletServerBatchWriter.java | 2 +-
.../core/client/impl/ThriftScanner.java | 2 +-
.../core/client/impl/ThriftTransportKey.java | 2 +-
.../core/client/impl/ThriftTransportPool.java | 2 +-
.../accumulo/core/client/impl/Writer.java | 2 +-
.../accumulo/core/rpc/SslConnectionParams.java | 275 ++++++++++++
.../accumulo/core/rpc/TBufferedSocket.java | 39 ++
.../accumulo/core/rpc/TTimeoutTransport.java | 60 +++
.../apache/accumulo/core/rpc/ThriftUtil.java | 434 +++++++++++++++++++
.../accumulo/core/util/SslConnectionParams.java | 275 ------------
.../accumulo/core/util/TBufferedSocket.java | 39 --
.../accumulo/core/util/TTimeoutTransport.java | 60 ---
.../apache/accumulo/core/util/ThriftUtil.java | 433 ------------------
.../java/org/apache/accumulo/proxy/Proxy.java | 2 +-
.../accumulo/server/AccumuloServerContext.java | 2 +-
.../accumulo/server/client/BulkImporter.java | 2 +-
.../accumulo/server/master/LiveTServerSet.java | 2 +-
.../server/master/balancer/TabletBalancer.java | 2 +-
.../server/rpc/ClientInfoProcessorFactory.java | 53 +++
.../server/rpc/CustomNonBlockingServer.java | 268 ++++++++++++
.../apache/accumulo/server/rpc/RpcWrapper.java | 62 +++
.../accumulo/server/rpc/ServerAddress.java | 42 ++
.../server/rpc/TBufferedServerSocket.java | 71 +++
.../server/rpc/TNonblockingServerSocket.java | 157 +++++++
.../accumulo/server/rpc/TServerUtils.java | 255 +++++++++++
.../accumulo/server/rpc/TimedProcessor.java | 69 +++
.../security/AuditedSecurityOperation.java | 2 +-
.../thrift/ClientInfoProcessorFactory.java | 53 ---
.../server/thrift/CustomNonBlockingServer.java | 268 ------------
.../accumulo/server/thrift/RpcWrapper.java | 62 ---
.../accumulo/server/thrift/ServerAddress.java | 42 --
.../server/thrift/TBufferedServerSocket.java | 71 ---
.../server/thrift/TNonblockingServerSocket.java | 157 -------
.../accumulo/server/thrift/TServerUtils.java | 255 -----------
.../accumulo/server/thrift/TimedProcessor.java | 69 ---
.../server/util/VerifyTabletAssignments.java | 2 +-
.../accumulo/server/util/TServerUtilsTest.java | 2 +-
.../gc/GarbageCollectWriteAheadLogs.java | 2 +-
.../accumulo/gc/SimpleGarbageCollector.java | 4 +-
.../CloseWriteAheadLogReferences.java | 2 +-
.../java/org/apache/accumulo/master/Master.java | 6 +-
.../org/apache/accumulo/monitor/Monitor.java | 2 +-
.../accumulo/monitor/ZooKeeperStatus.java | 2 +-
.../monitor/servlets/TServersServlet.java | 2 +-
.../apache/accumulo/tserver/TabletServer.java | 8 +-
.../accumulo/tserver/session/Session.java | 2 +-
.../apache/accumulo/test/WrongTabletTest.java | 2 +-
.../accumulo/test/functional/ZombieTServer.java | 4 +-
.../test/performance/thrift/NullTserver.java | 2 +-
.../org/apache/accumulo/test/TotalQueuedIT.java | 2 +-
...bageCollectorCommunicatesWithTServersIT.java | 2 +-
59 files changed, 1828 insertions(+), 1827 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/accumulo/blob/de1d3ee3/core/src/main/java/org/apache/accumulo/core/client/impl/ClientContext.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/client/impl/ClientContext.java b/core/src/main/java/org/apache/accumulo/core/client/impl/ClientContext.java
index 8fd12f2..e75bec6 100644
--- a/core/src/main/java/org/apache/accumulo/core/client/impl/ClientContext.java
+++ b/core/src/main/java/org/apache/accumulo/core/client/impl/ClientContext.java
@@ -33,9 +33,9 @@ import org.apache.accumulo.core.conf.AccumuloConfiguration;
import org.apache.accumulo.core.conf.CredentialProviderFactoryShim;
import org.apache.accumulo.core.conf.DefaultConfiguration;
import org.apache.accumulo.core.conf.Property;
+import org.apache.accumulo.core.rpc.SslConnectionParams;
import org.apache.accumulo.core.security.Credentials;
import org.apache.accumulo.core.security.thrift.TCredentials;
-import org.apache.accumulo.core.util.SslConnectionParams;
import org.apache.commons.configuration.Configuration;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
http://git-wip-us.apache.org/repos/asf/accumulo/blob/de1d3ee3/core/src/main/java/org/apache/accumulo/core/client/impl/ConditionalWriterImpl.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/client/impl/ConditionalWriterImpl.java b/core/src/main/java/org/apache/accumulo/core/client/impl/ConditionalWriterImpl.java
index c831915..ee56fff 100644
--- a/core/src/main/java/org/apache/accumulo/core/client/impl/ConditionalWriterImpl.java
+++ b/core/src/main/java/org/apache/accumulo/core/client/impl/ConditionalWriterImpl.java
@@ -60,6 +60,7 @@ import org.apache.accumulo.core.data.thrift.TConditionalSession;
import org.apache.accumulo.core.data.thrift.TKeyExtent;
import org.apache.accumulo.core.data.thrift.TMutation;
import org.apache.accumulo.core.master.state.tables.TableState;
+import org.apache.accumulo.core.rpc.ThriftUtil;
import org.apache.accumulo.core.security.Authorizations;
import org.apache.accumulo.core.security.ColumnVisibility;
import org.apache.accumulo.core.security.VisibilityEvaluator;
@@ -72,7 +73,6 @@ import org.apache.accumulo.core.trace.thrift.TInfo;
import org.apache.accumulo.core.util.BadArgumentException;
import org.apache.accumulo.core.util.ByteBufferUtil;
import org.apache.accumulo.core.util.LoggingRunnable;
-import org.apache.accumulo.core.util.ThriftUtil;
import org.apache.accumulo.core.util.UtilWaitThread;
import org.apache.accumulo.core.zookeeper.ZooUtil;
import org.apache.accumulo.fate.zookeeper.ZooCacheFactory;
http://git-wip-us.apache.org/repos/asf/accumulo/blob/de1d3ee3/core/src/main/java/org/apache/accumulo/core/client/impl/InstanceOperationsImpl.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/client/impl/InstanceOperationsImpl.java b/core/src/main/java/org/apache/accumulo/core/client/impl/InstanceOperationsImpl.java
index 4e74069..a62496b 100644
--- a/core/src/main/java/org/apache/accumulo/core/client/impl/InstanceOperationsImpl.java
+++ b/core/src/main/java/org/apache/accumulo/core/client/impl/InstanceOperationsImpl.java
@@ -36,11 +36,11 @@ import org.apache.accumulo.core.client.impl.thrift.ClientService;
import org.apache.accumulo.core.client.impl.thrift.ConfigurationType;
import org.apache.accumulo.core.client.impl.thrift.ThriftSecurityException;
import org.apache.accumulo.core.master.thrift.MasterClientService;
+import org.apache.accumulo.core.rpc.ThriftUtil;
import org.apache.accumulo.core.tabletserver.thrift.TabletClientService;
import org.apache.accumulo.core.tabletserver.thrift.TabletClientService.Client;
import org.apache.accumulo.core.trace.Tracer;
import org.apache.accumulo.core.util.AddressUtil;
-import org.apache.accumulo.core.util.ThriftUtil;
import org.apache.accumulo.core.zookeeper.ZooUtil;
import org.apache.accumulo.fate.zookeeper.ZooCache;
import org.apache.accumulo.fate.zookeeper.ZooCacheFactory;
http://git-wip-us.apache.org/repos/asf/accumulo/blob/de1d3ee3/core/src/main/java/org/apache/accumulo/core/client/impl/MasterClient.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/client/impl/MasterClient.java b/core/src/main/java/org/apache/accumulo/core/client/impl/MasterClient.java
index 092eec6..74b8ea9 100644
--- a/core/src/main/java/org/apache/accumulo/core/client/impl/MasterClient.java
+++ b/core/src/main/java/org/apache/accumulo/core/client/impl/MasterClient.java
@@ -28,7 +28,7 @@ import org.apache.accumulo.core.client.TableNotFoundException;
import org.apache.accumulo.core.client.impl.thrift.ThriftSecurityException;
import org.apache.accumulo.core.client.impl.thrift.ThriftTableOperationException;
import org.apache.accumulo.core.master.thrift.MasterClientService;
-import org.apache.accumulo.core.util.ThriftUtil;
+import org.apache.accumulo.core.rpc.ThriftUtil;
import org.apache.accumulo.core.util.UtilWaitThread;
import org.apache.log4j.Logger;
import org.apache.thrift.TServiceClient;
http://git-wip-us.apache.org/repos/asf/accumulo/blob/de1d3ee3/core/src/main/java/org/apache/accumulo/core/client/impl/ReplicationClient.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/client/impl/ReplicationClient.java b/core/src/main/java/org/apache/accumulo/core/client/impl/ReplicationClient.java
index f2007e9..edfba50 100644
--- a/core/src/main/java/org/apache/accumulo/core/client/impl/ReplicationClient.java
+++ b/core/src/main/java/org/apache/accumulo/core/client/impl/ReplicationClient.java
@@ -28,7 +28,7 @@ import org.apache.accumulo.core.client.Instance;
import org.apache.accumulo.core.client.impl.thrift.ThriftSecurityException;
import org.apache.accumulo.core.replication.thrift.ReplicationCoordinator;
import org.apache.accumulo.core.replication.thrift.ReplicationServicer;
-import org.apache.accumulo.core.util.ThriftUtil;
+import org.apache.accumulo.core.rpc.ThriftUtil;
import org.apache.accumulo.core.util.UtilWaitThread;
import org.apache.accumulo.core.zookeeper.ZooUtil;
import org.apache.accumulo.fate.zookeeper.ZooReader;
http://git-wip-us.apache.org/repos/asf/accumulo/blob/de1d3ee3/core/src/main/java/org/apache/accumulo/core/client/impl/ServerClient.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/client/impl/ServerClient.java b/core/src/main/java/org/apache/accumulo/core/client/impl/ServerClient.java
index ebccdab..1e44727 100644
--- a/core/src/main/java/org/apache/accumulo/core/client/impl/ServerClient.java
+++ b/core/src/main/java/org/apache/accumulo/core/client/impl/ServerClient.java
@@ -28,10 +28,10 @@ import org.apache.accumulo.core.client.Instance;
import org.apache.accumulo.core.client.impl.thrift.ClientService;
import org.apache.accumulo.core.client.impl.thrift.ClientService.Client;
import org.apache.accumulo.core.client.impl.thrift.ThriftSecurityException;
+import org.apache.accumulo.core.rpc.ThriftUtil;
import org.apache.accumulo.core.util.Pair;
import org.apache.accumulo.core.util.ServerServices;
import org.apache.accumulo.core.util.ServerServices.Service;
-import org.apache.accumulo.core.util.ThriftUtil;
import org.apache.accumulo.core.util.UtilWaitThread;
import org.apache.accumulo.core.zookeeper.ZooUtil;
import org.apache.accumulo.fate.zookeeper.ZooCache;
http://git-wip-us.apache.org/repos/asf/accumulo/blob/de1d3ee3/core/src/main/java/org/apache/accumulo/core/client/impl/TableOperationsImpl.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/client/impl/TableOperationsImpl.java b/core/src/main/java/org/apache/accumulo/core/client/impl/TableOperationsImpl.java
index 1def091..4843a9c 100644
--- a/core/src/main/java/org/apache/accumulo/core/client/impl/TableOperationsImpl.java
+++ b/core/src/main/java/org/apache/accumulo/core/client/impl/TableOperationsImpl.java
@@ -49,7 +49,6 @@ import java.util.zip.ZipEntry;
import java.util.zip.ZipInputStream;
import org.apache.accumulo.core.client.admin.CompactionConfig;
-
import org.apache.accumulo.core.Constants;
import org.apache.accumulo.core.client.AccumuloException;
import org.apache.accumulo.core.client.AccumuloSecurityException;
@@ -92,6 +91,7 @@ import org.apache.accumulo.core.metadata.MetadataServicer;
import org.apache.accumulo.core.metadata.MetadataTable;
import org.apache.accumulo.core.metadata.RootTable;
import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection;
+import org.apache.accumulo.core.rpc.ThriftUtil;
import org.apache.accumulo.core.security.Authorizations;
import org.apache.accumulo.core.tabletserver.thrift.NotServingTabletException;
import org.apache.accumulo.core.tabletserver.thrift.TabletClientService;
@@ -104,7 +104,6 @@ import org.apache.accumulo.core.util.NamingThreadFactory;
import org.apache.accumulo.core.util.OpTimer;
import org.apache.accumulo.core.util.Pair;
import org.apache.accumulo.core.util.TextUtil;
-import org.apache.accumulo.core.util.ThriftUtil;
import org.apache.accumulo.core.util.UtilWaitThread;
import org.apache.accumulo.core.volume.VolumeConfiguration;
import org.apache.hadoop.fs.FileStatus;
@@ -116,6 +115,7 @@ import org.apache.log4j.Logger;
import org.apache.thrift.TApplicationException;
import org.apache.thrift.TException;
import org.apache.thrift.transport.TTransportException;
+
import com.google.common.base.Joiner;
public class TableOperationsImpl extends TableOperationsHelper {
http://git-wip-us.apache.org/repos/asf/accumulo/blob/de1d3ee3/core/src/main/java/org/apache/accumulo/core/client/impl/TabletServerBatchReaderIterator.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/client/impl/TabletServerBatchReaderIterator.java b/core/src/main/java/org/apache/accumulo/core/client/impl/TabletServerBatchReaderIterator.java
index cd3dfd0..fb5c20b 100644
--- a/core/src/main/java/org/apache/accumulo/core/client/impl/TabletServerBatchReaderIterator.java
+++ b/core/src/main/java/org/apache/accumulo/core/client/impl/TabletServerBatchReaderIterator.java
@@ -54,6 +54,7 @@ import org.apache.accumulo.core.data.thrift.TKeyExtent;
import org.apache.accumulo.core.data.thrift.TKeyValue;
import org.apache.accumulo.core.data.thrift.TRange;
import org.apache.accumulo.core.master.state.tables.TableState;
+import org.apache.accumulo.core.rpc.ThriftUtil;
import org.apache.accumulo.core.security.Authorizations;
import org.apache.accumulo.core.tabletserver.thrift.NoSuchScanIDException;
import org.apache.accumulo.core.tabletserver.thrift.TabletClientService;
@@ -61,7 +62,6 @@ import org.apache.accumulo.core.trace.Tracer;
import org.apache.accumulo.core.trace.wrappers.TraceRunnable;
import org.apache.accumulo.core.util.ByteBufferUtil;
import org.apache.accumulo.core.util.OpTimer;
-import org.apache.accumulo.core.util.ThriftUtil;
import org.apache.hadoop.io.Text;
import org.apache.log4j.Level;
import org.apache.log4j.Logger;
http://git-wip-us.apache.org/repos/asf/accumulo/blob/de1d3ee3/core/src/main/java/org/apache/accumulo/core/client/impl/TabletServerBatchWriter.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/client/impl/TabletServerBatchWriter.java b/core/src/main/java/org/apache/accumulo/core/client/impl/TabletServerBatchWriter.java
index a6112da..30a707e 100644
--- a/core/src/main/java/org/apache/accumulo/core/client/impl/TabletServerBatchWriter.java
+++ b/core/src/main/java/org/apache/accumulo/core/client/impl/TabletServerBatchWriter.java
@@ -54,6 +54,7 @@ import org.apache.accumulo.core.data.Mutation;
import org.apache.accumulo.core.data.thrift.TMutation;
import org.apache.accumulo.core.data.thrift.UpdateErrors;
import org.apache.accumulo.core.master.state.tables.TableState;
+import org.apache.accumulo.core.rpc.ThriftUtil;
import org.apache.accumulo.core.tabletserver.thrift.ConstraintViolationException;
import org.apache.accumulo.core.tabletserver.thrift.NoSuchScanIDException;
import org.apache.accumulo.core.tabletserver.thrift.NotServingTabletException;
@@ -63,7 +64,6 @@ import org.apache.accumulo.core.trace.Trace;
import org.apache.accumulo.core.trace.Tracer;
import org.apache.accumulo.core.trace.thrift.TInfo;
import org.apache.accumulo.core.util.SimpleThreadPool;
-import org.apache.accumulo.core.util.ThriftUtil;
import org.apache.hadoop.io.Text;
import org.apache.log4j.Logger;
import org.apache.thrift.TApplicationException;
http://git-wip-us.apache.org/repos/asf/accumulo/blob/de1d3ee3/core/src/main/java/org/apache/accumulo/core/client/impl/ThriftScanner.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/client/impl/ThriftScanner.java b/core/src/main/java/org/apache/accumulo/core/client/impl/ThriftScanner.java
index 4b7d9ae..90e4421 100644
--- a/core/src/main/java/org/apache/accumulo/core/client/impl/ThriftScanner.java
+++ b/core/src/main/java/org/apache/accumulo/core/client/impl/ThriftScanner.java
@@ -48,6 +48,7 @@ import org.apache.accumulo.core.data.thrift.IterInfo;
import org.apache.accumulo.core.data.thrift.ScanResult;
import org.apache.accumulo.core.data.thrift.TKeyValue;
import org.apache.accumulo.core.master.state.tables.TableState;
+import org.apache.accumulo.core.rpc.ThriftUtil;
import org.apache.accumulo.core.security.Authorizations;
import org.apache.accumulo.core.tabletserver.thrift.NoSuchScanIDException;
import org.apache.accumulo.core.tabletserver.thrift.NotServingTabletException;
@@ -58,7 +59,6 @@ import org.apache.accumulo.core.trace.Trace;
import org.apache.accumulo.core.trace.Tracer;
import org.apache.accumulo.core.trace.thrift.TInfo;
import org.apache.accumulo.core.util.OpTimer;
-import org.apache.accumulo.core.util.ThriftUtil;
import org.apache.hadoop.io.Text;
import org.apache.log4j.Level;
import org.apache.log4j.Logger;
http://git-wip-us.apache.org/repos/asf/accumulo/blob/de1d3ee3/core/src/main/java/org/apache/accumulo/core/client/impl/ThriftTransportKey.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/client/impl/ThriftTransportKey.java b/core/src/main/java/org/apache/accumulo/core/client/impl/ThriftTransportKey.java
index a3a4f25..176e947 100644
--- a/core/src/main/java/org/apache/accumulo/core/client/impl/ThriftTransportKey.java
+++ b/core/src/main/java/org/apache/accumulo/core/client/impl/ThriftTransportKey.java
@@ -18,7 +18,7 @@ package org.apache.accumulo.core.client.impl;
import static com.google.common.base.Preconditions.checkArgument;
-import org.apache.accumulo.core.util.SslConnectionParams;
+import org.apache.accumulo.core.rpc.SslConnectionParams;
class ThriftTransportKey {
private final String location;
http://git-wip-us.apache.org/repos/asf/accumulo/blob/de1d3ee3/core/src/main/java/org/apache/accumulo/core/client/impl/ThriftTransportPool.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/client/impl/ThriftTransportPool.java b/core/src/main/java/org/apache/accumulo/core/client/impl/ThriftTransportPool.java
index 3380e13..bdb04d9 100644
--- a/core/src/main/java/org/apache/accumulo/core/client/impl/ThriftTransportPool.java
+++ b/core/src/main/java/org/apache/accumulo/core/client/impl/ThriftTransportPool.java
@@ -31,9 +31,9 @@ import java.util.Set;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicBoolean;
+import org.apache.accumulo.core.rpc.ThriftUtil;
import org.apache.accumulo.core.util.Daemon;
import org.apache.accumulo.core.util.Pair;
-import org.apache.accumulo.core.util.ThriftUtil;
import org.apache.log4j.Logger;
import org.apache.thrift.transport.TTransport;
import org.apache.thrift.transport.TTransportException;
http://git-wip-us.apache.org/repos/asf/accumulo/blob/de1d3ee3/core/src/main/java/org/apache/accumulo/core/client/impl/Writer.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/client/impl/Writer.java b/core/src/main/java/org/apache/accumulo/core/client/impl/Writer.java
index df7074d..d7761e9 100644
--- a/core/src/main/java/org/apache/accumulo/core/client/impl/Writer.java
+++ b/core/src/main/java/org/apache/accumulo/core/client/impl/Writer.java
@@ -26,12 +26,12 @@ import org.apache.accumulo.core.client.impl.TabletLocator.TabletLocation;
import org.apache.accumulo.core.client.impl.thrift.ThriftSecurityException;
import org.apache.accumulo.core.data.KeyExtent;
import org.apache.accumulo.core.data.Mutation;
+import org.apache.accumulo.core.rpc.ThriftUtil;
import org.apache.accumulo.core.tabletserver.thrift.ConstraintViolationException;
import org.apache.accumulo.core.tabletserver.thrift.NotServingTabletException;
import org.apache.accumulo.core.tabletserver.thrift.TDurability;
import org.apache.accumulo.core.tabletserver.thrift.TabletClientService;
import org.apache.accumulo.core.trace.Tracer;
-import org.apache.accumulo.core.util.ThriftUtil;
import org.apache.accumulo.core.util.UtilWaitThread;
import org.apache.hadoop.io.Text;
import org.apache.log4j.Logger;
http://git-wip-us.apache.org/repos/asf/accumulo/blob/de1d3ee3/core/src/main/java/org/apache/accumulo/core/rpc/SslConnectionParams.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/rpc/SslConnectionParams.java b/core/src/main/java/org/apache/accumulo/core/rpc/SslConnectionParams.java
new file mode 100644
index 0000000..718bf85
--- /dev/null
+++ b/core/src/main/java/org/apache/accumulo/core/rpc/SslConnectionParams.java
@@ -0,0 +1,275 @@
+/*
+ * 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.core.rpc;
+
+import java.io.File;
+import java.io.FileNotFoundException;
+import java.net.URL;
+import java.util.Arrays;
+
+import org.apache.accumulo.core.conf.AccumuloConfiguration;
+import org.apache.accumulo.core.conf.Property;
+import org.apache.commons.lang.StringUtils;
+import org.apache.log4j.Logger;
+import org.apache.thrift.transport.TSSLTransportFactory.TSSLTransportParameters;
+
+public class SslConnectionParams {
+ private static final Logger log = Logger.getLogger(SslConnectionParams.class);
+
+ private boolean useJsse = false;
+ private boolean clientAuth = false;
+
+ private boolean keyStoreSet;
+ private String keyStorePath;
+ private String keyStorePass;
+ private String keyStoreType;
+
+ private boolean trustStoreSet;
+ private String trustStorePath;
+ private String trustStorePass;
+ private String trustStoreType;
+
+ private String[] cipherSuites;
+ private String[] serverProtocols;
+ private String clientProtocol;
+
+ // Use the static construction methods
+ private SslConnectionParams() {}
+
+ public static SslConnectionParams forConfig(AccumuloConfiguration conf, boolean server) {
+ if (!conf.getBoolean(Property.INSTANCE_RPC_SSL_ENABLED))
+ return null;
+
+ SslConnectionParams result = new SslConnectionParams();
+ boolean requireClientAuth = conf.getBoolean(Property.INSTANCE_RPC_SSL_CLIENT_AUTH);
+ if (server) {
+ result.setClientAuth(requireClientAuth);
+ }
+ if (conf.getBoolean(Property.RPC_USE_JSSE)) {
+ result.setUseJsse(true);
+ return result;
+ }
+
+ try {
+ if (!server || requireClientAuth) {
+ result.setTrustStoreFromConf(conf);
+ }
+ if (server || requireClientAuth) {
+ result.setKeyStoreFromConf(conf);
+ }
+ } catch (FileNotFoundException e) {
+ throw new IllegalArgumentException("Could not load configured keystore file", e);
+ }
+
+ String ciphers = conf.get(Property.RPC_SSL_CIPHER_SUITES);
+ if (null != ciphers && !ciphers.isEmpty()) {
+ result.cipherSuites = StringUtils.split(ciphers, ',');
+ }
+
+ String enabledProtocols = conf.get(Property.RPC_SSL_ENABLED_PROTOCOLS);
+ result.serverProtocols = StringUtils.split(enabledProtocols, ',');
+
+ result.clientProtocol = conf.get(Property.RPC_SSL_CLIENT_PROTOCOL);
+
+ return result;
+ }
+
+ private static String passwordFromConf(AccumuloConfiguration conf, String defaultPassword, Property passwordOverrideProperty) {
+ String keystorePassword = conf.get(passwordOverrideProperty);
+ if (!keystorePassword.isEmpty()) {
+ if (log.isTraceEnabled())
+ log.trace("Using explicit SSL private key password from " + passwordOverrideProperty.getKey());
+ } else {
+ keystorePassword = defaultPassword;
+ }
+ return keystorePassword;
+ }
+
+ private static String storePathFromConf(AccumuloConfiguration conf, Property pathProperty) throws FileNotFoundException {
+ return findKeystore(conf.getPath(pathProperty));
+ }
+
+ public void setKeyStoreFromConf(AccumuloConfiguration conf) throws FileNotFoundException {
+ keyStoreSet = true;
+ keyStorePath = storePathFromConf(conf, Property.RPC_SSL_KEYSTORE_PATH);
+ keyStorePass = passwordFromConf(conf, conf.get(Property.INSTANCE_SECRET), Property.RPC_SSL_KEYSTORE_PASSWORD);
+ keyStoreType = conf.get(Property.RPC_SSL_KEYSTORE_TYPE);
+ }
+
+ public void setTrustStoreFromConf(AccumuloConfiguration conf) throws FileNotFoundException {
+ trustStoreSet = true;
+ trustStorePath = storePathFromConf(conf, Property.RPC_SSL_TRUSTSTORE_PATH);
+ trustStorePass = passwordFromConf(conf, "", Property.RPC_SSL_TRUSTSTORE_PASSWORD);
+ trustStoreType = conf.get(Property.RPC_SSL_TRUSTSTORE_TYPE);
+ }
+
+ public static SslConnectionParams forServer(AccumuloConfiguration configuration) {
+ return forConfig(configuration, true);
+ }
+
+ public static SslConnectionParams forClient(AccumuloConfiguration configuration) {
+ return forConfig(configuration, false);
+ }
+
+ private static String findKeystore(String keystorePath) throws FileNotFoundException {
+ try {
+ // first just try the file
+ File file = new File(keystorePath);
+ if (file.exists())
+ return file.getAbsolutePath();
+ if (!file.isAbsolute()) {
+ // try classpath
+ URL url = SslConnectionParams.class.getClassLoader().getResource(keystorePath);
+ if (url != null) {
+ file = new File(url.toURI());
+ if (file.exists())
+ return file.getAbsolutePath();
+ }
+ }
+ } catch (Exception e) {
+ log.warn("Exception finding keystore", e);
+ }
+ throw new FileNotFoundException("Failed to load SSL keystore from " + keystorePath);
+ }
+
+ public void setUseJsse(boolean useJsse) {
+ this.useJsse = useJsse;
+ }
+
+ public boolean useJsse() {
+ return useJsse;
+ }
+
+ public void setClientAuth(boolean clientAuth) {
+ this.clientAuth = clientAuth;
+ }
+
+ public boolean isClientAuth() {
+ return clientAuth;
+ }
+
+ public String[] getServerProtocols() {
+ return serverProtocols;
+ }
+
+ public String getClientProtocol() {
+ return clientProtocol;
+ }
+
+ public boolean isKeyStoreSet() {
+ return keyStoreSet;
+ }
+
+ public String getKeyStorePath() {
+ return keyStorePath;
+ }
+
+ /**
+ * @return the keyStorePass
+ */
+ public String getKeyStorePass() {
+ return keyStorePass;
+ }
+
+ public String getKeyStoreType() {
+ return keyStoreType;
+ }
+
+ public boolean isTrustStoreSet() {
+ return trustStoreSet;
+ }
+
+ public String getTrustStorePath() {
+ return trustStorePath;
+ }
+
+ public String getTrustStorePass() {
+ return trustStorePass;
+ }
+
+ /**
+ * @return the trustStoreType
+ */
+ public String getTrustStoreType() {
+ return trustStoreType;
+ }
+
+ public TSSLTransportParameters getTTransportParams() {
+ if (useJsse)
+ throw new IllegalStateException("Cannot get TTransportParams for JSEE configuration.");
+
+ // Null cipherSuites is implicitly handled
+ TSSLTransportParameters params = new TSSLTransportParameters(clientProtocol, cipherSuites);
+
+ params.requireClientAuth(clientAuth);
+ if (keyStoreSet) {
+ params.setKeyStore(keyStorePath, keyStorePass, null, keyStoreType);
+ }
+ if (trustStoreSet) {
+ params.setTrustStore(trustStorePath, trustStorePass, null, trustStoreType);
+ }
+ return params;
+ }
+
+ @Override
+ public int hashCode() {
+ int hash = 0;
+ hash = 31 * hash + (clientAuth ? 0 : 1);
+ hash = 31 * hash + (useJsse ? 0 : 1);
+ if (useJsse)
+ return hash;
+ hash = 31 * hash + (keyStoreSet ? 0 : 1);
+ hash = 31 * hash + (trustStoreSet ? 0 : 1);
+ if (keyStoreSet) {
+ hash = 31 * hash + keyStorePath.hashCode();
+ }
+ if (trustStoreSet) {
+ hash = 31 * hash + trustStorePath.hashCode();
+ }
+ hash = 31 * hash + clientProtocol.hashCode();
+ hash = 31 * hash + Arrays.hashCode(serverProtocols);
+ return super.hashCode();
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ if (!(obj instanceof SslConnectionParams))
+ return false;
+
+ SslConnectionParams other = (SslConnectionParams) obj;
+ if (clientAuth != other.clientAuth)
+ return false;
+ if (useJsse)
+ return other.useJsse;
+ if (keyStoreSet) {
+ if (!other.keyStoreSet)
+ return false;
+ if (!keyStorePath.equals(other.keyStorePath) || !keyStorePass.equals(other.keyStorePass) || !keyStoreType.equals(other.keyStoreType))
+ return false;
+ }
+ if (trustStoreSet) {
+ if (!other.trustStoreSet)
+ return false;
+ if (!trustStorePath.equals(other.trustStorePath) || !trustStorePass.equals(other.trustStorePass) || !trustStoreType.equals(other.trustStoreType))
+ return false;
+ }
+ if (!Arrays.equals(serverProtocols, other.serverProtocols)) {
+ return false;
+ }
+ return clientProtocol.equals(other.clientProtocol);
+ }
+}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/de1d3ee3/core/src/main/java/org/apache/accumulo/core/rpc/TBufferedSocket.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/rpc/TBufferedSocket.java b/core/src/main/java/org/apache/accumulo/core/rpc/TBufferedSocket.java
new file mode 100644
index 0000000..87b7c13
--- /dev/null
+++ b/core/src/main/java/org/apache/accumulo/core/rpc/TBufferedSocket.java
@@ -0,0 +1,39 @@
+/*
+ * 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.core.rpc;
+
+import java.io.BufferedInputStream;
+import java.io.BufferedOutputStream;
+import java.io.IOException;
+
+import org.apache.thrift.transport.TIOStreamTransport;
+import org.apache.thrift.transport.TSocket;
+
+public class TBufferedSocket extends TIOStreamTransport {
+
+ String client;
+
+ public TBufferedSocket(TSocket sock, int bufferSize) throws IOException {
+ super(new BufferedInputStream(sock.getSocket().getInputStream(), bufferSize), new BufferedOutputStream(sock.getSocket().getOutputStream(), bufferSize));
+ client = sock.getSocket().getInetAddress().getHostAddress() + ":" + sock.getSocket().getPort();
+ }
+
+ public String getClientString() {
+ return client;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/de1d3ee3/core/src/main/java/org/apache/accumulo/core/rpc/TTimeoutTransport.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/rpc/TTimeoutTransport.java b/core/src/main/java/org/apache/accumulo/core/rpc/TTimeoutTransport.java
new file mode 100644
index 0000000..6eace77
--- /dev/null
+++ b/core/src/main/java/org/apache/accumulo/core/rpc/TTimeoutTransport.java
@@ -0,0 +1,60 @@
+/*
+ * 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.core.rpc;
+
+import java.io.BufferedInputStream;
+import java.io.BufferedOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.lang.reflect.Method;
+import java.net.InetSocketAddress;
+import java.net.Socket;
+import java.net.SocketAddress;
+import java.nio.channels.spi.SelectorProvider;
+
+import org.apache.hadoop.net.NetUtils;
+import org.apache.thrift.transport.TIOStreamTransport;
+import org.apache.thrift.transport.TTransport;
+
+import com.google.common.net.HostAndPort;
+
+public class TTimeoutTransport {
+
+ private static InputStream getInputStream(Socket socket, long timeout) {
+ try {
+ Method m = NetUtils.class.getMethod("getInputStream", Socket.class, Long.TYPE);
+ return (InputStream)m.invoke(null, socket, timeout);
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ public static TTransport create(HostAndPort addr, long timeoutMillis) throws IOException {
+ return create(new InetSocketAddress(addr.getHostText(), addr.getPort()), timeoutMillis);
+ }
+
+ public static TTransport create(SocketAddress addr, long timeoutMillis) throws IOException {
+ Socket socket = SelectorProvider.provider().openSocketChannel().socket();
+ socket.setSoLinger(false, 0);
+ socket.setTcpNoDelay(true);
+ socket.connect(addr);
+ InputStream input = new BufferedInputStream(getInputStream(socket, timeoutMillis), 1024 * 10);
+ OutputStream output = new BufferedOutputStream(NetUtils.getOutputStream(socket, timeoutMillis), 1024 * 10);
+ return new TIOStreamTransport(input, output);
+ }
+}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/de1d3ee3/core/src/main/java/org/apache/accumulo/core/rpc/ThriftUtil.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/rpc/ThriftUtil.java b/core/src/main/java/org/apache/accumulo/core/rpc/ThriftUtil.java
new file mode 100644
index 0000000..a3cb252
--- /dev/null
+++ b/core/src/main/java/org/apache/accumulo/core/rpc/ThriftUtil.java
@@ -0,0 +1,434 @@
+/*
+ * 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.core.rpc;
+
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.net.InetAddress;
+import java.net.ServerSocket;
+import java.net.Socket;
+import java.net.UnknownHostException;
+import java.security.KeyStore;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+
+import javax.net.ssl.KeyManagerFactory;
+import javax.net.ssl.SSLContext;
+import javax.net.ssl.SSLServerSocket;
+import javax.net.ssl.SSLSocket;
+import javax.net.ssl.SSLSocketFactory;
+import javax.net.ssl.TrustManagerFactory;
+
+import org.apache.accumulo.core.client.AccumuloException;
+import org.apache.accumulo.core.client.AccumuloSecurityException;
+import org.apache.accumulo.core.client.impl.ClientContext;
+import org.apache.accumulo.core.client.impl.ClientExec;
+import org.apache.accumulo.core.client.impl.ClientExecReturn;
+import org.apache.accumulo.core.client.impl.ThriftTransportPool;
+import org.apache.accumulo.core.client.impl.thrift.ThriftSecurityException;
+import org.apache.accumulo.core.tabletserver.thrift.TabletClientService;
+import org.apache.accumulo.core.trace.Span;
+import org.apache.accumulo.core.trace.Trace;
+import org.apache.accumulo.core.util.UtilWaitThread;
+import org.apache.log4j.Logger;
+import org.apache.thrift.TException;
+import org.apache.thrift.TServiceClient;
+import org.apache.thrift.TServiceClientFactory;
+import org.apache.thrift.protocol.TCompactProtocol;
+import org.apache.thrift.protocol.TMessage;
+import org.apache.thrift.protocol.TProtocol;
+import org.apache.thrift.protocol.TProtocolFactory;
+import org.apache.thrift.transport.TFramedTransport;
+import org.apache.thrift.transport.TSSLTransportFactory;
+import org.apache.thrift.transport.TServerSocket;
+import org.apache.thrift.transport.TSocket;
+import org.apache.thrift.transport.TTransport;
+import org.apache.thrift.transport.TTransportException;
+import org.apache.thrift.transport.TTransportFactory;
+
+import com.google.common.base.Preconditions;
+import com.google.common.net.HostAndPort;
+
+public class ThriftUtil {
+ private static final Logger log = Logger.getLogger(ThriftUtil.class);
+
+ public static class TraceProtocol extends TCompactProtocol {
+ private Span span = null;
+
+ @Override
+ public void writeMessageBegin(TMessage message) throws TException {
+ span = Trace.start("client:" + message.name);
+ super.writeMessageBegin(message);
+ }
+
+ @Override
+ public void writeMessageEnd() throws TException {
+ super.writeMessageEnd();
+ span.stop();
+ }
+
+ public TraceProtocol(TTransport transport) {
+ super(transport);
+ }
+ }
+
+ public static class TraceProtocolFactory extends TCompactProtocol.Factory {
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public TProtocol getProtocol(TTransport trans) {
+ return new TraceProtocol(trans);
+ }
+ }
+
+ static private TProtocolFactory protocolFactory = new TraceProtocolFactory();
+ static private TTransportFactory transportFactory = new TFramedTransport.Factory(Integer.MAX_VALUE);
+
+ static public <T extends TServiceClient> T createClient(TServiceClientFactory<T> factory, TTransport transport) {
+ return factory.getClient(protocolFactory.getProtocol(transport), protocolFactory.getProtocol(transport));
+ }
+
+ static public <T extends TServiceClient> T getClient(TServiceClientFactory<T> factory, HostAndPort address, ClientContext context)
+ throws TTransportException {
+ return createClient(factory, ThriftTransportPool.getInstance().getTransportWithDefaultTimeout(address, context));
+ }
+
+ static public <T extends TServiceClient> T getClientNoTimeout(TServiceClientFactory<T> factory, String address, ClientContext context)
+ throws TTransportException {
+ return getClient(factory, address, context, 0);
+ }
+
+ static public <T extends TServiceClient> T getClient(TServiceClientFactory<T> factory, String address, ClientContext context)
+ throws TTransportException {
+ TTransport transport = ThriftTransportPool.getInstance().getTransport(address, context.getClientTimeoutInMillis(), context);
+ return createClient(factory, transport);
+ }
+
+ static private <T extends TServiceClient> T getClient(TServiceClientFactory<T> factory, String address, ClientContext context, long timeout)
+ throws TTransportException {
+ TTransport transport = ThriftTransportPool.getInstance().getTransport(address, timeout, context);
+ return createClient(factory, transport);
+ }
+
+ static public void returnClient(TServiceClient iface) { // Eew... the typing here is horrible
+ if (iface != null) {
+ ThriftTransportPool.getInstance().returnTransport(iface.getInputProtocol().getTransport());
+ }
+ }
+
+ static public TabletClientService.Client getTServerClient(String address, ClientContext context) throws TTransportException {
+ return getClient(new TabletClientService.Client.Factory(), address, context);
+ }
+
+ static public TabletClientService.Client getTServerClient(String address, ClientContext context, long timeout) throws TTransportException {
+ return getClient(new TabletClientService.Client.Factory(), address, context, timeout);
+ }
+
+ public static void execute(String address, ClientContext context, ClientExec<TabletClientService.Client> exec) throws AccumuloException,
+ AccumuloSecurityException {
+ while (true) {
+ TabletClientService.Client client = null;
+ try {
+ exec.execute(client = getTServerClient(address, context));
+ break;
+ } catch (TTransportException tte) {
+ log.debug("getTServerClient request failed, retrying ... ", tte);
+ UtilWaitThread.sleep(100);
+ } catch (ThriftSecurityException e) {
+ throw new AccumuloSecurityException(e.user, e.code, e);
+ } catch (Exception e) {
+ throw new AccumuloException(e);
+ } finally {
+ if (client != null)
+ returnClient(client);
+ }
+ }
+ }
+
+ public static <T> T execute(String address, ClientContext context, ClientExecReturn<T,TabletClientService.Client> exec) throws AccumuloException,
+ AccumuloSecurityException {
+ while (true) {
+ TabletClientService.Client client = null;
+ try {
+ return exec.execute(client = getTServerClient(address, context));
+ } catch (TTransportException tte) {
+ log.debug("getTServerClient request failed, retrying ... ", tte);
+ UtilWaitThread.sleep(100);
+ } catch (ThriftSecurityException e) {
+ throw new AccumuloSecurityException(e.user, e.code, e);
+ } catch (Exception e) {
+ throw new AccumuloException(e);
+ } finally {
+ if (client != null)
+ returnClient(client);
+ }
+ }
+ }
+
+ /**
+ * create a transport that is not pooled
+ */
+ public static TTransport createTransport(HostAndPort address, ClientContext context) throws TException {
+ return createClientTransport(address, (int) context.getClientTimeoutInMillis(), context.getClientSslParams());
+ }
+
+ public static TTransportFactory transportFactory() {
+ return transportFactory;
+ }
+
+ private final static Map<Integer,TTransportFactory> factoryCache = new HashMap<Integer,TTransportFactory>();
+
+ synchronized public static TTransportFactory transportFactory(int maxFrameSize) {
+ TTransportFactory factory = factoryCache.get(maxFrameSize);
+ if (factory == null) {
+ factory = new TFramedTransport.Factory(maxFrameSize);
+ factoryCache.put(maxFrameSize, factory);
+ }
+ return factory;
+ }
+
+ synchronized public static TTransportFactory transportFactory(long maxFrameSize) {
+ if (maxFrameSize > Integer.MAX_VALUE || maxFrameSize < 1)
+ throw new RuntimeException("Thrift transport frames are limited to " + Integer.MAX_VALUE);
+ return transportFactory((int) maxFrameSize);
+ }
+
+ public static TProtocolFactory protocolFactory() {
+ return protocolFactory;
+ }
+
+ public static TServerSocket getServerSocket(int port, int timeout, InetAddress address, SslConnectionParams params) throws TTransportException {
+ TServerSocket tServerSock;
+ if (params.useJsse()) {
+ tServerSock = TSSLTransportFactory.getServerSocket(port, timeout, params.isClientAuth(), address);
+ } else {
+ tServerSock = TSSLTransportFactory.getServerSocket(port, timeout, address, params.getTTransportParams());
+ }
+
+ ServerSocket serverSock = tServerSock.getServerSocket();
+ if (serverSock instanceof SSLServerSocket) {
+ SSLServerSocket sslServerSock = (SSLServerSocket) serverSock;
+ String[] protocols = params.getServerProtocols();
+
+ // Be nice for the user and automatically remove protocols that might not exist in their JVM. Keeps us from forcing config alterations too
+ // e.g. TLSv1.1 and TLSv1.2 don't exist in JDK6
+ Set<String> socketEnabledProtocols = new HashSet<String>(Arrays.asList(sslServerSock.getEnabledProtocols()));
+ // Keep only the enabled protocols that were specified by the configuration
+ socketEnabledProtocols.retainAll(Arrays.asList(protocols));
+ if (socketEnabledProtocols.isEmpty()) {
+ // Bad configuration...
+ throw new RuntimeException("No available protocols available for secure socket. Availaable protocols: "
+ + Arrays.toString(sslServerSock.getEnabledProtocols()) + ", allowed protocols: " + Arrays.toString(protocols));
+ }
+
+ // Set the protocol(s) on the server socket
+ sslServerSock.setEnabledProtocols(socketEnabledProtocols.toArray(new String[0]));
+ }
+
+ return tServerSock;
+ }
+
+ public static TTransport createClientTransport(HostAndPort address, int timeout, SslConnectionParams sslParams) throws TTransportException {
+ boolean success = false;
+ TTransport transport = null;
+ try {
+ if (sslParams != null) {
+ // TSSLTransportFactory handles timeout 0 -> forever natively
+ if (sslParams.useJsse()) {
+ transport = TSSLTransportFactory.getClientSocket(address.getHostText(), address.getPort(), timeout);
+ } else {
+ // JDK6's factory doesn't appear to pass the protocol onto the Socket properly so we have
+ // to do some magic to make sure that happens. Not an issue in JDK7
+
+ // Taken from thrift-0.9.1 to make the SSLContext
+ SSLContext sslContext = createSSLContext(sslParams);
+
+ // Create the factory from it
+ SSLSocketFactory sslSockFactory = sslContext.getSocketFactory();
+
+ // Wrap the real factory with our own that will set the protocol on the Socket before returning it
+ ProtocolOverridingSSLSocketFactory wrappingSslSockFactory = new ProtocolOverridingSSLSocketFactory(sslSockFactory,
+ new String[] {sslParams.getClientProtocol()});
+
+ // Create the TSocket from that
+ transport = createClient(wrappingSslSockFactory, address.getHostText(), address.getPort(), timeout);
+ }
+ // TSSLTransportFactory leaves transports open, so no need to open here
+ } else if (timeout == 0) {
+ transport = new TSocket(address.getHostText(), address.getPort());
+ transport.open();
+ } else {
+ try {
+ transport = TTimeoutTransport.create(address, timeout);
+ } catch (IOException ex) {
+ throw new TTransportException(ex);
+ }
+ transport.open();
+ }
+ transport = ThriftUtil.transportFactory().getTransport(transport);
+ success = true;
+ } finally {
+ if (!success && transport != null) {
+ transport.close();
+ }
+ }
+ return transport;
+ }
+
+ /**
+ * Lifted from TSSLTransportFactory in Thrift-0.9.1. The method to create a client socket with an SSLContextFactory object is not visibile to us. Have to use
+ * SslConnectionParams instead of TSSLTransportParameters because no getters exist on TSSLTransportParameters.
+ *
+ * @param params
+ * Parameters to use to create the SSLContext
+ */
+ private static SSLContext createSSLContext(SslConnectionParams params) throws TTransportException {
+ SSLContext ctx;
+ try {
+ ctx = SSLContext.getInstance(params.getClientProtocol());
+ TrustManagerFactory tmf = null;
+ KeyManagerFactory kmf = null;
+
+ if (params.isTrustStoreSet()) {
+ tmf = TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm());
+ KeyStore ts = KeyStore.getInstance(params.getTrustStoreType());
+ ts.load(new FileInputStream(params.getTrustStorePath()), params.getTrustStorePass().toCharArray());
+ tmf.init(ts);
+ }
+
+ if (params.isKeyStoreSet()) {
+ kmf = KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm());
+ KeyStore ks = KeyStore.getInstance(params.getKeyStoreType());
+ ks.load(new FileInputStream(params.getKeyStorePath()), params.getKeyStorePass().toCharArray());
+ kmf.init(ks, params.getKeyStorePass().toCharArray());
+ }
+
+ if (params.isKeyStoreSet() && params.isTrustStoreSet()) {
+ ctx.init(kmf.getKeyManagers(), tmf.getTrustManagers(), null);
+ } else if (params.isKeyStoreSet()) {
+ ctx.init(kmf.getKeyManagers(), null, null);
+ } else {
+ ctx.init(null, tmf.getTrustManagers(), null);
+ }
+
+ } catch (Exception e) {
+ throw new TTransportException("Error creating the transport", e);
+ }
+ return ctx;
+ }
+
+ /**
+ * Lifted from Thrift-0.9.1 because it was private. Create an SSLSocket with the given factory, host:port, and timeout.
+ *
+ * @param factory
+ * Factory to create the socket from
+ * @param host
+ * Destination host
+ * @param port
+ * Destination port
+ * @param timeout
+ * Socket timeout
+ */
+ private static TSocket createClient(SSLSocketFactory factory, String host, int port, int timeout) throws TTransportException {
+ try {
+ SSLSocket socket = (SSLSocket) factory.createSocket(host, port);
+ socket.setSoTimeout(timeout);
+ return new TSocket(socket);
+ } catch (Exception e) {
+ throw new TTransportException("Could not connect to " + host + " on port " + port, e);
+ }
+ }
+
+ /**
+ * JDK6's SSLSocketFactory doesn't seem to properly set the protocols on the Sockets that it creates which causes an SSLv2 client hello message during
+ * handshake, even when only TLSv1 is enabled. This only appears to be an issue on the client sockets, not the server sockets.
+ *
+ * This class wraps the SSLSocketFactory ensuring that the Socket is properly configured.
+ * http://www.coderanch.com/t/637177/Security/Disabling-handshake-message-Java
+ *
+ * This class can be removed when JDK6 support is officially unsupported by Accumulo
+ */
+ private static class ProtocolOverridingSSLSocketFactory extends SSLSocketFactory {
+
+ private final SSLSocketFactory delegate;
+ private final String[] enabledProtocols;
+
+ public ProtocolOverridingSSLSocketFactory(final SSLSocketFactory delegate, final String[] enabledProtocols) {
+ Preconditions.checkNotNull(enabledProtocols);
+ Preconditions.checkArgument(0 != enabledProtocols.length, "Expected at least one protocol");
+ this.delegate = delegate;
+ this.enabledProtocols = enabledProtocols;
+ }
+
+ @Override
+ public String[] getDefaultCipherSuites() {
+ return delegate.getDefaultCipherSuites();
+ }
+
+ @Override
+ public String[] getSupportedCipherSuites() {
+ return delegate.getSupportedCipherSuites();
+ }
+
+ @Override
+ public Socket createSocket(final Socket socket, final String host, final int port, final boolean autoClose) throws IOException {
+ final Socket underlyingSocket = delegate.createSocket(socket, host, port, autoClose);
+ return overrideProtocol(underlyingSocket);
+ }
+
+ @Override
+ public Socket createSocket(final String host, final int port) throws IOException, UnknownHostException {
+ final Socket underlyingSocket = delegate.createSocket(host, port);
+ return overrideProtocol(underlyingSocket);
+ }
+
+ @Override
+ public Socket createSocket(final String host, final int port, final InetAddress localAddress, final int localPort) throws IOException, UnknownHostException {
+ final Socket underlyingSocket = delegate.createSocket(host, port, localAddress, localPort);
+ return overrideProtocol(underlyingSocket);
+ }
+
+ @Override
+ public Socket createSocket(final InetAddress host, final int port) throws IOException {
+ final Socket underlyingSocket = delegate.createSocket(host, port);
+ return overrideProtocol(underlyingSocket);
+ }
+
+ @Override
+ public Socket createSocket(final InetAddress host, final int port, final InetAddress localAddress, final int localPort) throws IOException {
+ final Socket underlyingSocket = delegate.createSocket(host, port, localAddress, localPort);
+ return overrideProtocol(underlyingSocket);
+ }
+
+ /**
+ * Set the {@link javax.net.ssl.SSLSocket#getEnabledProtocols() enabled protocols} to {@link #enabledProtocols} if the <code>socket</code> is a
+ * {@link SSLSocket}
+ *
+ * @param socket
+ * The Socket
+ */
+ private Socket overrideProtocol(final Socket socket) {
+ if (socket instanceof SSLSocket) {
+ ((SSLSocket) socket).setEnabledProtocols(enabledProtocols);
+ }
+ return socket;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/de1d3ee3/core/src/main/java/org/apache/accumulo/core/util/SslConnectionParams.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/util/SslConnectionParams.java b/core/src/main/java/org/apache/accumulo/core/util/SslConnectionParams.java
deleted file mode 100644
index cd433c6..0000000
--- a/core/src/main/java/org/apache/accumulo/core/util/SslConnectionParams.java
+++ /dev/null
@@ -1,275 +0,0 @@
-/*
- * 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.core.util;
-
-import java.io.File;
-import java.io.FileNotFoundException;
-import java.net.URL;
-import java.util.Arrays;
-
-import org.apache.accumulo.core.conf.AccumuloConfiguration;
-import org.apache.accumulo.core.conf.Property;
-import org.apache.commons.lang.StringUtils;
-import org.apache.log4j.Logger;
-import org.apache.thrift.transport.TSSLTransportFactory.TSSLTransportParameters;
-
-public class SslConnectionParams {
- private static final Logger log = Logger.getLogger(SslConnectionParams.class);
-
- private boolean useJsse = false;
- private boolean clientAuth = false;
-
- private boolean keyStoreSet;
- private String keyStorePath;
- private String keyStorePass;
- private String keyStoreType;
-
- private boolean trustStoreSet;
- private String trustStorePath;
- private String trustStorePass;
- private String trustStoreType;
-
- private String[] cipherSuites;
- private String[] serverProtocols;
- private String clientProtocol;
-
- // Use the static construction methods
- private SslConnectionParams() {}
-
- public static SslConnectionParams forConfig(AccumuloConfiguration conf, boolean server) {
- if (!conf.getBoolean(Property.INSTANCE_RPC_SSL_ENABLED))
- return null;
-
- SslConnectionParams result = new SslConnectionParams();
- boolean requireClientAuth = conf.getBoolean(Property.INSTANCE_RPC_SSL_CLIENT_AUTH);
- if (server) {
- result.setClientAuth(requireClientAuth);
- }
- if (conf.getBoolean(Property.RPC_USE_JSSE)) {
- result.setUseJsse(true);
- return result;
- }
-
- try {
- if (!server || requireClientAuth) {
- result.setTrustStoreFromConf(conf);
- }
- if (server || requireClientAuth) {
- result.setKeyStoreFromConf(conf);
- }
- } catch (FileNotFoundException e) {
- throw new IllegalArgumentException("Could not load configured keystore file", e);
- }
-
- String ciphers = conf.get(Property.RPC_SSL_CIPHER_SUITES);
- if (null != ciphers && !ciphers.isEmpty()) {
- result.cipherSuites = StringUtils.split(ciphers, ',');
- }
-
- String enabledProtocols = conf.get(Property.RPC_SSL_ENABLED_PROTOCOLS);
- result.serverProtocols = StringUtils.split(enabledProtocols, ',');
-
- result.clientProtocol = conf.get(Property.RPC_SSL_CLIENT_PROTOCOL);
-
- return result;
- }
-
- private static String passwordFromConf(AccumuloConfiguration conf, String defaultPassword, Property passwordOverrideProperty) {
- String keystorePassword = conf.get(passwordOverrideProperty);
- if (!keystorePassword.isEmpty()) {
- if (log.isTraceEnabled())
- log.trace("Using explicit SSL private key password from " + passwordOverrideProperty.getKey());
- } else {
- keystorePassword = defaultPassword;
- }
- return keystorePassword;
- }
-
- private static String storePathFromConf(AccumuloConfiguration conf, Property pathProperty) throws FileNotFoundException {
- return findKeystore(conf.getPath(pathProperty));
- }
-
- public void setKeyStoreFromConf(AccumuloConfiguration conf) throws FileNotFoundException {
- keyStoreSet = true;
- keyStorePath = storePathFromConf(conf, Property.RPC_SSL_KEYSTORE_PATH);
- keyStorePass = passwordFromConf(conf, conf.get(Property.INSTANCE_SECRET), Property.RPC_SSL_KEYSTORE_PASSWORD);
- keyStoreType = conf.get(Property.RPC_SSL_KEYSTORE_TYPE);
- }
-
- public void setTrustStoreFromConf(AccumuloConfiguration conf) throws FileNotFoundException {
- trustStoreSet = true;
- trustStorePath = storePathFromConf(conf, Property.RPC_SSL_TRUSTSTORE_PATH);
- trustStorePass = passwordFromConf(conf, "", Property.RPC_SSL_TRUSTSTORE_PASSWORD);
- trustStoreType = conf.get(Property.RPC_SSL_TRUSTSTORE_TYPE);
- }
-
- public static SslConnectionParams forServer(AccumuloConfiguration configuration) {
- return forConfig(configuration, true);
- }
-
- public static SslConnectionParams forClient(AccumuloConfiguration configuration) {
- return forConfig(configuration, false);
- }
-
- private static String findKeystore(String keystorePath) throws FileNotFoundException {
- try {
- // first just try the file
- File file = new File(keystorePath);
- if (file.exists())
- return file.getAbsolutePath();
- if (!file.isAbsolute()) {
- // try classpath
- URL url = SslConnectionParams.class.getClassLoader().getResource(keystorePath);
- if (url != null) {
- file = new File(url.toURI());
- if (file.exists())
- return file.getAbsolutePath();
- }
- }
- } catch (Exception e) {
- log.warn("Exception finding keystore", e);
- }
- throw new FileNotFoundException("Failed to load SSL keystore from " + keystorePath);
- }
-
- public void setUseJsse(boolean useJsse) {
- this.useJsse = useJsse;
- }
-
- public boolean useJsse() {
- return useJsse;
- }
-
- public void setClientAuth(boolean clientAuth) {
- this.clientAuth = clientAuth;
- }
-
- public boolean isClientAuth() {
- return clientAuth;
- }
-
- public String[] getServerProtocols() {
- return serverProtocols;
- }
-
- public String getClientProtocol() {
- return clientProtocol;
- }
-
- public boolean isKeyStoreSet() {
- return keyStoreSet;
- }
-
- public String getKeyStorePath() {
- return keyStorePath;
- }
-
- /**
- * @return the keyStorePass
- */
- public String getKeyStorePass() {
- return keyStorePass;
- }
-
- public String getKeyStoreType() {
- return keyStoreType;
- }
-
- public boolean isTrustStoreSet() {
- return trustStoreSet;
- }
-
- public String getTrustStorePath() {
- return trustStorePath;
- }
-
- public String getTrustStorePass() {
- return trustStorePass;
- }
-
- /**
- * @return the trustStoreType
- */
- public String getTrustStoreType() {
- return trustStoreType;
- }
-
- public TSSLTransportParameters getTTransportParams() {
- if (useJsse)
- throw new IllegalStateException("Cannot get TTransportParams for JSEE configuration.");
-
- // Null cipherSuites is implicitly handled
- TSSLTransportParameters params = new TSSLTransportParameters(clientProtocol, cipherSuites);
-
- params.requireClientAuth(clientAuth);
- if (keyStoreSet) {
- params.setKeyStore(keyStorePath, keyStorePass, null, keyStoreType);
- }
- if (trustStoreSet) {
- params.setTrustStore(trustStorePath, trustStorePass, null, trustStoreType);
- }
- return params;
- }
-
- @Override
- public int hashCode() {
- int hash = 0;
- hash = 31 * hash + (clientAuth ? 0 : 1);
- hash = 31 * hash + (useJsse ? 0 : 1);
- if (useJsse)
- return hash;
- hash = 31 * hash + (keyStoreSet ? 0 : 1);
- hash = 31 * hash + (trustStoreSet ? 0 : 1);
- if (keyStoreSet) {
- hash = 31 * hash + keyStorePath.hashCode();
- }
- if (trustStoreSet) {
- hash = 31 * hash + trustStorePath.hashCode();
- }
- hash = 31 * hash + clientProtocol.hashCode();
- hash = 31 * hash + Arrays.hashCode(serverProtocols);
- return super.hashCode();
- }
-
- @Override
- public boolean equals(Object obj) {
- if (!(obj instanceof SslConnectionParams))
- return false;
-
- SslConnectionParams other = (SslConnectionParams) obj;
- if (clientAuth != other.clientAuth)
- return false;
- if (useJsse)
- return other.useJsse;
- if (keyStoreSet) {
- if (!other.keyStoreSet)
- return false;
- if (!keyStorePath.equals(other.keyStorePath) || !keyStorePass.equals(other.keyStorePass) || !keyStoreType.equals(other.keyStoreType))
- return false;
- }
- if (trustStoreSet) {
- if (!other.trustStoreSet)
- return false;
- if (!trustStorePath.equals(other.trustStorePath) || !trustStorePass.equals(other.trustStorePass) || !trustStoreType.equals(other.trustStoreType))
- return false;
- }
- if (!Arrays.equals(serverProtocols, other.serverProtocols)) {
- return false;
- }
- return clientProtocol.equals(other.clientProtocol);
- }
-}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/de1d3ee3/core/src/main/java/org/apache/accumulo/core/util/TBufferedSocket.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/util/TBufferedSocket.java b/core/src/main/java/org/apache/accumulo/core/util/TBufferedSocket.java
deleted file mode 100644
index c2d9400..0000000
--- a/core/src/main/java/org/apache/accumulo/core/util/TBufferedSocket.java
+++ /dev/null
@@ -1,39 +0,0 @@
-/*
- * 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.core.util;
-
-import java.io.BufferedInputStream;
-import java.io.BufferedOutputStream;
-import java.io.IOException;
-
-import org.apache.thrift.transport.TIOStreamTransport;
-import org.apache.thrift.transport.TSocket;
-
-public class TBufferedSocket extends TIOStreamTransport {
-
- String client;
-
- public TBufferedSocket(TSocket sock, int bufferSize) throws IOException {
- super(new BufferedInputStream(sock.getSocket().getInputStream(), bufferSize), new BufferedOutputStream(sock.getSocket().getOutputStream(), bufferSize));
- client = sock.getSocket().getInetAddress().getHostAddress() + ":" + sock.getSocket().getPort();
- }
-
- public String getClientString() {
- return client;
- }
-
-}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/de1d3ee3/core/src/main/java/org/apache/accumulo/core/util/TTimeoutTransport.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/util/TTimeoutTransport.java b/core/src/main/java/org/apache/accumulo/core/util/TTimeoutTransport.java
deleted file mode 100644
index a66b268..0000000
--- a/core/src/main/java/org/apache/accumulo/core/util/TTimeoutTransport.java
+++ /dev/null
@@ -1,60 +0,0 @@
-/*
- * 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.core.util;
-
-import java.io.BufferedInputStream;
-import java.io.BufferedOutputStream;
-import java.io.IOException;
-import java.io.InputStream;
-import java.io.OutputStream;
-import java.lang.reflect.Method;
-import java.net.InetSocketAddress;
-import java.net.Socket;
-import java.net.SocketAddress;
-import java.nio.channels.spi.SelectorProvider;
-
-import org.apache.hadoop.net.NetUtils;
-import org.apache.thrift.transport.TIOStreamTransport;
-import org.apache.thrift.transport.TTransport;
-
-import com.google.common.net.HostAndPort;
-
-public class TTimeoutTransport {
-
- private static InputStream getInputStream(Socket socket, long timeout) {
- try {
- Method m = NetUtils.class.getMethod("getInputStream", Socket.class, Long.TYPE);
- return (InputStream)m.invoke(null, socket, timeout);
- } catch (Exception e) {
- throw new RuntimeException(e);
- }
- }
-
- public static TTransport create(HostAndPort addr, long timeoutMillis) throws IOException {
- return create(new InetSocketAddress(addr.getHostText(), addr.getPort()), timeoutMillis);
- }
-
- public static TTransport create(SocketAddress addr, long timeoutMillis) throws IOException {
- Socket socket = SelectorProvider.provider().openSocketChannel().socket();
- socket.setSoLinger(false, 0);
- socket.setTcpNoDelay(true);
- socket.connect(addr);
- InputStream input = new BufferedInputStream(getInputStream(socket, timeoutMillis), 1024 * 10);
- OutputStream output = new BufferedOutputStream(NetUtils.getOutputStream(socket, timeoutMillis), 1024 * 10);
- return new TIOStreamTransport(input, output);
- }
-}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/de1d3ee3/core/src/main/java/org/apache/accumulo/core/util/ThriftUtil.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/util/ThriftUtil.java b/core/src/main/java/org/apache/accumulo/core/util/ThriftUtil.java
deleted file mode 100644
index 619131a..0000000
--- a/core/src/main/java/org/apache/accumulo/core/util/ThriftUtil.java
+++ /dev/null
@@ -1,433 +0,0 @@
-/*
- * 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.core.util;
-
-import java.io.FileInputStream;
-import java.io.IOException;
-import java.net.InetAddress;
-import java.net.ServerSocket;
-import java.net.Socket;
-import java.net.UnknownHostException;
-import java.security.KeyStore;
-import java.util.Arrays;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.Map;
-import java.util.Set;
-
-import javax.net.ssl.KeyManagerFactory;
-import javax.net.ssl.SSLContext;
-import javax.net.ssl.SSLServerSocket;
-import javax.net.ssl.SSLSocket;
-import javax.net.ssl.SSLSocketFactory;
-import javax.net.ssl.TrustManagerFactory;
-
-import org.apache.accumulo.core.client.AccumuloException;
-import org.apache.accumulo.core.client.AccumuloSecurityException;
-import org.apache.accumulo.core.client.impl.ClientContext;
-import org.apache.accumulo.core.client.impl.ClientExec;
-import org.apache.accumulo.core.client.impl.ClientExecReturn;
-import org.apache.accumulo.core.client.impl.ThriftTransportPool;
-import org.apache.accumulo.core.client.impl.thrift.ThriftSecurityException;
-import org.apache.accumulo.core.tabletserver.thrift.TabletClientService;
-import org.apache.accumulo.core.trace.Span;
-import org.apache.accumulo.core.trace.Trace;
-import org.apache.log4j.Logger;
-import org.apache.thrift.TException;
-import org.apache.thrift.TServiceClient;
-import org.apache.thrift.TServiceClientFactory;
-import org.apache.thrift.protocol.TCompactProtocol;
-import org.apache.thrift.protocol.TMessage;
-import org.apache.thrift.protocol.TProtocol;
-import org.apache.thrift.protocol.TProtocolFactory;
-import org.apache.thrift.transport.TFramedTransport;
-import org.apache.thrift.transport.TSSLTransportFactory;
-import org.apache.thrift.transport.TServerSocket;
-import org.apache.thrift.transport.TSocket;
-import org.apache.thrift.transport.TTransport;
-import org.apache.thrift.transport.TTransportException;
-import org.apache.thrift.transport.TTransportFactory;
-
-import com.google.common.base.Preconditions;
-import com.google.common.net.HostAndPort;
-
-public class ThriftUtil {
- private static final Logger log = Logger.getLogger(ThriftUtil.class);
-
- public static class TraceProtocol extends TCompactProtocol {
- private Span span = null;
-
- @Override
- public void writeMessageBegin(TMessage message) throws TException {
- span = Trace.start("client:" + message.name);
- super.writeMessageBegin(message);
- }
-
- @Override
- public void writeMessageEnd() throws TException {
- super.writeMessageEnd();
- span.stop();
- }
-
- public TraceProtocol(TTransport transport) {
- super(transport);
- }
- }
-
- public static class TraceProtocolFactory extends TCompactProtocol.Factory {
- private static final long serialVersionUID = 1L;
-
- @Override
- public TProtocol getProtocol(TTransport trans) {
- return new TraceProtocol(trans);
- }
- }
-
- static private TProtocolFactory protocolFactory = new TraceProtocolFactory();
- static private TTransportFactory transportFactory = new TFramedTransport.Factory(Integer.MAX_VALUE);
-
- static public <T extends TServiceClient> T createClient(TServiceClientFactory<T> factory, TTransport transport) {
- return factory.getClient(protocolFactory.getProtocol(transport), protocolFactory.getProtocol(transport));
- }
-
- static public <T extends TServiceClient> T getClient(TServiceClientFactory<T> factory, HostAndPort address, ClientContext context)
- throws TTransportException {
- return createClient(factory, ThriftTransportPool.getInstance().getTransportWithDefaultTimeout(address, context));
- }
-
- static public <T extends TServiceClient> T getClientNoTimeout(TServiceClientFactory<T> factory, String address, ClientContext context)
- throws TTransportException {
- return getClient(factory, address, context, 0);
- }
-
- static public <T extends TServiceClient> T getClient(TServiceClientFactory<T> factory, String address, ClientContext context)
- throws TTransportException {
- TTransport transport = ThriftTransportPool.getInstance().getTransport(address, context.getClientTimeoutInMillis(), context);
- return createClient(factory, transport);
- }
-
- static private <T extends TServiceClient> T getClient(TServiceClientFactory<T> factory, String address, ClientContext context, long timeout)
- throws TTransportException {
- TTransport transport = ThriftTransportPool.getInstance().getTransport(address, timeout, context);
- return createClient(factory, transport);
- }
-
- static public void returnClient(TServiceClient iface) { // Eew... the typing here is horrible
- if (iface != null) {
- ThriftTransportPool.getInstance().returnTransport(iface.getInputProtocol().getTransport());
- }
- }
-
- static public TabletClientService.Client getTServerClient(String address, ClientContext context) throws TTransportException {
- return getClient(new TabletClientService.Client.Factory(), address, context);
- }
-
- static public TabletClientService.Client getTServerClient(String address, ClientContext context, long timeout) throws TTransportException {
- return getClient(new TabletClientService.Client.Factory(), address, context, timeout);
- }
-
- public static void execute(String address, ClientContext context, ClientExec<TabletClientService.Client> exec) throws AccumuloException,
- AccumuloSecurityException {
- while (true) {
- TabletClientService.Client client = null;
- try {
- exec.execute(client = getTServerClient(address, context));
- break;
- } catch (TTransportException tte) {
- log.debug("getTServerClient request failed, retrying ... ", tte);
- UtilWaitThread.sleep(100);
- } catch (ThriftSecurityException e) {
- throw new AccumuloSecurityException(e.user, e.code, e);
- } catch (Exception e) {
- throw new AccumuloException(e);
- } finally {
- if (client != null)
- returnClient(client);
- }
- }
- }
-
- public static <T> T execute(String address, ClientContext context, ClientExecReturn<T,TabletClientService.Client> exec) throws AccumuloException,
- AccumuloSecurityException {
- while (true) {
- TabletClientService.Client client = null;
- try {
- return exec.execute(client = getTServerClient(address, context));
- } catch (TTransportException tte) {
- log.debug("getTServerClient request failed, retrying ... ", tte);
- UtilWaitThread.sleep(100);
- } catch (ThriftSecurityException e) {
- throw new AccumuloSecurityException(e.user, e.code, e);
- } catch (Exception e) {
- throw new AccumuloException(e);
- } finally {
- if (client != null)
- returnClient(client);
- }
- }
- }
-
- /**
- * create a transport that is not pooled
- */
- public static TTransport createTransport(HostAndPort address, ClientContext context) throws TException {
- return createClientTransport(address, (int) context.getClientTimeoutInMillis(), context.getClientSslParams());
- }
-
- public static TTransportFactory transportFactory() {
- return transportFactory;
- }
-
- private final static Map<Integer,TTransportFactory> factoryCache = new HashMap<Integer,TTransportFactory>();
-
- synchronized public static TTransportFactory transportFactory(int maxFrameSize) {
- TTransportFactory factory = factoryCache.get(maxFrameSize);
- if (factory == null) {
- factory = new TFramedTransport.Factory(maxFrameSize);
- factoryCache.put(maxFrameSize, factory);
- }
- return factory;
- }
-
- synchronized public static TTransportFactory transportFactory(long maxFrameSize) {
- if (maxFrameSize > Integer.MAX_VALUE || maxFrameSize < 1)
- throw new RuntimeException("Thrift transport frames are limited to " + Integer.MAX_VALUE);
- return transportFactory((int) maxFrameSize);
- }
-
- public static TProtocolFactory protocolFactory() {
- return protocolFactory;
- }
-
- public static TServerSocket getServerSocket(int port, int timeout, InetAddress address, SslConnectionParams params) throws TTransportException {
- TServerSocket tServerSock;
- if (params.useJsse()) {
- tServerSock = TSSLTransportFactory.getServerSocket(port, timeout, params.isClientAuth(), address);
- } else {
- tServerSock = TSSLTransportFactory.getServerSocket(port, timeout, address, params.getTTransportParams());
- }
-
- ServerSocket serverSock = tServerSock.getServerSocket();
- if (serverSock instanceof SSLServerSocket) {
- SSLServerSocket sslServerSock = (SSLServerSocket) serverSock;
- String[] protocols = params.getServerProtocols();
-
- // Be nice for the user and automatically remove protocols that might not exist in their JVM. Keeps us from forcing config alterations too
- // e.g. TLSv1.1 and TLSv1.2 don't exist in JDK6
- Set<String> socketEnabledProtocols = new HashSet<String>(Arrays.asList(sslServerSock.getEnabledProtocols()));
- // Keep only the enabled protocols that were specified by the configuration
- socketEnabledProtocols.retainAll(Arrays.asList(protocols));
- if (socketEnabledProtocols.isEmpty()) {
- // Bad configuration...
- throw new RuntimeException("No available protocols available for secure socket. Availaable protocols: "
- + Arrays.toString(sslServerSock.getEnabledProtocols()) + ", allowed protocols: " + Arrays.toString(protocols));
- }
-
- // Set the protocol(s) on the server socket
- sslServerSock.setEnabledProtocols(socketEnabledProtocols.toArray(new String[0]));
- }
-
- return tServerSock;
- }
-
- public static TTransport createClientTransport(HostAndPort address, int timeout, SslConnectionParams sslParams) throws TTransportException {
- boolean success = false;
- TTransport transport = null;
- try {
- if (sslParams != null) {
- // TSSLTransportFactory handles timeout 0 -> forever natively
- if (sslParams.useJsse()) {
- transport = TSSLTransportFactory.getClientSocket(address.getHostText(), address.getPort(), timeout);
- } else {
- // JDK6's factory doesn't appear to pass the protocol onto the Socket properly so we have
- // to do some magic to make sure that happens. Not an issue in JDK7
-
- // Taken from thrift-0.9.1 to make the SSLContext
- SSLContext sslContext = createSSLContext(sslParams);
-
- // Create the factory from it
- SSLSocketFactory sslSockFactory = sslContext.getSocketFactory();
-
- // Wrap the real factory with our own that will set the protocol on the Socket before returning it
- ProtocolOverridingSSLSocketFactory wrappingSslSockFactory = new ProtocolOverridingSSLSocketFactory(sslSockFactory,
- new String[] {sslParams.getClientProtocol()});
-
- // Create the TSocket from that
- transport = createClient(wrappingSslSockFactory, address.getHostText(), address.getPort(), timeout);
- }
- // TSSLTransportFactory leaves transports open, so no need to open here
- } else if (timeout == 0) {
- transport = new TSocket(address.getHostText(), address.getPort());
- transport.open();
- } else {
- try {
- transport = TTimeoutTransport.create(address, timeout);
- } catch (IOException ex) {
- throw new TTransportException(ex);
- }
- transport.open();
- }
- transport = ThriftUtil.transportFactory().getTransport(transport);
- success = true;
- } finally {
- if (!success && transport != null) {
- transport.close();
- }
- }
- return transport;
- }
-
- /**
- * Lifted from TSSLTransportFactory in Thrift-0.9.1. The method to create a client socket with an SSLContextFactory object is not visibile to us. Have to use
- * SslConnectionParams instead of TSSLTransportParameters because no getters exist on TSSLTransportParameters.
- *
- * @param params
- * Parameters to use to create the SSLContext
- */
- private static SSLContext createSSLContext(SslConnectionParams params) throws TTransportException {
- SSLContext ctx;
- try {
- ctx = SSLContext.getInstance(params.getClientProtocol());
- TrustManagerFactory tmf = null;
- KeyManagerFactory kmf = null;
-
- if (params.isTrustStoreSet()) {
- tmf = TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm());
- KeyStore ts = KeyStore.getInstance(params.getTrustStoreType());
- ts.load(new FileInputStream(params.getTrustStorePath()), params.getTrustStorePass().toCharArray());
- tmf.init(ts);
- }
-
- if (params.isKeyStoreSet()) {
- kmf = KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm());
- KeyStore ks = KeyStore.getInstance(params.getKeyStoreType());
- ks.load(new FileInputStream(params.getKeyStorePath()), params.getKeyStorePass().toCharArray());
- kmf.init(ks, params.getKeyStorePass().toCharArray());
- }
-
- if (params.isKeyStoreSet() && params.isTrustStoreSet()) {
- ctx.init(kmf.getKeyManagers(), tmf.getTrustManagers(), null);
- } else if (params.isKeyStoreSet()) {
- ctx.init(kmf.getKeyManagers(), null, null);
- } else {
- ctx.init(null, tmf.getTrustManagers(), null);
- }
-
- } catch (Exception e) {
- throw new TTransportException("Error creating the transport", e);
- }
- return ctx;
- }
-
- /**
- * Lifted from Thrift-0.9.1 because it was private. Create an SSLSocket with the given factory, host:port, and timeout.
- *
- * @param factory
- * Factory to create the socket from
- * @param host
- * Destination host
- * @param port
- * Destination port
- * @param timeout
- * Socket timeout
- */
- private static TSocket createClient(SSLSocketFactory factory, String host, int port, int timeout) throws TTransportException {
- try {
- SSLSocket socket = (SSLSocket) factory.createSocket(host, port);
- socket.setSoTimeout(timeout);
- return new TSocket(socket);
- } catch (Exception e) {
- throw new TTransportException("Could not connect to " + host + " on port " + port, e);
- }
- }
-
- /**
- * JDK6's SSLSocketFactory doesn't seem to properly set the protocols on the Sockets that it creates which causes an SSLv2 client hello message during
- * handshake, even when only TLSv1 is enabled. This only appears to be an issue on the client sockets, not the server sockets.
- *
- * This class wraps the SSLSocketFactory ensuring that the Socket is properly configured.
- * http://www.coderanch.com/t/637177/Security/Disabling-handshake-message-Java
- *
- * This class can be removed when JDK6 support is officially unsupported by Accumulo
- */
- private static class ProtocolOverridingSSLSocketFactory extends SSLSocketFactory {
-
- private final SSLSocketFactory delegate;
- private final String[] enabledProtocols;
-
- public ProtocolOverridingSSLSocketFactory(final SSLSocketFactory delegate, final String[] enabledProtocols) {
- Preconditions.checkNotNull(enabledProtocols);
- Preconditions.checkArgument(0 != enabledProtocols.length, "Expected at least one protocol");
- this.delegate = delegate;
- this.enabledProtocols = enabledProtocols;
- }
-
- @Override
- public String[] getDefaultCipherSuites() {
- return delegate.getDefaultCipherSuites();
- }
-
- @Override
- public String[] getSupportedCipherSuites() {
- return delegate.getSupportedCipherSuites();
- }
-
- @Override
- public Socket createSocket(final Socket socket, final String host, final int port, final boolean autoClose) throws IOException {
- final Socket underlyingSocket = delegate.createSocket(socket, host, port, autoClose);
- return overrideProtocol(underlyingSocket);
- }
-
- @Override
- public Socket createSocket(final String host, final int port) throws IOException, UnknownHostException {
- final Socket underlyingSocket = delegate.createSocket(host, port);
- return overrideProtocol(underlyingSocket);
- }
-
- @Override
- public Socket createSocket(final String host, final int port, final InetAddress localAddress, final int localPort) throws IOException, UnknownHostException {
- final Socket underlyingSocket = delegate.createSocket(host, port, localAddress, localPort);
- return overrideProtocol(underlyingSocket);
- }
-
- @Override
- public Socket createSocket(final InetAddress host, final int port) throws IOException {
- final Socket underlyingSocket = delegate.createSocket(host, port);
- return overrideProtocol(underlyingSocket);
- }
-
- @Override
- public Socket createSocket(final InetAddress host, final int port, final InetAddress localAddress, final int localPort) throws IOException {
- final Socket underlyingSocket = delegate.createSocket(host, port, localAddress, localPort);
- return overrideProtocol(underlyingSocket);
- }
-
- /**
- * Set the {@link javax.net.ssl.SSLSocket#getEnabledProtocols() enabled protocols} to {@link #enabledProtocols} if the <code>socket</code> is a
- * {@link SSLSocket}
- *
- * @param socket
- * The Socket
- */
- private Socket overrideProtocol(final Socket socket) {
- if (socket instanceof SSLSocket) {
- ((SSLSocket) socket).setEnabledProtocols(enabledProtocols);
- }
- return socket;
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/de1d3ee3/proxy/src/main/java/org/apache/accumulo/proxy/Proxy.java
----------------------------------------------------------------------
diff --git a/proxy/src/main/java/org/apache/accumulo/proxy/Proxy.java b/proxy/src/main/java/org/apache/accumulo/proxy/Proxy.java
index 0a7b301..4b048eb 100644
--- a/proxy/src/main/java/org/apache/accumulo/proxy/Proxy.java
+++ b/proxy/src/main/java/org/apache/accumulo/proxy/Proxy.java
@@ -27,7 +27,7 @@ import org.apache.accumulo.core.cli.Help;
import org.apache.accumulo.core.conf.AccumuloConfiguration;
import org.apache.accumulo.minicluster.MiniAccumuloCluster;
import org.apache.accumulo.proxy.thrift.AccumuloProxy;
-import org.apache.accumulo.server.thrift.RpcWrapper;
+import org.apache.accumulo.server.rpc.RpcWrapper;
import org.apache.log4j.Logger;
import org.apache.thrift.TProcessor;
import org.apache.thrift.protocol.TCompactProtocol;
[2/3] accumulo git commit: ACCUMULO-3394 Change thrift package to rpc
and make an rpc package in core too
Posted by el...@apache.org.
http://git-wip-us.apache.org/repos/asf/accumulo/blob/de1d3ee3/server/base/src/main/java/org/apache/accumulo/server/AccumuloServerContext.java
----------------------------------------------------------------------
diff --git a/server/base/src/main/java/org/apache/accumulo/server/AccumuloServerContext.java b/server/base/src/main/java/org/apache/accumulo/server/AccumuloServerContext.java
index 3a435c7..09ae4f4 100644
--- a/server/base/src/main/java/org/apache/accumulo/server/AccumuloServerContext.java
+++ b/server/base/src/main/java/org/apache/accumulo/server/AccumuloServerContext.java
@@ -20,8 +20,8 @@ import org.apache.accumulo.core.client.Instance;
import org.apache.accumulo.core.client.impl.ClientContext;
import org.apache.accumulo.core.client.mock.MockInstance;
import org.apache.accumulo.core.client.security.tokens.PasswordToken;
+import org.apache.accumulo.core.rpc.SslConnectionParams;
import org.apache.accumulo.core.security.Credentials;
-import org.apache.accumulo.core.util.SslConnectionParams;
import org.apache.accumulo.server.conf.ServerConfigurationFactory;
import org.apache.accumulo.server.security.SystemCredentials;
http://git-wip-us.apache.org/repos/asf/accumulo/blob/de1d3ee3/server/base/src/main/java/org/apache/accumulo/server/client/BulkImporter.java
----------------------------------------------------------------------
diff --git a/server/base/src/main/java/org/apache/accumulo/server/client/BulkImporter.java b/server/base/src/main/java/org/apache/accumulo/server/client/BulkImporter.java
index 593d9b7..f338db8 100644
--- a/server/base/src/main/java/org/apache/accumulo/server/client/BulkImporter.java
+++ b/server/base/src/main/java/org/apache/accumulo/server/client/BulkImporter.java
@@ -52,6 +52,7 @@ import org.apache.accumulo.core.data.thrift.TKeyExtent;
import org.apache.accumulo.core.file.FileOperations;
import org.apache.accumulo.core.file.FileSKVIterator;
import org.apache.accumulo.core.metadata.MetadataTable;
+import org.apache.accumulo.core.rpc.ThriftUtil;
import org.apache.accumulo.core.tabletserver.thrift.TabletClientService;
import org.apache.accumulo.core.trace.Tracer;
import org.apache.accumulo.core.trace.wrappers.TraceRunnable;
@@ -59,7 +60,6 @@ import org.apache.accumulo.core.util.CachedConfiguration;
import org.apache.accumulo.core.util.LoggingRunnable;
import org.apache.accumulo.core.util.NamingThreadFactory;
import org.apache.accumulo.core.util.StopWatch;
-import org.apache.accumulo.core.util.ThriftUtil;
import org.apache.accumulo.core.util.UtilWaitThread;
import org.apache.accumulo.server.fs.VolumeManager;
import org.apache.accumulo.server.fs.VolumeManagerImpl;
http://git-wip-us.apache.org/repos/asf/accumulo/blob/de1d3ee3/server/base/src/main/java/org/apache/accumulo/server/master/LiveTServerSet.java
----------------------------------------------------------------------
diff --git a/server/base/src/main/java/org/apache/accumulo/server/master/LiveTServerSet.java b/server/base/src/main/java/org/apache/accumulo/server/master/LiveTServerSet.java
index d392bde..5eea41c 100644
--- a/server/base/src/main/java/org/apache/accumulo/server/master/LiveTServerSet.java
+++ b/server/base/src/main/java/org/apache/accumulo/server/master/LiveTServerSet.java
@@ -31,12 +31,12 @@ import org.apache.accumulo.core.client.impl.ClientContext;
import org.apache.accumulo.core.client.impl.thrift.ThriftSecurityException;
import org.apache.accumulo.core.data.KeyExtent;
import org.apache.accumulo.core.master.thrift.TabletServerStatus;
+import org.apache.accumulo.core.rpc.ThriftUtil;
import org.apache.accumulo.core.tabletserver.thrift.NotServingTabletException;
import org.apache.accumulo.core.tabletserver.thrift.TabletClientService;
import org.apache.accumulo.core.trace.Tracer;
import org.apache.accumulo.core.util.AddressUtil;
import org.apache.accumulo.core.util.ServerServices;
-import org.apache.accumulo.core.util.ThriftUtil;
import org.apache.accumulo.core.zookeeper.ZooUtil;
import org.apache.accumulo.server.master.state.TServerInstance;
import org.apache.accumulo.server.util.Halt;
http://git-wip-us.apache.org/repos/asf/accumulo/blob/de1d3ee3/server/base/src/main/java/org/apache/accumulo/server/master/balancer/TabletBalancer.java
----------------------------------------------------------------------
diff --git a/server/base/src/main/java/org/apache/accumulo/server/master/balancer/TabletBalancer.java b/server/base/src/main/java/org/apache/accumulo/server/master/balancer/TabletBalancer.java
index cd25d49..9822d0f 100644
--- a/server/base/src/main/java/org/apache/accumulo/server/master/balancer/TabletBalancer.java
+++ b/server/base/src/main/java/org/apache/accumulo/server/master/balancer/TabletBalancer.java
@@ -26,11 +26,11 @@ import java.util.SortedMap;
import org.apache.accumulo.core.client.impl.thrift.ThriftSecurityException;
import org.apache.accumulo.core.data.KeyExtent;
import org.apache.accumulo.core.master.thrift.TabletServerStatus;
+import org.apache.accumulo.core.rpc.ThriftUtil;
import org.apache.accumulo.core.tabletserver.thrift.TabletClientService;
import org.apache.accumulo.core.tabletserver.thrift.TabletClientService.Client;
import org.apache.accumulo.core.tabletserver.thrift.TabletStats;
import org.apache.accumulo.core.trace.Tracer;
-import org.apache.accumulo.core.util.ThriftUtil;
import org.apache.accumulo.server.AccumuloServerContext;
import org.apache.accumulo.server.conf.ServerConfiguration;
import org.apache.accumulo.server.conf.ServerConfigurationFactory;
http://git-wip-us.apache.org/repos/asf/accumulo/blob/de1d3ee3/server/base/src/main/java/org/apache/accumulo/server/rpc/ClientInfoProcessorFactory.java
----------------------------------------------------------------------
diff --git a/server/base/src/main/java/org/apache/accumulo/server/rpc/ClientInfoProcessorFactory.java b/server/base/src/main/java/org/apache/accumulo/server/rpc/ClientInfoProcessorFactory.java
new file mode 100644
index 0000000..5f630c2
--- /dev/null
+++ b/server/base/src/main/java/org/apache/accumulo/server/rpc/ClientInfoProcessorFactory.java
@@ -0,0 +1,53 @@
+/*
+ * 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 org.apache.accumulo.core.rpc.TBufferedSocket;
+import org.apache.thrift.TProcessor;
+import org.apache.thrift.TProcessorFactory;
+import org.apache.thrift.transport.TSocket;
+import org.apache.thrift.transport.TTransport;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Sets the address of a client in a ThreadLocal to allow for more informative log messages.
+ */
+public class ClientInfoProcessorFactory extends TProcessorFactory {
+ private static final Logger log = LoggerFactory.getLogger(ClientInfoProcessorFactory.class);
+
+ private final ThreadLocal<String> clientAddress;
+
+ public ClientInfoProcessorFactory(ThreadLocal<String> clientAddress, TProcessor processor) {
+ super(processor);
+ this.clientAddress = clientAddress;
+ }
+
+ @Override
+ public TProcessor getProcessor(TTransport trans) {
+ if (trans instanceof TBufferedSocket) {
+ TBufferedSocket tsock = (TBufferedSocket) trans;
+ clientAddress.set(tsock.getClientString());
+ } else if (trans instanceof TSocket) {
+ TSocket tsock = (TSocket) trans;
+ clientAddress.set(tsock.getSocket().getInetAddress().getHostAddress() + ":" + tsock.getSocket().getPort());
+ } else {
+ log.warn("Unable to extract clientAddress from transport of type {}", trans.getClass());
+ }
+ return super.getProcessor(trans);
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/accumulo/blob/de1d3ee3/server/base/src/main/java/org/apache/accumulo/server/rpc/CustomNonBlockingServer.java
----------------------------------------------------------------------
diff --git a/server/base/src/main/java/org/apache/accumulo/server/rpc/CustomNonBlockingServer.java b/server/base/src/main/java/org/apache/accumulo/server/rpc/CustomNonBlockingServer.java
new file mode 100644
index 0000000..21f55b3
--- /dev/null
+++ b/server/base/src/main/java/org/apache/accumulo/server/rpc/CustomNonBlockingServer.java
@@ -0,0 +1,268 @@
+/*
+ * 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.io.IOException;
+import java.net.Socket;
+import java.nio.channels.SelectionKey;
+import java.util.Iterator;
+
+import org.apache.log4j.Logger;
+import org.apache.thrift.server.THsHaServer;
+import org.apache.thrift.server.TNonblockingServer;
+import org.apache.thrift.transport.TNonblockingServerTransport;
+import org.apache.thrift.transport.TNonblockingSocket;
+import org.apache.thrift.transport.TNonblockingTransport;
+import org.apache.thrift.transport.TTransportException;
+
+/**
+ * This class implements a custom non-blocking thrift server, incorporating the {@link THsHaServer} features, and overriding the underlying
+ * {@link TNonblockingServer} methods, especially {@link org.apache.thrift.server.TNonblockingServer.SelectAcceptThread}, in order to override the
+ * {@link org.apache.thrift.server.AbstractNonblockingServer.FrameBuffer} and {@link org.apache.thrift.server.AbstractNonblockingServer.AsyncFrameBuffer} with
+ * one that reveals the client address from its transport.
+ *
+ * <p>
+ * The justification for this is explained in https://issues.apache.org/jira/browse/ACCUMULO-1691, and is needed due to the repeated regressions:
+ * <ul>
+ * <li>https://issues.apache.org/jira/browse/THRIFT-958</li>
+ * <li>https://issues.apache.org/jira/browse/THRIFT-1464</li>
+ * <li>https://issues.apache.org/jira/browse/THRIFT-2173</li>
+ * </ul>
+ *
+ * <p>
+ * This class contains a copy of {@link org.apache.thrift.server.TNonblockingServer.SelectAcceptThread} from Thrift 0.9.1, with the slight modification of
+ * instantiating a custom FrameBuffer, rather than the {@link org.apache.thrift.server.AbstractNonblockingServer.FrameBuffer} and
+ * {@link org.apache.thrift.server.AbstractNonblockingServer.AsyncFrameBuffer}. Because of this, any change in the implementation upstream will require a review
+ * of this implementation here, to ensure any new bugfixes/features in the upstream Thrift class are also applied here, at least until
+ * https://issues.apache.org/jira/browse/THRIFT-2173 is implemented. In the meantime, the maven-enforcer-plugin ensures that Thrift remains at version 0.9.1,
+ * which has been reviewed and tested.
+ */
+public class CustomNonBlockingServer extends THsHaServer {
+
+ private static final Logger LOGGER = Logger.getLogger(CustomNonBlockingServer.class);
+ private SelectAcceptThread selectAcceptThread_;
+ private volatile boolean stopped_ = false;
+
+ public CustomNonBlockingServer(Args args) {
+ super(args);
+ }
+
+ @Override
+ protected Runnable getRunnable(final FrameBuffer frameBuffer) {
+ return new Runnable() {
+ @Override
+ public void run() {
+ if (frameBuffer instanceof CustomNonblockingFrameBuffer) {
+ TNonblockingTransport trans = ((CustomNonblockingFrameBuffer) frameBuffer).getTransport();
+ if (trans instanceof TNonblockingSocket) {
+ TNonblockingSocket tsock = (TNonblockingSocket) trans;
+ Socket sock = tsock.getSocketChannel().socket();
+ TServerUtils.clientAddress.set(sock.getInetAddress().getHostAddress() + ":" + sock.getPort());
+ }
+ }
+ frameBuffer.invoke();
+ }
+ };
+ }
+
+ @Override
+ protected boolean startThreads() {
+ // start the selector
+ try {
+ selectAcceptThread_ = new SelectAcceptThread((TNonblockingServerTransport) serverTransport_);
+ selectAcceptThread_.start();
+ return true;
+ } catch (IOException e) {
+ LOGGER.error("Failed to start selector thread!", e);
+ return false;
+ }
+ }
+
+ @Override
+ public void stop() {
+ stopped_ = true;
+ if (selectAcceptThread_ != null) {
+ selectAcceptThread_.wakeupSelector();
+ }
+ }
+
+ @Override
+ public boolean isStopped() {
+ return selectAcceptThread_.isStopped();
+ }
+
+ @Override
+ protected void joinSelector() {
+ // wait until the selector thread exits
+ try {
+ selectAcceptThread_.join();
+ } catch (InterruptedException e) {
+ // for now, just silently ignore. technically this means we'll have less of
+ // a graceful shutdown as a result.
+ }
+ }
+
+ private interface CustomNonblockingFrameBuffer {
+ TNonblockingTransport getTransport();
+ }
+
+ private class CustomAsyncFrameBuffer extends AsyncFrameBuffer implements CustomNonblockingFrameBuffer {
+ private TNonblockingTransport trans;
+
+ public CustomAsyncFrameBuffer(TNonblockingTransport trans, SelectionKey selectionKey, AbstractSelectThread selectThread) {
+ super(trans, selectionKey, selectThread);
+ this.trans = trans;
+ }
+
+ @Override
+ public TNonblockingTransport getTransport() {
+ return trans;
+ }
+ }
+
+ private class CustomFrameBuffer extends FrameBuffer implements CustomNonblockingFrameBuffer {
+ private TNonblockingTransport trans;
+
+ public CustomFrameBuffer(TNonblockingTransport trans, SelectionKey selectionKey, AbstractSelectThread selectThread) {
+ super(trans, selectionKey, selectThread);
+ this.trans = trans;
+ }
+
+ @Override
+ public TNonblockingTransport getTransport() {
+ return trans;
+ }
+ }
+
+ // @formatter:off
+ private class SelectAcceptThread extends AbstractSelectThread {
+
+ // The server transport on which new client transports will be accepted
+ private final TNonblockingServerTransport serverTransport;
+
+ /**
+ * Set up the thread that will handle the non-blocking accepts, reads, and
+ * writes.
+ */
+ public SelectAcceptThread(final TNonblockingServerTransport serverTransport)
+ throws IOException {
+ this.serverTransport = serverTransport;
+ serverTransport.registerSelector(selector);
+ }
+
+ public boolean isStopped() {
+ return stopped_;
+ }
+
+ /**
+ * The work loop. Handles both selecting (all IO operations) and managing
+ * the selection preferences of all existing connections.
+ */
+ @Override
+ public void run() {
+ try {
+ if (eventHandler_ != null) {
+ eventHandler_.preServe();
+ }
+
+ while (!stopped_) {
+ select();
+ processInterestChanges();
+ }
+ for (SelectionKey selectionKey : selector.keys()) {
+ cleanupSelectionKey(selectionKey);
+ }
+ } catch (Throwable t) {
+ LOGGER.error("run() exiting due to uncaught error", t);
+ } finally {
+ stopped_ = true;
+ }
+ }
+
+ /**
+ * Select and process IO events appropriately:
+ * If there are connections to be accepted, accept them.
+ * If there are existing connections with data waiting to be read, read it,
+ * buffering until a whole frame has been read.
+ * If there are any pending responses, buffer them until their target client
+ * is available, and then send the data.
+ */
+ private void select() {
+ try {
+ // wait for io events.
+ selector.select();
+
+ // process the io events we received
+ Iterator<SelectionKey> selectedKeys = selector.selectedKeys().iterator();
+ while (!stopped_ && selectedKeys.hasNext()) {
+ SelectionKey key = selectedKeys.next();
+ selectedKeys.remove();
+
+ // skip if not valid
+ if (!key.isValid()) {
+ cleanupSelectionKey(key);
+ continue;
+ }
+
+ // if the key is marked Accept, then it has to be the server
+ // transport.
+ if (key.isAcceptable()) {
+ handleAccept();
+ } else if (key.isReadable()) {
+ // deal with reads
+ handleRead(key);
+ } else if (key.isWritable()) {
+ // deal with writes
+ handleWrite(key);
+ } else {
+ LOGGER.warn("Unexpected state in select! " + key.interestOps());
+ }
+ }
+ } catch (IOException e) {
+ LOGGER.warn("Got an IOException while selecting!", e);
+ }
+ }
+
+ /**
+ * Accept a new connection.
+ */
+ @SuppressWarnings("unused")
+ private void handleAccept() throws IOException {
+ SelectionKey clientKey = null;
+ TNonblockingTransport client = null;
+ try {
+ // accept the connection
+ client = (TNonblockingTransport)serverTransport.accept();
+ clientKey = client.registerSelector(selector, SelectionKey.OP_READ);
+
+ // add this key to the map
+ FrameBuffer frameBuffer = processorFactory_.isAsyncProcessor() ?
+ new CustomAsyncFrameBuffer(client, clientKey,SelectAcceptThread.this) :
+ new CustomFrameBuffer(client, clientKey,SelectAcceptThread.this);
+
+ clientKey.attach(frameBuffer);
+ } catch (TTransportException tte) {
+ // something went wrong accepting.
+ LOGGER.warn("Exception trying to accept!", tte);
+ tte.printStackTrace();
+ if (clientKey != null) cleanupSelectionKey(clientKey);
+ if (client != null) client.close();
+ }
+ }
+ } // SelectAcceptThread
+ // @formatter:on
+}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/de1d3ee3/server/base/src/main/java/org/apache/accumulo/server/rpc/RpcWrapper.java
----------------------------------------------------------------------
diff --git 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
new file mode 100644
index 0000000..7b34986
--- /dev/null
+++ b/server/base/src/main/java/org/apache/accumulo/server/rpc/RpcWrapper.java
@@ -0,0 +1,62 @@
+/*
+ * 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 org.apache.accumulo.core.trace.wrappers.RpcServerInvocationHandler;
+import org.apache.accumulo.core.trace.wrappers.TraceWrap;
+import org.apache.thrift.TApplicationException;
+import org.apache.thrift.TException;
+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.
+ *
+ * @since 1.6.1
+ */
+public class RpcWrapper {
+
+ public static <T> T service(final T instance) {
+ InvocationHandler handler = new RpcServerInvocationHandler<T>(instance) {
+ @Override
+ public Object invoke(Object obj, Method method, Object[] args) throws Throwable {
+ try {
+ return super.invoke(obj, method, args);
+ } catch (RuntimeException e) {
+ String msg = e.getMessage();
+ LoggerFactory.getLogger(instance.getClass()).error(msg, e);
+ throw new TException(msg);
+ } catch (Error e) {
+ String msg = e.getMessage();
+ LoggerFactory.getLogger(instance.getClass()).error(msg, e);
+ throw new TException(msg);
+ }
+ }
+ };
+
+ @SuppressWarnings("unchecked")
+ T proxiedInstance = (T) Proxy.newProxyInstance(instance.getClass().getClassLoader(), instance.getClass().getInterfaces(), handler);
+ return proxiedInstance;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/de1d3ee3/server/base/src/main/java/org/apache/accumulo/server/rpc/ServerAddress.java
----------------------------------------------------------------------
diff --git a/server/base/src/main/java/org/apache/accumulo/server/rpc/ServerAddress.java b/server/base/src/main/java/org/apache/accumulo/server/rpc/ServerAddress.java
new file mode 100644
index 0000000..b655287
--- /dev/null
+++ b/server/base/src/main/java/org/apache/accumulo/server/rpc/ServerAddress.java
@@ -0,0 +1,42 @@
+/*
+ * 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 org.apache.thrift.server.TServer;
+
+import com.google.common.net.HostAndPort;
+
+/**
+ * Encapsulate a Thrift server and the address, host and port, to which it is bound.
+ */
+public class ServerAddress {
+ public final TServer server;
+ public final HostAndPort address;
+
+ public ServerAddress(TServer server, HostAndPort address) {
+ this.server = server;
+ this.address = address;
+ }
+
+ public TServer getServer() {
+ return server;
+ }
+
+ public HostAndPort getAddress() {
+ return address;
+ }
+}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/de1d3ee3/server/base/src/main/java/org/apache/accumulo/server/rpc/TBufferedServerSocket.java
----------------------------------------------------------------------
diff --git a/server/base/src/main/java/org/apache/accumulo/server/rpc/TBufferedServerSocket.java b/server/base/src/main/java/org/apache/accumulo/server/rpc/TBufferedServerSocket.java
new file mode 100644
index 0000000..2887f48
--- /dev/null
+++ b/server/base/src/main/java/org/apache/accumulo/server/rpc/TBufferedServerSocket.java
@@ -0,0 +1,71 @@
+/*
+ * 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.io.IOException;
+import java.net.ServerSocket;
+
+import org.apache.accumulo.core.rpc.TBufferedSocket;
+import org.apache.thrift.transport.TServerTransport;
+import org.apache.thrift.transport.TSocket;
+import org.apache.thrift.transport.TTransport;
+import org.apache.thrift.transport.TTransportException;
+
+// Thrift-959 removed the small buffer from TSocket; this adds it back for servers
+public class TBufferedServerSocket extends TServerTransport {
+
+ // expose acceptImpl
+ static class TServerSocket extends org.apache.thrift.transport.TServerSocket {
+ public TServerSocket(ServerSocket serverSocket) {
+ super(serverSocket);
+ }
+
+ public TSocket acceptImplPublic() throws TTransportException {
+ return acceptImpl();
+ }
+ }
+
+ final TServerSocket impl;
+ final int bufferSize;
+
+ public TBufferedServerSocket(ServerSocket serverSocket, int bufferSize) {
+ this.impl = new TServerSocket(serverSocket);
+ this.bufferSize = bufferSize;
+ }
+
+ @Override
+ public void listen() throws TTransportException {
+ impl.listen();
+ }
+
+ @Override
+ public void close() {
+ impl.close();
+ }
+
+ // Wrap accepted sockets using buffered IO
+ @Override
+ protected TTransport acceptImpl() throws TTransportException {
+ TSocket sock = impl.acceptImplPublic();
+ try {
+ return new TBufferedSocket(sock, this.bufferSize);
+ } catch (IOException e) {
+ throw new TTransportException(e);
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/de1d3ee3/server/base/src/main/java/org/apache/accumulo/server/rpc/TNonblockingServerSocket.java
----------------------------------------------------------------------
diff --git a/server/base/src/main/java/org/apache/accumulo/server/rpc/TNonblockingServerSocket.java b/server/base/src/main/java/org/apache/accumulo/server/rpc/TNonblockingServerSocket.java
new file mode 100644
index 0000000..3afe149
--- /dev/null
+++ b/server/base/src/main/java/org/apache/accumulo/server/rpc/TNonblockingServerSocket.java
@@ -0,0 +1,157 @@
+/*
+ * 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 org.apache.log4j.Logger;
+import org.apache.thrift.transport.TNonblockingServerTransport;
+import org.apache.thrift.transport.TNonblockingSocket;
+import org.apache.thrift.transport.TTransportException;
+
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.net.ServerSocket;
+import java.net.SocketException;
+import java.nio.channels.ClosedChannelException;
+import java.nio.channels.SelectionKey;
+import java.nio.channels.Selector;
+import java.nio.channels.ServerSocketChannel;
+import java.nio.channels.SocketChannel;
+
+/**
+ * Wrapper around ServerSocketChannel.
+ *
+ * This class is copied from org.apache.thrift.transport.TNonblockingServerSocket version 0.9.
+ * The only change (apart from the logging statements) is the addition of the {@link #getPort()} method to retrieve the port used by the ServerSocket.
+ */
+public class TNonblockingServerSocket extends TNonblockingServerTransport {
+ private static final Logger log = Logger.getLogger(TNonblockingServerTransport.class.getName());
+
+ /**
+ * This channel is where all the nonblocking magic happens.
+ */
+ private ServerSocketChannel serverSocketChannel = null;
+
+ /**
+ * Underlying ServerSocket object
+ */
+ private ServerSocket serverSocket_ = null;
+
+ /**
+ * Timeout for client sockets from accept
+ */
+ private int clientTimeout_ = 0;
+
+ /**
+ * Creates just a port listening server socket
+ */
+ public TNonblockingServerSocket(int port) throws TTransportException {
+ this(port, 0);
+ }
+
+ /**
+ * Creates just a port listening server socket
+ */
+ public TNonblockingServerSocket(int port, int clientTimeout) throws TTransportException {
+ this(new InetSocketAddress(port), clientTimeout);
+ }
+
+ public TNonblockingServerSocket(InetSocketAddress bindAddr) throws TTransportException {
+ this(bindAddr, 0);
+ }
+
+ public TNonblockingServerSocket(InetSocketAddress bindAddr, int clientTimeout) throws TTransportException {
+ clientTimeout_ = clientTimeout;
+ try {
+ serverSocketChannel = ServerSocketChannel.open();
+ serverSocketChannel.configureBlocking(false);
+
+ // Make server socket
+ serverSocket_ = serverSocketChannel.socket();
+ // Prevent 2MSL delay problem on server restarts
+ serverSocket_.setReuseAddress(true);
+ // Bind to listening port
+ serverSocket_.bind(bindAddr);
+ } catch (IOException ioe) {
+ serverSocket_ = null;
+ throw new TTransportException("Could not create ServerSocket on address " + bindAddr.toString() + ".");
+ }
+ }
+
+ public void listen() throws TTransportException {
+ // Make sure not to block on accept
+ if (serverSocket_ != null) {
+ try {
+ serverSocket_.setSoTimeout(0);
+ } catch (SocketException sx) {
+ log.error("SocketException caused by serverSocket in listen()", sx);
+ }
+ }
+ }
+
+ protected TNonblockingSocket acceptImpl() throws TTransportException {
+ if (serverSocket_ == null) {
+ throw new TTransportException(TTransportException.NOT_OPEN, "No underlying server socket.");
+ }
+ try {
+ SocketChannel socketChannel = serverSocketChannel.accept();
+ if (socketChannel == null) {
+ return null;
+ }
+
+ TNonblockingSocket tsocket = new TNonblockingSocket(socketChannel);
+ tsocket.setTimeout(clientTimeout_);
+ return tsocket;
+ } catch (IOException iox) {
+ throw new TTransportException(iox);
+ }
+ }
+
+ public void registerSelector(Selector selector) {
+ try {
+ // Register the server socket channel, indicating an interest in
+ // accepting new connections
+ serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
+ } catch (ClosedChannelException e) {
+ // this shouldn't happen, ideally...
+ // TODO: decide what to do with this.
+ }
+ }
+
+ public void close() {
+ if (serverSocket_ != null) {
+ try {
+ serverSocket_.close();
+ } catch (IOException iox) {
+ log.warn("WARNING: Could not close server socket: " + iox.getMessage());
+ }
+ serverSocket_ = null;
+ }
+ }
+
+ public void interrupt() {
+ // The thread-safeness of this is dubious, but Java documentation suggests
+ // that it is safe to do this from a different thread context
+ close();
+ }
+
+ public int getPort() {
+ return serverSocket_.getLocalPort();
+ }
+}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/de1d3ee3/server/base/src/main/java/org/apache/accumulo/server/rpc/TServerUtils.java
----------------------------------------------------------------------
diff --git a/server/base/src/main/java/org/apache/accumulo/server/rpc/TServerUtils.java b/server/base/src/main/java/org/apache/accumulo/server/rpc/TServerUtils.java
new file mode 100644
index 0000000..ebf2c84
--- /dev/null
+++ b/server/base/src/main/java/org/apache/accumulo/server/rpc/TServerUtils.java
@@ -0,0 +1,255 @@
+/*
+ * 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.Field;
+import java.net.BindException;
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.net.UnknownHostException;
+import java.util.Random;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.ThreadPoolExecutor;
+
+import org.apache.accumulo.core.conf.AccumuloConfiguration;
+import org.apache.accumulo.core.conf.Property;
+import org.apache.accumulo.core.rpc.SslConnectionParams;
+import org.apache.accumulo.core.rpc.ThriftUtil;
+import org.apache.accumulo.core.util.Daemon;
+import org.apache.accumulo.core.util.LoggingRunnable;
+import org.apache.accumulo.core.util.SimpleThreadPool;
+import org.apache.accumulo.core.util.UtilWaitThread;
+import org.apache.accumulo.server.AccumuloServerContext;
+import org.apache.accumulo.server.util.Halt;
+import org.apache.accumulo.server.util.time.SimpleTimer;
+import org.apache.log4j.Logger;
+import org.apache.thrift.TProcessor;
+import org.apache.thrift.TProcessorFactory;
+import org.apache.thrift.server.TServer;
+import org.apache.thrift.server.TThreadPoolServer;
+import org.apache.thrift.transport.TServerTransport;
+import org.apache.thrift.transport.TTransportException;
+
+import com.google.common.net.HostAndPort;
+
+public class TServerUtils {
+ private static final Logger log = Logger.getLogger(TServerUtils.class);
+
+ public static final ThreadLocal<String> clientAddress = new ThreadLocal<String>();
+
+ /**
+ * Start a server, at the given port, or higher, if that port is not available.
+ *
+ * @param portHintProperty
+ * the port to attempt to open, can be zero, meaning "any available port"
+ * @param processor
+ * the service to be started
+ * @param serverName
+ * the name of the class that is providing the service
+ * @param threadName
+ * name this service's thread for better debugging
+ * @return the server object created, and the port actually used
+ * @throws UnknownHostException
+ * when we don't know our own address
+ */
+ public static ServerAddress startServer(AccumuloServerContext service, String address, Property portHintProperty, TProcessor processor, String serverName,
+ String threadName, Property portSearchProperty, Property minThreadProperty, Property timeBetweenThreadChecksProperty, Property maxMessageSizeProperty)
+ throws UnknownHostException {
+ int portHint = service.getConfiguration().getPort(portHintProperty);
+ int minThreads = 2;
+ if (minThreadProperty != null)
+ minThreads = service.getConfiguration().getCount(minThreadProperty);
+ long timeBetweenThreadChecks = 1000;
+ if (timeBetweenThreadChecksProperty != null)
+ timeBetweenThreadChecks = service.getConfiguration().getTimeInMillis(timeBetweenThreadChecksProperty);
+ long maxMessageSize = 10 * 1000 * 1000;
+ if (maxMessageSizeProperty != null)
+ maxMessageSize = service.getConfiguration().getMemoryInBytes(maxMessageSizeProperty);
+ boolean portSearch = false;
+ if (portSearchProperty != null)
+ portSearch = service.getConfiguration().getBoolean(portSearchProperty);
+ // create the TimedProcessor outside the port search loop so we don't try to register the same metrics mbean more than once
+ TimedProcessor timedProcessor = new TimedProcessor(service.getConfiguration(), processor, serverName, threadName);
+ Random random = new Random();
+ for (int j = 0; j < 100; j++) {
+
+ // Are we going to slide around, looking for an open port?
+ int portsToSearch = 1;
+ if (portSearch)
+ portsToSearch = 1000;
+
+ for (int i = 0; i < portsToSearch; i++) {
+ int port = portHint + i;
+ if (portHint != 0 && i > 0)
+ port = 1024 + random.nextInt(65535 - 1024);
+ if (port > 65535)
+ port = 1024 + port % (65535 - 1024);
+ try {
+ HostAndPort addr = HostAndPort.fromParts(address, port);
+ return TServerUtils.startTServer(addr, timedProcessor, serverName, threadName, minThreads,
+ service.getConfiguration().getCount(Property.GENERAL_SIMPLETIMER_THREADPOOL_SIZE), timeBetweenThreadChecks, maxMessageSize,
+ service.getServerSslParams(), service.getClientTimeoutInMillis());
+ } catch (TTransportException ex) {
+ log.error("Unable to start TServer", ex);
+ if (ex.getCause() == null || ex.getCause().getClass() == BindException.class) {
+ // Note: with a TNonblockingServerSocket a "port taken" exception is a cause-less
+ // TTransportException, and with a TSocket created by TSSLTransportFactory, it
+ // comes through as caused by a BindException.
+ log.info("Unable to use port " + port + ", retrying. (Thread Name = " + threadName + ")");
+ UtilWaitThread.sleep(250);
+ } else {
+ // thrift is passing up a nested exception that isn't a BindException,
+ // so no reason to believe retrying on a different port would help.
+ log.error("Unable to start TServer", ex);
+ break;
+ }
+ }
+ }
+ }
+ throw new UnknownHostException("Unable to find a listen port");
+ }
+
+ /**
+ * Create a NonBlockingServer with a custom thread pool that can dynamically resize itself.
+ */
+ public static ServerAddress createNonBlockingServer(HostAndPort address, TProcessor processor, final String serverName, String threadName,
+ final int numThreads, final int numSTThreads, long timeBetweenThreadChecks, long maxMessageSize) throws TTransportException {
+ TNonblockingServerSocket transport = new TNonblockingServerSocket(new InetSocketAddress(address.getHostText(), address.getPort()));
+ CustomNonBlockingServer.Args options = new CustomNonBlockingServer.Args(transport);
+ options.protocolFactory(ThriftUtil.protocolFactory());
+ options.transportFactory(ThriftUtil.transportFactory(maxMessageSize));
+ options.maxReadBufferBytes = maxMessageSize;
+ options.stopTimeoutVal(5);
+ /*
+ * Create our own very special thread pool.
+ */
+ final ThreadPoolExecutor pool = new SimpleThreadPool(numThreads, "ClientPool");
+ // periodically adjust the number of threads we need by checking how busy our threads are
+ SimpleTimer.getInstance(numSTThreads).schedule(new Runnable() {
+ @Override
+ public void run() {
+ if (pool.getCorePoolSize() <= pool.getActiveCount()) {
+ int larger = pool.getCorePoolSize() + Math.min(pool.getQueue().size(), 2);
+ log.info("Increasing server thread pool size on " + serverName + " to " + larger);
+ pool.setMaximumPoolSize(larger);
+ pool.setCorePoolSize(larger);
+ } else {
+ if (pool.getCorePoolSize() > pool.getActiveCount() + 3) {
+ int smaller = Math.max(numThreads, pool.getCorePoolSize() - 1);
+ if (smaller != pool.getCorePoolSize()) {
+ // ACCUMULO-2997 there is a race condition here... the active count could be higher by the time
+ // we decrease the core pool size... so the active count could end up higher than
+ // the core pool size, in which case everything will be queued... the increase case
+ // should handle this and prevent deadlock
+ log.info("Decreasing server thread pool size on " + serverName + " to " + smaller);
+ pool.setCorePoolSize(smaller);
+ }
+ }
+ }
+ }
+ }, timeBetweenThreadChecks, timeBetweenThreadChecks);
+ options.executorService(pool);
+ options.processorFactory(new TProcessorFactory(processor));
+ if (address.getPort() == 0) {
+ address = HostAndPort.fromParts(address.getHostText(), transport.getPort());
+ }
+ return new ServerAddress(new CustomNonBlockingServer(options), address);
+ }
+
+ public static TServer createThreadPoolServer(TServerTransport transport, TProcessor processor) {
+ TThreadPoolServer.Args options = new TThreadPoolServer.Args(transport);
+ options.protocolFactory(ThriftUtil.protocolFactory());
+ options.transportFactory(ThriftUtil.transportFactory());
+ options.processorFactory(new ClientInfoProcessorFactory(clientAddress, processor));
+ return new TThreadPoolServer(options);
+ }
+
+ public static ServerAddress createSslThreadPoolServer(HostAndPort address, TProcessor processor, long socketTimeout, SslConnectionParams sslParams)
+ throws TTransportException {
+ org.apache.thrift.transport.TServerSocket transport;
+ try {
+ transport = ThriftUtil.getServerSocket(address.getPort(), (int) socketTimeout, InetAddress.getByName(address.getHostText()), sslParams);
+ } catch (UnknownHostException e) {
+ throw new TTransportException(e);
+ }
+ if (address.getPort() == 0) {
+ address = HostAndPort.fromParts(address.getHostText(), transport.getServerSocket().getLocalPort());
+ }
+ return new ServerAddress(createThreadPoolServer(transport, processor), address);
+ }
+
+ public static ServerAddress startTServer(AccumuloConfiguration conf, HostAndPort address, TProcessor processor, String serverName, String threadName, int numThreads, int numSTThreads,
+ long timeBetweenThreadChecks, long maxMessageSize, SslConnectionParams sslParams, long sslSocketTimeout) throws TTransportException {
+ return startTServer(address, new TimedProcessor(conf, processor, serverName, threadName), serverName, threadName, numThreads, numSTThreads,
+ timeBetweenThreadChecks, maxMessageSize, sslParams, sslSocketTimeout);
+ }
+
+ /**
+ * Start the appropriate Thrift server (SSL or non-blocking server) for the given parameters. Non-null SSL parameters will cause an SSL server to be started.
+ *
+ * @return A ServerAddress encapsulating the Thrift server created and the host/port which it is bound to.
+ */
+ public static ServerAddress startTServer(HostAndPort address, TimedProcessor processor, String serverName, String threadName, int numThreads,
+ int numSTThreads, long timeBetweenThreadChecks, long maxMessageSize, SslConnectionParams sslParams, long sslSocketTimeout) throws TTransportException {
+
+ ServerAddress serverAddress;
+ if (sslParams != null) {
+ serverAddress = createSslThreadPoolServer(address, processor, sslSocketTimeout, sslParams);
+ } else {
+ serverAddress = createNonBlockingServer(address, processor, serverName, threadName, numThreads, numSTThreads, timeBetweenThreadChecks, maxMessageSize);
+ }
+ final TServer finalServer = serverAddress.server;
+ Runnable serveTask = new Runnable() {
+ @Override
+ public void run() {
+ try {
+ finalServer.serve();
+ } catch (Error e) {
+ Halt.halt("Unexpected error in TThreadPoolServer " + e + ", halting.");
+ }
+ }
+ };
+ serveTask = new LoggingRunnable(TServerUtils.log, serveTask);
+ Thread thread = new Daemon(serveTask, threadName);
+ thread.start();
+ // check for the special "bind to everything address"
+ if (serverAddress.address.getHostText().equals("0.0.0.0")) {
+ // can't get the address from the bind, so we'll do our best to invent our hostname
+ try {
+ serverAddress = new ServerAddress(finalServer, HostAndPort.fromParts(InetAddress.getLocalHost().getHostName(), serverAddress.address.getPort()));
+ } catch (UnknownHostException e) {
+ throw new TTransportException(e);
+ }
+ }
+ return serverAddress;
+ }
+
+ // Existing connections will keep our thread running: reach in with reflection and insist that they shutdown.
+ public static void stopTServer(TServer s) {
+ if (s == null)
+ return;
+ s.stop();
+ try {
+ Field f = s.getClass().getDeclaredField("executorService_");
+ f.setAccessible(true);
+ ExecutorService es = (ExecutorService) f.get(s);
+ es.shutdownNow();
+ } catch (Exception e) {
+ TServerUtils.log.error("Unable to call shutdownNow", e);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/de1d3ee3/server/base/src/main/java/org/apache/accumulo/server/rpc/TimedProcessor.java
----------------------------------------------------------------------
diff --git a/server/base/src/main/java/org/apache/accumulo/server/rpc/TimedProcessor.java b/server/base/src/main/java/org/apache/accumulo/server/rpc/TimedProcessor.java
new file mode 100644
index 0000000..a842572
--- /dev/null
+++ b/server/base/src/main/java/org/apache/accumulo/server/rpc/TimedProcessor.java
@@ -0,0 +1,69 @@
+/*
+ * 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 org.apache.accumulo.core.conf.AccumuloConfiguration;
+import org.apache.accumulo.server.metrics.Metrics;
+import org.apache.accumulo.server.metrics.MetricsFactory;
+import org.apache.accumulo.server.metrics.ThriftMetrics;
+import org.apache.thrift.TException;
+import org.apache.thrift.TProcessor;
+import org.apache.thrift.protocol.TProtocol;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * A {@link TProcessor} which tracks the duration of an RPC and adds it to the metrics subsystem.
+ */
+public class TimedProcessor implements TProcessor {
+ private static final Logger log = LoggerFactory.getLogger(TimedProcessor.class);
+
+ private final TProcessor other;
+ private final Metrics metrics;
+ private long idleStart = 0;
+
+ public TimedProcessor(AccumuloConfiguration conf, TProcessor next, String serverName, String threadName) {
+ this.other = next;
+ // Register the metrics MBean
+ MetricsFactory factory = new MetricsFactory(conf);
+ metrics = factory.createThriftMetrics(serverName, threadName);
+ try {
+ metrics.register();
+ } catch (Exception e) {
+ log.error("Exception registering MBean with MBean Server", e);
+ }
+ idleStart = System.currentTimeMillis();
+ }
+
+ @Override
+ public boolean process(TProtocol in, TProtocol out) throws TException {
+ long now = 0;
+ final boolean metricsEnabled = metrics.isEnabled();
+ if (metricsEnabled) {
+ now = System.currentTimeMillis();
+ metrics.add(ThriftMetrics.idle, (now - idleStart));
+ }
+ try {
+ return other.process(in, out);
+ } finally {
+ if (metricsEnabled) {
+ idleStart = System.currentTimeMillis();
+ metrics.add(ThriftMetrics.execute, idleStart - now);
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/de1d3ee3/server/base/src/main/java/org/apache/accumulo/server/security/AuditedSecurityOperation.java
----------------------------------------------------------------------
diff --git a/server/base/src/main/java/org/apache/accumulo/server/security/AuditedSecurityOperation.java b/server/base/src/main/java/org/apache/accumulo/server/security/AuditedSecurityOperation.java
index e09f7fd..a2afeac 100644
--- a/server/base/src/main/java/org/apache/accumulo/server/security/AuditedSecurityOperation.java
+++ b/server/base/src/main/java/org/apache/accumulo/server/security/AuditedSecurityOperation.java
@@ -44,10 +44,10 @@ import org.apache.accumulo.core.security.TablePermission;
import org.apache.accumulo.core.security.thrift.TCredentials;
import org.apache.accumulo.core.util.ByteBufferUtil;
import org.apache.accumulo.server.AccumuloServerContext;
+import org.apache.accumulo.server.rpc.TServerUtils;
import org.apache.accumulo.server.security.handler.Authenticator;
import org.apache.accumulo.server.security.handler.Authorizor;
import org.apache.accumulo.server.security.handler.PermissionHandler;
-import org.apache.accumulo.server.thrift.TServerUtils;
import org.apache.hadoop.io.Text;
import org.apache.log4j.Level;
import org.apache.log4j.Logger;
http://git-wip-us.apache.org/repos/asf/accumulo/blob/de1d3ee3/server/base/src/main/java/org/apache/accumulo/server/thrift/ClientInfoProcessorFactory.java
----------------------------------------------------------------------
diff --git a/server/base/src/main/java/org/apache/accumulo/server/thrift/ClientInfoProcessorFactory.java b/server/base/src/main/java/org/apache/accumulo/server/thrift/ClientInfoProcessorFactory.java
deleted file mode 100644
index 208fde5..0000000
--- a/server/base/src/main/java/org/apache/accumulo/server/thrift/ClientInfoProcessorFactory.java
+++ /dev/null
@@ -1,53 +0,0 @@
-/*
- * 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.thrift;
-
-import org.apache.accumulo.core.util.TBufferedSocket;
-import org.apache.thrift.TProcessor;
-import org.apache.thrift.TProcessorFactory;
-import org.apache.thrift.transport.TSocket;
-import org.apache.thrift.transport.TTransport;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * Sets the address of a client in a ThreadLocal to allow for more informative log messages.
- */
-public class ClientInfoProcessorFactory extends TProcessorFactory {
- private static final Logger log = LoggerFactory.getLogger(ClientInfoProcessorFactory.class);
-
- private final ThreadLocal<String> clientAddress;
-
- public ClientInfoProcessorFactory(ThreadLocal<String> clientAddress, TProcessor processor) {
- super(processor);
- this.clientAddress = clientAddress;
- }
-
- @Override
- public TProcessor getProcessor(TTransport trans) {
- if (trans instanceof TBufferedSocket) {
- TBufferedSocket tsock = (TBufferedSocket) trans;
- clientAddress.set(tsock.getClientString());
- } else if (trans instanceof TSocket) {
- TSocket tsock = (TSocket) trans;
- clientAddress.set(tsock.getSocket().getInetAddress().getHostAddress() + ":" + tsock.getSocket().getPort());
- } else {
- log.warn("Unable to extract clientAddress from transport of type {}", trans.getClass());
- }
- return super.getProcessor(trans);
- }
-}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/accumulo/blob/de1d3ee3/server/base/src/main/java/org/apache/accumulo/server/thrift/CustomNonBlockingServer.java
----------------------------------------------------------------------
diff --git a/server/base/src/main/java/org/apache/accumulo/server/thrift/CustomNonBlockingServer.java b/server/base/src/main/java/org/apache/accumulo/server/thrift/CustomNonBlockingServer.java
deleted file mode 100644
index a96f7b5..0000000
--- a/server/base/src/main/java/org/apache/accumulo/server/thrift/CustomNonBlockingServer.java
+++ /dev/null
@@ -1,268 +0,0 @@
-/*
- * 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.thrift;
-
-import java.io.IOException;
-import java.net.Socket;
-import java.nio.channels.SelectionKey;
-import java.util.Iterator;
-
-import org.apache.log4j.Logger;
-import org.apache.thrift.server.THsHaServer;
-import org.apache.thrift.server.TNonblockingServer;
-import org.apache.thrift.transport.TNonblockingServerTransport;
-import org.apache.thrift.transport.TNonblockingSocket;
-import org.apache.thrift.transport.TNonblockingTransport;
-import org.apache.thrift.transport.TTransportException;
-
-/**
- * This class implements a custom non-blocking thrift server, incorporating the {@link THsHaServer} features, and overriding the underlying
- * {@link TNonblockingServer} methods, especially {@link org.apache.thrift.server.TNonblockingServer.SelectAcceptThread}, in order to override the
- * {@link org.apache.thrift.server.AbstractNonblockingServer.FrameBuffer} and {@link org.apache.thrift.server.AbstractNonblockingServer.AsyncFrameBuffer} with
- * one that reveals the client address from its transport.
- *
- * <p>
- * The justification for this is explained in https://issues.apache.org/jira/browse/ACCUMULO-1691, and is needed due to the repeated regressions:
- * <ul>
- * <li>https://issues.apache.org/jira/browse/THRIFT-958</li>
- * <li>https://issues.apache.org/jira/browse/THRIFT-1464</li>
- * <li>https://issues.apache.org/jira/browse/THRIFT-2173</li>
- * </ul>
- *
- * <p>
- * This class contains a copy of {@link org.apache.thrift.server.TNonblockingServer.SelectAcceptThread} from Thrift 0.9.1, with the slight modification of
- * instantiating a custom FrameBuffer, rather than the {@link org.apache.thrift.server.AbstractNonblockingServer.FrameBuffer} and
- * {@link org.apache.thrift.server.AbstractNonblockingServer.AsyncFrameBuffer}. Because of this, any change in the implementation upstream will require a review
- * of this implementation here, to ensure any new bugfixes/features in the upstream Thrift class are also applied here, at least until
- * https://issues.apache.org/jira/browse/THRIFT-2173 is implemented. In the meantime, the maven-enforcer-plugin ensures that Thrift remains at version 0.9.1,
- * which has been reviewed and tested.
- */
-public class CustomNonBlockingServer extends THsHaServer {
-
- private static final Logger LOGGER = Logger.getLogger(CustomNonBlockingServer.class);
- private SelectAcceptThread selectAcceptThread_;
- private volatile boolean stopped_ = false;
-
- public CustomNonBlockingServer(Args args) {
- super(args);
- }
-
- @Override
- protected Runnable getRunnable(final FrameBuffer frameBuffer) {
- return new Runnable() {
- @Override
- public void run() {
- if (frameBuffer instanceof CustomNonblockingFrameBuffer) {
- TNonblockingTransport trans = ((CustomNonblockingFrameBuffer) frameBuffer).getTransport();
- if (trans instanceof TNonblockingSocket) {
- TNonblockingSocket tsock = (TNonblockingSocket) trans;
- Socket sock = tsock.getSocketChannel().socket();
- TServerUtils.clientAddress.set(sock.getInetAddress().getHostAddress() + ":" + sock.getPort());
- }
- }
- frameBuffer.invoke();
- }
- };
- }
-
- @Override
- protected boolean startThreads() {
- // start the selector
- try {
- selectAcceptThread_ = new SelectAcceptThread((TNonblockingServerTransport) serverTransport_);
- selectAcceptThread_.start();
- return true;
- } catch (IOException e) {
- LOGGER.error("Failed to start selector thread!", e);
- return false;
- }
- }
-
- @Override
- public void stop() {
- stopped_ = true;
- if (selectAcceptThread_ != null) {
- selectAcceptThread_.wakeupSelector();
- }
- }
-
- @Override
- public boolean isStopped() {
- return selectAcceptThread_.isStopped();
- }
-
- @Override
- protected void joinSelector() {
- // wait until the selector thread exits
- try {
- selectAcceptThread_.join();
- } catch (InterruptedException e) {
- // for now, just silently ignore. technically this means we'll have less of
- // a graceful shutdown as a result.
- }
- }
-
- private interface CustomNonblockingFrameBuffer {
- TNonblockingTransport getTransport();
- }
-
- private class CustomAsyncFrameBuffer extends AsyncFrameBuffer implements CustomNonblockingFrameBuffer {
- private TNonblockingTransport trans;
-
- public CustomAsyncFrameBuffer(TNonblockingTransport trans, SelectionKey selectionKey, AbstractSelectThread selectThread) {
- super(trans, selectionKey, selectThread);
- this.trans = trans;
- }
-
- @Override
- public TNonblockingTransport getTransport() {
- return trans;
- }
- }
-
- private class CustomFrameBuffer extends FrameBuffer implements CustomNonblockingFrameBuffer {
- private TNonblockingTransport trans;
-
- public CustomFrameBuffer(TNonblockingTransport trans, SelectionKey selectionKey, AbstractSelectThread selectThread) {
- super(trans, selectionKey, selectThread);
- this.trans = trans;
- }
-
- @Override
- public TNonblockingTransport getTransport() {
- return trans;
- }
- }
-
- // @formatter:off
- private class SelectAcceptThread extends AbstractSelectThread {
-
- // The server transport on which new client transports will be accepted
- private final TNonblockingServerTransport serverTransport;
-
- /**
- * Set up the thread that will handle the non-blocking accepts, reads, and
- * writes.
- */
- public SelectAcceptThread(final TNonblockingServerTransport serverTransport)
- throws IOException {
- this.serverTransport = serverTransport;
- serverTransport.registerSelector(selector);
- }
-
- public boolean isStopped() {
- return stopped_;
- }
-
- /**
- * The work loop. Handles both selecting (all IO operations) and managing
- * the selection preferences of all existing connections.
- */
- @Override
- public void run() {
- try {
- if (eventHandler_ != null) {
- eventHandler_.preServe();
- }
-
- while (!stopped_) {
- select();
- processInterestChanges();
- }
- for (SelectionKey selectionKey : selector.keys()) {
- cleanupSelectionKey(selectionKey);
- }
- } catch (Throwable t) {
- LOGGER.error("run() exiting due to uncaught error", t);
- } finally {
- stopped_ = true;
- }
- }
-
- /**
- * Select and process IO events appropriately:
- * If there are connections to be accepted, accept them.
- * If there are existing connections with data waiting to be read, read it,
- * buffering until a whole frame has been read.
- * If there are any pending responses, buffer them until their target client
- * is available, and then send the data.
- */
- private void select() {
- try {
- // wait for io events.
- selector.select();
-
- // process the io events we received
- Iterator<SelectionKey> selectedKeys = selector.selectedKeys().iterator();
- while (!stopped_ && selectedKeys.hasNext()) {
- SelectionKey key = selectedKeys.next();
- selectedKeys.remove();
-
- // skip if not valid
- if (!key.isValid()) {
- cleanupSelectionKey(key);
- continue;
- }
-
- // if the key is marked Accept, then it has to be the server
- // transport.
- if (key.isAcceptable()) {
- handleAccept();
- } else if (key.isReadable()) {
- // deal with reads
- handleRead(key);
- } else if (key.isWritable()) {
- // deal with writes
- handleWrite(key);
- } else {
- LOGGER.warn("Unexpected state in select! " + key.interestOps());
- }
- }
- } catch (IOException e) {
- LOGGER.warn("Got an IOException while selecting!", e);
- }
- }
-
- /**
- * Accept a new connection.
- */
- @SuppressWarnings("unused")
- private void handleAccept() throws IOException {
- SelectionKey clientKey = null;
- TNonblockingTransport client = null;
- try {
- // accept the connection
- client = (TNonblockingTransport)serverTransport.accept();
- clientKey = client.registerSelector(selector, SelectionKey.OP_READ);
-
- // add this key to the map
- FrameBuffer frameBuffer = processorFactory_.isAsyncProcessor() ?
- new CustomAsyncFrameBuffer(client, clientKey,SelectAcceptThread.this) :
- new CustomFrameBuffer(client, clientKey,SelectAcceptThread.this);
-
- clientKey.attach(frameBuffer);
- } catch (TTransportException tte) {
- // something went wrong accepting.
- LOGGER.warn("Exception trying to accept!", tte);
- tte.printStackTrace();
- if (clientKey != null) cleanupSelectionKey(clientKey);
- if (client != null) client.close();
- }
- }
- } // SelectAcceptThread
- // @formatter:on
-}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/de1d3ee3/server/base/src/main/java/org/apache/accumulo/server/thrift/RpcWrapper.java
----------------------------------------------------------------------
diff --git a/server/base/src/main/java/org/apache/accumulo/server/thrift/RpcWrapper.java b/server/base/src/main/java/org/apache/accumulo/server/thrift/RpcWrapper.java
deleted file mode 100644
index 53ed709..0000000
--- a/server/base/src/main/java/org/apache/accumulo/server/thrift/RpcWrapper.java
+++ /dev/null
@@ -1,62 +0,0 @@
-/*
- * 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.thrift;
-
-import java.lang.reflect.InvocationHandler;
-import java.lang.reflect.Method;
-import java.lang.reflect.Proxy;
-
-import org.apache.accumulo.core.trace.wrappers.RpcServerInvocationHandler;
-import org.apache.accumulo.core.trace.wrappers.TraceWrap;
-import org.apache.thrift.TApplicationException;
-import org.apache.thrift.TException;
-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.
- *
- * @since 1.6.1
- */
-public class RpcWrapper {
-
- public static <T> T service(final T instance) {
- InvocationHandler handler = new RpcServerInvocationHandler<T>(instance) {
- @Override
- public Object invoke(Object obj, Method method, Object[] args) throws Throwable {
- try {
- return super.invoke(obj, method, args);
- } catch (RuntimeException e) {
- String msg = e.getMessage();
- LoggerFactory.getLogger(instance.getClass()).error(msg, e);
- throw new TException(msg);
- } catch (Error e) {
- String msg = e.getMessage();
- LoggerFactory.getLogger(instance.getClass()).error(msg, e);
- throw new TException(msg);
- }
- }
- };
-
- @SuppressWarnings("unchecked")
- T proxiedInstance = (T) Proxy.newProxyInstance(instance.getClass().getClassLoader(), instance.getClass().getInterfaces(), handler);
- return proxiedInstance;
- }
-
-}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/de1d3ee3/server/base/src/main/java/org/apache/accumulo/server/thrift/ServerAddress.java
----------------------------------------------------------------------
diff --git a/server/base/src/main/java/org/apache/accumulo/server/thrift/ServerAddress.java b/server/base/src/main/java/org/apache/accumulo/server/thrift/ServerAddress.java
deleted file mode 100644
index f52951e..0000000
--- a/server/base/src/main/java/org/apache/accumulo/server/thrift/ServerAddress.java
+++ /dev/null
@@ -1,42 +0,0 @@
-/*
- * 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.thrift;
-
-import org.apache.thrift.server.TServer;
-
-import com.google.common.net.HostAndPort;
-
-/**
- * Encapsulate a Thrift server and the address, host and port, to which it is bound.
- */
-public class ServerAddress {
- public final TServer server;
- public final HostAndPort address;
-
- public ServerAddress(TServer server, HostAndPort address) {
- this.server = server;
- this.address = address;
- }
-
- public TServer getServer() {
- return server;
- }
-
- public HostAndPort getAddress() {
- return address;
- }
-}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/de1d3ee3/server/base/src/main/java/org/apache/accumulo/server/thrift/TBufferedServerSocket.java
----------------------------------------------------------------------
diff --git a/server/base/src/main/java/org/apache/accumulo/server/thrift/TBufferedServerSocket.java b/server/base/src/main/java/org/apache/accumulo/server/thrift/TBufferedServerSocket.java
deleted file mode 100644
index 5534313..0000000
--- a/server/base/src/main/java/org/apache/accumulo/server/thrift/TBufferedServerSocket.java
+++ /dev/null
@@ -1,71 +0,0 @@
-/*
- * 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.thrift;
-
-import java.io.IOException;
-import java.net.ServerSocket;
-
-import org.apache.accumulo.core.util.TBufferedSocket;
-import org.apache.thrift.transport.TServerTransport;
-import org.apache.thrift.transport.TSocket;
-import org.apache.thrift.transport.TTransport;
-import org.apache.thrift.transport.TTransportException;
-
-// Thrift-959 removed the small buffer from TSocket; this adds it back for servers
-public class TBufferedServerSocket extends TServerTransport {
-
- // expose acceptImpl
- static class TServerSocket extends org.apache.thrift.transport.TServerSocket {
- public TServerSocket(ServerSocket serverSocket) {
- super(serverSocket);
- }
-
- public TSocket acceptImplPublic() throws TTransportException {
- return acceptImpl();
- }
- }
-
- final TServerSocket impl;
- final int bufferSize;
-
- public TBufferedServerSocket(ServerSocket serverSocket, int bufferSize) {
- this.impl = new TServerSocket(serverSocket);
- this.bufferSize = bufferSize;
- }
-
- @Override
- public void listen() throws TTransportException {
- impl.listen();
- }
-
- @Override
- public void close() {
- impl.close();
- }
-
- // Wrap accepted sockets using buffered IO
- @Override
- protected TTransport acceptImpl() throws TTransportException {
- TSocket sock = impl.acceptImplPublic();
- try {
- return new TBufferedSocket(sock, this.bufferSize);
- } catch (IOException e) {
- throw new TTransportException(e);
- }
- }
-
-}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/de1d3ee3/server/base/src/main/java/org/apache/accumulo/server/thrift/TNonblockingServerSocket.java
----------------------------------------------------------------------
diff --git a/server/base/src/main/java/org/apache/accumulo/server/thrift/TNonblockingServerSocket.java b/server/base/src/main/java/org/apache/accumulo/server/thrift/TNonblockingServerSocket.java
deleted file mode 100644
index 77c5ca6..0000000
--- a/server/base/src/main/java/org/apache/accumulo/server/thrift/TNonblockingServerSocket.java
+++ /dev/null
@@ -1,157 +0,0 @@
-/*
- * 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.thrift;
-
-import org.apache.log4j.Logger;
-import org.apache.thrift.transport.TNonblockingServerTransport;
-import org.apache.thrift.transport.TNonblockingSocket;
-import org.apache.thrift.transport.TTransportException;
-
-import java.io.IOException;
-import java.net.InetSocketAddress;
-import java.net.ServerSocket;
-import java.net.SocketException;
-import java.nio.channels.ClosedChannelException;
-import java.nio.channels.SelectionKey;
-import java.nio.channels.Selector;
-import java.nio.channels.ServerSocketChannel;
-import java.nio.channels.SocketChannel;
-
-/**
- * Wrapper around ServerSocketChannel.
- *
- * This class is copied from org.apache.thrift.transport.TNonblockingServerSocket version 0.9.
- * The only change (apart from the logging statements) is the addition of the {@link #getPort()} method to retrieve the port used by the ServerSocket.
- */
-public class TNonblockingServerSocket extends TNonblockingServerTransport {
- private static final Logger log = Logger.getLogger(TNonblockingServerTransport.class.getName());
-
- /**
- * This channel is where all the nonblocking magic happens.
- */
- private ServerSocketChannel serverSocketChannel = null;
-
- /**
- * Underlying ServerSocket object
- */
- private ServerSocket serverSocket_ = null;
-
- /**
- * Timeout for client sockets from accept
- */
- private int clientTimeout_ = 0;
-
- /**
- * Creates just a port listening server socket
- */
- public TNonblockingServerSocket(int port) throws TTransportException {
- this(port, 0);
- }
-
- /**
- * Creates just a port listening server socket
- */
- public TNonblockingServerSocket(int port, int clientTimeout) throws TTransportException {
- this(new InetSocketAddress(port), clientTimeout);
- }
-
- public TNonblockingServerSocket(InetSocketAddress bindAddr) throws TTransportException {
- this(bindAddr, 0);
- }
-
- public TNonblockingServerSocket(InetSocketAddress bindAddr, int clientTimeout) throws TTransportException {
- clientTimeout_ = clientTimeout;
- try {
- serverSocketChannel = ServerSocketChannel.open();
- serverSocketChannel.configureBlocking(false);
-
- // Make server socket
- serverSocket_ = serverSocketChannel.socket();
- // Prevent 2MSL delay problem on server restarts
- serverSocket_.setReuseAddress(true);
- // Bind to listening port
- serverSocket_.bind(bindAddr);
- } catch (IOException ioe) {
- serverSocket_ = null;
- throw new TTransportException("Could not create ServerSocket on address " + bindAddr.toString() + ".");
- }
- }
-
- public void listen() throws TTransportException {
- // Make sure not to block on accept
- if (serverSocket_ != null) {
- try {
- serverSocket_.setSoTimeout(0);
- } catch (SocketException sx) {
- log.error("SocketException caused by serverSocket in listen()", sx);
- }
- }
- }
-
- protected TNonblockingSocket acceptImpl() throws TTransportException {
- if (serverSocket_ == null) {
- throw new TTransportException(TTransportException.NOT_OPEN, "No underlying server socket.");
- }
- try {
- SocketChannel socketChannel = serverSocketChannel.accept();
- if (socketChannel == null) {
- return null;
- }
-
- TNonblockingSocket tsocket = new TNonblockingSocket(socketChannel);
- tsocket.setTimeout(clientTimeout_);
- return tsocket;
- } catch (IOException iox) {
- throw new TTransportException(iox);
- }
- }
-
- public void registerSelector(Selector selector) {
- try {
- // Register the server socket channel, indicating an interest in
- // accepting new connections
- serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
- } catch (ClosedChannelException e) {
- // this shouldn't happen, ideally...
- // TODO: decide what to do with this.
- }
- }
-
- public void close() {
- if (serverSocket_ != null) {
- try {
- serverSocket_.close();
- } catch (IOException iox) {
- log.warn("WARNING: Could not close server socket: " + iox.getMessage());
- }
- serverSocket_ = null;
- }
- }
-
- public void interrupt() {
- // The thread-safeness of this is dubious, but Java documentation suggests
- // that it is safe to do this from a different thread context
- close();
- }
-
- public int getPort() {
- return serverSocket_.getLocalPort();
- }
-}