You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@accumulo.apache.org by ct...@apache.org on 2014/11/25 23:36:51 UTC
[07/10] accumulo git commit: ACCUMULO-3199 Internal refactor to add
ClientContext
http://git-wip-us.apache.org/repos/asf/accumulo/blob/42c25faa/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 21bf2b9..2181451 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
@@ -16,6 +16,11 @@
*/
package org.apache.accumulo.core.client.impl;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Arrays;
@@ -29,10 +34,9 @@ import java.util.Set;
import java.util.SortedMap;
import java.util.TreeMap;
-import junit.framework.TestCase;
-
import org.apache.accumulo.core.client.AccumuloException;
import org.apache.accumulo.core.client.AccumuloSecurityException;
+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.TabletLocator.TabletLocation;
@@ -53,12 +57,13 @@ import org.apache.accumulo.core.metadata.RootTable;
import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection;
import org.apache.accumulo.core.security.Credentials;
import org.apache.hadoop.io.Text;
+import org.junit.Before;
+import org.junit.Test;
-public class TabletLocatorImplTest extends TestCase {
+public class TabletLocatorImplTest {
private static final KeyExtent RTE = RootTable.EXTENT;
private static final KeyExtent MTE = new KeyExtent(new Text(MetadataTable.ID), null, RTE.getEndRow());
- private static Credentials credentials = null;
static KeyExtent nke(String t, String er, String per) {
return new KeyExtent(new Text(t), er == null ? null : new Text(er), per == null ? null : new Text(per));
@@ -136,9 +141,8 @@ public class TabletLocatorImplTest extends TestCase {
TreeMap<KeyExtent,TabletLocation> mcke = createMetaCacheKE(data);
TestTabletLocationObtainer ttlo = new TestTabletLocationObtainer(tservers);
- TestInstance testInstance = new TestInstance("instance1", "tserver1");
- RootTabletLocator rtl = new TestRootTabletLocator(testInstance);
+ RootTabletLocator rtl = new TestRootTabletLocator();
TabletLocatorImpl rootTabletCache = new TabletLocatorImpl(new Text(MetadataTable.ID), rtl, ttlo, new YesLockChecker());
TabletLocatorImpl tab1TabletCache = new TabletLocatorImpl(new Text(table), rootTabletCache, ttlo, tslc);
@@ -161,6 +165,15 @@ public class TabletLocatorImplTest extends TestCase {
return createLocators(tservers, "tserver1", "tserver2", table, data);
}
+ private TestInstance testInstance;
+ private ClientContext context;
+
+ @Before
+ public void setUp() {
+ testInstance = new TestInstance("instance1", "tserver1");
+ context = new ClientContext(testInstance, new Credentials("test", null), new ClientConfiguration());
+ }
+
private void runTest(Text tableName, List<Range> ranges, TabletLocatorImpl tab1TabletCache, Map<String,Map<KeyExtent,List<Range>>> expected) throws Exception {
List<Range> failures = Collections.emptyList();
runTest(tableName, ranges, tab1TabletCache, expected, failures);
@@ -170,7 +183,7 @@ public class TabletLocatorImplTest extends TestCase {
List<Range> efailures) throws Exception {
Map<String,Map<KeyExtent,List<Range>>> binnedRanges = new HashMap<String,Map<KeyExtent,List<Range>>>();
- List<Range> f = tab1TabletCache.binRanges(credentials, ranges, binnedRanges);
+ List<Range> f = tab1TabletCache.binRanges(context, ranges, binnedRanges);
assertEquals(expected, binnedRanges);
HashSet<Range> f1 = new HashSet<Range>(f);
@@ -224,7 +237,7 @@ public class TabletLocatorImplTest extends TestCase {
private void runTest(TabletLocatorImpl metaCache, List<Mutation> ml, Map<String,Map<KeyExtent,List<String>>> emb, String... efailures) throws Exception {
Map<String,TabletServerMutations<Mutation>> binnedMutations = new HashMap<String,TabletServerMutations<Mutation>>();
List<Mutation> afailures = new ArrayList<Mutation>();
- metaCache.binMutations(credentials, ml, binnedMutations, afailures);
+ metaCache.binMutations(context, ml, binnedMutations, afailures);
verify(emb, binnedMutations);
@@ -295,6 +308,7 @@ public class TabletLocatorImplTest extends TestCase {
return emb;
}
+ @Test
public void testRemoveOverlapping1() {
TreeMap<Text,TabletLocation> mc = createMetaCache(nke("0", null, null), "l1");
@@ -339,6 +353,7 @@ public class TabletLocatorImplTest extends TestCase {
}
+ @Test
public void testRemoveOverlapping2() {
// test removes when cache does not contain all tablets in a table
@@ -461,7 +476,7 @@ public class TabletLocatorImplTest extends TestCase {
}
@Override
- public TabletLocations lookupTablet(Credentials credentials, TabletLocation src, Text row, Text stopRow, TabletLocator parent)
+ public TabletLocations lookupTablet(ClientContext context, TabletLocation src, Text row, Text stopRow, TabletLocator parent)
throws AccumuloSecurityException {
// System.out.println("lookupTablet("+src+","+row+","+stopRow+","+ parent+")");
@@ -470,7 +485,7 @@ public class TabletLocatorImplTest extends TestCase {
Map<KeyExtent,SortedMap<Key,Value>> tablets = tservers.get(src.tablet_location);
if (tablets == null) {
- parent.invalidateCache(src.tablet_location);
+ parent.invalidateCache(context.getInstance(), src.tablet_location);
return null;
}
@@ -493,7 +508,7 @@ public class TabletLocatorImplTest extends TestCase {
}
@Override
- public List<TabletLocation> lookupTablets(Credentials credentials, String tserver, Map<KeyExtent,List<Range>> map, TabletLocator parent)
+ public List<TabletLocation> lookupTablets(ClientContext context, String tserver, Map<KeyExtent,List<Range>> map, TabletLocator parent)
throws AccumuloSecurityException {
ArrayList<TabletLocation> list = new ArrayList<TabletLocation>();
@@ -501,7 +516,7 @@ public class TabletLocatorImplTest extends TestCase {
Map<KeyExtent,SortedMap<Key,Value>> tablets = tservers.get(tserver);
if (tablets == null) {
- parent.invalidateCache(tserver);
+ parent.invalidateCache(context.getInstance(), tserver);
return list;
}
@@ -557,20 +572,17 @@ public class TabletLocatorImplTest extends TestCase {
static class TestRootTabletLocator extends RootTabletLocator {
- private Instance instance;
-
- TestRootTabletLocator(Instance instance) {
- super(instance, new YesLockChecker());
- this.instance = instance;
+ TestRootTabletLocator() {
+ super(new YesLockChecker());
}
@Override
- protected TabletLocation getRootTabletLocation() {
- return new TabletLocation(RootTable.EXTENT, instance.getRootTabletLocation(), "1");
+ protected TabletLocation getRootTabletLocation(ClientContext context) {
+ return new TabletLocation(RootTable.EXTENT, context.getInstance().getRootTabletLocation(), "1");
}
@Override
- public void invalidateCache(String server) {}
+ public void invalidateCache(Instance instance, String server) {}
}
@@ -644,9 +656,8 @@ public class TabletLocatorImplTest extends TestCase {
}
- private void locateTabletTest(TabletLocatorImpl cache, String row, boolean skipRow, KeyExtent expected, String server, Credentials credentials)
- throws Exception {
- TabletLocation tl = cache.locateTablet(credentials, new Text(row), skipRow, false);
+ private void locateTabletTest(TabletLocatorImpl cache, String row, boolean skipRow, KeyExtent expected, String server) throws Exception {
+ TabletLocation tl = cache.locateTablet(context, new Text(row), skipRow, false);
if (expected == null) {
if (tl != null)
@@ -659,28 +670,28 @@ public class TabletLocatorImplTest extends TestCase {
}
}
- private void locateTabletTest(TabletLocatorImpl cache, String row, KeyExtent expected, String server, Credentials credentials) throws Exception {
- locateTabletTest(cache, row, false, expected, server, credentials);
+ private void locateTabletTest(TabletLocatorImpl cache, String row, KeyExtent expected, String server) throws Exception {
+ locateTabletTest(cache, row, false, expected, server);
}
+ @Test
public void test1() throws Exception {
TServers tservers = new TServers();
TestTabletLocationObtainer ttlo = new TestTabletLocationObtainer(tservers);
- TestInstance testInstance = new TestInstance("instance1", "tserver1");
- RootTabletLocator rtl = new TestRootTabletLocator(testInstance);
+ RootTabletLocator rtl = new TestRootTabletLocator();
TabletLocatorImpl rootTabletCache = new TabletLocatorImpl(new Text(MetadataTable.ID), rtl, ttlo, new YesLockChecker());
TabletLocatorImpl tab1TabletCache = new TabletLocatorImpl(new Text("tab1"), rootTabletCache, ttlo, new YesLockChecker());
- locateTabletTest(tab1TabletCache, "r1", null, null, credentials);
+ locateTabletTest(tab1TabletCache, "r1", null, null);
KeyExtent tab1e = nke("tab1", null, null);
setLocation(tservers, "tserver1", RTE, MTE, "tserver2");
setLocation(tservers, "tserver2", MTE, tab1e, "tserver3");
- locateTabletTest(tab1TabletCache, "r1", tab1e, "tserver3", credentials);
- locateTabletTest(tab1TabletCache, "r2", tab1e, "tserver3", credentials);
+ locateTabletTest(tab1TabletCache, "r1", tab1e, "tserver3");
+ locateTabletTest(tab1TabletCache, "r2", tab1e, "tserver3");
// simulate a split
KeyExtent tab1e1 = nke("tab1", "g", null);
@@ -689,52 +700,52 @@ public class TabletLocatorImplTest extends TestCase {
setLocation(tservers, "tserver2", MTE, tab1e1, "tserver4");
setLocation(tservers, "tserver2", MTE, tab1e2, "tserver5");
- locateTabletTest(tab1TabletCache, "r1", tab1e, "tserver3", credentials);
+ locateTabletTest(tab1TabletCache, "r1", tab1e, "tserver3");
tab1TabletCache.invalidateCache(tab1e);
- locateTabletTest(tab1TabletCache, "r1", tab1e2, "tserver5", credentials);
- locateTabletTest(tab1TabletCache, "a", tab1e1, "tserver4", credentials);
- locateTabletTest(tab1TabletCache, "a", true, tab1e1, "tserver4", credentials);
- locateTabletTest(tab1TabletCache, "g", tab1e1, "tserver4", credentials);
- locateTabletTest(tab1TabletCache, "g", true, tab1e2, "tserver5", credentials);
+ locateTabletTest(tab1TabletCache, "r1", tab1e2, "tserver5");
+ locateTabletTest(tab1TabletCache, "a", tab1e1, "tserver4");
+ locateTabletTest(tab1TabletCache, "a", true, tab1e1, "tserver4");
+ locateTabletTest(tab1TabletCache, "g", tab1e1, "tserver4");
+ locateTabletTest(tab1TabletCache, "g", true, tab1e2, "tserver5");
// simulate a partial split
KeyExtent tab1e22 = nke("tab1", null, "m");
setLocation(tservers, "tserver2", MTE, tab1e22, "tserver6");
- locateTabletTest(tab1TabletCache, "r1", tab1e2, "tserver5", credentials);
+ locateTabletTest(tab1TabletCache, "r1", tab1e2, "tserver5");
tab1TabletCache.invalidateCache(tab1e2);
- locateTabletTest(tab1TabletCache, "r1", tab1e22, "tserver6", credentials);
- locateTabletTest(tab1TabletCache, "h", null, null, credentials);
- locateTabletTest(tab1TabletCache, "a", tab1e1, "tserver4", credentials);
+ locateTabletTest(tab1TabletCache, "r1", tab1e22, "tserver6");
+ locateTabletTest(tab1TabletCache, "h", null, null);
+ locateTabletTest(tab1TabletCache, "a", tab1e1, "tserver4");
KeyExtent tab1e21 = nke("tab1", "m", "g");
setLocation(tservers, "tserver2", MTE, tab1e21, "tserver7");
- locateTabletTest(tab1TabletCache, "r1", tab1e22, "tserver6", credentials);
- locateTabletTest(tab1TabletCache, "h", tab1e21, "tserver7", credentials);
- locateTabletTest(tab1TabletCache, "a", tab1e1, "tserver4", credentials);
+ locateTabletTest(tab1TabletCache, "r1", tab1e22, "tserver6");
+ locateTabletTest(tab1TabletCache, "h", tab1e21, "tserver7");
+ locateTabletTest(tab1TabletCache, "a", tab1e1, "tserver4");
// simulate a migration
setLocation(tservers, "tserver2", MTE, tab1e21, "tserver8");
tab1TabletCache.invalidateCache(tab1e21);
- locateTabletTest(tab1TabletCache, "r1", tab1e22, "tserver6", credentials);
- locateTabletTest(tab1TabletCache, "h", tab1e21, "tserver8", credentials);
- locateTabletTest(tab1TabletCache, "a", tab1e1, "tserver4", credentials);
+ locateTabletTest(tab1TabletCache, "r1", tab1e22, "tserver6");
+ locateTabletTest(tab1TabletCache, "h", tab1e21, "tserver8");
+ locateTabletTest(tab1TabletCache, "a", tab1e1, "tserver4");
// simulate a server failure
setLocation(tservers, "tserver2", MTE, tab1e21, "tserver9");
- tab1TabletCache.invalidateCache("tserver8");
- locateTabletTest(tab1TabletCache, "r1", tab1e22, "tserver6", credentials);
- locateTabletTest(tab1TabletCache, "h", tab1e21, "tserver9", credentials);
- locateTabletTest(tab1TabletCache, "a", tab1e1, "tserver4", credentials);
+ tab1TabletCache.invalidateCache(context.getInstance(), "tserver8");
+ locateTabletTest(tab1TabletCache, "r1", tab1e22, "tserver6");
+ locateTabletTest(tab1TabletCache, "h", tab1e21, "tserver9");
+ locateTabletTest(tab1TabletCache, "a", tab1e1, "tserver4");
// simulate all servers failing
deleteServer(tservers, "tserver1");
deleteServer(tservers, "tserver2");
- tab1TabletCache.invalidateCache("tserver4");
- tab1TabletCache.invalidateCache("tserver6");
- tab1TabletCache.invalidateCache("tserver9");
+ tab1TabletCache.invalidateCache(context.getInstance(), "tserver4");
+ tab1TabletCache.invalidateCache(context.getInstance(), "tserver6");
+ tab1TabletCache.invalidateCache(context.getInstance(), "tserver9");
- locateTabletTest(tab1TabletCache, "r1", null, null, credentials);
- locateTabletTest(tab1TabletCache, "h", null, null, credentials);
- locateTabletTest(tab1TabletCache, "a", null, null, credentials);
+ locateTabletTest(tab1TabletCache, "r1", null, null);
+ locateTabletTest(tab1TabletCache, "h", null, null);
+ locateTabletTest(tab1TabletCache, "a", null, null);
testInstance.setRootTabletLocation("tserver4");
setLocation(tservers, "tserver4", RTE, MTE, "tserver5");
@@ -742,9 +753,9 @@ public class TabletLocatorImplTest extends TestCase {
setLocation(tservers, "tserver5", MTE, tab1e21, "tserver2");
setLocation(tservers, "tserver5", MTE, tab1e22, "tserver3");
- locateTabletTest(tab1TabletCache, "a", tab1e1, "tserver1", credentials);
- locateTabletTest(tab1TabletCache, "h", tab1e21, "tserver2", credentials);
- locateTabletTest(tab1TabletCache, "r", tab1e22, "tserver3", credentials);
+ locateTabletTest(tab1TabletCache, "a", tab1e1, "tserver1");
+ locateTabletTest(tab1TabletCache, "h", tab1e21, "tserver2");
+ locateTabletTest(tab1TabletCache, "r", tab1e22, "tserver3");
// simulate the metadata table splitting
KeyExtent mte1 = new KeyExtent(new Text(MetadataTable.ID), tab1e21.getMetadataEntry(), RTE.getEndRow());
@@ -761,29 +772,29 @@ public class TabletLocatorImplTest extends TestCase {
tab1TabletCache.invalidateCache(tab1e21);
tab1TabletCache.invalidateCache(tab1e22);
- locateTabletTest(tab1TabletCache, "a", tab1e1, "tserver7", credentials);
- locateTabletTest(tab1TabletCache, "h", tab1e21, "tserver8", credentials);
- locateTabletTest(tab1TabletCache, "r", tab1e22, "tserver9", credentials);
+ locateTabletTest(tab1TabletCache, "a", tab1e1, "tserver7");
+ locateTabletTest(tab1TabletCache, "h", tab1e21, "tserver8");
+ locateTabletTest(tab1TabletCache, "r", tab1e22, "tserver9");
// simulate metadata and regular server down and the reassigned
deleteServer(tservers, "tserver5");
- tab1TabletCache.invalidateCache("tserver7");
- locateTabletTest(tab1TabletCache, "a", null, null, credentials);
- locateTabletTest(tab1TabletCache, "h", tab1e21, "tserver8", credentials);
- locateTabletTest(tab1TabletCache, "r", tab1e22, "tserver9", credentials);
+ tab1TabletCache.invalidateCache(context.getInstance(), "tserver7");
+ locateTabletTest(tab1TabletCache, "a", null, null);
+ locateTabletTest(tab1TabletCache, "h", tab1e21, "tserver8");
+ locateTabletTest(tab1TabletCache, "r", tab1e22, "tserver9");
setLocation(tservers, "tserver4", RTE, mte1, "tserver10");
setLocation(tservers, "tserver10", mte1, tab1e1, "tserver7");
setLocation(tservers, "tserver10", mte1, tab1e21, "tserver8");
- locateTabletTest(tab1TabletCache, "a", tab1e1, "tserver7", credentials);
- locateTabletTest(tab1TabletCache, "h", tab1e21, "tserver8", credentials);
- locateTabletTest(tab1TabletCache, "r", tab1e22, "tserver9", credentials);
- tab1TabletCache.invalidateCache("tserver7");
+ locateTabletTest(tab1TabletCache, "a", tab1e1, "tserver7");
+ locateTabletTest(tab1TabletCache, "h", tab1e21, "tserver8");
+ locateTabletTest(tab1TabletCache, "r", tab1e22, "tserver9");
+ tab1TabletCache.invalidateCache(context.getInstance(), "tserver7");
setLocation(tservers, "tserver10", mte1, tab1e1, "tserver2");
- locateTabletTest(tab1TabletCache, "a", tab1e1, "tserver2", credentials);
- locateTabletTest(tab1TabletCache, "h", tab1e21, "tserver8", credentials);
- locateTabletTest(tab1TabletCache, "r", tab1e22, "tserver9", credentials);
+ locateTabletTest(tab1TabletCache, "a", tab1e1, "tserver2");
+ locateTabletTest(tab1TabletCache, "h", tab1e21, "tserver8");
+ locateTabletTest(tab1TabletCache, "r", tab1e22, "tserver9");
// simulate a hole in the metadata, caused by a partial split
KeyExtent mte11 = new KeyExtent(new Text(MetadataTable.ID), tab1e1.getMetadataEntry(), RTE.getEndRow());
@@ -795,18 +806,19 @@ public class TabletLocatorImplTest extends TestCase {
// at this point should be no table1 metadata
tab1TabletCache.invalidateCache(tab1e1);
tab1TabletCache.invalidateCache(tab1e21);
- locateTabletTest(tab1TabletCache, "a", null, null, credentials);
- locateTabletTest(tab1TabletCache, "h", tab1e21, "tserver12", credentials);
- locateTabletTest(tab1TabletCache, "r", tab1e22, "tserver9", credentials);
+ locateTabletTest(tab1TabletCache, "a", null, null);
+ locateTabletTest(tab1TabletCache, "h", tab1e21, "tserver12");
+ locateTabletTest(tab1TabletCache, "r", tab1e22, "tserver9");
setLocation(tservers, "tserver4", RTE, mte11, "tserver5");
setLocation(tservers, "tserver5", mte11, tab1e1, "tserver13");
- locateTabletTest(tab1TabletCache, "a", tab1e1, "tserver13", credentials);
- locateTabletTest(tab1TabletCache, "h", tab1e21, "tserver12", credentials);
- locateTabletTest(tab1TabletCache, "r", tab1e22, "tserver9", credentials);
+ locateTabletTest(tab1TabletCache, "a", tab1e1, "tserver13");
+ locateTabletTest(tab1TabletCache, "h", tab1e21, "tserver12");
+ locateTabletTest(tab1TabletCache, "r", tab1e22, "tserver9");
}
+ @Test
public void test2() throws Exception {
TServers tservers = new TServers();
TabletLocatorImpl metaCache = createLocators(tservers, "tserver1", "tserver2", "foo");
@@ -817,15 +829,16 @@ public class TabletLocatorImplTest extends TestCase {
setLocation(tservers, "tserver2", MTE, ke1, null);
setLocation(tservers, "tserver2", MTE, ke2, "L1");
- locateTabletTest(metaCache, "a", null, null, credentials);
- locateTabletTest(metaCache, "r", ke2, "L1", credentials);
+ locateTabletTest(metaCache, "a", null, null);
+ locateTabletTest(metaCache, "r", ke2, "L1");
setLocation(tservers, "tserver2", MTE, ke1, "L2");
- locateTabletTest(metaCache, "a", ke1, "L2", credentials);
- locateTabletTest(metaCache, "r", ke2, "L1", credentials);
+ locateTabletTest(metaCache, "a", ke1, "L2");
+ locateTabletTest(metaCache, "r", ke2, "L1");
}
+ @Test
public void testBinRanges1() throws Exception {
Text tableName = new Text("foo");
@@ -853,6 +866,7 @@ public class TabletLocatorImplTest extends TestCase {
runTest(tableName, ranges, metaCache, expected);
}
+ @Test
public void testBinRanges2() throws Exception {
Text tableName = new Text("foo");
@@ -868,6 +882,7 @@ public class TabletLocatorImplTest extends TestCase {
runTest(tableName, ranges, metaCache, expected);
}
+ @Test
public void testBinRanges3() throws Exception {
Text tableName = new Text("foo");
@@ -933,6 +948,7 @@ public class TabletLocatorImplTest extends TestCase {
runTest(tableName, ranges, metaCache, expected);
}
+ @Test
public void testBinRanges4() throws Exception {
Text tableName = new Text("foo");
@@ -998,6 +1014,7 @@ public class TabletLocatorImplTest extends TestCase {
}
+ @Test
public void testBinRanges5() throws Exception {
// Test binning when there is a hole in the metadata
Text tableName = new Text("foo");
@@ -1038,6 +1055,7 @@ public class TabletLocatorImplTest extends TestCase {
runTest(tableName, ranges, metaCache, expected4, nrl(nr("0", "11"), nr("1", "2"), nr("0", "4"), nr("2", "4")));
}
+ @Test
public void testBinMutations1() throws Exception {
// one tablet table
KeyExtent ke1 = nke("foo", null, null);
@@ -1057,6 +1075,7 @@ public class TabletLocatorImplTest extends TestCase {
}
+ @Test
public void testBinMutations2() throws Exception {
// no tablets for table
TabletLocatorImpl metaCache = createLocators("foo");
@@ -1066,6 +1085,7 @@ public class TabletLocatorImplTest extends TestCase {
runTest(metaCache, ml, emb, "a", "c");
}
+ @Test
public void testBinMutations3() throws Exception {
// three tablet table
KeyExtent ke1 = nke("foo", "h", null);
@@ -1099,6 +1119,7 @@ public class TabletLocatorImplTest extends TestCase {
runTest(metaCache, ml, emb);
}
+ @Test
public void testBinMutations4() throws Exception {
// three table with hole
KeyExtent ke1 = nke("foo", "h", null);
@@ -1132,6 +1153,7 @@ public class TabletLocatorImplTest extends TestCase {
runTest(metaCache, ml, emb, "t");
}
+ @Test
public void testBinSplit() throws Exception {
// try binning mutations and ranges when a tablet splits
@@ -1191,6 +1213,7 @@ public class TabletLocatorImplTest extends TestCase {
}
}
+ @Test
public void testBug1() throws Exception {
// a bug that occurred while running continuous ingest
KeyExtent mte1 = new KeyExtent(new Text(MetadataTable.ID), new Text("0;0bc"), RTE.getEndRow());
@@ -1198,9 +1221,8 @@ public class TabletLocatorImplTest extends TestCase {
TServers tservers = new TServers();
TestTabletLocationObtainer ttlo = new TestTabletLocationObtainer(tservers);
- TestInstance testInstance = new TestInstance("instance1", "tserver1");
- RootTabletLocator rtl = new TestRootTabletLocator(testInstance);
+ RootTabletLocator rtl = new TestRootTabletLocator();
TabletLocatorImpl rootTabletCache = new TabletLocatorImpl(new Text(MetadataTable.ID), rtl, ttlo, new YesLockChecker());
TabletLocatorImpl tab0TabletCache = new TabletLocatorImpl(new Text("0"), rootTabletCache, ttlo, new YesLockChecker());
@@ -1215,9 +1237,10 @@ public class TabletLocatorImplTest extends TestCase {
setLocation(tservers, "tserver3", mte2, ke2, "tserver5");
// look up something that comes after the last entry in mte1
- locateTabletTest(tab0TabletCache, "0bbff", ke2, "tserver5", credentials);
+ locateTabletTest(tab0TabletCache, "0bbff", ke2, "tserver5");
}
+ @Test
public void testBug2() throws Exception {
// a bug that occurred while running a functional test
KeyExtent mte1 = new KeyExtent(new Text(MetadataTable.ID), new Text("~"), RTE.getEndRow());
@@ -1225,9 +1248,8 @@ public class TabletLocatorImplTest extends TestCase {
TServers tservers = new TServers();
TestTabletLocationObtainer ttlo = new TestTabletLocationObtainer(tservers);
- TestInstance testInstance = new TestInstance("instance1", "tserver1");
- RootTabletLocator rtl = new TestRootTabletLocator(testInstance);
+ RootTabletLocator rtl = new TestRootTabletLocator();
TabletLocatorImpl rootTabletCache = new TabletLocatorImpl(new Text(MetadataTable.ID), rtl, ttlo, new YesLockChecker());
TabletLocatorImpl tab0TabletCache = new TabletLocatorImpl(new Text("0"), rootTabletCache, ttlo, new YesLockChecker());
@@ -1239,11 +1261,12 @@ public class TabletLocatorImplTest extends TestCase {
ts3.put(mte2, new TreeMap<Key,Value>());
tservers.tservers.put("tserver3", ts3);
- assertNull(tab0TabletCache.locateTablet(credentials, new Text("row_0000000000"), false, false));
+ assertNull(tab0TabletCache.locateTablet(context, new Text("row_0000000000"), false, false));
}
// this test reproduces a problem where empty metadata tablets, that were created by user tablets being merged away, caused locating tablets to fail
+ @Test
public void testBug3() throws Exception {
KeyExtent mte1 = new KeyExtent(new Text(MetadataTable.ID), new Text("1;c"), RTE.getEndRow());
KeyExtent mte2 = new KeyExtent(new Text(MetadataTable.ID), new Text("1;f"), new Text("1;c"));
@@ -1255,9 +1278,8 @@ public class TabletLocatorImplTest extends TestCase {
TServers tservers = new TServers();
TestTabletLocationObtainer ttlo = new TestTabletLocationObtainer(tservers);
- TestInstance testInstance = new TestInstance("instance1", "tserver1");
- RootTabletLocator rtl = new TestRootTabletLocator(testInstance);
+ RootTabletLocator rtl = new TestRootTabletLocator();
TabletLocatorImpl rootTabletCache = new TabletLocatorImpl(new Text(MetadataTable.ID), rtl, ttlo, new YesLockChecker());
TabletLocatorImpl tab0TabletCache = new TabletLocatorImpl(new Text("1"), rootTabletCache, ttlo, new YesLockChecker());
@@ -1274,10 +1296,11 @@ public class TabletLocatorImplTest extends TestCase {
createEmptyTablet(tservers, "tserver5", mte4);
setLocation(tservers, "tserver6", mte5, ke1, "tserver7");
- locateTabletTest(tab0TabletCache, "a", ke1, "tserver7", credentials);
+ locateTabletTest(tab0TabletCache, "a", ke1, "tserver7");
}
+ @Test
public void testAccumulo1248() throws Exception {
TServers tservers = new TServers();
TabletLocatorImpl metaCache = createLocators(tservers, "tserver1", "tserver2", "foo");
@@ -1290,7 +1313,7 @@ public class TabletLocatorImplTest extends TestCase {
setLocation(tservers, "tserver2", MTE, ke1, "L2", "I2");
try {
- metaCache.locateTablet(credentials, new Text("a"), false, false);
+ metaCache.locateTablet(context, new Text("a"), false, false);
assertTrue(false);
} catch (Exception e) {
@@ -1298,6 +1321,7 @@ public class TabletLocatorImplTest extends TestCase {
}
+ @Test
public void testLostLock() throws Exception {
final HashSet<String> activeLocks = new HashSet<String>();
@@ -1318,40 +1342,40 @@ public class TabletLocatorImplTest extends TestCase {
activeLocks.add("L1:5");
- locateTabletTest(metaCache, "a", ke1, "L1", credentials);
- locateTabletTest(metaCache, "a", ke1, "L1", credentials);
+ locateTabletTest(metaCache, "a", ke1, "L1");
+ locateTabletTest(metaCache, "a", ke1, "L1");
activeLocks.clear();
- locateTabletTest(metaCache, "a", null, null, credentials);
- locateTabletTest(metaCache, "a", null, null, credentials);
- locateTabletTest(metaCache, "a", null, null, credentials);
+ locateTabletTest(metaCache, "a", null, null);
+ locateTabletTest(metaCache, "a", null, null);
+ locateTabletTest(metaCache, "a", null, null);
clearLocation(tservers, "tserver2", MTE, ke1, "5");
setLocation(tservers, "tserver2", MTE, ke1, "L2", "6");
activeLocks.add("L2:6");
- locateTabletTest(metaCache, "a", ke1, "L2", credentials);
- locateTabletTest(metaCache, "a", ke1, "L2", credentials);
+ locateTabletTest(metaCache, "a", ke1, "L2");
+ locateTabletTest(metaCache, "a", ke1, "L2");
clearLocation(tservers, "tserver2", MTE, ke1, "6");
- locateTabletTest(metaCache, "a", ke1, "L2", credentials);
+ locateTabletTest(metaCache, "a", ke1, "L2");
setLocation(tservers, "tserver2", MTE, ke1, "L3", "7");
- locateTabletTest(metaCache, "a", ke1, "L2", credentials);
+ locateTabletTest(metaCache, "a", ke1, "L2");
activeLocks.clear();
- locateTabletTest(metaCache, "a", null, null, credentials);
- locateTabletTest(metaCache, "a", null, null, credentials);
+ locateTabletTest(metaCache, "a", null, null);
+ locateTabletTest(metaCache, "a", null, null);
activeLocks.add("L3:7");
- locateTabletTest(metaCache, "a", ke1, "L3", credentials);
- locateTabletTest(metaCache, "a", ke1, "L3", credentials);
+ locateTabletTest(metaCache, "a", ke1, "L3");
+ locateTabletTest(metaCache, "a", ke1, "L3");
List<Mutation> ml = nml(nm("a", "cf1:cq1=v1", "cf1:cq2=v2"), nm("w", "cf1:cq3=v3"));
Map<String,Map<KeyExtent,List<String>>> emb = cemb(nol("a", "L3", ke1), nol("w", "L3", ke1));
http://git-wip-us.apache.org/repos/asf/accumulo/blob/42c25faa/core/src/test/java/org/apache/accumulo/core/metadata/MetadataServicerTest.java
----------------------------------------------------------------------
diff --git a/core/src/test/java/org/apache/accumulo/core/metadata/MetadataServicerTest.java b/core/src/test/java/org/apache/accumulo/core/metadata/MetadataServicerTest.java
index 687e543..7501c31 100644
--- a/core/src/test/java/org/apache/accumulo/core/metadata/MetadataServicerTest.java
+++ b/core/src/test/java/org/apache/accumulo/core/metadata/MetadataServicerTest.java
@@ -23,9 +23,11 @@ import static org.junit.Assert.assertTrue;
import org.apache.accumulo.core.client.AccumuloException;
import org.apache.accumulo.core.client.AccumuloSecurityException;
+import org.apache.accumulo.core.client.ClientConfiguration;
import org.apache.accumulo.core.client.Connector;
import org.apache.accumulo.core.client.TableExistsException;
import org.apache.accumulo.core.client.TableNotFoundException;
+import org.apache.accumulo.core.client.impl.ClientContext;
import org.apache.accumulo.core.client.mock.MockInstance;
import org.apache.accumulo.core.client.security.tokens.PasswordToken;
import org.apache.accumulo.core.security.Credentials;
@@ -47,36 +49,37 @@ public class MetadataServicerTest {
connector.tableOperations().create(userTableName);
String userTableId = connector.tableOperations().tableIdMap().get(userTableName);
Credentials credentials = new Credentials("root", new PasswordToken(""));
+ ClientContext context = new ClientContext(instance, credentials, new ClientConfiguration());
- MetadataServicer ms = MetadataServicer.forTableId(instance, credentials, RootTable.ID);
+ MetadataServicer ms = MetadataServicer.forTableId(context, RootTable.ID);
assertTrue(ms instanceof ServicerForRootTable);
assertFalse(ms instanceof TableMetadataServicer);
assertEquals(RootTable.ID, ms.getServicedTableId());
- ms = MetadataServicer.forTableId(instance, credentials, MetadataTable.ID);
+ ms = MetadataServicer.forTableId(context, MetadataTable.ID);
assertTrue(ms instanceof ServicerForMetadataTable);
assertTrue(ms instanceof TableMetadataServicer);
assertEquals(RootTable.NAME, ((TableMetadataServicer) ms).getServicingTableName());
assertEquals(MetadataTable.ID, ms.getServicedTableId());
- ms = MetadataServicer.forTableId(instance, credentials, userTableId);
+ ms = MetadataServicer.forTableId(context, userTableId);
assertTrue(ms instanceof ServicerForUserTables);
assertTrue(ms instanceof TableMetadataServicer);
assertEquals(MetadataTable.NAME, ((TableMetadataServicer) ms).getServicingTableName());
assertEquals(userTableId, ms.getServicedTableId());
- ms = MetadataServicer.forTableName(instance, credentials, RootTable.NAME);
+ ms = MetadataServicer.forTableName(context, RootTable.NAME);
assertTrue(ms instanceof ServicerForRootTable);
assertFalse(ms instanceof TableMetadataServicer);
assertEquals(RootTable.ID, ms.getServicedTableId());
- ms = MetadataServicer.forTableName(instance, credentials, MetadataTable.NAME);
+ ms = MetadataServicer.forTableName(context, MetadataTable.NAME);
assertTrue(ms instanceof ServicerForMetadataTable);
assertTrue(ms instanceof TableMetadataServicer);
assertEquals(RootTable.NAME, ((TableMetadataServicer) ms).getServicingTableName());
assertEquals(MetadataTable.ID, ms.getServicedTableId());
- ms = MetadataServicer.forTableName(instance, credentials, userTableName);
+ ms = MetadataServicer.forTableName(context, userTableName);
assertTrue(ms instanceof ServicerForUserTables);
assertTrue(ms instanceof TableMetadataServicer);
assertEquals(MetadataTable.NAME, ((TableMetadataServicer) ms).getServicingTableName());
http://git-wip-us.apache.org/repos/asf/accumulo/blob/42c25faa/core/src/test/java/org/apache/accumulo/core/replication/ReplicationOperationsImplTest.java
----------------------------------------------------------------------
diff --git a/core/src/test/java/org/apache/accumulo/core/replication/ReplicationOperationsImplTest.java b/core/src/test/java/org/apache/accumulo/core/replication/ReplicationOperationsImplTest.java
index b404d3d..df9816e 100644
--- a/core/src/test/java/org/apache/accumulo/core/replication/ReplicationOperationsImplTest.java
+++ b/core/src/test/java/org/apache/accumulo/core/replication/ReplicationOperationsImplTest.java
@@ -23,7 +23,9 @@ import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.accumulo.core.client.BatchWriter;
import org.apache.accumulo.core.client.BatchWriterConfig;
+import org.apache.accumulo.core.client.ClientConfiguration;
import org.apache.accumulo.core.client.Connector;
+import org.apache.accumulo.core.client.impl.ClientContext;
import org.apache.accumulo.core.client.impl.ReplicationOperationsImpl;
import org.apache.accumulo.core.client.mock.MockInstance;
import org.apache.accumulo.core.client.security.tokens.PasswordToken;
@@ -98,7 +100,8 @@ public class ReplicationOperationsImplTest {
final AtomicBoolean done = new AtomicBoolean(false);
final AtomicBoolean exception = new AtomicBoolean(false);
- final ReplicationOperationsImpl roi = new ReplicationOperationsImpl(inst, new Credentials("root", new PasswordToken("")));
+ ClientContext context = new ClientContext(inst, new Credentials("root", new PasswordToken("")), new ClientConfiguration());
+ final ReplicationOperationsImpl roi = new ReplicationOperationsImpl(context);
Thread t = new Thread(new Runnable() {
@Override
public void run() {
@@ -196,7 +199,8 @@ public class ReplicationOperationsImplTest {
final AtomicBoolean done = new AtomicBoolean(false);
final AtomicBoolean exception = new AtomicBoolean(false);
- final ReplicationOperationsImpl roi = new ReplicationOperationsImpl(inst, new Credentials("root", new PasswordToken("")));
+ ClientContext context = new ClientContext(inst, new Credentials("root", new PasswordToken("")), new ClientConfiguration());
+ final ReplicationOperationsImpl roi = new ReplicationOperationsImpl(context);
Thread t = new Thread(new Runnable() {
@Override
public void run() {
@@ -280,7 +284,8 @@ public class ReplicationOperationsImplTest {
final AtomicBoolean done = new AtomicBoolean(false);
final AtomicBoolean exception = new AtomicBoolean(false);
- final ReplicationOperationsImpl roi = new ReplicationOperationsImpl(inst, new Credentials("root", new PasswordToken("")));
+ ClientContext context = new ClientContext(inst, new Credentials("root", new PasswordToken("")), new ClientConfiguration());
+ final ReplicationOperationsImpl roi = new ReplicationOperationsImpl(context);
Thread t = new Thread(new Runnable() {
@Override
public void run() {
@@ -357,7 +362,8 @@ public class ReplicationOperationsImplTest {
final AtomicBoolean done = new AtomicBoolean(false);
final AtomicBoolean exception = new AtomicBoolean(false);
- final ReplicationOperationsImpl roi = new ReplicationOperationsImpl(inst, new Credentials("root", new PasswordToken("")));
+ ClientContext context = new ClientContext(inst, new Credentials("root", new PasswordToken("")), new ClientConfiguration());
+ final ReplicationOperationsImpl roi = new ReplicationOperationsImpl(context);
Thread t = new Thread(new Runnable() {
@Override
public void run() {
http://git-wip-us.apache.org/repos/asf/accumulo/blob/42c25faa/mapreduce/src/main/java/org/apache/accumulo/core/client/mapred/AbstractInputFormat.java
----------------------------------------------------------------------
diff --git a/mapreduce/src/main/java/org/apache/accumulo/core/client/mapred/AbstractInputFormat.java b/mapreduce/src/main/java/org/apache/accumulo/core/client/mapred/AbstractInputFormat.java
index 5af78d2..3c86243 100644
--- a/mapreduce/src/main/java/org/apache/accumulo/core/client/mapred/AbstractInputFormat.java
+++ b/mapreduce/src/main/java/org/apache/accumulo/core/client/mapred/AbstractInputFormat.java
@@ -39,6 +39,7 @@ import org.apache.accumulo.core.client.Scanner;
import org.apache.accumulo.core.client.TableDeletedException;
import org.apache.accumulo.core.client.TableNotFoundException;
import org.apache.accumulo.core.client.TableOfflineException;
+import org.apache.accumulo.core.client.impl.ClientContext;
import org.apache.accumulo.core.client.impl.OfflineScanner;
import org.apache.accumulo.core.client.impl.ScannerImpl;
import org.apache.accumulo.core.client.impl.Tables;
@@ -394,7 +395,8 @@ public abstract class AbstractInputFormat<K,V> implements InputFormat<K,V> {
} else if (instance instanceof MockInstance) {
scanner = instance.getConnector(principal, token).createScanner(split.getTableName(), authorizations);
} else {
- scanner = new ScannerImpl(instance, new Credentials(principal, token), split.getTableId(), authorizations);
+ ClientContext context = new ClientContext(instance, new Credentials(principal, token), ClientConfiguration.loadDefault());
+ scanner = new ScannerImpl(context, split.getTableId(), authorizations);
}
if (isIsolated) {
log.info("Creating isolated scanner");
@@ -514,9 +516,10 @@ public abstract class AbstractInputFormat<K,V> implements InputFormat<K,V> {
tl = getTabletLocator(job, tableId);
// its possible that the cache could contain complete, but old information about a tables tablets... so clear it
tl.invalidateCache();
- Credentials creds = new Credentials(getPrincipal(job), getAuthenticationToken(job));
- while (!tl.binRanges(creds, ranges, binnedRanges).isEmpty()) {
+ ClientContext context = new ClientContext(getInstance(job), new Credentials(getPrincipal(job), getAuthenticationToken(job)),
+ ClientConfiguration.loadDefault());
+ while (!tl.binRanges(context, ranges, binnedRanges).isEmpty()) {
if (!(instance instanceof MockInstance)) {
if (!Tables.exists(instance, tableId))
throw new TableDeletedException(tableId);
http://git-wip-us.apache.org/repos/asf/accumulo/blob/42c25faa/mapreduce/src/main/java/org/apache/accumulo/core/client/mapreduce/AbstractInputFormat.java
----------------------------------------------------------------------
diff --git a/mapreduce/src/main/java/org/apache/accumulo/core/client/mapreduce/AbstractInputFormat.java b/mapreduce/src/main/java/org/apache/accumulo/core/client/mapreduce/AbstractInputFormat.java
index 836cff9..40cdb14 100644
--- a/mapreduce/src/main/java/org/apache/accumulo/core/client/mapreduce/AbstractInputFormat.java
+++ b/mapreduce/src/main/java/org/apache/accumulo/core/client/mapreduce/AbstractInputFormat.java
@@ -40,6 +40,7 @@ import org.apache.accumulo.core.client.Scanner;
import org.apache.accumulo.core.client.TableDeletedException;
import org.apache.accumulo.core.client.TableNotFoundException;
import org.apache.accumulo.core.client.TableOfflineException;
+import org.apache.accumulo.core.client.impl.ClientContext;
import org.apache.accumulo.core.client.impl.OfflineScanner;
import org.apache.accumulo.core.client.impl.ScannerImpl;
import org.apache.accumulo.core.client.impl.Tables;
@@ -425,7 +426,8 @@ public abstract class AbstractInputFormat<K,V> extends InputFormat<K,V> {
} else if (instance instanceof MockInstance) {
scanner = instance.getConnector(principal, token).createScanner(split.getTableName(), authorizations);
} else {
- scanner = new ScannerImpl(instance, new Credentials(principal, token), split.getTableId(), authorizations);
+ ClientContext context = new ClientContext(instance, new Credentials(principal, token), ClientConfiguration.loadDefault());
+ scanner = new ScannerImpl(context, split.getTableId(), authorizations);
}
if (isIsolated) {
log.info("Creating isolated scanner");
@@ -569,9 +571,10 @@ public abstract class AbstractInputFormat<K,V> extends InputFormat<K,V> {
tl = getTabletLocator(context, tableId);
// its possible that the cache could contain complete, but old information about a tables tablets... so clear it
tl.invalidateCache();
- Credentials creds = new Credentials(getPrincipal(context), getAuthenticationToken(context));
- while (!tl.binRanges(creds, ranges, binnedRanges).isEmpty()) {
+ ClientContext clientContext = new ClientContext(getInstance(context), new Credentials(getPrincipal(context), getAuthenticationToken(context)),
+ ClientConfiguration.loadDefault());
+ while (!tl.binRanges(clientContext, ranges, binnedRanges).isEmpty()) {
if (!(instance instanceof MockInstance)) {
if (!Tables.exists(instance, tableId))
throw new TableDeletedException(tableId);
http://git-wip-us.apache.org/repos/asf/accumulo/blob/42c25faa/mapreduce/src/main/java/org/apache/accumulo/core/client/mapreduce/lib/impl/InputConfigurator.java
----------------------------------------------------------------------
diff --git a/mapreduce/src/main/java/org/apache/accumulo/core/client/mapreduce/lib/impl/InputConfigurator.java b/mapreduce/src/main/java/org/apache/accumulo/core/client/mapreduce/lib/impl/InputConfigurator.java
index 90a8622..af84bb4 100644
--- a/mapreduce/src/main/java/org/apache/accumulo/core/client/mapreduce/lib/impl/InputConfigurator.java
+++ b/mapreduce/src/main/java/org/apache/accumulo/core/client/mapreduce/lib/impl/InputConfigurator.java
@@ -36,6 +36,7 @@ import java.util.StringTokenizer;
import org.apache.accumulo.core.client.AccumuloException;
import org.apache.accumulo.core.client.AccumuloSecurityException;
+import org.apache.accumulo.core.client.ClientConfiguration;
import org.apache.accumulo.core.client.ClientSideIteratorScanner;
import org.apache.accumulo.core.client.Connector;
import org.apache.accumulo.core.client.Instance;
@@ -44,6 +45,7 @@ import org.apache.accumulo.core.client.IteratorSetting;
import org.apache.accumulo.core.client.RowIterator;
import org.apache.accumulo.core.client.Scanner;
import org.apache.accumulo.core.client.TableNotFoundException;
+import org.apache.accumulo.core.client.impl.ClientContext;
import org.apache.accumulo.core.client.impl.Tables;
import org.apache.accumulo.core.client.impl.TabletLocator;
import org.apache.accumulo.core.client.mapreduce.InputTableConfig;
@@ -59,6 +61,7 @@ import org.apache.accumulo.core.master.state.tables.TableState;
import org.apache.accumulo.core.metadata.MetadataTable;
import org.apache.accumulo.core.metadata.schema.MetadataSchema;
import org.apache.accumulo.core.security.Authorizations;
+import org.apache.accumulo.core.security.Credentials;
import org.apache.accumulo.core.security.TablePermission;
import org.apache.accumulo.core.util.Base64;
import org.apache.accumulo.core.util.Pair;
@@ -314,8 +317,7 @@ public class InputConfigurator extends ConfiguratorBase {
for (String col : serialized) {
int idx = col.indexOf(":");
- Text cf = new Text(idx < 0 ? Base64.decodeBase64(col.getBytes(UTF_8)) : Base64.decodeBase64(col.substring(0, idx).getBytes(
- UTF_8)));
+ Text cf = new Text(idx < 0 ? Base64.decodeBase64(col.getBytes(UTF_8)) : Base64.decodeBase64(col.substring(0, idx).getBytes(UTF_8)));
Text cq = idx < 0 ? null : new Text(Base64.decodeBase64(col.substring(idx + 1).getBytes(UTF_8)));
columns.add(new Pair<Text,Text>(cf, cq));
}
@@ -609,7 +611,9 @@ public class InputConfigurator extends ConfiguratorBase {
if ("MockInstance".equals(instanceType))
return new MockTabletLocator();
Instance instance = getInstance(implementingClass, conf);
- return TabletLocator.getLocator(instance, new Text(tableId));
+ ClientContext context = new ClientContext(instance,
+ new Credentials(getPrincipal(implementingClass, conf), getAuthenticationToken(implementingClass, conf)), ClientConfiguration.loadDefault());
+ return TabletLocator.getLocator(context, new Text(tableId));
}
// InputFormat doesn't have the equivalent of OutputFormat's checkOutputSpecs(JobContext job)
http://git-wip-us.apache.org/repos/asf/accumulo/blob/42c25faa/minicluster/src/main/java/org/apache/accumulo/minicluster/impl/MiniAccumuloClusterImpl.java
----------------------------------------------------------------------
diff --git a/minicluster/src/main/java/org/apache/accumulo/minicluster/impl/MiniAccumuloClusterImpl.java b/minicluster/src/main/java/org/apache/accumulo/minicluster/impl/MiniAccumuloClusterImpl.java
index 5149d9d..448746f 100644
--- a/minicluster/src/main/java/org/apache/accumulo/minicluster/impl/MiniAccumuloClusterImpl.java
+++ b/minicluster/src/main/java/org/apache/accumulo/minicluster/impl/MiniAccumuloClusterImpl.java
@@ -59,6 +59,7 @@ 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.ZooKeeperInstance;
+import org.apache.accumulo.core.client.impl.ClientContext;
import org.apache.accumulo.core.client.impl.MasterClient;
import org.apache.accumulo.core.client.impl.thrift.ThriftSecurityException;
import org.apache.accumulo.core.client.security.tokens.PasswordToken;
@@ -67,6 +68,7 @@ import org.apache.accumulo.core.conf.Property;
import org.apache.accumulo.core.master.thrift.MasterClientService;
import org.apache.accumulo.core.master.thrift.MasterGoalState;
import org.apache.accumulo.core.master.thrift.MasterMonitorInfo;
+import org.apache.accumulo.core.security.Credentials;
import org.apache.accumulo.core.trace.Tracer;
import org.apache.accumulo.core.util.CachedConfiguration;
import org.apache.accumulo.core.util.Daemon;
@@ -82,7 +84,6 @@ import org.apache.accumulo.server.Accumulo;
import org.apache.accumulo.server.fs.VolumeManager;
import org.apache.accumulo.server.fs.VolumeManagerImpl;
import org.apache.accumulo.server.init.Initialize;
-import org.apache.accumulo.server.security.SystemCredentials;
import org.apache.accumulo.server.util.AccumuloStatus;
import org.apache.accumulo.server.util.PortUtils;
import org.apache.accumulo.server.util.time.SimpleTimer;
@@ -856,8 +857,9 @@ public class MiniAccumuloClusterImpl implements AccumuloCluster {
MasterMonitorInfo stats = null;
try {
Instance instance = new ZooKeeperInstance(getClientConfig());
- client = MasterClient.getConnectionWithRetry(instance);
- stats = client.getMasterStats(Tracer.traceInfo(), SystemCredentials.get(instance).toThrift(instance));
+ ClientContext context = new ClientContext(instance, new Credentials("root", new PasswordToken("unchecked")), getClientConfig());
+ client = MasterClient.getConnectionWithRetry(context);
+ stats = client.getMasterStats(Tracer.traceInfo(), context.rpcCreds());
} catch (ThriftSecurityException exception) {
throw new AccumuloSecurityException(exception);
} catch (TException exception) {
http://git-wip-us.apache.org/repos/asf/accumulo/blob/42c25faa/minicluster/src/test/java/org/apache/accumulo/minicluster/MiniAccumuloClusterStartStopTest.java
----------------------------------------------------------------------
diff --git a/minicluster/src/test/java/org/apache/accumulo/minicluster/MiniAccumuloClusterStartStopTest.java b/minicluster/src/test/java/org/apache/accumulo/minicluster/MiniAccumuloClusterStartStopTest.java
index 2031b11..e60b41c 100644
--- a/minicluster/src/test/java/org/apache/accumulo/minicluster/MiniAccumuloClusterStartStopTest.java
+++ b/minicluster/src/test/java/org/apache/accumulo/minicluster/MiniAccumuloClusterStartStopTest.java
@@ -16,35 +16,52 @@
*/
package org.apache.accumulo.minicluster;
+import java.io.File;
import java.io.IOException;
import org.apache.accumulo.core.client.Connector;
+import org.apache.commons.io.FileUtils;
+import org.apache.log4j.Logger;
import org.junit.After;
import org.junit.Before;
+import org.junit.Rule;
import org.junit.Test;
-import org.junit.rules.TemporaryFolder;
+import org.junit.rules.TestName;
public class MiniAccumuloClusterStartStopTest {
- public TemporaryFolder folder = new TemporaryFolder();
+ private static final Logger log = Logger.getLogger(MiniAccumuloClusterStartStopTest.class);
+ private File baseDir = new File(System.getProperty("user.dir") + "/target/mini-tests/" + this.getClass().getName());
+ private MiniAccumuloCluster accumulo;
+
+ @Rule
+ public TestName testName = new TestName();
@Before
- public void createMacDir() throws IOException {
- folder.create();
+ public void setupTestCluster() throws IOException {
+ baseDir.mkdirs();
+ File testDir = new File(baseDir, testName.getMethodName());
+ FileUtils.deleteQuietly(testDir);
+ testDir.mkdir();
+ accumulo = new MiniAccumuloCluster(testDir, "superSecret");
}
@After
- public void deleteMacDir() {
- folder.delete();
+ public void teardownTestCluster() {
+ if (accumulo != null) {
+ try {
+ accumulo.stop();
+ } catch (IOException | InterruptedException e) {
+ log.warn("Failure during tear down", e);
+ }
+ }
}
@Test
public void multipleStartsThrowsAnException() throws Exception {
- MiniAccumuloCluster accumulo = new MiniAccumuloCluster(folder.getRoot(), "superSecret");
-
// In 1.6.0, multiple start's did not throw an exception as advertised
- try {
accumulo.start();
+ try {
accumulo.start();
} finally {
accumulo.stop();
@@ -53,7 +70,6 @@ public class MiniAccumuloClusterStartStopTest {
@Test
public void multipleStopsIsAllowed() throws Exception {
- MiniAccumuloCluster accumulo = new MiniAccumuloCluster(folder.getRoot(), "superSecret");
accumulo.start();
Connector conn = accumulo.getConnector("root", "superSecret");
http://git-wip-us.apache.org/repos/asf/accumulo/blob/42c25faa/server/base/src/main/java/org/apache/accumulo/server/AccumuloServerContext.java
----------------------------------------------------------------------
diff --git a/server/base/src/main/java/org/apache/accumulo/server/AccumuloServerContext.java b/server/base/src/main/java/org/apache/accumulo/server/AccumuloServerContext.java
new file mode 100644
index 0000000..3a435c7
--- /dev/null
+++ b/server/base/src/main/java/org/apache/accumulo/server/AccumuloServerContext.java
@@ -0,0 +1,67 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.accumulo.server;
+
+import org.apache.accumulo.core.client.Instance;
+import org.apache.accumulo.core.client.impl.ClientContext;
+import org.apache.accumulo.core.client.mock.MockInstance;
+import org.apache.accumulo.core.client.security.tokens.PasswordToken;
+import org.apache.accumulo.core.security.Credentials;
+import org.apache.accumulo.core.util.SslConnectionParams;
+import org.apache.accumulo.server.conf.ServerConfigurationFactory;
+import org.apache.accumulo.server.security.SystemCredentials;
+
+/**
+ * Provides a server context for Accumulo server components that operate with the system credentials and have access to the system files and configuration.
+ */
+public class AccumuloServerContext extends ClientContext {
+
+ private final ServerConfigurationFactory confFactory;
+
+ /**
+ * Construct a server context from the server's configuration
+ */
+ public AccumuloServerContext(ServerConfigurationFactory confFactory) {
+ super(confFactory.getInstance(), getCredentials(confFactory.getInstance()), confFactory.getConfiguration());
+ this.confFactory = confFactory;
+ }
+
+ /**
+ * Get the credentials to use for this instance so it can be passed to the superclass during construction.
+ */
+ private static Credentials getCredentials(Instance instance) {
+ if (instance instanceof MockInstance) {
+ return new Credentials("mockSystemUser", new PasswordToken("mockSystemPassword"));
+ }
+ return SystemCredentials.get(instance);
+ }
+
+ /**
+ * Retrieve the configuration factory used to construct this context
+ */
+ public ServerConfigurationFactory getServerConfigurationFactory() {
+ return confFactory;
+ }
+
+ /**
+ * Retrieve the SSL/TLS configuration for starting up a listening service
+ */
+ public SslConnectionParams getServerSslParams() {
+ return SslConnectionParams.forServer(getConfiguration());
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/42c25faa/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 a2ff172..5496eb4 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
@@ -34,7 +34,7 @@ import java.util.concurrent.TimeUnit;
import org.apache.accumulo.core.client.AccumuloException;
import org.apache.accumulo.core.client.AccumuloSecurityException;
-import org.apache.accumulo.core.client.Instance;
+import org.apache.accumulo.core.client.impl.ClientContext;
import org.apache.accumulo.core.client.impl.ServerClient;
import org.apache.accumulo.core.client.impl.TabletLocator;
import org.apache.accumulo.core.client.impl.TabletLocator.TabletLocation;
@@ -52,17 +52,15 @@ import org.apache.accumulo.core.data.thrift.TKeyExtent;
import org.apache.accumulo.core.file.FileOperations;
import org.apache.accumulo.core.file.FileSKVIterator;
import org.apache.accumulo.core.metadata.MetadataTable;
-import org.apache.accumulo.core.security.Credentials;
import org.apache.accumulo.core.tabletserver.thrift.TabletClientService;
-import org.apache.accumulo.core.trace.wrappers.TraceRunnable;
import org.apache.accumulo.core.trace.Tracer;
+import org.apache.accumulo.core.trace.wrappers.TraceRunnable;
import org.apache.accumulo.core.util.CachedConfiguration;
import org.apache.accumulo.core.util.LoggingRunnable;
import org.apache.accumulo.core.util.NamingThreadFactory;
import org.apache.accumulo.core.util.StopWatch;
import org.apache.accumulo.core.util.ThriftUtil;
import org.apache.accumulo.core.util.UtilWaitThread;
-import org.apache.accumulo.server.conf.ServerConfigurationFactory;
import org.apache.accumulo.server.fs.VolumeManager;
import org.apache.accumulo.server.fs.VolumeManagerImpl;
import org.apache.accumulo.server.util.FileUtil;
@@ -77,9 +75,9 @@ public class BulkImporter {
private static final Logger log = Logger.getLogger(BulkImporter.class);
- public static List<String> bulkLoad(AccumuloConfiguration conf, Instance instance, Credentials creds, long tid, String tableId, List<String> files,
- String errorDir, boolean setTime) throws IOException, AccumuloException, AccumuloSecurityException, ThriftTableOperationException {
- AssignmentStats stats = new BulkImporter(conf, instance, creds, tid, tableId, setTime).importFiles(files, new Path(errorDir));
+ public static List<String> bulkLoad(ClientContext context, long tid, String tableId, List<String> files, String errorDir, boolean setTime)
+ throws IOException, AccumuloException, AccumuloSecurityException, ThriftTableOperationException {
+ AssignmentStats stats = new BulkImporter(context, tid, tableId, setTime).importFiles(files, new Path(errorDir));
List<String> result = new ArrayList<String>();
for (Path p : stats.completeFailures.keySet()) {
result.add(p.toString());
@@ -93,35 +91,30 @@ public class BulkImporter {
EXAMINE_MAP_FILES, QUERY_METADATA, IMPORT_MAP_FILES, SLEEP, TOTAL
}
- private Instance instance;
- private Credentials credentials;
+ private final ClientContext context;
private String tableId;
private long tid;
- private AccumuloConfiguration acuConf;
private boolean setTime;
- public BulkImporter(AccumuloConfiguration conf, Instance instance, Credentials credentials, long tid, String tableId, boolean setTime) {
- this.instance = instance;
- this.credentials = credentials;
+ public BulkImporter(ClientContext context, long tid, String tableId, boolean setTime) {
+ this.context = context;
this.tid = tid;
this.tableId = tableId;
- this.acuConf = conf;
this.setTime = setTime;
}
public AssignmentStats importFiles(List<String> files, Path failureDir) throws IOException, AccumuloException, AccumuloSecurityException,
ThriftTableOperationException {
- int numThreads = acuConf.getCount(Property.TSERV_BULK_PROCESS_THREADS);
- int numAssignThreads = acuConf.getCount(Property.TSERV_BULK_ASSIGNMENT_THREADS);
- final ServerConfigurationFactory confFactory = new ServerConfigurationFactory(instance);
+ int numThreads = context.getConfiguration().getCount(Property.TSERV_BULK_PROCESS_THREADS);
+ int numAssignThreads = context.getConfiguration().getCount(Property.TSERV_BULK_ASSIGNMENT_THREADS);
timer = new StopWatch<Timers>(Timers.class);
timer.start(Timers.TOTAL);
Configuration conf = CachedConfiguration.getInstance();
- VolumeManagerImpl.get(acuConf);
- final VolumeManager fs = VolumeManagerImpl.get(acuConf);
+ VolumeManagerImpl.get(context.getConfiguration());
+ final VolumeManager fs = VolumeManagerImpl.get(context.getConfiguration());
Set<Path> paths = new HashSet<Path>();
for (String file : files) {
@@ -132,7 +125,7 @@ public class BulkImporter {
final Map<Path,List<KeyExtent>> completeFailures = Collections.synchronizedSortedMap(new TreeMap<Path,List<KeyExtent>>());
ClientService.Client client = null;
- final TabletLocator locator = TabletLocator.getLocator(instance, new Text(tableId));
+ final TabletLocator locator = TabletLocator.getLocator(context, new Text(tableId));
try {
final Map<Path,List<TabletLocation>> assignments = Collections.synchronizedSortedMap(new TreeMap<Path,List<TabletLocation>>());
@@ -147,7 +140,7 @@ public class BulkImporter {
public void run() {
List<TabletLocation> tabletsToAssignMapFileTo = Collections.emptyList();
try {
- tabletsToAssignMapFileTo = findOverlappingTablets(confFactory.getConfiguration(), fs, locator, mapFile, credentials);
+ tabletsToAssignMapFileTo = findOverlappingTablets(context, fs, locator, mapFile);
} catch (Exception ex) {
log.warn("Unable to find tablets that overlap file " + mapFile.toString());
}
@@ -172,8 +165,7 @@ public class BulkImporter {
timer.stop(Timers.EXAMINE_MAP_FILES);
assignmentStats.attemptingAssignments(assignments);
- Map<Path,List<KeyExtent>> assignmentFailures = assignMapFiles(acuConf, instance, conf, credentials, fs, tableId, assignments, paths, numAssignThreads,
- numThreads);
+ Map<Path,List<KeyExtent>> assignmentFailures = assignMapFiles(context, conf, fs, tableId, assignments, paths, numAssignThreads, numThreads);
assignmentStats.assignmentsFailed(assignmentFailures);
Map<Path,Integer> failureCount = new TreeMap<Path,Integer>();
@@ -210,8 +202,7 @@ public class BulkImporter {
try {
timer.start(Timers.QUERY_METADATA);
- tabletsToAssignMapFileTo.addAll(findOverlappingTablets(confFactory.getConfiguration(), fs, locator, entry.getKey(), ke,
- credentials));
+ tabletsToAssignMapFileTo.addAll(findOverlappingTablets(context, fs, locator, entry.getKey(), ke));
timer.stop(Timers.QUERY_METADATA);
keListIter.remove();
} catch (Exception ex) {
@@ -224,8 +215,7 @@ public class BulkImporter {
}
assignmentStats.attemptingAssignments(assignments);
- Map<Path,List<KeyExtent>> assignmentFailures2 = assignMapFiles(acuConf, instance, conf, credentials, fs, tableId, assignments, paths, numAssignThreads,
- numThreads);
+ Map<Path,List<KeyExtent>> assignmentFailures2 = assignMapFiles(context, conf, fs, tableId, assignments, paths, numAssignThreads, numThreads);
assignmentStats.assignmentsFailed(assignmentFailures2);
// merge assignmentFailures2 into assignmentFailures
@@ -249,7 +239,7 @@ public class BulkImporter {
Set<Entry<Path,Integer>> failureIter = failureCount.entrySet();
for (Entry<Path,Integer> entry : failureIter) {
- int retries = acuConf.getCount(Property.TSERV_BULK_RETRY);
+ int retries = context.getConfiguration().getCount(Property.TSERV_BULK_RETRY);
if (entry.getValue() > retries && assignmentFailures.get(entry.getKey()) != null) {
log.error("Map file " + entry.getKey() + " failed more than " + retries + " times, giving up.");
completeFailures.put(entry.getKey(), assignmentFailures.get(entry.getKey()));
@@ -428,16 +418,16 @@ public class BulkImporter {
return result;
}
- private Map<Path,List<KeyExtent>> assignMapFiles(AccumuloConfiguration acuConf, Instance instance, Configuration conf, Credentials credentials,
- VolumeManager fs, String tableId, Map<Path,List<TabletLocation>> assignments, Collection<Path> paths, int numThreads, int numMapThreads) {
+ private Map<Path,List<KeyExtent>> assignMapFiles(ClientContext context, Configuration conf, VolumeManager fs, String tableId,
+ Map<Path,List<TabletLocation>> assignments, Collection<Path> paths, int numThreads, int numMapThreads) {
timer.start(Timers.EXAMINE_MAP_FILES);
- Map<Path,List<AssignmentInfo>> assignInfo = estimateSizes(acuConf, conf, fs, assignments, paths, numMapThreads);
+ Map<Path,List<AssignmentInfo>> assignInfo = estimateSizes(context.getConfiguration(), conf, fs, assignments, paths, numMapThreads);
timer.stop(Timers.EXAMINE_MAP_FILES);
Map<Path,List<KeyExtent>> ret;
timer.start(Timers.IMPORT_MAP_FILES);
- ret = assignMapFiles(acuConf, credentials, tableId, assignInfo, locationsOf(assignments), numThreads);
+ ret = assignMapFiles(tableId, assignInfo, locationsOf(assignments), numThreads);
timer.stop(Timers.IMPORT_MAP_FILES);
return ret;
@@ -446,17 +436,12 @@ public class BulkImporter {
private class AssignmentTask implements Runnable {
final Map<Path,List<KeyExtent>> assignmentFailures;
String location;
- Credentials credentials;
private Map<KeyExtent,List<PathSize>> assignmentsPerTablet;
- private final AccumuloConfiguration acuConf;
- public AssignmentTask(AccumuloConfiguration acuConf, Credentials credentials, Map<Path,List<KeyExtent>> assignmentFailures, String tableName, String location,
- Map<KeyExtent,List<PathSize>> assignmentsPerTablet) {
- this.acuConf = acuConf;
+ public AssignmentTask(Map<Path,List<KeyExtent>> assignmentFailures, String tableName, String location, Map<KeyExtent,List<PathSize>> assignmentsPerTablet) {
this.assignmentFailures = assignmentFailures;
this.location = location;
this.assignmentsPerTablet = assignmentsPerTablet;
- this.credentials = credentials;
}
private void handleFailures(Collection<KeyExtent> failures, String message) {
@@ -488,7 +473,7 @@ public class BulkImporter {
log.debug("Assigning " + uniqMapFiles.size() + " map files to " + assignmentsPerTablet.size() + " tablets at " + location);
try {
- List<KeyExtent> failures = assignMapFiles(acuConf, credentials, location, assignmentsPerTablet);
+ List<KeyExtent> failures = assignMapFiles(context, location, assignmentsPerTablet);
handleFailures(failures, "Not Serving Tablet");
} catch (AccumuloException e) {
handleFailures(assignmentsPerTablet.keySet(), e.getMessage());
@@ -514,8 +499,7 @@ public class BulkImporter {
}
}
- private Map<Path,List<KeyExtent>> assignMapFiles(AccumuloConfiguration acuConf, Credentials credentials, String tableName, Map<Path,List<AssignmentInfo>> assignments,
- Map<KeyExtent,String> locations, int numThreads) {
+ private Map<Path,List<KeyExtent>> assignMapFiles(String tableName, Map<Path,List<AssignmentInfo>> assignments, Map<KeyExtent,String> locations, int numThreads) {
// group assignments by tablet
Map<KeyExtent,List<PathSize>> assignmentsPerTablet = new TreeMap<KeyExtent,List<PathSize>>();
@@ -575,7 +559,7 @@ public class BulkImporter {
for (Entry<String,Map<KeyExtent,List<PathSize>>> entry : assignmentsPerTabletServer.entrySet()) {
String location = entry.getKey();
- threadPool.submit(new AssignmentTask(acuConf, credentials, assignmentFailures, tableName, location, entry.getValue()));
+ threadPool.submit(new AssignmentTask(assignmentFailures, tableName, location, entry.getValue()));
}
threadPool.shutdown();
@@ -592,11 +576,11 @@ public class BulkImporter {
return assignmentFailures;
}
- private List<KeyExtent> assignMapFiles(AccumuloConfiguration acuConf, Credentials credentials, String location, Map<KeyExtent,List<PathSize>> assignmentsPerTablet)
+ private List<KeyExtent> assignMapFiles(ClientContext context, String location, Map<KeyExtent,List<PathSize>> assignmentsPerTablet)
throws AccumuloException, AccumuloSecurityException {
try {
- long timeInMillis = acuConf.getTimeInMillis(Property.TSERV_BULK_TIMEOUT);
- TabletClientService.Iface client = ThriftUtil.getTServerClient(location, acuConf, timeInMillis);
+ long timeInMillis = context.getConfiguration().getTimeInMillis(Property.TSERV_BULK_TIMEOUT);
+ TabletClientService.Iface client = ThriftUtil.getTServerClient(location, context, 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()) {
@@ -610,8 +594,7 @@ public class BulkImporter {
}
log.debug("Asking " + location + " to bulk load " + files);
- List<TKeyExtent> failures = client.bulkImport(Tracer.traceInfo(), credentials.toThrift(instance), tid, Translator.translate(files, Translators.KET),
- setTime);
+ List<TKeyExtent> failures = client.bulkImport(Tracer.traceInfo(), context.rpcCreds(), tid, Translator.translate(files, Translators.KET), setTime);
return Translator.translate(failures, Translators.TKET);
} finally {
@@ -625,31 +608,29 @@ public class BulkImporter {
}
}
- public static List<TabletLocation> findOverlappingTablets(AccumuloConfiguration acuConf, VolumeManager fs, TabletLocator locator, Path file,
- Credentials credentials) throws Exception {
- return findOverlappingTablets(acuConf, fs, locator, file, null, null, credentials);
+ public static List<TabletLocation> findOverlappingTablets(ClientContext context, VolumeManager fs, TabletLocator locator, Path file) throws Exception {
+ return findOverlappingTablets(context, fs, locator, file, null, null);
}
- public static List<TabletLocation> findOverlappingTablets(AccumuloConfiguration acuConf, VolumeManager fs, TabletLocator locator, Path file,
- KeyExtent failed,
- Credentials credentials) throws Exception {
+ public static List<TabletLocation> findOverlappingTablets(ClientContext context, VolumeManager fs, TabletLocator locator, Path file, KeyExtent failed)
+ throws Exception {
locator.invalidateCache(failed);
Text start = failed.getPrevEndRow();
if (start != null)
start = Range.followingPrefix(start);
- return findOverlappingTablets(acuConf, fs, locator, file, start, failed.getEndRow(), credentials);
+ return findOverlappingTablets(context, fs, locator, file, start, failed.getEndRow());
}
final static byte[] byte0 = {0};
- public static List<TabletLocation> findOverlappingTablets(AccumuloConfiguration acuConf, VolumeManager vm, TabletLocator locator, Path file, Text startRow,
- Text endRow, Credentials credentials) throws Exception {
+ public static List<TabletLocation> findOverlappingTablets(ClientContext context, VolumeManager vm, TabletLocator locator, Path file, Text startRow,
+ Text endRow) throws Exception {
List<TabletLocation> result = new ArrayList<TabletLocation>();
Collection<ByteSequence> columnFamilies = Collections.emptyList();
String filename = file.toString();
// log.debug(filename + " finding overlapping tablets " + startRow + " -> " + endRow);
FileSystem fs = vm.getVolumeByPath(file).getFileSystem();
- FileSKVIterator reader = FileOperations.getInstance().openReader(filename, true, fs, fs.getConf(), acuConf);
+ FileSKVIterator reader = FileOperations.getInstance().openReader(filename, true, fs, fs.getConf(), context.getConfiguration());
try {
Text row = startRow;
if (row == null)
@@ -662,7 +643,7 @@ public class BulkImporter {
break;
}
row = reader.getTopKey().getRow();
- TabletLocation tabletLocation = locator.locateTablet(credentials, row, false, true);
+ TabletLocation tabletLocation = locator.locateTablet(context, row, false, true);
// log.debug(filename + " found row " + row + " at location " + tabletLocation);
result.add(tabletLocation);
row = tabletLocation.tablet_extent.getEndRow();
http://git-wip-us.apache.org/repos/asf/accumulo/blob/42c25faa/server/base/src/main/java/org/apache/accumulo/server/client/ClientServiceHandler.java
----------------------------------------------------------------------
diff --git a/server/base/src/main/java/org/apache/accumulo/server/client/ClientServiceHandler.java b/server/base/src/main/java/org/apache/accumulo/server/client/ClientServiceHandler.java
index 1b54ebd..e4d6040 100644
--- a/server/base/src/main/java/org/apache/accumulo/server/client/ClientServiceHandler.java
+++ b/server/base/src/main/java/org/apache/accumulo/server/client/ClientServiceHandler.java
@@ -55,6 +55,7 @@ import org.apache.accumulo.core.security.SystemPermission;
import org.apache.accumulo.core.security.TablePermission;
import org.apache.accumulo.core.security.thrift.TCredentials;
import org.apache.accumulo.core.trace.thrift.TInfo;
+import org.apache.accumulo.server.AccumuloServerContext;
import org.apache.accumulo.server.conf.ServerConfigurationFactory;
import org.apache.accumulo.server.fs.VolumeManager;
import org.apache.accumulo.server.security.AuditedSecurityOperation;
@@ -67,15 +68,18 @@ import org.apache.thrift.TException;
public class ClientServiceHandler implements ClientService.Iface {
private static final Logger log = Logger.getLogger(ClientServiceHandler.class);
- private static SecurityOperation security = AuditedSecurityOperation.getInstance();
protected final TransactionWatcher transactionWatcher;
+ private final AccumuloServerContext context;
private final Instance instance;
private final VolumeManager fs;
+ private final SecurityOperation security;
- public ClientServiceHandler(Instance instance, TransactionWatcher transactionWatcher, VolumeManager fs) {
- this.instance = instance;
+ public ClientServiceHandler(AccumuloServerContext context, TransactionWatcher transactionWatcher, VolumeManager fs) {
+ this.context = context;
+ this.instance = context.getInstance();
this.transactionWatcher = transactionWatcher;
this.fs = fs;
+ this.security = AuditedSecurityOperation.getInstance(context);
}
public static String checkTableId(Instance instance, String tableName, TableOperation operation) throws ThriftTableOperationException {
@@ -238,7 +242,7 @@ public class ClientServiceHandler implements ClientService.Iface {
return security.listUsers(credentials);
}
- private static Map<String,String> conf(TCredentials credentials, AccumuloConfiguration conf) throws TException {
+ private Map<String,String> conf(TCredentials credentials, AccumuloConfiguration conf) throws TException {
security.authenticateUser(credentials, credentials);
conf.invalidateCache();
@@ -253,7 +257,7 @@ public class ClientServiceHandler implements ClientService.Iface {
@Override
public Map<String,String> getConfiguration(TInfo tinfo, TCredentials credentials, ConfigurationType type) throws TException {
- ServerConfigurationFactory factory = new ServerConfigurationFactory(instance);
+ ServerConfigurationFactory factory = context.getServerConfigurationFactory();
switch (type) {
case CURRENT:
return conf(credentials, factory.getConfiguration());
@@ -268,7 +272,7 @@ public class ClientServiceHandler implements ClientService.Iface {
@Override
public Map<String,String> getTableConfiguration(TInfo tinfo, TCredentials credentials, String tableName) throws TException, ThriftTableOperationException {
String tableId = checkTableId(instance, tableName, null);
- AccumuloConfiguration config = new ServerConfigurationFactory(instance).getTableConfiguration(tableId);
+ AccumuloConfiguration config = context.getServerConfigurationFactory().getTableConfiguration(tableId);
return conf(credentials, config);
}
@@ -282,7 +286,7 @@ public class ClientServiceHandler implements ClientService.Iface {
return transactionWatcher.run(Constants.BULK_ARBITRATOR_TYPE, tid, new Callable<List<String>>() {
@Override
public List<String> call() throws Exception {
- return BulkImporter.bulkLoad(new ServerConfigurationFactory(instance).getConfiguration(), instance, Credentials.fromThrift(credentials), tid, tableId, files, errorDir, setTime);
+ return BulkImporter.bulkLoad(context, tid, tableId, files, errorDir, setTime);
}
});
} catch (AccumuloSecurityException e) {
@@ -337,7 +341,7 @@ public class ClientServiceHandler implements ClientService.Iface {
try {
shouldMatch = loader.loadClass(interfaceMatch);
- AccumuloConfiguration conf = new ServerConfigurationFactory(instance).getTableConfiguration(tableId);
+ AccumuloConfiguration conf = context.getServerConfigurationFactory().getTableConfiguration(tableId);
String context = conf.get(Property.TABLE_CLASSPATH);
@@ -371,7 +375,7 @@ public class ClientServiceHandler implements ClientService.Iface {
try {
shouldMatch = loader.loadClass(interfaceMatch);
- AccumuloConfiguration conf = new ServerConfigurationFactory(instance).getNamespaceConfiguration(namespaceId);
+ AccumuloConfiguration conf = context.getServerConfigurationFactory().getNamespaceConfiguration(namespaceId);
String context = conf.get(Property.TABLE_CLASSPATH);
@@ -410,7 +414,7 @@ public class ClientServiceHandler implements ClientService.Iface {
}
// use the same set of tableIds that were validated above to avoid race conditions
- Map<TreeSet<String>,Long> diskUsage = TableDiskUsage.getDiskUsage(new ServerConfigurationFactory(instance).getConfiguration(), tableIds, fs, conn);
+ Map<TreeSet<String>,Long> diskUsage = TableDiskUsage.getDiskUsage(context.getServerConfigurationFactory().getConfiguration(), tableIds, fs, conn);
List<TDiskUsage> retUsages = new ArrayList<TDiskUsage>();
for (Map.Entry<TreeSet<String>,Long> usageItem : diskUsage.entrySet()) {
retUsages.add(new TDiskUsage(new ArrayList<String>(usageItem.getKey()), usageItem.getValue()));
@@ -435,7 +439,7 @@ public class ClientServiceHandler implements ClientService.Iface {
String why = "Could not find namespace while getting configuration.";
throw new ThriftTableOperationException(null, ns, null, TableOperationExceptionType.NAMESPACE_NOTFOUND, why);
}
- AccumuloConfiguration config = new ServerConfigurationFactory(instance).getNamespaceConfiguration(namespaceId);
+ AccumuloConfiguration config = context.getServerConfigurationFactory().getNamespaceConfiguration(namespaceId);
return conf(credentials, config);
}
}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/42c25faa/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 efbd882..4ab9f90 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
@@ -30,6 +30,7 @@ import org.apache.accumulo.core.client.AccumuloSecurityException;
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.impl.ClientContext;
import org.apache.accumulo.core.client.impl.ConnectorImpl;
import org.apache.accumulo.core.client.security.tokens.AuthenticationToken;
import org.apache.accumulo.core.client.security.tokens.PasswordToken;
@@ -154,7 +155,7 @@ public class HdfsZooInstance implements Instance {
@Override
public Connector getConnector(String principal, AuthenticationToken token) throws AccumuloException, AccumuloSecurityException {
- return new ConnectorImpl(this, new Credentials(principal, token));
+ return new ConnectorImpl(new ClientContext(this, new Credentials(principal, token), SiteConfiguration.getInstance()));
}
@Deprecated
http://git-wip-us.apache.org/repos/asf/accumulo/blob/42c25faa/server/base/src/main/java/org/apache/accumulo/server/conf/ServerConfigurationFactory.java
----------------------------------------------------------------------
diff --git a/server/base/src/main/java/org/apache/accumulo/server/conf/ServerConfigurationFactory.java b/server/base/src/main/java/org/apache/accumulo/server/conf/ServerConfigurationFactory.java
index 7e36b40..8bcb5a7 100644
--- a/server/base/src/main/java/org/apache/accumulo/server/conf/ServerConfigurationFactory.java
+++ b/server/base/src/main/java/org/apache/accumulo/server/conf/ServerConfigurationFactory.java
@@ -138,6 +138,7 @@ public class ServerConfigurationFactory extends ServerConfiguration {
return defaultConfig;
}
+ @Override
public synchronized AccumuloConfiguration getConfiguration() {
if (systemConfig == null) {
checkPermissions();
@@ -146,12 +147,13 @@ public class ServerConfigurationFactory extends ServerConfiguration {
return systemConfig;
}
+ @Override
public TableConfiguration getTableConfiguration(String tableId) {
checkPermissions();
synchronized (tableConfigs) {
TableConfiguration conf = tableConfigs.get(instanceID).get(tableId);
if (conf == null && Tables.exists(instance, tableId)) {
- conf = new TableConfiguration(instance.getInstanceID(), tableId, getNamespaceConfigurationForTable(tableId));
+ conf = new TableConfiguration(instance, tableId, getNamespaceConfigurationForTable(tableId));
ConfigSanityCheck.validate(conf);
tableConfigs.get(instanceID).put(tableId, conf);
}
@@ -159,6 +161,7 @@ public class ServerConfigurationFactory extends ServerConfiguration {
}
}
+ @Override
public TableConfiguration getTableConfiguration(KeyExtent extent) {
return getTableConfiguration(extent.getTableId().toString());
}
@@ -177,6 +180,7 @@ public class ServerConfigurationFactory extends ServerConfiguration {
}
}
+ @Override
public NamespaceConfiguration getNamespaceConfiguration(String namespaceId) {
checkPermissions();
synchronized (namespaceConfigs) {
@@ -192,6 +196,7 @@ public class ServerConfigurationFactory extends ServerConfiguration {
}
}
+ @Override
public Instance getInstance() {
return instance;
}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/42c25faa/server/base/src/main/java/org/apache/accumulo/server/conf/TableConfiguration.java
----------------------------------------------------------------------
diff --git a/server/base/src/main/java/org/apache/accumulo/server/conf/TableConfiguration.java b/server/base/src/main/java/org/apache/accumulo/server/conf/TableConfiguration.java
index 8cf46c7..7dced80 100644
--- a/server/base/src/main/java/org/apache/accumulo/server/conf/TableConfiguration.java
+++ b/server/base/src/main/java/org/apache/accumulo/server/conf/TableConfiguration.java
@@ -26,7 +26,6 @@ import org.apache.accumulo.core.conf.Property;
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.server.client.HdfsZooInstance;
import org.apache.accumulo.server.conf.ZooCachePropertyAccessor.PropCacheKey;
import org.apache.log4j.Logger;
@@ -36,19 +35,13 @@ public class TableConfiguration extends ObservableConfiguration {
private static final Map<PropCacheKey,ZooCache> propCaches = new java.util.HashMap<PropCacheKey,ZooCache>();
private ZooCachePropertyAccessor propCacheAccessor = null;
- private final String instanceId;
private final Instance instance;
private final NamespaceConfiguration parent;
private ZooCacheFactory zcf = new ZooCacheFactory();
- private String table = null;
+ private final String table;
- public TableConfiguration(String instanceId, String table, NamespaceConfiguration parent) {
- this(instanceId, HdfsZooInstance.getInstance(), table, parent);
- }
-
- public TableConfiguration(String instanceId, Instance instance, String table, NamespaceConfiguration parent) {
- this.instanceId = instanceId;
+ public TableConfiguration(Instance instance, String table, NamespaceConfiguration parent) {
this.instance = instance;
this.table = table;
this.parent = parent;
@@ -95,7 +88,7 @@ public class TableConfiguration extends ObservableConfiguration {
}
private String getPath() {
- return ZooUtil.getRoot(instanceId) + Constants.ZTABLES + "/" + table + Constants.ZTABLE_CONF;
+ return ZooUtil.getRoot(instance.getInstanceID()) + Constants.ZTABLES + "/" + table + Constants.ZTABLE_CONF;
}
@Override
http://git-wip-us.apache.org/repos/asf/accumulo/blob/42c25faa/server/base/src/main/java/org/apache/accumulo/server/fs/VolumeUtil.java
----------------------------------------------------------------------
diff --git a/server/base/src/main/java/org/apache/accumulo/server/fs/VolumeUtil.java b/server/base/src/main/java/org/apache/accumulo/server/fs/VolumeUtil.java
index baad742..6ebbe1e 100644
--- a/server/base/src/main/java/org/apache/accumulo/server/fs/VolumeUtil.java
+++ b/server/base/src/main/java/org/apache/accumulo/server/fs/VolumeUtil.java
@@ -31,13 +31,12 @@ import org.apache.accumulo.core.metadata.schema.DataFileValue;
import org.apache.accumulo.core.protobuf.ProtobufUtil;
import org.apache.accumulo.core.replication.StatusUtil;
import org.apache.accumulo.core.replication.proto.Replication.Status;
-import org.apache.accumulo.core.security.Credentials;
import org.apache.accumulo.core.tabletserver.log.LogEntry;
import org.apache.accumulo.core.util.CachedConfiguration;
import org.apache.accumulo.core.util.Pair;
+import org.apache.accumulo.server.AccumuloServerContext;
import org.apache.accumulo.server.ServerConstants;
import org.apache.accumulo.server.fs.VolumeManager.FileType;
-import org.apache.accumulo.server.security.SystemCredentials;
import org.apache.accumulo.server.util.MetadataTableUtil;
import org.apache.accumulo.server.util.ReplicationTableUtil;
import org.apache.accumulo.server.zookeeper.ZooLock;
@@ -184,7 +183,8 @@ public class VolumeUtil {
* This method does two things. First, it switches any volumes a tablet is using that are configured in instance.volumes.replacements. Second, if a tablet dir
* is no longer configured for use it chooses a new tablet directory.
*/
- public static TabletFiles updateTabletVolumes(ZooLock zooLock, VolumeManager vm, KeyExtent extent, TabletFiles tabletFiles, boolean replicate) throws IOException {
+ public static TabletFiles updateTabletVolumes(AccumuloServerContext context, ZooLock zooLock, VolumeManager vm, KeyExtent extent, TabletFiles tabletFiles,
+ boolean replicate) throws IOException {
List<Pair<Path,Path>> replacements = ServerConstants.getVolumeReplacements();
log.trace("Using volume replacements: " + replacements);
@@ -235,26 +235,26 @@ public class VolumeUtil {
}
if (logsToRemove.size() + filesToRemove.size() > 0 || switchedDir != null) {
- Credentials creds = SystemCredentials.get();
- MetadataTableUtil.updateTabletVolumes(extent, logsToRemove, logsToAdd, filesToRemove, filesToAdd, switchedDir, zooLock, creds);
+ MetadataTableUtil.updateTabletVolumes(extent, logsToRemove, logsToAdd, filesToRemove, filesToAdd, switchedDir, zooLock, context);
if (replicate) {
Status status = StatusUtil.fileClosed();
log.debug("Tablet directory switched, need to record old log files " + logsToRemove + " " + ProtobufUtil.toString(status));
// Before deleting these logs, we need to mark them for replication
for (LogEntry logEntry : logsToRemove) {
- ReplicationTableUtil.updateFiles(creds, extent, logEntry.logSet, status);
+ ReplicationTableUtil.updateFiles(context, extent, logEntry.logSet, status);
}
}
}
- ret.dir = decommisionedTabletDir(zooLock, vm, extent, tabletDir);
+ ret.dir = decommisionedTabletDir(context, zooLock, vm, extent, tabletDir);
// method this should return the exact strings that are in the metadata table
return ret;
}
- private static String decommisionedTabletDir(ZooLock zooLock, VolumeManager vm, KeyExtent extent, String metaDir) throws IOException {
+ private static String decommisionedTabletDir(AccumuloServerContext context, ZooLock zooLock, VolumeManager vm, KeyExtent extent, String metaDir)
+ throws IOException {
Path dir = new Path(metaDir);
if (isActiveVolume(dir))
return metaDir;
@@ -309,7 +309,7 @@ public class VolumeUtil {
return newDir.toString();
} else {
- MetadataTableUtil.updateTabletDir(extent, newDir.toString(), SystemCredentials.get(), zooLock);
+ MetadataTableUtil.updateTabletDir(extent, newDir.toString(), context, zooLock);
return newDir.toString();
}
}