You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@accumulo.apache.org by ec...@apache.org on 2013/11/19 20:58:15 UTC
[2/3] ACCUMULO-1009
http://git-wip-us.apache.org/repos/asf/accumulo/blob/7038755b/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
index e8dd6a2..fab02b2 100644
--- a/core/src/main/java/org/apache/accumulo/core/util/ThriftUtil.java
+++ b/core/src/main/java/org/apache/accumulo/core/util/ThriftUtil.java
@@ -17,6 +17,7 @@
package org.apache.accumulo.core.util;
import java.io.IOException;
+import java.net.InetAddress;
import java.util.HashMap;
import java.util.Map;
@@ -40,6 +41,9 @@ 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;
@@ -48,15 +52,15 @@ import com.google.common.net.HostAndPort;
public class ThriftUtil {
private static final Logger log = Logger.getLogger(ThriftUtil.class);
-
+
public static class TraceProtocol extends TCompactProtocol {
-
+
@Override
public void writeMessageBegin(TMessage message) throws TException {
Trace.start("client:" + message.name);
super.writeMessageBegin(message);
}
-
+
@Override
public void writeMessageEnd() throws TException {
super.writeMessageEnd();
@@ -64,63 +68,65 @@ public class ThriftUtil {
if (currentTrace != null)
currentTrace.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, AccumuloConfiguration conf)
throws TTransportException {
return createClient(factory, ThriftTransportPool.getInstance().getTransportWithDefaultTimeout(address, conf));
}
-
- static public <T extends TServiceClient> T getClientNoTimeout(TServiceClientFactory<T> factory, String address) throws TTransportException {
- return createClient(factory, ThriftTransportPool.getInstance().getTransport(address, 0));
- }
-
- static public <T extends TServiceClient> T getClient(TServiceClientFactory<T> factory, String address, Property property, AccumuloConfiguration configuration)
+
+ static public <T extends TServiceClient> T getClientNoTimeout(TServiceClientFactory<T> factory, String address, AccumuloConfiguration configuration)
throws TTransportException {
- long timeout = configuration.getTimeInMillis(property);
- TTransport transport = ThriftTransportPool.getInstance().getTransport(address, timeout);
+ return getClient(factory, address, 0, configuration);
+ }
+
+ static public <T extends TServiceClient> T getClient(TServiceClientFactory<T> factory, String address, Property timeoutProperty,
+ AccumuloConfiguration configuration) throws TTransportException {
+ long timeout = configuration.getTimeInMillis(timeoutProperty);
+ TTransport transport = ThriftTransportPool.getInstance().getTransport(address, timeout, SslConnectionParams.forClient(configuration));
return createClient(factory, transport);
}
-
- static public <T extends TServiceClient> T getClient(TServiceClientFactory<T> factory, String address, long timeout) throws TTransportException {
- TTransport transport = ThriftTransportPool.getInstance().getTransport(address, timeout);
+
+ static public <T extends TServiceClient> T getClient(TServiceClientFactory<T> factory, String address, long timeout, AccumuloConfiguration configuration)
+ throws TTransportException {
+ TTransport transport = ThriftTransportPool.getInstance().getTransport(address, timeout, SslConnectionParams.forClient(configuration));
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, AccumuloConfiguration conf) throws TTransportException {
return getClient(new TabletClientService.Client.Factory(), address, Property.GENERAL_RPC_TIMEOUT, conf);
}
-
- static public TabletClientService.Client getTServerClient(String address, long timeout) throws TTransportException {
- return getClient(new TabletClientService.Client.Factory(), address, timeout);
+
+ static public TabletClientService.Client getTServerClient(String address, AccumuloConfiguration conf, long timeout) throws TTransportException {
+ return getClient(new TabletClientService.Client.Factory(), address, timeout, conf);
}
-
+
public static void execute(String address, AccumuloConfiguration conf, ClientExec<TabletClientService.Client> exec) throws AccumuloException,
AccumuloSecurityException {
while (true) {
@@ -141,7 +147,7 @@ public class ThriftUtil {
}
}
}
-
+
public static <T> T execute(String address, AccumuloConfiguration conf, ClientExecReturn<T,TabletClientService.Client> exec) throws AccumuloException,
AccumuloSecurityException {
while (true) {
@@ -161,34 +167,20 @@ public class ThriftUtil {
}
}
}
-
+
/**
* create a transport that is not pooled
*/
public static TTransport createTransport(HostAndPort address, AccumuloConfiguration conf) throws TException {
- TTransport transport = null;
-
- try {
- transport = TTimeoutTransport.create(address, conf.getTimeInMillis(Property.GENERAL_RPC_TIMEOUT));
- transport = ThriftUtil.transportFactory().getTransport(transport);
- transport.open();
- TTransport tmp = transport;
- transport = null;
- return tmp;
- } catch (IOException ex) {
- throw new TTransportException(ex);
- } finally {
- if (transport != null)
- transport.close();
- }
+ return createClientTransport(address, (int) conf.getTimeInMillis(Property.GENERAL_RPC_TIMEOUT), SslConnectionParams.forClient(conf));
}
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) {
@@ -197,18 +189,59 @@ public class ThriftUtil {
}
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 {
+ if (params.useJsse()) {
+ return TSSLTransportFactory.getServerSocket(port, timeout, params.isClientAuth(), address);
+ } else {
+ return TSSLTransportFactory.getServerSocket(port, timeout, address, params.getTTransportParams());
+ }
+ }
+
public static void close() {
ThriftTransportPool.close();
}
+
+ 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 {
+ transport = TSSLTransportFactory.getClientSocket(address.getHostText(), address.getPort(), timeout, sslParams.getTTransportParams());
+ }
+ // 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;
+ }
}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/7038755b/core/src/main/java/org/apache/accumulo/core/util/shell/Shell.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/util/shell/Shell.java b/core/src/main/java/org/apache/accumulo/core/util/shell/Shell.java
index 52e1d04..b518400 100644
--- a/core/src/main/java/org/apache/accumulo/core/util/shell/Shell.java
+++ b/core/src/main/java/org/apache/accumulo/core/util/shell/Shell.java
@@ -50,7 +50,10 @@ import org.apache.accumulo.core.client.mock.MockInstance;
import org.apache.accumulo.core.client.security.tokens.AuthenticationToken;
import org.apache.accumulo.core.client.security.tokens.PasswordToken;
import org.apache.accumulo.core.conf.AccumuloConfiguration;
+import org.apache.accumulo.core.conf.ClientConfiguration;
+import org.apache.accumulo.core.conf.SiteConfiguration;
import org.apache.accumulo.core.conf.Property;
+import org.apache.accumulo.core.conf.ClientConfiguration.ClientProperty;
import org.apache.accumulo.core.data.Key;
import org.apache.accumulo.core.data.Value;
import org.apache.accumulo.core.data.thrift.TConstraintViolationSummary;
@@ -398,15 +401,25 @@ public class Shell extends ShellOptions {
instanceName = options.getZooKeeperInstanceName();
hosts = options.getZooKeeperHosts();
}
- instance = getZooInstance(instanceName, hosts);
+ try {
+ instance = getZooInstance(instanceName, hosts, options.getClientConfiguration());
+ } catch (Exception e) {
+ throw new IllegalArgumentException("Unable to load client config from " + options.getClientConfigFile(), e);
+ }
}
}
- private static Instance getZooInstance(String instanceName, String keepers) {
+ /*
+ * Takes instanceName and keepers as separate arguments, rather than just packaged into the clientConfig,
+ * so that we can fail over to accumulo-site.xml or HDFS config if they're unspecified.
+ */
+ private static Instance getZooInstance(String instanceName, String keepers, ClientConfiguration clientConfig) {
UUID instanceId = null;
+ if (instanceName == null) {
+ instanceName = clientConfig.get(ClientProperty.INSTANCE_NAME);
+ }
if (instanceName == null || keepers == null) {
- @SuppressWarnings("deprecation")
- AccumuloConfiguration conf = AccumuloConfiguration.getSiteConfiguration();
+ AccumuloConfiguration conf = SiteConfiguration.getInstance(clientConfig.getAccumuloConfiguration());
if (instanceName == null) {
Path instanceDir = new Path(conf.get(Property.INSTANCE_DFS_DIR), "instance_id");
instanceId = UUID.fromString(ZooUtil.getInstanceIDFromHdfs(instanceDir));
@@ -416,9 +429,9 @@ public class Shell extends ShellOptions {
}
}
if (instanceId != null) {
- return new ZooKeeperInstance(instanceId, keepers);
+ return new ZooKeeperInstance(clientConfig.withInstance(instanceId).withZkHosts(keepers));
} else {
- return new ZooKeeperInstance(instanceName, keepers);
+ return new ZooKeeperInstance(clientConfig.withInstance(instanceName).withZkHosts(keepers));
}
}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/7038755b/core/src/main/java/org/apache/accumulo/core/util/shell/ShellOptionsJC.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/util/shell/ShellOptionsJC.java b/core/src/main/java/org/apache/accumulo/core/util/shell/ShellOptionsJC.java
index cb1f1c8..2f30a87 100644
--- a/core/src/main/java/org/apache/accumulo/core/util/shell/ShellOptionsJC.java
+++ b/core/src/main/java/org/apache/accumulo/core/util/shell/ShellOptionsJC.java
@@ -25,6 +25,9 @@ import java.util.Scanner;
import java.util.TreeMap;
import org.apache.accumulo.core.client.security.tokens.AuthenticationToken;
+import org.apache.accumulo.core.conf.ClientConfiguration;
+import org.apache.accumulo.core.conf.ClientConfiguration.ClientProperty;
+import org.apache.commons.configuration.ConfigurationException;
import org.apache.log4j.Logger;
import com.beust.jcommander.DynamicParameter;
@@ -161,6 +164,12 @@ public class ShellOptionsJC {
@Parameter(names = {"-z", "--zooKeeperInstance"}, description = "use a zookeeper instance with the given instance name and list of zoo hosts", arity = 2)
private List<String> zooKeeperInstance = new ArrayList<String>();
+ @Parameter(names = {"--ssl"}, description = "use ssl to connect to accumulo")
+ private boolean useSsl = false;
+
+ @Parameter(names = "--config-file", description = "read the given client config file. If omitted, the path searched can be specified with $ACCUMULO_CLIENT_CONF_PATH, which defaults to ~/.accumulo/config:$ACCUMULO_CONF_DIR/client.conf:/etc/accumulo/client.conf")
+ private String clientConfigFile = null;
+
@Parameter(names = {"-zi", "--zooKeeperInstanceName"}, description="use a zookeeper instance with the given instance name")
private String zooKeeperInstanceName;
@@ -247,4 +256,22 @@ public class ShellOptionsJC {
public List<String> getUnrecognizedOptions() {
return unrecognizedOptions;
}
+
+ public boolean useSsl() {
+ return useSsl;
+ }
+
+ public String getClientConfigFile() {
+ return clientConfigFile;
+ }
+
+ public ClientConfiguration getClientConfiguration() throws ConfigurationException,
+ FileNotFoundException {
+ ClientConfiguration clientConfig = ClientConfiguration.loadDefault(getClientConfigFile());
+ if (useSsl()) {
+ clientConfig.setProperty(ClientProperty.INSTANCE_RPC_SSL_ENABLED, "true");
+ }
+ return clientConfig;
+ }
+
}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/7038755b/core/src/test/java/org/apache/accumulo/core/client/impl/TabletLocatorImplTest.java
----------------------------------------------------------------------
diff --git a/core/src/test/java/org/apache/accumulo/core/client/impl/TabletLocatorImplTest.java b/core/src/test/java/org/apache/accumulo/core/client/impl/TabletLocatorImplTest.java
index 8f3fa1d..7abacb8 100644
--- a/core/src/test/java/org/apache/accumulo/core/client/impl/TabletLocatorImplTest.java
+++ b/core/src/test/java/org/apache/accumulo/core/client/impl/TabletLocatorImplTest.java
@@ -445,6 +445,7 @@ public class TabletLocatorImplTest extends TestCase {
}
@Override
+ @Deprecated
public void setConfiguration(AccumuloConfiguration conf) {
this.conf = conf;
}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/7038755b/core/src/test/java/org/apache/accumulo/core/client/mapreduce/lib/util/ConfiguratorBaseTest.java
----------------------------------------------------------------------
diff --git a/core/src/test/java/org/apache/accumulo/core/client/mapreduce/lib/util/ConfiguratorBaseTest.java b/core/src/test/java/org/apache/accumulo/core/client/mapreduce/lib/util/ConfiguratorBaseTest.java
index 62564fa..50fc0a9 100644
--- a/core/src/test/java/org/apache/accumulo/core/client/mapreduce/lib/util/ConfiguratorBaseTest.java
+++ b/core/src/test/java/org/apache/accumulo/core/client/mapreduce/lib/util/ConfiguratorBaseTest.java
@@ -27,6 +27,8 @@ import org.apache.accumulo.core.client.mock.MockInstance;
import org.apache.accumulo.core.client.security.tokens.AuthenticationToken;
import org.apache.accumulo.core.client.security.tokens.AuthenticationToken.AuthenticationTokenSerializer;
import org.apache.accumulo.core.client.security.tokens.PasswordToken;
+import org.apache.accumulo.core.conf.ClientConfiguration;
+import org.apache.accumulo.core.conf.ClientConfiguration.ClientProperty;
import org.apache.commons.codec.binary.Base64;
import org.apache.hadoop.conf.Configuration;
import org.apache.log4j.Level;
@@ -75,16 +77,47 @@ public class ConfiguratorBaseTest {
assertEquals("file:testFile", conf.get(ConfiguratorBase.enumToConfKey(this.getClass(), ConfiguratorBase.ConnectorInfo.TOKEN)));
}
+ @SuppressWarnings("deprecation")
@Test
- public void testSetZooKeeperInstance() {
+ public void testSetZooKeeperInstance_legacy() {
Configuration conf = new Configuration();
ConfiguratorBase.setZooKeeperInstance(this.getClass(), conf, "testInstanceName", "testZooKeepers");
- assertEquals("testInstanceName", conf.get(ConfiguratorBase.enumToConfKey(this.getClass(), ConfiguratorBase.InstanceOpts.NAME)));
- assertEquals("testZooKeepers", conf.get(ConfiguratorBase.enumToConfKey(this.getClass(), ConfiguratorBase.InstanceOpts.ZOO_KEEPERS)));
+ ClientConfiguration clientConf = ClientConfiguration.deserialize(conf.get(ConfiguratorBase.enumToConfKey(this.getClass(), ConfiguratorBase.InstanceOpts.CLIENT_CONFIG)));
+ assertEquals("testInstanceName", clientConf.get(ClientProperty.INSTANCE_NAME));
+ assertEquals("testZooKeepers", clientConf.get(ClientProperty.INSTANCE_ZK_HOST));
assertEquals(ZooKeeperInstance.class.getSimpleName(), conf.get(ConfiguratorBase.enumToConfKey(this.getClass(), ConfiguratorBase.InstanceOpts.TYPE)));
- // TODO uncomment this after ACCUMULO-1699
- // Instance instance = ConfiguratorBase.getInstance(this.getClass(), conf);
- // assertEquals(ZooKeeperInstance.class.getName(), instance.getClass().getName());
+
+ Instance instance = ConfiguratorBase.getInstance(this.getClass(), conf);
+ assertEquals(ZooKeeperInstance.class.getName(), instance.getClass().getName());
+ assertEquals("testInstanceName", ((ZooKeeperInstance)instance).getInstanceName());
+ assertEquals("testZooKeepers", ((ZooKeeperInstance)instance).getZooKeepers());
+
+ // Also make sure we can still deserialize job configurations with the old keys
+ conf = new Configuration();
+ conf.set(ConfiguratorBase.enumToConfKey(this.getClass(), ConfiguratorBase.InstanceOpts.TYPE), ZooKeeperInstance.class.getSimpleName());
+ conf.set(ConfiguratorBase.enumToConfKey(this.getClass(), ConfiguratorBase.InstanceOpts.NAME), "testInstanceName");
+ conf.set(ConfiguratorBase.enumToConfKey(this.getClass(), ConfiguratorBase.InstanceOpts.ZOO_KEEPERS), "testZooKeepers");
+ instance = ConfiguratorBase.getInstance(this.getClass(), conf);
+ assertEquals("testInstanceName", ((ZooKeeperInstance)instance).getInstanceName());
+ assertEquals("testZooKeepers", ((ZooKeeperInstance)instance).getZooKeepers());
+ }
+
+ @Test
+ public void testSetZooKeeperInstance() {
+ Configuration conf = new Configuration();
+ ConfiguratorBase.setZooKeeperInstance(this.getClass(), conf, new ClientConfiguration().withInstance("testInstanceName").withZkHosts("testZooKeepers").withSsl(true).withZkTimeout(1234));
+ ClientConfiguration clientConf = ClientConfiguration.deserialize(conf.get(ConfiguratorBase.enumToConfKey(this.getClass(), ConfiguratorBase.InstanceOpts.CLIENT_CONFIG)));
+ assertEquals("testInstanceName", clientConf.get(ClientProperty.INSTANCE_NAME));
+ assertEquals("testZooKeepers", clientConf.get(ClientProperty.INSTANCE_ZK_HOST));
+ assertEquals("true", clientConf.get(ClientProperty.INSTANCE_RPC_SSL_ENABLED));
+ assertEquals("1234", clientConf.get(ClientProperty.INSTANCE_ZK_TIMEOUT));
+ assertEquals(ZooKeeperInstance.class.getSimpleName(), conf.get(ConfiguratorBase.enumToConfKey(this.getClass(), ConfiguratorBase.InstanceOpts.TYPE)));
+
+ Instance instance = ConfiguratorBase.getInstance(this.getClass(), conf);
+ assertEquals(ZooKeeperInstance.class.getName(), instance.getClass().getName());
+ assertEquals("testInstanceName", ((ZooKeeperInstance)instance).getInstanceName());
+ assertEquals("testZooKeepers", ((ZooKeeperInstance)instance).getZooKeepers());
+ assertEquals(1234000, ((ZooKeeperInstance)instance).getZooKeepersSessionTimeOut());
}
@Test
http://git-wip-us.apache.org/repos/asf/accumulo/blob/7038755b/core/src/test/java/org/apache/accumulo/core/conf/ClientConfigurationTest.java
----------------------------------------------------------------------
diff --git a/core/src/test/java/org/apache/accumulo/core/conf/ClientConfigurationTest.java b/core/src/test/java/org/apache/accumulo/core/conf/ClientConfigurationTest.java
new file mode 100644
index 0000000..55cf9d3
--- /dev/null
+++ b/core/src/test/java/org/apache/accumulo/core/conf/ClientConfigurationTest.java
@@ -0,0 +1,65 @@
+/*
+ * 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.conf;
+
+import static org.junit.Assert.assertEquals;
+
+import java.util.Arrays;
+
+import org.apache.accumulo.core.conf.ClientConfiguration.ClientProperty;
+import org.apache.commons.configuration.Configuration;
+import org.apache.commons.configuration.PropertiesConfiguration;
+import org.junit.Test;
+
+public class ClientConfigurationTest {
+ @Test
+ public void testOverrides() throws Exception {
+ ClientConfiguration clientConfig = createConfig();
+ assertExpectedConfig(clientConfig);
+ }
+
+ @Test
+ public void testSerialization() throws Exception {
+ ClientConfiguration clientConfig = createConfig();
+ // sanity check that we're starting with what we're expecting
+ assertExpectedConfig(clientConfig);
+
+ String serialized = clientConfig.serialize();
+ ClientConfiguration deserializedClientConfig = ClientConfiguration.deserialize(serialized);
+ assertExpectedConfig(deserializedClientConfig);
+ }
+
+ private void assertExpectedConfig(ClientConfiguration clientConfig) {
+ assertEquals("firstZkHosts", clientConfig.get(ClientProperty.INSTANCE_ZK_HOST));
+ assertEquals("secondInstanceName", clientConfig.get(ClientProperty.INSTANCE_NAME));
+ assertEquals("123s", clientConfig.get(ClientProperty.INSTANCE_ZK_TIMEOUT));
+ assertEquals(ClientProperty.RPC_SSL_TRUSTSTORE_TYPE.getDefaultValue(), clientConfig.get(ClientProperty.RPC_SSL_TRUSTSTORE_TYPE));
+ }
+
+ private ClientConfiguration createConfig() {
+ Configuration first = new PropertiesConfiguration();
+ first.addProperty(ClientProperty.INSTANCE_ZK_HOST.getKey(), "firstZkHosts");
+ Configuration second = new PropertiesConfiguration();
+ second.addProperty(ClientProperty.INSTANCE_ZK_HOST.getKey(), "secondZkHosts");
+ second.addProperty(ClientProperty.INSTANCE_NAME.getKey(), "secondInstanceName");
+ Configuration third = new PropertiesConfiguration();
+ third.addProperty(ClientProperty.INSTANCE_ZK_HOST.getKey(), "thirdZkHosts");
+ third.addProperty(ClientProperty.INSTANCE_NAME.getKey(), "thirdInstanceName");
+ third.addProperty(ClientProperty.INSTANCE_ZK_TIMEOUT.getKey(), "123s");
+ return new ClientConfiguration(Arrays.asList(first, second, third));
+ }
+}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/7038755b/core/src/test/java/org/apache/accumulo/core/util/shell/ShellSetInstanceTest.java
----------------------------------------------------------------------
diff --git a/core/src/test/java/org/apache/accumulo/core/util/shell/ShellSetInstanceTest.java b/core/src/test/java/org/apache/accumulo/core/util/shell/ShellSetInstanceTest.java
index 23ca13a..5ce1320 100644
--- a/core/src/test/java/org/apache/accumulo/core/util/shell/ShellSetInstanceTest.java
+++ b/core/src/test/java/org/apache/accumulo/core/util/shell/ShellSetInstanceTest.java
@@ -38,10 +38,14 @@ import jline.console.ConsoleReader;
import org.apache.accumulo.core.client.ZooKeeperInstance;
import org.apache.accumulo.core.client.mock.MockInstance;
import org.apache.accumulo.core.conf.AccumuloConfiguration;
+import org.apache.accumulo.core.conf.ClientConfiguration;
+import org.apache.accumulo.core.conf.ClientConfiguration.ClientProperty;
+import org.apache.accumulo.core.conf.ConfigSanityCheck;
import org.apache.accumulo.core.conf.Property;
import org.apache.accumulo.core.zookeeper.ZooUtil;
import org.apache.hadoop.fs.Path;
import org.apache.log4j.Level;
+import org.easymock.EasyMock;
import org.junit.AfterClass;
import org.junit.Before;
import org.junit.BeforeClass;
@@ -51,7 +55,7 @@ import org.powermock.core.classloader.annotations.PrepareForTest;
import org.powermock.modules.junit4.PowerMockRunner;
@RunWith(PowerMockRunner.class)
-@PrepareForTest({Shell.class, AccumuloConfiguration.class, ZooUtil.class})
+@PrepareForTest({Shell.class, ZooUtil.class, ConfigSanityCheck.class})
public class ShellSetInstanceTest {
public static class TestOutputStream extends OutputStream {
StringBuilder sb = new StringBuilder();
@@ -121,38 +125,49 @@ public class ShellSetInstanceTest {
testSetInstance_HdfsZooInstance(false, false, false);
}
- @SuppressWarnings("deprecation")
private void testSetInstance_HdfsZooInstance(boolean explicitHdfs, boolean onlyInstance, boolean onlyHosts)
throws Exception {
+ ClientConfiguration clientConf = createMock(ClientConfiguration.class);
ShellOptionsJC opts = createMock(ShellOptionsJC.class);
expect(opts.isFake()).andReturn(false);
+ expect(opts.getClientConfiguration()).andReturn(clientConf);
expect(opts.isHdfsZooInstance()).andReturn(explicitHdfs);
if (!explicitHdfs) {
expect(opts.getZooKeeperInstance())
.andReturn(Collections.<String>emptyList());
if (onlyInstance) {
expect(opts.getZooKeeperInstanceName()).andReturn("instance");
+ expect(clientConf.withInstance("instance")).andReturn(clientConf);
} else {
expect(opts.getZooKeeperInstanceName()).andReturn(null);
}
if (onlyHosts) {
expect(opts.getZooKeeperHosts()).andReturn("host3,host4");
+ expect(clientConf.withZkHosts("host3,host4")).andReturn(clientConf);
} else {
expect(opts.getZooKeeperHosts()).andReturn(null);
}
}
replay(opts);
+ if (!onlyInstance) {
+ expect(clientConf.get(ClientProperty.INSTANCE_NAME)).andReturn(null);
+ }
+
AccumuloConfiguration conf = createMock(AccumuloConfiguration.class);
- mockStatic(AccumuloConfiguration.class);
- expect(AccumuloConfiguration.getSiteConfiguration()).andReturn(conf);
- replay(AccumuloConfiguration.class);
+ expect(clientConf.getAccumuloConfiguration()).andReturn(conf);
+
+ mockStatic(ConfigSanityCheck.class);
+ ConfigSanityCheck.validate(EasyMock.<AccumuloConfiguration>anyObject());
+ expectLastCall();
+ replay(ConfigSanityCheck.class);
if (!onlyHosts) {
- expect(conf.get(Property.INSTANCE_ZK_HOST)).andReturn("host1,host2");
+ expect(conf.get(Property.INSTANCE_ZK_HOST)).andReturn("host1,host2").atLeastOnce();
+ expect(clientConf.withZkHosts("host1,host2")).andReturn(clientConf);
}
if (!onlyInstance) {
- expect(conf.get(Property.INSTANCE_DFS_DIR)).andReturn("/dfs");
+ expect(conf.get(Property.INSTANCE_DFS_DIR)).andReturn("/dfs").atLeastOnce();
}
replay(conf);
UUID randomUUID = null;
@@ -162,21 +177,13 @@ public class ShellSetInstanceTest {
expect(ZooUtil.getInstanceIDFromHdfs(anyObject(Path.class)))
.andReturn(randomUUID.toString());
replay(ZooUtil.class);
+ expect(clientConf.withInstance(randomUUID)).andReturn(clientConf);
}
+ replay(clientConf);
ZooKeeperInstance theInstance = createMock(ZooKeeperInstance.class);
- String expectedHosts = "host1,host2";
- if (onlyHosts)
- expectedHosts = "host3,host4";
-
- if (!onlyInstance) {
- expectNew(ZooKeeperInstance.class, randomUUID, expectedHosts)
- .andReturn(theInstance);
- } else {
- expectNew(ZooKeeperInstance.class, "instance", expectedHosts)
- .andReturn(theInstance);
- }
+ expectNew(ZooKeeperInstance.class, clientConf).andReturn(theInstance);
replay(theInstance, ZooKeeperInstance.class);
shell.setInstance(opts);
@@ -191,28 +198,31 @@ public class ShellSetInstanceTest {
testSetInstance_ZKInstance(false);
}
private void testSetInstance_ZKInstance(boolean dashZ) throws Exception {
+ ClientConfiguration clientConf = createMock(ClientConfiguration.class);
ShellOptionsJC opts = createMock(ShellOptionsJC.class);
expect(opts.isFake()).andReturn(false);
+ expect(opts.getClientConfiguration()).andReturn(clientConf);
expect(opts.isHdfsZooInstance()).andReturn(false);
if (dashZ) {
+ expect(clientConf.withInstance("foo")).andReturn(clientConf);
+ expect(clientConf.withZkHosts("host1,host2")).andReturn(clientConf);
List<String> zl = new java.util.ArrayList<String>();
zl.add("foo");
zl.add("host1,host2");
expect(opts.getZooKeeperInstance()).andReturn(zl);
expectLastCall().anyTimes();
} else {
+ expect(clientConf.withInstance("bar")).andReturn(clientConf);
+ expect(clientConf.withZkHosts("host3,host4")).andReturn(clientConf);
expect(opts.getZooKeeperInstance()).andReturn(Collections.<String>emptyList());
expect(opts.getZooKeeperInstanceName()).andReturn("bar");
expect(opts.getZooKeeperHosts()).andReturn("host3,host4");
}
+ replay(clientConf);
replay(opts);
ZooKeeperInstance theInstance = createMock(ZooKeeperInstance.class);
- if (dashZ)
- expectNew(ZooKeeperInstance.class, "foo", "host1,host2").andReturn(theInstance);
- else
- expectNew(ZooKeeperInstance.class, "bar", "host3,host4")
- .andReturn(theInstance);
+ expectNew(ZooKeeperInstance.class, clientConf).andReturn(theInstance);
replay(theInstance, ZooKeeperInstance.class);
shell.setInstance(opts);
http://git-wip-us.apache.org/repos/asf/accumulo/blob/7038755b/examples/simple/src/main/java/org/apache/accumulo/examples/simple/filedata/FileDataQuery.java
----------------------------------------------------------------------
diff --git a/examples/simple/src/main/java/org/apache/accumulo/examples/simple/filedata/FileDataQuery.java b/examples/simple/src/main/java/org/apache/accumulo/examples/simple/filedata/FileDataQuery.java
index ecb42c7..a66438e 100644
--- a/examples/simple/src/main/java/org/apache/accumulo/examples/simple/filedata/FileDataQuery.java
+++ b/examples/simple/src/main/java/org/apache/accumulo/examples/simple/filedata/FileDataQuery.java
@@ -28,6 +28,7 @@ import org.apache.accumulo.core.client.Scanner;
import org.apache.accumulo.core.client.TableNotFoundException;
import org.apache.accumulo.core.client.ZooKeeperInstance;
import org.apache.accumulo.core.client.security.tokens.AuthenticationToken;
+import org.apache.accumulo.core.conf.ClientConfiguration;
import org.apache.accumulo.core.data.Key;
import org.apache.accumulo.core.data.Range;
import org.apache.accumulo.core.data.Value;
@@ -46,7 +47,7 @@ public class FileDataQuery {
public FileDataQuery(String instanceName, String zooKeepers, String user, AuthenticationToken token, String tableName, Authorizations auths) throws AccumuloException,
AccumuloSecurityException, TableNotFoundException {
- ZooKeeperInstance instance = new ZooKeeperInstance(instanceName, zooKeepers);
+ ZooKeeperInstance instance = new ZooKeeperInstance(ClientConfiguration.loadDefault().withInstance(instanceName).withZkHosts(zooKeepers));
conn = instance.getConnector(user, token);
lastRefs = new ArrayList<Entry<Key,Value>>();
cis = new ChunkInputStream();
http://git-wip-us.apache.org/repos/asf/accumulo/blob/7038755b/examples/simple/src/main/java/org/apache/accumulo/examples/simple/mapreduce/TokenFileWordCount.java
----------------------------------------------------------------------
diff --git a/examples/simple/src/main/java/org/apache/accumulo/examples/simple/mapreduce/TokenFileWordCount.java b/examples/simple/src/main/java/org/apache/accumulo/examples/simple/mapreduce/TokenFileWordCount.java
index 16e0356..1114a7e 100644
--- a/examples/simple/src/main/java/org/apache/accumulo/examples/simple/mapreduce/TokenFileWordCount.java
+++ b/examples/simple/src/main/java/org/apache/accumulo/examples/simple/mapreduce/TokenFileWordCount.java
@@ -19,6 +19,7 @@ package org.apache.accumulo.examples.simple.mapreduce;
import java.io.IOException;
import org.apache.accumulo.core.client.mapreduce.AccumuloOutputFormat;
+import org.apache.accumulo.core.conf.ClientConfiguration;
import org.apache.accumulo.core.data.Mutation;
import org.apache.accumulo.core.data.Value;
import org.apache.accumulo.core.util.CachedConfiguration;
@@ -82,7 +83,7 @@ public class TokenFileWordCount extends Configured implements Tool {
job.setOutputValueClass(Mutation.class);
// AccumuloInputFormat not used here, but it uses the same functions.
- AccumuloOutputFormat.setZooKeeperInstance(job, instance, zookeepers);
+ AccumuloOutputFormat.setZooKeeperInstance(job, ClientConfiguration.loadDefault().withInstance(instance).withZkHosts(zookeepers));
AccumuloOutputFormat.setConnectorInfo(job, user, tokenFile);
AccumuloOutputFormat.setCreateTables(job, true);
AccumuloOutputFormat.setDefaultTableName(job, tableName);
http://git-wip-us.apache.org/repos/asf/accumulo/blob/7038755b/minicluster/src/main/java/org/apache/accumulo/minicluster/MiniAccumuloCluster.java
----------------------------------------------------------------------
diff --git a/minicluster/src/main/java/org/apache/accumulo/minicluster/MiniAccumuloCluster.java b/minicluster/src/main/java/org/apache/accumulo/minicluster/MiniAccumuloCluster.java
index 42882bb..8e7e2ba 100644
--- a/minicluster/src/main/java/org/apache/accumulo/minicluster/MiniAccumuloCluster.java
+++ b/minicluster/src/main/java/org/apache/accumulo/minicluster/MiniAccumuloCluster.java
@@ -48,6 +48,7 @@ import org.apache.accumulo.core.client.Connector;
import org.apache.accumulo.core.client.Instance;
import org.apache.accumulo.core.client.ZooKeeperInstance;
import org.apache.accumulo.core.client.security.tokens.PasswordToken;
+import org.apache.accumulo.core.conf.ClientConfiguration;
import org.apache.accumulo.core.conf.Property;
import org.apache.accumulo.core.master.thrift.MasterGoalState;
import org.apache.accumulo.core.util.Daemon;
@@ -60,6 +61,7 @@ import org.apache.accumulo.server.init.Initialize;
import org.apache.accumulo.server.util.PortUtils;
import org.apache.accumulo.server.util.time.SimpleTimer;
import org.apache.accumulo.start.Main;
+import org.apache.commons.configuration.MapConfiguration;
import org.apache.accumulo.start.classloader.vfs.MiniDFSUtil;
import org.apache.accumulo.tserver.TabletServer;
import org.apache.commons.io.FileUtils;
@@ -71,6 +73,9 @@ import org.apache.hadoop.hdfs.DFSConfigKeys;
import org.apache.hadoop.hdfs.MiniDFSCluster;
import org.apache.zookeeper.server.ZooKeeperServerMain;
+import com.google.common.base.Predicate;
+import com.google.common.collect.Maps;
+
/**
* A utility class that will create Zookeeper and Accumulo processes that write all of their data to a single local directory. This class makes it easy to test
* code against a real Accumulo instance. Its much more accurate for testing than {@link org.apache.accumulo.core.client.mock.MockAccumulo}, but much slower.
@@ -232,6 +237,9 @@ public class MiniAccumuloCluster {
argList.addAll(Arrays.asList(javaBin, "-Dproc=" + clazz.getSimpleName(), "-cp", classpath));
argList.add("-Djava.library.path=" + config.getLibDir());
argList.addAll(extraJvmOpts);
+ for (Entry<String,String> sysProp : config.getSystemProperties().entrySet()) {
+ argList.add(String.format("-D%s=%s", sysProp.getKey(), sysProp.getValue()));
+ }
argList.addAll(Arrays.asList("-XX:+UseConcMarkSweepGC", "-XX:CMSInitiatingOccupancyFraction=75", Main.class.getName(), className));
argList.addAll(Arrays.asList(args));
@@ -239,6 +247,15 @@ public class MiniAccumuloCluster {
builder.environment().put("ACCUMULO_HOME", config.getDir().getAbsolutePath());
builder.environment().put("ACCUMULO_LOG_DIR", config.getLogDir().getAbsolutePath());
+ builder.environment().put("ACCUMULO_CLIENT_CONF_PATH", config.getClientConfFile().getAbsolutePath());
+
+ // if we're running under accumulo.start, we forward these env vars
+ String env = System.getenv("HADOOP_PREFIX");
+ if (env != null)
+ builder.environment().put("HADOOP_PREFIX", env);
+ env = System.getenv("ZOOKEEPER_HOME");
+ if (env != null)
+ builder.environment().put("ZOOKEEPER_HOME", env);
builder.environment().put("ACCUMULO_CONF_DIR", config.getConfDir().getAbsolutePath());
// hadoop-2.2 puts error messages in the logs if this is not set
builder.environment().put("HADOOP_HOME", config.getDir().getAbsolutePath());
@@ -333,19 +350,19 @@ public class MiniAccumuloCluster {
dfsUri = "file://";
}
+ File clientConfFile = config.getClientConfFile();
+ // Write only the properties that correspond to ClientConfiguration properties
+ writeConfigProperties(clientConfFile, Maps.filterEntries(config.getSiteConfig(), new Predicate<Entry<String,String>>() {
+ public boolean apply(Entry<String,String> v) {
+ return ClientConfiguration.ClientProperty.getPropertyByKey(v.getKey()) != null;
+ }
+ }));
+
File siteFile = new File(config.getConfDir(), "accumulo-site.xml");
writeConfig(siteFile, config.getSiteConfig().entrySet());
- FileWriter fileWriter = new FileWriter(siteFile);
- fileWriter.append("<configuration>\n");
-
- for (Entry<String,String> entry : config.getSiteConfig().entrySet())
- fileWriter.append("<property><name>" + entry.getKey() + "</name><value>" + entry.getValue() + "</value></property>\n");
- fileWriter.append("</configuration>\n");
- fileWriter.close();
-
zooCfgFile = new File(config.getConfDir(), "zoo.cfg");
- fileWriter = new FileWriter(zooCfgFile);
+ FileWriter fileWriter = new FileWriter(zooCfgFile);
// zookeeper uses Properties to read its config, so use that to write in order to properly escape things like Windows paths
Properties zooCfg = new Properties();
@@ -383,6 +400,13 @@ public class MiniAccumuloCluster {
fileWriter.append("</configuration>\n");
fileWriter.close();
}
+ private void writeConfigProperties(File file, Map<String,String> settings) throws IOException {
+ FileWriter fileWriter = new FileWriter(file);
+
+ for (Entry<String,String> entry : settings.entrySet())
+ fileWriter.append(entry.getKey() + "=" + entry.getValue() + "\n");
+ fileWriter.close();
+ }
/**
* Starts Accumulo and Zookeeper processes. Can only be called once.
@@ -597,10 +621,14 @@ public class MiniAccumuloCluster {
* @since 1.6.0
*/
public Connector getConnector(String user, String passwd) throws AccumuloException, AccumuloSecurityException {
- Instance instance = new ZooKeeperInstance(this.getInstanceName(), this.getZooKeepers());
+ Instance instance = new ZooKeeperInstance(getClientConfig());
return instance.getConnector(user, new PasswordToken(passwd));
}
+ public ClientConfiguration getClientConfig() {
+ return new ClientConfiguration(Arrays.asList(new MapConfiguration(config.getSiteConfig()))).withInstance(this.getInstanceName()).withZkHosts(this.getZooKeepers());
+ }
+
public FileSystem getFileSystem() {
try {
return FileSystem.get(new URI(dfsUri), new Configuration());
http://git-wip-us.apache.org/repos/asf/accumulo/blob/7038755b/minicluster/src/main/java/org/apache/accumulo/minicluster/MiniAccumuloConfig.java
----------------------------------------------------------------------
diff --git a/minicluster/src/main/java/org/apache/accumulo/minicluster/MiniAccumuloConfig.java b/minicluster/src/main/java/org/apache/accumulo/minicluster/MiniAccumuloConfig.java
index bfa7922..5c7050a 100644
--- a/minicluster/src/main/java/org/apache/accumulo/minicluster/MiniAccumuloConfig.java
+++ b/minicluster/src/main/java/org/apache/accumulo/minicluster/MiniAccumuloConfig.java
@@ -39,8 +39,8 @@ public class MiniAccumuloConfig {
private Map<String,String> siteConfig = new HashMap<String,String>();
private int numTservers = 2;
private Map<ServerType,Long> memoryConfig = new HashMap<ServerType,Long>();
-
private boolean jdwpEnabled = false;
+ private Map<String,String> systemProperties = new HashMap<String,String>();
private String instanceName = "miniInstance";
@@ -360,6 +360,15 @@ public class MiniAccumuloConfig {
}
/**
+ * @return location of client conf file containing connection parameters for connecting to this minicluster
+ *
+ * @since 1.6.0
+ */
+ public File getClientConfFile() {
+ return new File(getConfDir(), "client.conf");
+ }
+
+ /**
* Whether or not the Accumulo garbage collector proces will run
*/
public boolean shouldRunGC() {
@@ -376,6 +385,24 @@ public class MiniAccumuloConfig {
}
/**
+ * @return sets system properties set for service processes
+ *
+ * @since 1.6.0
+ */
+ public void setSystemProperties(Map<String,String> systemProperties) {
+ this.systemProperties = new HashMap<String,String>(systemProperties);
+ }
+
+ /**
+ * @return a copy of the system properties for service processes
+ *
+ * @since 1.6.0
+ */
+ public Map<String,String> getSystemProperties() {
+ return new HashMap<String,String>(systemProperties);
+ }
+
+ /*
* Gets the classpath elements to use when spawning processes.
*
* @return the classpathItems, if set
http://git-wip-us.apache.org/repos/asf/accumulo/blob/7038755b/minicluster/src/main/java/org/apache/accumulo/minicluster/MiniAccumuloInstance.java
----------------------------------------------------------------------
diff --git a/minicluster/src/main/java/org/apache/accumulo/minicluster/MiniAccumuloInstance.java b/minicluster/src/main/java/org/apache/accumulo/minicluster/MiniAccumuloInstance.java
index 540d7ae..1e1c464 100644
--- a/minicluster/src/main/java/org/apache/accumulo/minicluster/MiniAccumuloInstance.java
+++ b/minicluster/src/main/java/org/apache/accumulo/minicluster/MiniAccumuloInstance.java
@@ -17,11 +17,15 @@
package org.apache.accumulo.minicluster;
import java.io.File;
+import java.io.FileNotFoundException;
import java.net.MalformedURLException;
import org.apache.accumulo.core.client.Instance;
import org.apache.accumulo.core.client.ZooKeeperInstance;
+import org.apache.accumulo.core.conf.ClientConfiguration;
import org.apache.accumulo.core.conf.Property;
+import org.apache.commons.configuration.ConfigurationException;
+import org.apache.commons.configuration.PropertiesConfiguration;
import org.apache.hadoop.conf.Configuration;
/**
@@ -32,11 +36,20 @@ public class MiniAccumuloInstance extends ZooKeeperInstance {
/**
* Construct an {@link Instance} entry point to Accumulo using a {@link MiniAccumuloCluster} directory
*/
- public MiniAccumuloInstance(String instanceName, File directory) {
- super(instanceName, getZooKeepersFromDir(directory));
+ public MiniAccumuloInstance(String instanceName, File directory) throws FileNotFoundException {
+ super(new ClientConfiguration(getConfigProperties(directory)).withInstance(instanceName).withZkHosts(getZooKeepersFromDir(directory)));
+ }
+
+ public static PropertiesConfiguration getConfigProperties(File directory) {
+ try {
+ return new PropertiesConfiguration(new File(new File(directory, "conf"), "client.conf"));
+ } catch (ConfigurationException e) {
+ // this should never happen since we wrote the config file ourselves
+ throw new IllegalArgumentException(e);
+ }
}
- private static String getZooKeepersFromDir(File directory) {
+ private static String getZooKeepersFromDir(File directory) throws FileNotFoundException {
if (!directory.isDirectory())
throw new IllegalArgumentException("Not a directory " + directory.getPath());
File configFile = new File(new File(directory, "conf"), "accumulo-site.xml");
@@ -44,7 +57,7 @@ public class MiniAccumuloInstance extends ZooKeeperInstance {
try {
conf.addResource(configFile.toURI().toURL());
} catch (MalformedURLException e) {
- throw new IllegalStateException("Missing file: " + configFile.getPath());
+ throw new FileNotFoundException("Missing file: " + configFile.getPath());
}
return conf.get(Property.INSTANCE_ZK_HOST.getKey());
}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/7038755b/minicluster/src/test/java/org/apache/accumulo/minicluster/MiniAccumuloClusterGCTest.java
----------------------------------------------------------------------
diff --git a/minicluster/src/test/java/org/apache/accumulo/minicluster/MiniAccumuloClusterGCTest.java b/minicluster/src/test/java/org/apache/accumulo/minicluster/MiniAccumuloClusterGCTest.java
index 281e805..a29bbc0 100644
--- a/minicluster/src/test/java/org/apache/accumulo/minicluster/MiniAccumuloClusterGCTest.java
+++ b/minicluster/src/test/java/org/apache/accumulo/minicluster/MiniAccumuloClusterGCTest.java
@@ -69,7 +69,6 @@ public class MiniAccumuloClusterGCTest {
}
}
-
private static File testDir = new File(System.getProperty("user.dir") + "/target/" + MiniAccumuloClusterGCTest.class.getName());
private static MiniAccumuloConfig macConfig;
private static MiniAccumuloCluster accumulo;
@@ -103,7 +102,7 @@ public class MiniAccumuloClusterGCTest {
// This test seems to be a little too unstable for a unit test
@Ignore
public void test() throws Exception {
- ZooKeeperInstance inst = new ZooKeeperInstance(accumulo.getInstanceName(), accumulo.getZooKeepers());
+ ZooKeeperInstance inst = new ZooKeeperInstance(accumulo.getClientConfig());
Connector c = inst.getConnector("root", new PasswordToken(passwd));
final String table = "foobar";
http://git-wip-us.apache.org/repos/asf/accumulo/blob/7038755b/minicluster/src/test/java/org/apache/accumulo/minicluster/MiniAccumuloClusterTest.java
----------------------------------------------------------------------
diff --git a/minicluster/src/test/java/org/apache/accumulo/minicluster/MiniAccumuloClusterTest.java b/minicluster/src/test/java/org/apache/accumulo/minicluster/MiniAccumuloClusterTest.java
index 26a1546..5de981d 100644
--- a/minicluster/src/test/java/org/apache/accumulo/minicluster/MiniAccumuloClusterTest.java
+++ b/minicluster/src/test/java/org/apache/accumulo/minicluster/MiniAccumuloClusterTest.java
@@ -31,7 +31,6 @@ import org.apache.accumulo.core.client.BatchWriterConfig;
import org.apache.accumulo.core.client.Connector;
import org.apache.accumulo.core.client.IteratorSetting;
import org.apache.accumulo.core.client.Scanner;
-import org.apache.accumulo.core.client.ZooKeeperInstance;
import org.apache.accumulo.core.client.security.tokens.PasswordToken;
import org.apache.accumulo.core.conf.Property;
import org.apache.accumulo.core.data.Key;
@@ -86,8 +85,7 @@ public class MiniAccumuloClusterTest {
@Test(timeout = 30000)
public void test() throws Exception {
- Connector conn = new ZooKeeperInstance(accumulo.getConfig().getInstanceName(), accumulo.getConfig().getZooKeepers()).getConnector("root",
- new PasswordToken("superSecret"));
+ Connector conn = accumulo.getConnector("root", "superSecret");
conn.tableOperations().create("table1");
@@ -102,8 +100,7 @@ public class MiniAccumuloClusterTest {
conn.tableOperations().attachIterator("table1", is);
- Connector uconn = new ZooKeeperInstance(accumulo.getConfig().getInstanceName(), accumulo.getConfig().getZooKeepers()).getConnector("user1",
- new PasswordToken("pass1"));
+ Connector uconn = accumulo.getConnector("user1", "pass1");
BatchWriter bw = uconn.createBatchWriter("table1", new BatchWriterConfig());
@@ -162,8 +159,7 @@ public class MiniAccumuloClusterTest {
@Test(timeout = 60000)
public void testPerTableClasspath() throws Exception {
- Connector conn = new ZooKeeperInstance(accumulo.getConfig().getInstanceName(), accumulo.getConfig().getZooKeepers()).getConnector("root",
- new PasswordToken("superSecret"));
+ Connector conn = accumulo.getConnector("root", "superSecret");
conn.tableOperations().create("table2");
http://git-wip-us.apache.org/repos/asf/accumulo/blob/7038755b/proxy/src/main/java/org/apache/accumulo/proxy/ProxyServer.java
----------------------------------------------------------------------
diff --git a/proxy/src/main/java/org/apache/accumulo/proxy/ProxyServer.java b/proxy/src/main/java/org/apache/accumulo/proxy/ProxyServer.java
index c92f73b..d979d40 100644
--- a/proxy/src/main/java/org/apache/accumulo/proxy/ProxyServer.java
+++ b/proxy/src/main/java/org/apache/accumulo/proxy/ProxyServer.java
@@ -61,6 +61,7 @@ import org.apache.accumulo.core.client.mock.MockInstance;
import org.apache.accumulo.core.client.security.SecurityErrorCode;
import org.apache.accumulo.core.client.security.tokens.AuthenticationToken;
import org.apache.accumulo.core.client.security.tokens.PasswordToken;
+import org.apache.accumulo.core.conf.ClientConfiguration;
import org.apache.accumulo.core.data.Column;
import org.apache.accumulo.core.data.ConditionalMutation;
import org.apache.accumulo.core.data.Key;
@@ -175,7 +176,7 @@ public class ProxyServer implements AccumuloProxy.Iface {
if (useMock != null && Boolean.parseBoolean(useMock))
instance = new MockInstance();
else
- instance = new ZooKeeperInstance(props.getProperty("instance"), props.getProperty("zookeepers"));
+ instance = new ZooKeeperInstance(ClientConfiguration.loadDefault().withInstance(props.getProperty("instance")).withZkHosts(props.getProperty("zookeepers")));
try {
String tokenProp = props.getProperty("tokenClass", PasswordToken.class.getName());
http://git-wip-us.apache.org/repos/asf/accumulo/blob/7038755b/server/base/src/main/java/org/apache/accumulo/server/cli/ClientOnDefaultTable.java
----------------------------------------------------------------------
diff --git a/server/base/src/main/java/org/apache/accumulo/server/cli/ClientOnDefaultTable.java b/server/base/src/main/java/org/apache/accumulo/server/cli/ClientOnDefaultTable.java
index 53f5ac2..588c35c 100644
--- a/server/base/src/main/java/org/apache/accumulo/server/cli/ClientOnDefaultTable.java
+++ b/server/base/src/main/java/org/apache/accumulo/server/cli/ClientOnDefaultTable.java
@@ -36,7 +36,7 @@ public class ClientOnDefaultTable extends org.apache.accumulo.core.cli.ClientOnD
if (instance == null) {
return cachedInstance = HdfsZooInstance.getInstance();
}
- return cachedInstance = new ZooKeeperInstance(this.instance, this.zookeepers);
+ return cachedInstance = new ZooKeeperInstance(this.getClientConfiguration());
}
public ClientOnDefaultTable(String table) {
super(table);
http://git-wip-us.apache.org/repos/asf/accumulo/blob/7038755b/server/base/src/main/java/org/apache/accumulo/server/cli/ClientOnRequiredTable.java
----------------------------------------------------------------------
diff --git a/server/base/src/main/java/org/apache/accumulo/server/cli/ClientOnRequiredTable.java b/server/base/src/main/java/org/apache/accumulo/server/cli/ClientOnRequiredTable.java
index e9e9bf1..f2e04e4 100644
--- a/server/base/src/main/java/org/apache/accumulo/server/cli/ClientOnRequiredTable.java
+++ b/server/base/src/main/java/org/apache/accumulo/server/cli/ClientOnRequiredTable.java
@@ -36,6 +36,6 @@ public class ClientOnRequiredTable extends org.apache.accumulo.core.cli.ClientOn
if (instance == null) {
return cachedInstance = HdfsZooInstance.getInstance();
}
- return cachedInstance = new ZooKeeperInstance(this.instance, this.zookeepers);
+ return cachedInstance = new ZooKeeperInstance(getClientConfiguration());
}
}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/7038755b/server/base/src/main/java/org/apache/accumulo/server/cli/ClientOpts.java
----------------------------------------------------------------------
diff --git a/server/base/src/main/java/org/apache/accumulo/server/cli/ClientOpts.java b/server/base/src/main/java/org/apache/accumulo/server/cli/ClientOpts.java
index 6f3516a..c19b7b0 100644
--- a/server/base/src/main/java/org/apache/accumulo/server/cli/ClientOpts.java
+++ b/server/base/src/main/java/org/apache/accumulo/server/cli/ClientOpts.java
@@ -34,6 +34,6 @@ public class ClientOpts extends org.apache.accumulo.core.cli.ClientOpts {
if (instance == null) {
return HdfsZooInstance.getInstance();
}
- return new ZooKeeperInstance(this.instance, this.zookeepers);
+ return new ZooKeeperInstance(this.getClientConfiguration());
}
}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/7038755b/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 606941d..ecbe0c4 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
@@ -585,7 +585,7 @@ public class BulkImporter {
throws AccumuloException, AccumuloSecurityException {
try {
long timeInMillis = instance.getConfiguration().getTimeInMillis(Property.TSERV_BULK_TIMEOUT);
- TabletClientService.Iface client = ThriftUtil.getTServerClient(location, timeInMillis);
+ TabletClientService.Iface client = ThriftUtil.getTServerClient(location, instance.getConfiguration(), timeInMillis);
try {
HashMap<KeyExtent,Map<String,org.apache.accumulo.core.data.thrift.MapFileInfo>> files = new HashMap<KeyExtent,Map<String,org.apache.accumulo.core.data.thrift.MapFileInfo>>();
for (Entry<KeyExtent,List<PathSize>> entry : assignmentsPerTablet.entrySet()) {
http://git-wip-us.apache.org/repos/asf/accumulo/blob/7038755b/server/base/src/main/java/org/apache/accumulo/server/client/HdfsZooInstance.java
----------------------------------------------------------------------
diff --git a/server/base/src/main/java/org/apache/accumulo/server/client/HdfsZooInstance.java b/server/base/src/main/java/org/apache/accumulo/server/client/HdfsZooInstance.java
index 9e6bbe7..00714f9 100644
--- a/server/base/src/main/java/org/apache/accumulo/server/client/HdfsZooInstance.java
+++ b/server/base/src/main/java/org/apache/accumulo/server/client/HdfsZooInstance.java
@@ -175,6 +175,7 @@ public class HdfsZooInstance implements Instance {
}
@Override
+ @Deprecated
public void setConfiguration(AccumuloConfiguration conf) {
this.conf = conf;
}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/7038755b/server/base/src/main/java/org/apache/accumulo/server/util/TServerUtils.java
----------------------------------------------------------------------
diff --git a/server/base/src/main/java/org/apache/accumulo/server/util/TServerUtils.java b/server/base/src/main/java/org/apache/accumulo/server/util/TServerUtils.java
index 8abd104..eec64ef 100644
--- a/server/base/src/main/java/org/apache/accumulo/server/util/TServerUtils.java
+++ b/server/base/src/main/java/org/apache/accumulo/server/util/TServerUtils.java
@@ -18,6 +18,7 @@ package org.apache.accumulo.server.util;
import java.io.IOException;
import java.lang.reflect.Field;
+import java.net.BindException;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
@@ -33,6 +34,7 @@ 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.TBufferedSocket;
import org.apache.accumulo.core.util.ThriftUtil;
import org.apache.accumulo.core.util.UtilWaitThread;
@@ -47,6 +49,7 @@ import org.apache.thrift.server.TServer;
import org.apache.thrift.server.TThreadPoolServer;
import org.apache.thrift.transport.TNonblockingSocket;
import org.apache.thrift.transport.TServerTransport;
+import org.apache.thrift.transport.TSocket;
import org.apache.thrift.transport.TTransport;
import org.apache.thrift.transport.TTransportException;
@@ -54,19 +57,19 @@ 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>();
-
+
public static class ServerAddress {
public final TServer server;
public final HostAndPort address;
-
+
public ServerAddress(TServer server, HostAndPort address) {
this.server = server;
this.address = address;
}
}
-
+
/**
* Start a server, at the given port, or higher, if that port is not available.
*
@@ -81,6 +84,7 @@ public class TServerUtils {
* @param portSearchProperty
* @param minThreadProperty
* @param timeBetweenThreadChecksProperty
+ * @param generalSslEnabled
* @return the server object created, and the port actually used
* @throws UnknownHostException
* when we don't know our own address
@@ -105,12 +109,12 @@ public class TServerUtils {
TServerUtils.TimedProcessor timedProcessor = new TServerUtils.TimedProcessor(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)
@@ -119,22 +123,34 @@ public class TServerUtils {
port = 1024 + port % (65535 - 1024);
try {
HostAndPort addr = HostAndPort.fromParts(address, port);
- return TServerUtils.startTServer(addr, timedProcessor, serverName, threadName, minThreads, timeBetweenThreadChecks, maxMessageSize);
- } catch (Exception ex) {
- log.info("Unable to use port " + port + ", retrying. (Thread Name = " + threadName + ")");
- UtilWaitThread.sleep(250);
+ return TServerUtils.startTServer(addr, timedProcessor, serverName, threadName, minThreads, timeBetweenThreadChecks, maxMessageSize,
+ SslConnectionParams.forServer(conf), conf.getTimeInMillis(Property.GENERAL_RPC_TIMEOUT));
+ } 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");
}
-
+
public static class TimedProcessor implements TProcessor {
-
+
final TProcessor other;
ThriftMetrics metrics = null;
long idleStart = 0;
-
+
TimedProcessor(TProcessor next, String serverName, String threadName) {
this.other = next;
// Register the metrics MBean
@@ -146,7 +162,7 @@ public class TServerUtils {
}
idleStart = System.currentTimeMillis();
}
-
+
@Override
public boolean process(TProtocol in, TProtocol out) throws TException {
long now = 0;
@@ -169,41 +185,46 @@ public class TServerUtils {
}
}
}
-
+
public static class ClientInfoProcessorFactory extends TProcessorFactory {
-
+
public ClientInfoProcessorFactory(TProcessor processor) {
super(processor);
}
-
+
@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);
}
}
-
+
public static class THsHaServer extends org.apache.thrift.server.THsHaServer {
public THsHaServer(Args args) {
super(args);
}
-
+
@Override
protected Runnable getRunnable(FrameBuffer frameBuffer) {
return new Invocation(frameBuffer);
}
-
+
private class Invocation implements Runnable {
-
+
private final FrameBuffer frameBuffer;
-
+
public Invocation(final FrameBuffer frameBuffer) {
this.frameBuffer = frameBuffer;
}
-
+
@Override
public void run() {
if (frameBuffer.trans_ instanceof TNonblockingSocket) {
@@ -215,24 +236,10 @@ public class TServerUtils {
}
}
}
-
- public static ServerAddress startHsHaServer(HostAndPort address, TProcessor processor, final String serverName, String threadName, final int numThreads,
+
+ public static ServerAddress createHsHaServer(HostAndPort address, TProcessor processor, final String serverName, String threadName, final int numThreads,
long timeBetweenThreadChecks, long maxMessageSize) throws TTransportException {
TNonblockingServerSocket transport = new TNonblockingServerSocket(new InetSocketAddress(address.getHostText(), address.getPort()));
- // check for the special "bind to everything address"
- String hostname = address.getHostText();
- if (hostname.equals("0.0.0.0")) {
- // can't get the address from the bind, so we'll do our best to invent our hostname
- try {
- hostname = InetAddress.getLocalHost().getHostName();
- } catch (UnknownHostException e) {
- throw new TTransportException(e);
- }
- }
- int port = address.getPort();
- if (port == 0) {
- port = transport.getPort();
- }
THsHaServer.Args options = new THsHaServer.Args(transport);
options.protocolFactory(ThriftUtil.protocolFactory());
options.transportFactory(ThriftUtil.transportFactory(maxMessageSize));
@@ -267,12 +274,15 @@ public class TServerUtils {
}, timeBetweenThreadChecks, timeBetweenThreadChecks);
options.executorService(pool);
options.processorFactory(new TProcessorFactory(processor));
- return new ServerAddress(new THsHaServer(options), HostAndPort.fromParts(hostname, port));
+ if (address.getPort() == 0) {
+ address = HostAndPort.fromParts(address.getHostText(), transport.getPort());
+ }
+ return new ServerAddress(new THsHaServer(options), address);
}
-
- public static ServerAddress startThreadPoolServer(HostAndPort address, TProcessor processor, String serverName, String threadName, int numThreads)
+
+ public static ServerAddress createThreadPoolServer(HostAndPort address, TProcessor processor, String serverName, String threadName, int numThreads)
throws TTransportException {
-
+
// if port is zero, then we must bind to get the port number
ServerSocket sock;
try {
@@ -284,23 +294,52 @@ public class TServerUtils {
throw new TTransportException(ex);
}
TServerTransport transport = new TBufferedServerSocket(sock, 32 * 1024);
+ return new ServerAddress(createThreadPoolServer(transport, processor), 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(processor));
- return new ServerAddress(new TThreadPoolServer(options), address);
+ return new TThreadPoolServer(options);
}
-
- public static ServerAddress startTServer(HostAndPort address, TProcessor processor, String serverName, String threadName, int numThreads, long timeBetweenThreadChecks, long maxMessageSize)
+
+ public static ServerAddress createSslThreadPoolServer(HostAndPort address, TProcessor processor, long socketTimeout, SslConnectionParams sslParams)
throws TTransportException {
- return startTServer(address, new TimedProcessor(processor, serverName, threadName), serverName, threadName, numThreads, timeBetweenThreadChecks, maxMessageSize);
+ 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(HostAndPort address, TimedProcessor processor, String serverName, String threadName, int numThreads, long timeBetweenThreadChecks, long maxMessageSize)
- throws TTransportException {
- ServerAddress result = startHsHaServer(address, processor, serverName, threadName, numThreads, timeBetweenThreadChecks, maxMessageSize);
- //ServerAddress result = startThreadPoolServer(address, processor, serverName, threadName, -1);
- final TServer finalServer = result.server;
+
+ public static ServerAddress startTServer(HostAndPort address, TProcessor processor, String serverName, String threadName, int numThreads,
+ long timeBetweenThreadChecks, long maxMessageSize) throws TTransportException {
+ return startTServer(address, processor, serverName, threadName, numThreads, timeBetweenThreadChecks, maxMessageSize, null, -1);
+ }
+
+ public static ServerAddress startTServer(HostAndPort address, TProcessor processor, String serverName, String threadName, int numThreads,
+ long timeBetweenThreadChecks, long maxMessageSize, SslConnectionParams sslParams, long sslSocketTimeout) throws TTransportException {
+ return startTServer(address, new TimedProcessor(processor, serverName, threadName), serverName, threadName, numThreads, timeBetweenThreadChecks,
+ maxMessageSize, sslParams, sslSocketTimeout);
+ }
+
+ public static ServerAddress startTServer(HostAndPort address, TimedProcessor processor, String serverName, String threadName, int numThreads,
+ long timeBetweenThreadChecks, long maxMessageSize, SslConnectionParams sslParams, long sslSocketTimeout) throws TTransportException {
+
+ ServerAddress serverAddress;
+ if (sslParams != null) {
+ serverAddress = createSslThreadPoolServer(address, processor, sslSocketTimeout, sslParams);
+ } else {
+ serverAddress = createHsHaServer(address, processor, serverName, threadName, numThreads, timeBetweenThreadChecks, maxMessageSize);
+ }
+ final TServer finalServer = serverAddress.server;
Runnable serveTask = new Runnable() {
@Override
public void run() {
@@ -314,9 +353,18 @@ public class TServerUtils {
serveTask = new LoggingRunnable(TServerUtils.log, serveTask);
Thread thread = new Daemon(serveTask, threadName);
thread.start();
- return result;
+ // 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)
http://git-wip-us.apache.org/repos/asf/accumulo/blob/7038755b/server/extras/src/main/java/org/apache/accumulo/utils/metanalysis/IndexMeta.java
----------------------------------------------------------------------
diff --git a/server/extras/src/main/java/org/apache/accumulo/utils/metanalysis/IndexMeta.java b/server/extras/src/main/java/org/apache/accumulo/utils/metanalysis/IndexMeta.java
index 8e7b221..5b85f18 100644
--- a/server/extras/src/main/java/org/apache/accumulo/utils/metanalysis/IndexMeta.java
+++ b/server/extras/src/main/java/org/apache/accumulo/utils/metanalysis/IndexMeta.java
@@ -26,6 +26,7 @@ import java.util.Map;
import org.apache.accumulo.core.client.Connector;
import org.apache.accumulo.core.client.TableExistsException;
import org.apache.accumulo.core.client.mapreduce.AccumuloOutputFormat;
+import org.apache.accumulo.core.conf.ClientConfiguration;
import org.apache.accumulo.core.data.ColumnUpdate;
import org.apache.accumulo.core.data.KeyExtent;
import org.apache.accumulo.core.data.Mutation;
@@ -120,6 +121,10 @@ public class IndexMeta extends Configured implements Tool {
static class Opts extends ClientOpts {
@Parameter(description = "<logfile> { <logfile> ...}")
List<String> logFiles = new ArrayList<String>();
+
+ public ClientConfiguration getConf() {
+ return this.getClientConfiguration();
+ }
}
@Override
@@ -146,7 +151,7 @@ public class IndexMeta extends Configured implements Tool {
job.setNumReduceTasks(0);
job.setOutputFormatClass(AccumuloOutputFormat.class);
- AccumuloOutputFormat.setZooKeeperInstance(job, opts.instance, opts.zookeepers);
+ AccumuloOutputFormat.setZooKeeperInstance(job, opts.getConf());
AccumuloOutputFormat.setConnectorInfo(job, opts.principal, opts.getToken());
AccumuloOutputFormat.setCreateTables(job, false);
http://git-wip-us.apache.org/repos/asf/accumulo/blob/7038755b/test/pom.xml
----------------------------------------------------------------------
diff --git a/test/pom.xml b/test/pom.xml
index 54b8733..3e2f033 100644
--- a/test/pom.xml
+++ b/test/pom.xml
@@ -25,6 +25,9 @@
<artifactId>accumulo-test</artifactId>
<name>Testing</name>
<description>Tests for Apache Accumulo.</description>
+ <properties>
+ <bouncycastle.version>1.49</bouncycastle.version>
+ </properties>
<dependencies>
<dependency>
<groupId>com.beust</groupId>
@@ -146,6 +149,18 @@
<scope>test</scope>
</dependency>
<dependency>
+ <groupId>org.bouncycastle</groupId>
+ <artifactId>bcpkix-jdk15on</artifactId>
+ <version>${bouncycastle.version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.bouncycastle</groupId>
+ <artifactId>bcprov-jdk15on</artifactId>
+ <version>${bouncycastle.version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
<groupId>org.mortbay.jetty</groupId>
<artifactId>jetty</artifactId>
<scope>test</scope>
@@ -223,6 +238,18 @@
</execution>
</executions>
</plugin>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-failsafe-plugin</artifactId>
+ <configuration>
+ <systemProperties>
+ <property>
+ <name>org.apache.accumulo.test.functional.useSslForIT</name>
+ <value>${useSslForIT}</value>
+ </property>
+ </systemProperties>
+ </configuration>
+ </plugin>
</plugins>
</build>
</profile>
http://git-wip-us.apache.org/repos/asf/accumulo/blob/7038755b/test/src/main/java/org/apache/accumulo/test/IMMLGBenchmark.java
----------------------------------------------------------------------
diff --git a/test/src/main/java/org/apache/accumulo/test/IMMLGBenchmark.java b/test/src/main/java/org/apache/accumulo/test/IMMLGBenchmark.java
index 948a741..7bee351 100644
--- a/test/src/main/java/org/apache/accumulo/test/IMMLGBenchmark.java
+++ b/test/src/main/java/org/apache/accumulo/test/IMMLGBenchmark.java
@@ -36,6 +36,7 @@ import org.apache.accumulo.core.client.Scanner;
import org.apache.accumulo.core.client.TableNotFoundException;
import org.apache.accumulo.core.client.ZooKeeperInstance;
import org.apache.accumulo.core.client.security.tokens.PasswordToken;
+import org.apache.accumulo.core.conf.ClientConfiguration;
import org.apache.accumulo.core.conf.Property;
import org.apache.accumulo.core.data.Key;
import org.apache.accumulo.core.data.Mutation;
@@ -51,7 +52,7 @@ import org.apache.hadoop.io.Text;
*/
public class IMMLGBenchmark {
public static void main(String[] args) throws Exception {
- ZooKeeperInstance zki = new ZooKeeperInstance("test16", "localhost");
+ ZooKeeperInstance zki = new ZooKeeperInstance(new ClientConfiguration().withInstance("test16").withZkHosts("localhost"));
Connector conn = zki.getConnector("root", new PasswordToken("secret"));
int numlg = Integer.parseInt(args[0]);
http://git-wip-us.apache.org/repos/asf/accumulo/blob/7038755b/test/src/main/java/org/apache/accumulo/test/TestIngest.java
----------------------------------------------------------------------
diff --git a/test/src/main/java/org/apache/accumulo/test/TestIngest.java b/test/src/main/java/org/apache/accumulo/test/TestIngest.java
index 972a20e..1efd872 100644
--- a/test/src/main/java/org/apache/accumulo/test/TestIngest.java
+++ b/test/src/main/java/org/apache/accumulo/test/TestIngest.java
@@ -49,7 +49,6 @@ import org.apache.accumulo.core.util.CachedConfiguration;
import org.apache.accumulo.core.util.FastFormat;
import org.apache.accumulo.fate.zookeeper.ZooReader;
import org.apache.accumulo.server.cli.ClientOnDefaultTable;
-import org.apache.accumulo.server.conf.ServerConfiguration;
import org.apache.accumulo.trace.instrument.Trace;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
@@ -180,8 +179,6 @@ public class TestIngest {
Opts opts = new Opts();
BatchWriterOpts bwOpts = new BatchWriterOpts();
opts.parseArgs(TestIngest.class.getName(), args, bwOpts);
- opts.getInstance().setConfiguration(ServerConfiguration.getSiteConfiguration());
-
Instance instance = opts.getInstance();
http://git-wip-us.apache.org/repos/asf/accumulo/blob/7038755b/test/src/main/java/org/apache/accumulo/test/performance/metadata/MetadataBatchScanTest.java
----------------------------------------------------------------------
diff --git a/test/src/main/java/org/apache/accumulo/test/performance/metadata/MetadataBatchScanTest.java b/test/src/main/java/org/apache/accumulo/test/performance/metadata/MetadataBatchScanTest.java
index c6ad74f..cf4f134 100644
--- a/test/src/main/java/org/apache/accumulo/test/performance/metadata/MetadataBatchScanTest.java
+++ b/test/src/main/java/org/apache/accumulo/test/performance/metadata/MetadataBatchScanTest.java
@@ -30,6 +30,7 @@ import org.apache.accumulo.core.client.BatchWriterConfig;
import org.apache.accumulo.core.client.Connector;
import org.apache.accumulo.core.client.Scanner;
import org.apache.accumulo.core.client.ZooKeeperInstance;
+import org.apache.accumulo.core.conf.ClientConfiguration;
import org.apache.accumulo.core.data.Key;
import org.apache.accumulo.core.data.KeyExtent;
import org.apache.accumulo.core.data.Mutation;
@@ -57,7 +58,7 @@ public class MetadataBatchScanTest {
public static void main(String[] args) throws Exception {
- final Connector connector = new ZooKeeperInstance("acu14", "localhost").getConnector(SystemCredentials.get().getPrincipal(), SystemCredentials.get()
+ final Connector connector = new ZooKeeperInstance(new ClientConfiguration().withInstance("acu14").withZkHosts("localhost")).getConnector(SystemCredentials.get().getPrincipal(), SystemCredentials.get()
.getToken());
TreeSet<Long> splits = new TreeSet<Long>();
http://git-wip-us.apache.org/repos/asf/accumulo/blob/7038755b/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 78b2564..05384d7 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
@@ -29,6 +29,7 @@ import org.apache.accumulo.core.client.ZooKeeperInstance;
import org.apache.accumulo.core.client.impl.Tables;
import org.apache.accumulo.core.client.impl.thrift.SecurityErrorCode;
import org.apache.accumulo.core.client.impl.thrift.ThriftSecurityException;
+import org.apache.accumulo.core.conf.ClientConfiguration;
import org.apache.accumulo.core.conf.DefaultConfiguration;
import org.apache.accumulo.core.conf.Property;
import org.apache.accumulo.core.data.KeyExtent;
@@ -247,7 +248,7 @@ public class NullTserver {
HostAndPort addr = HostAndPort.fromParts(InetAddress.getLocalHost().getHostName(), opts.port);
// modify !METADATA
- ZooKeeperInstance zki = new ZooKeeperInstance(opts.iname, opts.keepers);
+ ZooKeeperInstance zki = new ZooKeeperInstance(new ClientConfiguration().withInstance(opts.iname).withZkHosts(opts.keepers));
String tableId = Tables.getTableId(zki, opts.tableName);
// read the locations for the table
http://git-wip-us.apache.org/repos/asf/accumulo/blob/7038755b/test/src/main/java/org/apache/accumulo/test/randomwalk/State.java
----------------------------------------------------------------------
diff --git a/test/src/main/java/org/apache/accumulo/test/randomwalk/State.java b/test/src/main/java/org/apache/accumulo/test/randomwalk/State.java
index ea6d8ed..5227b2a 100644
--- a/test/src/main/java/org/apache/accumulo/test/randomwalk/State.java
+++ b/test/src/main/java/org/apache/accumulo/test/randomwalk/State.java
@@ -31,6 +31,7 @@ import org.apache.accumulo.core.client.MultiTableBatchWriter;
import org.apache.accumulo.core.client.ZooKeeperInstance;
import org.apache.accumulo.core.client.security.tokens.AuthenticationToken;
import org.apache.accumulo.core.client.security.tokens.PasswordToken;
+import org.apache.accumulo.core.conf.ClientConfiguration;
import org.apache.accumulo.core.security.Credentials;
import org.apache.log4j.Logger;
@@ -124,7 +125,7 @@ public class State {
if (instance == null) {
String instance = props.getProperty("INSTANCE");
String zookeepers = props.getProperty("ZOOKEEPERS");
- this.instance = new ZooKeeperInstance(instance, zookeepers);
+ this.instance = new ZooKeeperInstance(new ClientConfiguration().withInstance(instance).withZkHosts(zookeepers));
}
return instance;
}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/7038755b/test/src/main/java/org/apache/accumulo/test/randomwalk/multitable/CopyTool.java
----------------------------------------------------------------------
diff --git a/test/src/main/java/org/apache/accumulo/test/randomwalk/multitable/CopyTool.java b/test/src/main/java/org/apache/accumulo/test/randomwalk/multitable/CopyTool.java
index e104c99..749209e 100644
--- a/test/src/main/java/org/apache/accumulo/test/randomwalk/multitable/CopyTool.java
+++ b/test/src/main/java/org/apache/accumulo/test/randomwalk/multitable/CopyTool.java
@@ -21,6 +21,7 @@ import java.io.IOException;
import org.apache.accumulo.core.client.mapreduce.AccumuloInputFormat;
import org.apache.accumulo.core.client.mapreduce.AccumuloOutputFormat;
import org.apache.accumulo.core.client.security.tokens.PasswordToken;
+import org.apache.accumulo.core.conf.ClientConfiguration;
import org.apache.accumulo.core.data.Key;
import org.apache.accumulo.core.data.Mutation;
import org.apache.accumulo.core.data.Value;
@@ -55,11 +56,13 @@ public class CopyTool extends Configured implements Tool {
return 1;
}
+ ClientConfiguration clientConf = new ClientConfiguration().withInstance(args[3]).withZkHosts(args[4]);
+
job.setInputFormatClass(AccumuloInputFormat.class);
AccumuloInputFormat.setConnectorInfo(job, args[0], new PasswordToken(args[1]));
AccumuloInputFormat.setInputTableName(job, args[2]);
AccumuloInputFormat.setScanAuthorizations(job, Authorizations.EMPTY);
- AccumuloInputFormat.setZooKeeperInstance(job, args[3], args[4]);
+ AccumuloInputFormat.setZooKeeperInstance(job, clientConf);
job.setMapperClass(SeqMapClass.class);
job.setMapOutputKeyClass(Text.class);
@@ -71,7 +74,7 @@ public class CopyTool extends Configured implements Tool {
AccumuloOutputFormat.setConnectorInfo(job, args[0], new PasswordToken(args[1]));
AccumuloOutputFormat.setCreateTables(job, true);
AccumuloOutputFormat.setDefaultTableName(job, args[5]);
- AccumuloOutputFormat.setZooKeeperInstance(job, args[3], args[4]);
+ AccumuloOutputFormat.setZooKeeperInstance(job, clientConf);
job.waitForCompletion(true);
return job.isSuccessful() ? 0 : 1;
http://git-wip-us.apache.org/repos/asf/accumulo/blob/7038755b/test/src/main/java/org/apache/accumulo/test/randomwalk/sequential/MapRedVerifyTool.java
----------------------------------------------------------------------
diff --git a/test/src/main/java/org/apache/accumulo/test/randomwalk/sequential/MapRedVerifyTool.java b/test/src/main/java/org/apache/accumulo/test/randomwalk/sequential/MapRedVerifyTool.java
index 6c7cc63..b0c5029 100644
--- a/test/src/main/java/org/apache/accumulo/test/randomwalk/sequential/MapRedVerifyTool.java
+++ b/test/src/main/java/org/apache/accumulo/test/randomwalk/sequential/MapRedVerifyTool.java
@@ -22,6 +22,7 @@ import java.util.Iterator;
import org.apache.accumulo.core.client.mapreduce.AccumuloInputFormat;
import org.apache.accumulo.core.client.mapreduce.AccumuloOutputFormat;
import org.apache.accumulo.core.client.security.tokens.PasswordToken;
+import org.apache.accumulo.core.conf.ClientConfiguration;
import org.apache.accumulo.core.data.Key;
import org.apache.accumulo.core.data.Mutation;
import org.apache.accumulo.core.data.Value;
@@ -86,10 +87,12 @@ public class MapRedVerifyTool extends Configured implements Tool {
return 1;
}
+ ClientConfiguration clientConf = new ClientConfiguration().withInstance(args[3]).withZkHosts(args[4]);
+
job.setInputFormatClass(AccumuloInputFormat.class);
AccumuloInputFormat.setConnectorInfo(job, args[0], new PasswordToken(args[1]));
AccumuloInputFormat.setInputTableName(job, args[2]);
- AccumuloInputFormat.setZooKeeperInstance(job, args[3], args[4]);
+ AccumuloInputFormat.setZooKeeperInstance(job, clientConf);
job.setMapperClass(SeqMapClass.class);
job.setMapOutputKeyClass(NullWritable.class);
@@ -102,7 +105,7 @@ public class MapRedVerifyTool extends Configured implements Tool {
AccumuloOutputFormat.setConnectorInfo(job, args[0], new PasswordToken(args[1]));
AccumuloOutputFormat.setCreateTables(job, true);
AccumuloOutputFormat.setDefaultTableName(job, args[5]);
- AccumuloOutputFormat.setZooKeeperInstance(job, args[3], args[4]);
+ AccumuloOutputFormat.setZooKeeperInstance(job, clientConf);
job.waitForCompletion(true);
return job.isSuccessful() ? 0 : 1;
http://git-wip-us.apache.org/repos/asf/accumulo/blob/7038755b/test/src/main/java/org/apache/accumulo/test/scalability/ScaleTest.java
----------------------------------------------------------------------
diff --git a/test/src/main/java/org/apache/accumulo/test/scalability/ScaleTest.java b/test/src/main/java/org/apache/accumulo/test/scalability/ScaleTest.java
index e6ba77b..c20d004 100644
--- a/test/src/main/java/org/apache/accumulo/test/scalability/ScaleTest.java
+++ b/test/src/main/java/org/apache/accumulo/test/scalability/ScaleTest.java
@@ -24,6 +24,7 @@ import org.apache.accumulo.core.client.AccumuloSecurityException;
import org.apache.accumulo.core.client.Connector;
import org.apache.accumulo.core.client.ZooKeeperInstance;
import org.apache.accumulo.core.client.security.tokens.PasswordToken;
+import org.apache.accumulo.core.conf.ClientConfiguration;
import org.apache.hadoop.io.Text;
public abstract class ScaleTest {
@@ -47,7 +48,7 @@ public abstract class ScaleTest {
String password = this.scaleProps.getProperty("PASSWORD");
System.out.println(password);
- conn = new ZooKeeperInstance(instanceName, zookeepers).getConnector(user, new PasswordToken(password));
+ conn = new ZooKeeperInstance(new ClientConfiguration().withInstance(instanceName).withZkHosts(zookeepers)).getConnector(user, new PasswordToken(password));
}
protected void startTimer() {