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 2015/02/13 04:28:10 UTC
[1/9] accumulo git commit: ACCUMULO-3574 Prevent cached connection
being returned when not requested.
Repository: accumulo
Updated Branches:
refs/heads/1.5 e5086f908 -> bd981c886
refs/heads/1.6 94b12f1e3 -> 1213ee2be
refs/heads/master b8a1b3106 -> 7ae2e5afd
ACCUMULO-3574 Prevent cached connection being returned when not requested.
Project: http://git-wip-us.apache.org/repos/asf/accumulo/repo
Commit: http://git-wip-us.apache.org/repos/asf/accumulo/commit/bd981c88
Tree: http://git-wip-us.apache.org/repos/asf/accumulo/tree/bd981c88
Diff: http://git-wip-us.apache.org/repos/asf/accumulo/diff/bd981c88
Branch: refs/heads/1.5
Commit: bd981c88654458bba41e754d861173533560294e
Parents: e5086f9
Author: Josh Elser <el...@apache.org>
Authored: Thu Feb 12 19:16:47 2015 -0500
Committer: Josh Elser <el...@apache.org>
Committed: Thu Feb 12 19:16:47 2015 -0500
----------------------------------------------------------------------
.../org/apache/accumulo/core/client/impl/ThriftTransportPool.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/accumulo/blob/bd981c88/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 a6e3a8c..4914261 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
@@ -440,7 +440,7 @@ public class ThriftTransportPool {
int index = random.nextInt(servers.size());
ThriftTransportKey ttk = servers.get(index);
- if (!preferCachedConnection) {
+ if (preferCachedConnection) {
synchronized (this) {
List<CachedConnection> cachedConnList = getCache().get(ttk);
if (cachedConnList != null) {
[3/9] accumulo git commit: ACCUMULO-3574 Prevent cached connection
being returned when not requested.
Posted by el...@apache.org.
ACCUMULO-3574 Prevent cached connection being returned when not requested.
Project: http://git-wip-us.apache.org/repos/asf/accumulo/repo
Commit: http://git-wip-us.apache.org/repos/asf/accumulo/commit/bd981c88
Tree: http://git-wip-us.apache.org/repos/asf/accumulo/tree/bd981c88
Diff: http://git-wip-us.apache.org/repos/asf/accumulo/diff/bd981c88
Branch: refs/heads/master
Commit: bd981c88654458bba41e754d861173533560294e
Parents: e5086f9
Author: Josh Elser <el...@apache.org>
Authored: Thu Feb 12 19:16:47 2015 -0500
Committer: Josh Elser <el...@apache.org>
Committed: Thu Feb 12 19:16:47 2015 -0500
----------------------------------------------------------------------
.../org/apache/accumulo/core/client/impl/ThriftTransportPool.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/accumulo/blob/bd981c88/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 a6e3a8c..4914261 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
@@ -440,7 +440,7 @@ public class ThriftTransportPool {
int index = random.nextInt(servers.size());
ThriftTransportKey ttk = servers.get(index);
- if (!preferCachedConnection) {
+ if (preferCachedConnection) {
synchronized (this) {
List<CachedConnection> cachedConnList = getCache().get(ttk);
if (cachedConnList != null) {
[7/9] accumulo git commit: ACCUMULO-3574 Add an IT for testing
transport caching.
Posted by el...@apache.org.
ACCUMULO-3574 Add an IT for testing transport caching.
Project: http://git-wip-us.apache.org/repos/asf/accumulo/repo
Commit: http://git-wip-us.apache.org/repos/asf/accumulo/commit/1213ee2b
Tree: http://git-wip-us.apache.org/repos/asf/accumulo/tree/1213ee2b
Diff: http://git-wip-us.apache.org/repos/asf/accumulo/diff/1213ee2b
Branch: refs/heads/1.6
Commit: 1213ee2beaa4ac970c7bb6ba84355bae049cb4c4
Parents: 6111043
Author: Josh Elser <el...@apache.org>
Authored: Thu Feb 12 19:18:36 2015 -0500
Committer: Josh Elser <el...@apache.org>
Committed: Thu Feb 12 19:18:36 2015 -0500
----------------------------------------------------------------------
.../core/client/impl/ThriftTransportKey.java | 8 +-
.../core/client/impl/ThriftTransportPool.java | 4 +-
.../accumulo/test/TransportCachingIT.java | 115 +++++++++++++++++++
3 files changed, 124 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/accumulo/blob/1213ee2b/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 de33941..8e3ee47 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
@@ -19,7 +19,10 @@ package org.apache.accumulo.core.client.impl;
import org.apache.accumulo.core.util.ArgumentChecker;
import org.apache.accumulo.core.util.SslConnectionParams;
-class ThriftTransportKey {
+import com.google.common.annotations.VisibleForTesting;
+
+@VisibleForTesting
+public class ThriftTransportKey {
private final String location;
private final int port;
private final long timeout;
@@ -27,7 +30,8 @@ class ThriftTransportKey {
private int hash = -1;
- ThriftTransportKey(String location, long timeout, SslConnectionParams sslParams) {
+ @VisibleForTesting
+ public ThriftTransportKey(String location, long timeout, SslConnectionParams sslParams) {
ArgumentChecker.notNull(location);
String[] locationAndPort = location.split(":", 2);
if (locationAndPort.length == 2) {
http://git-wip-us.apache.org/repos/asf/accumulo/blob/1213ee2b/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 575a537..33997e0 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
@@ -41,6 +41,7 @@ import org.apache.log4j.Logger;
import org.apache.thrift.transport.TTransport;
import org.apache.thrift.transport.TTransportException;
+import com.google.common.annotations.VisibleForTesting;
import com.google.common.net.HostAndPort;
public class ThriftTransportPool {
@@ -420,7 +421,8 @@ public class ThriftTransportPool {
return createNewTransport(cacheKey);
}
- Pair<String,TTransport> getAnyTransport(List<ThriftTransportKey> servers, boolean preferCachedConnection) throws TTransportException {
+ @VisibleForTesting
+ public Pair<String,TTransport> getAnyTransport(List<ThriftTransportKey> servers, boolean preferCachedConnection) throws TTransportException {
servers = new ArrayList<ThriftTransportKey>(servers);
http://git-wip-us.apache.org/repos/asf/accumulo/blob/1213ee2b/test/src/test/java/org/apache/accumulo/test/TransportCachingIT.java
----------------------------------------------------------------------
diff --git a/test/src/test/java/org/apache/accumulo/test/TransportCachingIT.java b/test/src/test/java/org/apache/accumulo/test/TransportCachingIT.java
new file mode 100644
index 0000000..ddbd3e8
--- /dev/null
+++ b/test/src/test/java/org/apache/accumulo/test/TransportCachingIT.java
@@ -0,0 +1,115 @@
+/*
+ * 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.test;
+
+import static com.google.common.base.Charsets.UTF_8;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+
+import java.util.ArrayList;
+
+import org.apache.accumulo.core.Constants;
+import org.apache.accumulo.core.client.Connector;
+import org.apache.accumulo.core.client.Instance;
+import org.apache.accumulo.core.client.impl.ServerConfigurationUtil;
+import org.apache.accumulo.core.client.impl.ThriftTransportKey;
+import org.apache.accumulo.core.client.impl.ThriftTransportPool;
+import org.apache.accumulo.core.conf.DefaultConfiguration;
+import org.apache.accumulo.core.conf.Property;
+import org.apache.accumulo.core.util.ServerServices;
+import org.apache.accumulo.core.util.ServerServices.Service;
+import org.apache.accumulo.core.util.SslConnectionParams;
+import org.apache.accumulo.core.zookeeper.ZooUtil;
+import org.apache.accumulo.fate.zookeeper.ZooCache;
+import org.apache.accumulo.fate.zookeeper.ZooCacheFactory;
+import org.apache.accumulo.harness.AccumuloClusterIT;
+import org.apache.thrift.transport.TTransport;
+import org.apache.thrift.transport.TTransportException;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Test that {@link ThriftTransportPool} actually adheres to the cachedConnection argument
+ */
+public class TransportCachingIT extends AccumuloClusterIT {
+ private static final Logger log = LoggerFactory.getLogger(TransportCachingIT.class);
+
+ @Test
+ public void testCachedTransport() {
+ Connector conn = getConnector();
+ Instance instance = conn.getInstance();
+ long rpcTimeout = DefaultConfiguration.getTimeInMillis(Property.GENERAL_RPC_TIMEOUT.getDefaultValue());
+
+ // create list of servers
+ ArrayList<ThriftTransportKey> servers = new ArrayList<ThriftTransportKey>();
+
+ // add tservers
+ ZooCache zc = new ZooCacheFactory().getZooCache(instance.getZooKeepers(), instance.getZooKeepersSessionTimeOut());
+ for (String tserver : zc.getChildren(ZooUtil.getRoot(instance) + Constants.ZTSERVERS)) {
+ String path = ZooUtil.getRoot(instance) + Constants.ZTSERVERS + "/" + tserver;
+ byte[] data = ZooUtil.getLockData(zc, path);
+ if (data != null && !new String(data, UTF_8).equals("master"))
+ servers.add(new ThriftTransportKey(new ServerServices(new String(data)).getAddressString(Service.TSERV_CLIENT), rpcTimeout, SslConnectionParams
+ .forClient(ServerConfigurationUtil.getConfiguration(instance))));
+ }
+
+ ThriftTransportPool pool = ThriftTransportPool.getInstance();
+ TTransport first = null;
+ while (null == first) {
+ try {
+ // Get a transport (cached or not)
+ first = pool.getAnyTransport(servers, true).getSecond();
+ } catch (TTransportException e) {
+ log.warn("Failed to obtain transport to " + servers);
+ }
+ }
+
+ assertNotNull(first);
+ // Return it to unreserve it
+ pool.returnTransport(first);
+
+ TTransport second = null;
+ while (null == second) {
+ try {
+ // Get a cached transport (should be the first)
+ second = pool.getAnyTransport(servers, true).getSecond();
+ } catch (TTransportException e) {
+ log.warn("Failed obtain 2nd transport to " + servers);
+ }
+ }
+
+ // We should get the same transport
+ assertTrue("Expected the first and second to be the same instance", first == second);
+ // Return the 2nd
+ pool.returnTransport(second);
+
+ TTransport third = null;
+ while (null == third) {
+ try {
+ // Get a non-cached transport
+ third = pool.getAnyTransport(servers, false).getSecond();
+ } catch (TTransportException e) {
+ log.warn("Failed obtain 2nd transport to " + servers);
+ }
+ }
+
+ assertFalse("Expected second and third transport to be different instances", second == third);
+ pool.returnTransport(third);
+ }
+}
[9/9] accumulo git commit: ACCUMULO-3574 Fix IT after the merge
Posted by el...@apache.org.
ACCUMULO-3574 Fix IT after the merge
Project: http://git-wip-us.apache.org/repos/asf/accumulo/repo
Commit: http://git-wip-us.apache.org/repos/asf/accumulo/commit/7ae2e5af
Tree: http://git-wip-us.apache.org/repos/asf/accumulo/tree/7ae2e5af
Diff: http://git-wip-us.apache.org/repos/asf/accumulo/diff/7ae2e5af
Branch: refs/heads/master
Commit: 7ae2e5afd44de1fbf38bb427cab2894af6c9b6be
Parents: 3d3bfe6
Author: Josh Elser <el...@apache.org>
Authored: Thu Feb 12 19:37:14 2015 -0500
Committer: Josh Elser <el...@apache.org>
Committed: Thu Feb 12 19:37:14 2015 -0500
----------------------------------------------------------------------
.../apache/accumulo/test/TransportCachingIT.java | 16 +++++++++++-----
1 file changed, 11 insertions(+), 5 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/accumulo/blob/7ae2e5af/test/src/test/java/org/apache/accumulo/test/TransportCachingIT.java
----------------------------------------------------------------------
diff --git a/test/src/test/java/org/apache/accumulo/test/TransportCachingIT.java b/test/src/test/java/org/apache/accumulo/test/TransportCachingIT.java
index ddbd3e8..e760295 100644
--- a/test/src/test/java/org/apache/accumulo/test/TransportCachingIT.java
+++ b/test/src/test/java/org/apache/accumulo/test/TransportCachingIT.java
@@ -24,16 +24,17 @@ import static org.junit.Assert.assertTrue;
import java.util.ArrayList;
import org.apache.accumulo.core.Constants;
+import org.apache.accumulo.core.client.ClientConfiguration;
import org.apache.accumulo.core.client.Connector;
import org.apache.accumulo.core.client.Instance;
-import org.apache.accumulo.core.client.impl.ServerConfigurationUtil;
+import org.apache.accumulo.core.client.impl.ClientContext;
import org.apache.accumulo.core.client.impl.ThriftTransportKey;
import org.apache.accumulo.core.client.impl.ThriftTransportPool;
import org.apache.accumulo.core.conf.DefaultConfiguration;
import org.apache.accumulo.core.conf.Property;
+import org.apache.accumulo.core.security.Credentials;
import org.apache.accumulo.core.util.ServerServices;
import org.apache.accumulo.core.util.ServerServices.Service;
-import org.apache.accumulo.core.util.SslConnectionParams;
import org.apache.accumulo.core.zookeeper.ZooUtil;
import org.apache.accumulo.fate.zookeeper.ZooCache;
import org.apache.accumulo.fate.zookeeper.ZooCacheFactory;
@@ -54,6 +55,9 @@ public class TransportCachingIT extends AccumuloClusterIT {
public void testCachedTransport() {
Connector conn = getConnector();
Instance instance = conn.getInstance();
+ ClientConfiguration clientConf = ClientConfiguration.loadDefault();
+ clientConf.withInstance(instance.getInstanceName()).withZkHosts(instance.getZooKeepers());
+ ClientContext context = new ClientContext(instance, new Credentials(getPrincipal(), getToken()), clientConf);
long rpcTimeout = DefaultConfiguration.getTimeInMillis(Property.GENERAL_RPC_TIMEOUT.getDefaultValue());
// create list of servers
@@ -64,9 +68,11 @@ public class TransportCachingIT extends AccumuloClusterIT {
for (String tserver : zc.getChildren(ZooUtil.getRoot(instance) + Constants.ZTSERVERS)) {
String path = ZooUtil.getRoot(instance) + Constants.ZTSERVERS + "/" + tserver;
byte[] data = ZooUtil.getLockData(zc, path);
- if (data != null && !new String(data, UTF_8).equals("master"))
- servers.add(new ThriftTransportKey(new ServerServices(new String(data)).getAddressString(Service.TSERV_CLIENT), rpcTimeout, SslConnectionParams
- .forClient(ServerConfigurationUtil.getConfiguration(instance))));
+ if (data != null) {
+ String strData = new String(data, UTF_8);
+ if (!strData.equals("master"))
+ servers.add(new ThriftTransportKey(new ServerServices(strData).getAddress(Service.TSERV_CLIENT), rpcTimeout, context));
+ }
}
ThriftTransportPool pool = ThriftTransportPool.getInstance();
[6/9] accumulo git commit: ACCUMULO-3574 Add an IT for testing
transport caching.
Posted by el...@apache.org.
ACCUMULO-3574 Add an IT for testing transport caching.
Project: http://git-wip-us.apache.org/repos/asf/accumulo/repo
Commit: http://git-wip-us.apache.org/repos/asf/accumulo/commit/1213ee2b
Tree: http://git-wip-us.apache.org/repos/asf/accumulo/tree/1213ee2b
Diff: http://git-wip-us.apache.org/repos/asf/accumulo/diff/1213ee2b
Branch: refs/heads/master
Commit: 1213ee2beaa4ac970c7bb6ba84355bae049cb4c4
Parents: 6111043
Author: Josh Elser <el...@apache.org>
Authored: Thu Feb 12 19:18:36 2015 -0500
Committer: Josh Elser <el...@apache.org>
Committed: Thu Feb 12 19:18:36 2015 -0500
----------------------------------------------------------------------
.../core/client/impl/ThriftTransportKey.java | 8 +-
.../core/client/impl/ThriftTransportPool.java | 4 +-
.../accumulo/test/TransportCachingIT.java | 115 +++++++++++++++++++
3 files changed, 124 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/accumulo/blob/1213ee2b/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 de33941..8e3ee47 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
@@ -19,7 +19,10 @@ package org.apache.accumulo.core.client.impl;
import org.apache.accumulo.core.util.ArgumentChecker;
import org.apache.accumulo.core.util.SslConnectionParams;
-class ThriftTransportKey {
+import com.google.common.annotations.VisibleForTesting;
+
+@VisibleForTesting
+public class ThriftTransportKey {
private final String location;
private final int port;
private final long timeout;
@@ -27,7 +30,8 @@ class ThriftTransportKey {
private int hash = -1;
- ThriftTransportKey(String location, long timeout, SslConnectionParams sslParams) {
+ @VisibleForTesting
+ public ThriftTransportKey(String location, long timeout, SslConnectionParams sslParams) {
ArgumentChecker.notNull(location);
String[] locationAndPort = location.split(":", 2);
if (locationAndPort.length == 2) {
http://git-wip-us.apache.org/repos/asf/accumulo/blob/1213ee2b/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 575a537..33997e0 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
@@ -41,6 +41,7 @@ import org.apache.log4j.Logger;
import org.apache.thrift.transport.TTransport;
import org.apache.thrift.transport.TTransportException;
+import com.google.common.annotations.VisibleForTesting;
import com.google.common.net.HostAndPort;
public class ThriftTransportPool {
@@ -420,7 +421,8 @@ public class ThriftTransportPool {
return createNewTransport(cacheKey);
}
- Pair<String,TTransport> getAnyTransport(List<ThriftTransportKey> servers, boolean preferCachedConnection) throws TTransportException {
+ @VisibleForTesting
+ public Pair<String,TTransport> getAnyTransport(List<ThriftTransportKey> servers, boolean preferCachedConnection) throws TTransportException {
servers = new ArrayList<ThriftTransportKey>(servers);
http://git-wip-us.apache.org/repos/asf/accumulo/blob/1213ee2b/test/src/test/java/org/apache/accumulo/test/TransportCachingIT.java
----------------------------------------------------------------------
diff --git a/test/src/test/java/org/apache/accumulo/test/TransportCachingIT.java b/test/src/test/java/org/apache/accumulo/test/TransportCachingIT.java
new file mode 100644
index 0000000..ddbd3e8
--- /dev/null
+++ b/test/src/test/java/org/apache/accumulo/test/TransportCachingIT.java
@@ -0,0 +1,115 @@
+/*
+ * 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.test;
+
+import static com.google.common.base.Charsets.UTF_8;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+
+import java.util.ArrayList;
+
+import org.apache.accumulo.core.Constants;
+import org.apache.accumulo.core.client.Connector;
+import org.apache.accumulo.core.client.Instance;
+import org.apache.accumulo.core.client.impl.ServerConfigurationUtil;
+import org.apache.accumulo.core.client.impl.ThriftTransportKey;
+import org.apache.accumulo.core.client.impl.ThriftTransportPool;
+import org.apache.accumulo.core.conf.DefaultConfiguration;
+import org.apache.accumulo.core.conf.Property;
+import org.apache.accumulo.core.util.ServerServices;
+import org.apache.accumulo.core.util.ServerServices.Service;
+import org.apache.accumulo.core.util.SslConnectionParams;
+import org.apache.accumulo.core.zookeeper.ZooUtil;
+import org.apache.accumulo.fate.zookeeper.ZooCache;
+import org.apache.accumulo.fate.zookeeper.ZooCacheFactory;
+import org.apache.accumulo.harness.AccumuloClusterIT;
+import org.apache.thrift.transport.TTransport;
+import org.apache.thrift.transport.TTransportException;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Test that {@link ThriftTransportPool} actually adheres to the cachedConnection argument
+ */
+public class TransportCachingIT extends AccumuloClusterIT {
+ private static final Logger log = LoggerFactory.getLogger(TransportCachingIT.class);
+
+ @Test
+ public void testCachedTransport() {
+ Connector conn = getConnector();
+ Instance instance = conn.getInstance();
+ long rpcTimeout = DefaultConfiguration.getTimeInMillis(Property.GENERAL_RPC_TIMEOUT.getDefaultValue());
+
+ // create list of servers
+ ArrayList<ThriftTransportKey> servers = new ArrayList<ThriftTransportKey>();
+
+ // add tservers
+ ZooCache zc = new ZooCacheFactory().getZooCache(instance.getZooKeepers(), instance.getZooKeepersSessionTimeOut());
+ for (String tserver : zc.getChildren(ZooUtil.getRoot(instance) + Constants.ZTSERVERS)) {
+ String path = ZooUtil.getRoot(instance) + Constants.ZTSERVERS + "/" + tserver;
+ byte[] data = ZooUtil.getLockData(zc, path);
+ if (data != null && !new String(data, UTF_8).equals("master"))
+ servers.add(new ThriftTransportKey(new ServerServices(new String(data)).getAddressString(Service.TSERV_CLIENT), rpcTimeout, SslConnectionParams
+ .forClient(ServerConfigurationUtil.getConfiguration(instance))));
+ }
+
+ ThriftTransportPool pool = ThriftTransportPool.getInstance();
+ TTransport first = null;
+ while (null == first) {
+ try {
+ // Get a transport (cached or not)
+ first = pool.getAnyTransport(servers, true).getSecond();
+ } catch (TTransportException e) {
+ log.warn("Failed to obtain transport to " + servers);
+ }
+ }
+
+ assertNotNull(first);
+ // Return it to unreserve it
+ pool.returnTransport(first);
+
+ TTransport second = null;
+ while (null == second) {
+ try {
+ // Get a cached transport (should be the first)
+ second = pool.getAnyTransport(servers, true).getSecond();
+ } catch (TTransportException e) {
+ log.warn("Failed obtain 2nd transport to " + servers);
+ }
+ }
+
+ // We should get the same transport
+ assertTrue("Expected the first and second to be the same instance", first == second);
+ // Return the 2nd
+ pool.returnTransport(second);
+
+ TTransport third = null;
+ while (null == third) {
+ try {
+ // Get a non-cached transport
+ third = pool.getAnyTransport(servers, false).getSecond();
+ } catch (TTransportException e) {
+ log.warn("Failed obtain 2nd transport to " + servers);
+ }
+ }
+
+ assertFalse("Expected second and third transport to be different instances", second == third);
+ pool.returnTransport(third);
+ }
+}
[8/9] accumulo git commit: Merge branch '1.6'
Posted by el...@apache.org.
Merge branch '1.6'
Conflicts:
core/src/main/java/org/apache/accumulo/core/client/impl/ThriftTransportKey.java
Project: http://git-wip-us.apache.org/repos/asf/accumulo/repo
Commit: http://git-wip-us.apache.org/repos/asf/accumulo/commit/3d3bfe68
Tree: http://git-wip-us.apache.org/repos/asf/accumulo/tree/3d3bfe68
Diff: http://git-wip-us.apache.org/repos/asf/accumulo/diff/3d3bfe68
Branch: refs/heads/master
Commit: 3d3bfe684efc20d7ef5b325ed6b6d369c887dd89
Parents: b8a1b31 1213ee2
Author: Josh Elser <el...@apache.org>
Authored: Thu Feb 12 19:21:35 2015 -0500
Committer: Josh Elser <el...@apache.org>
Committed: Thu Feb 12 19:21:35 2015 -0500
----------------------------------------------------------------------
.../core/client/impl/ThriftTransportKey.java | 7 +-
.../core/client/impl/ThriftTransportPool.java | 6 +-
.../accumulo/test/TransportCachingIT.java | 115 +++++++++++++++++++
3 files changed, 124 insertions(+), 4 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/accumulo/blob/3d3bfe68/core/src/main/java/org/apache/accumulo/core/client/impl/ThriftTransportKey.java
----------------------------------------------------------------------
diff --cc core/src/main/java/org/apache/accumulo/core/client/impl/ThriftTransportKey.java
index 072724b,8e3ee47..a843111
--- 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
@@@ -16,48 -16,40 +16,51 @@@
*/
package org.apache.accumulo.core.client.impl;
-import org.apache.accumulo.core.util.ArgumentChecker;
-import org.apache.accumulo.core.util.SslConnectionParams;
+import static com.google.common.base.Preconditions.checkNotNull;
+
+import org.apache.accumulo.core.rpc.SaslConnectionParams;
+import org.apache.accumulo.core.rpc.SslConnectionParams;
+ import com.google.common.annotations.VisibleForTesting;
+import com.google.common.net.HostAndPort;
- class ThriftTransportKey {
+ @VisibleForTesting
+ public class ThriftTransportKey {
- private final String location;
- private final int port;
+ private final HostAndPort server;
private final long timeout;
private final SslConnectionParams sslParams;
+ private final SaslConnectionParams saslParams;
private int hash = -1;
- ThriftTransportKey(HostAndPort server, long timeout, ClientContext context) {
+ @VisibleForTesting
- public ThriftTransportKey(String location, long timeout, SslConnectionParams sslParams) {
- ArgumentChecker.notNull(location);
- String[] locationAndPort = location.split(":", 2);
- if (locationAndPort.length == 2) {
- this.location = locationAndPort[0];
- this.port = Integer.parseInt(locationAndPort[1]);
- } else
- throw new IllegalArgumentException("Location was expected to contain port but did not. location=" + location);
-
++ public ThriftTransportKey(HostAndPort server, long timeout, ClientContext context) {
+ checkNotNull(server, "location is null");
+ this.server = server;
this.timeout = timeout;
- this.sslParams = sslParams;
+ this.sslParams = context.getClientSslParams();
+ this.saslParams = context.getClientSaslParams();
+ if (null != saslParams) {
+ // TSasl and TSSL transport factories don't play nicely together
+ if (null != sslParams) {
+ throw new RuntimeException("Cannot use both SSL and SASL thrift transports");
+ }
+ }
}
- String getLocation() {
- return location;
+ /**
+ * Visible only for testing
+ */
+ ThriftTransportKey(HostAndPort server, long timeout, SslConnectionParams sslParams, SaslConnectionParams saslParams) {
+ checkNotNull(server, "location is null");
+ this.server = server;
+ this.timeout = timeout;
+ this.sslParams = sslParams;
+ this.saslParams = saslParams;
}
- int getPort() {
- return port;
+ HostAndPort getServer() {
+ return server;
}
long getTimeout() {
http://git-wip-us.apache.org/repos/asf/accumulo/blob/3d3bfe68/core/src/main/java/org/apache/accumulo/core/client/impl/ThriftTransportPool.java
----------------------------------------------------------------------
diff --cc core/src/main/java/org/apache/accumulo/core/client/impl/ThriftTransportPool.java
index bc1cdbb,33997e0..730fd73
--- 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,14 -31,17 +31,15 @@@ import java.util.Set
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicBoolean;
-import org.apache.accumulo.core.conf.AccumuloConfiguration;
-import org.apache.accumulo.core.conf.Property;
+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.SslConnectionParams;
-import org.apache.accumulo.core.util.ThriftUtil;
-import org.apache.log4j.Logger;
import org.apache.thrift.transport.TTransport;
import org.apache.thrift.transport.TTransportException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+ import com.google.common.annotations.VisibleForTesting;
import com.google.common.net.HostAndPort;
public class ThriftTransportPool {
[4/9] accumulo git commit: Merge branch '1.5' into 1.6
Posted by el...@apache.org.
Merge branch '1.5' into 1.6
Project: http://git-wip-us.apache.org/repos/asf/accumulo/repo
Commit: http://git-wip-us.apache.org/repos/asf/accumulo/commit/6111043f
Tree: http://git-wip-us.apache.org/repos/asf/accumulo/tree/6111043f
Diff: http://git-wip-us.apache.org/repos/asf/accumulo/diff/6111043f
Branch: refs/heads/1.6
Commit: 6111043f3fb84ecd28dd91b7be6956eff11ce675
Parents: 94b12f1 bd981c8
Author: Josh Elser <el...@apache.org>
Authored: Thu Feb 12 19:18:30 2015 -0500
Committer: Josh Elser <el...@apache.org>
Committed: Thu Feb 12 19:18:30 2015 -0500
----------------------------------------------------------------------
.../org/apache/accumulo/core/client/impl/ThriftTransportPool.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/accumulo/blob/6111043f/core/src/main/java/org/apache/accumulo/core/client/impl/ThriftTransportPool.java
----------------------------------------------------------------------
[2/9] accumulo git commit: ACCUMULO-3574 Prevent cached connection
being returned when not requested.
Posted by el...@apache.org.
ACCUMULO-3574 Prevent cached connection being returned when not requested.
Project: http://git-wip-us.apache.org/repos/asf/accumulo/repo
Commit: http://git-wip-us.apache.org/repos/asf/accumulo/commit/bd981c88
Tree: http://git-wip-us.apache.org/repos/asf/accumulo/tree/bd981c88
Diff: http://git-wip-us.apache.org/repos/asf/accumulo/diff/bd981c88
Branch: refs/heads/1.6
Commit: bd981c88654458bba41e754d861173533560294e
Parents: e5086f9
Author: Josh Elser <el...@apache.org>
Authored: Thu Feb 12 19:16:47 2015 -0500
Committer: Josh Elser <el...@apache.org>
Committed: Thu Feb 12 19:16:47 2015 -0500
----------------------------------------------------------------------
.../org/apache/accumulo/core/client/impl/ThriftTransportPool.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/accumulo/blob/bd981c88/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 a6e3a8c..4914261 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
@@ -440,7 +440,7 @@ public class ThriftTransportPool {
int index = random.nextInt(servers.size());
ThriftTransportKey ttk = servers.get(index);
- if (!preferCachedConnection) {
+ if (preferCachedConnection) {
synchronized (this) {
List<CachedConnection> cachedConnList = getCache().get(ttk);
if (cachedConnList != null) {
[5/9] accumulo git commit: Merge branch '1.5' into 1.6
Posted by el...@apache.org.
Merge branch '1.5' into 1.6
Project: http://git-wip-us.apache.org/repos/asf/accumulo/repo
Commit: http://git-wip-us.apache.org/repos/asf/accumulo/commit/6111043f
Tree: http://git-wip-us.apache.org/repos/asf/accumulo/tree/6111043f
Diff: http://git-wip-us.apache.org/repos/asf/accumulo/diff/6111043f
Branch: refs/heads/master
Commit: 6111043f3fb84ecd28dd91b7be6956eff11ce675
Parents: 94b12f1 bd981c8
Author: Josh Elser <el...@apache.org>
Authored: Thu Feb 12 19:18:30 2015 -0500
Committer: Josh Elser <el...@apache.org>
Committed: Thu Feb 12 19:18:30 2015 -0500
----------------------------------------------------------------------
.../org/apache/accumulo/core/client/impl/ThriftTransportPool.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/accumulo/blob/6111043f/core/src/main/java/org/apache/accumulo/core/client/impl/ThriftTransportPool.java
----------------------------------------------------------------------