You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@accumulo.apache.org by GitBox <gi...@apache.org> on 2019/01/11 16:21:17 UTC
[GitHub] milleruntime closed pull request #862: Fix TokenFile in MapReduce
milleruntime closed pull request #862: Fix TokenFile in MapReduce
URL: https://github.com/apache/accumulo/pull/862
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git a/core/src/main/java/org/apache/accumulo/core/client/mapred/AbstractInputFormat.java b/core/src/main/java/org/apache/accumulo/core/client/mapred/AbstractInputFormat.java
index 60f824e8b1..fe7d5b8483 100644
--- a/core/src/main/java/org/apache/accumulo/core/client/mapred/AbstractInputFormat.java
+++ b/core/src/main/java/org/apache/accumulo/core/client/mapred/AbstractInputFormat.java
@@ -34,7 +34,6 @@
import org.apache.accumulo.core.client.Accumulo;
import org.apache.accumulo.core.client.AccumuloClient;
-import org.apache.accumulo.core.client.AccumuloException;
import org.apache.accumulo.core.client.AccumuloSecurityException;
import org.apache.accumulo.core.client.BatchScanner;
import org.apache.accumulo.core.client.ClientSideIteratorScanner;
@@ -53,14 +52,15 @@
import org.apache.accumulo.core.client.security.tokens.KerberosToken;
import org.apache.accumulo.core.client.security.tokens.PasswordToken;
import org.apache.accumulo.core.clientImpl.AuthenticationTokenIdentifier;
+import org.apache.accumulo.core.clientImpl.ClientConfConverter;
import org.apache.accumulo.core.clientImpl.ClientContext;
-import org.apache.accumulo.core.clientImpl.ClientInfo;
import org.apache.accumulo.core.clientImpl.DelegationTokenImpl;
import org.apache.accumulo.core.clientImpl.OfflineScanner;
import org.apache.accumulo.core.clientImpl.ScannerImpl;
import org.apache.accumulo.core.clientImpl.Table;
import org.apache.accumulo.core.clientImpl.Tables;
import org.apache.accumulo.core.clientImpl.TabletLocator;
+import org.apache.accumulo.core.clientImpl.mapreduce.lib.InputConfigurator;
import org.apache.accumulo.core.data.Key;
import org.apache.accumulo.core.data.Range;
import org.apache.accumulo.core.data.Value;
@@ -630,13 +630,6 @@ public float getProgress() throws IOException {
}
- Map<String,Map<KeyExtent,List<Range>>> binOfflineTable(JobConf job, Table.ID tableId,
- List<Range> ranges)
- throws TableNotFoundException, AccumuloException, AccumuloSecurityException {
- return Configurator.binOffline(tableId, ranges,
- ClientInfo.from(Configurator.getClientProperties(CLASS, job)));
- }
-
/**
* Gets the splits of the tables that have been set on the job by reading the metadata table for
* the specified ranges.
@@ -662,8 +655,8 @@ public float getProgress() throws IOException {
org.apache.accumulo.core.client.mapreduce.InputTableConfig tableConfig = tableConfigEntry
.getValue();
- try (
- ClientContext context = new ClientContext(Configurator.getClientProperties(CLASS, job))) {
+ try (ClientContext context = new ClientContext(
+ ClientConfConverter.toProperties(getClientConfiguration(job)))) {
Table.ID tableId;
// resolve table name to id once, and use id from this point forward
try {
@@ -696,12 +689,12 @@ public float getProgress() throws IOException {
TabletLocator tl;
try {
if (tableConfig.isOfflineScan()) {
- binnedRanges = binOfflineTable(job, tableId, ranges);
+ binnedRanges = InputConfigurator.binOffline(tableId, ranges, context);
while (binnedRanges == null) {
// Some tablets were still online, try again
// sleep randomly between 100 and 200 ms
sleepUninterruptibly(100 + random.nextInt(100), TimeUnit.MILLISECONDS);
- binnedRanges = binOfflineTable(job, tableId, ranges);
+ binnedRanges = InputConfigurator.binOffline(tableId, ranges, context);
}
} else {
tl = TabletLocator.getLocator(context, tableId);
diff --git a/core/src/main/java/org/apache/accumulo/core/client/mapreduce/AbstractInputFormat.java b/core/src/main/java/org/apache/accumulo/core/client/mapreduce/AbstractInputFormat.java
index 7f9aab6460..92b13ac113 100644
--- a/core/src/main/java/org/apache/accumulo/core/client/mapreduce/AbstractInputFormat.java
+++ b/core/src/main/java/org/apache/accumulo/core/client/mapreduce/AbstractInputFormat.java
@@ -28,13 +28,11 @@
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
+import java.util.Properties;
import java.util.Random;
import java.util.concurrent.TimeUnit;
import java.util.stream.Stream;
-import org.apache.accumulo.core.client.Accumulo;
-import org.apache.accumulo.core.client.AccumuloClient;
-import org.apache.accumulo.core.client.AccumuloException;
import org.apache.accumulo.core.client.AccumuloSecurityException;
import org.apache.accumulo.core.client.BatchScanner;
import org.apache.accumulo.core.client.ClientSideIteratorScanner;
@@ -53,14 +51,15 @@
import org.apache.accumulo.core.client.security.tokens.KerberosToken;
import org.apache.accumulo.core.client.security.tokens.PasswordToken;
import org.apache.accumulo.core.clientImpl.AuthenticationTokenIdentifier;
+import org.apache.accumulo.core.clientImpl.ClientConfConverter;
import org.apache.accumulo.core.clientImpl.ClientContext;
-import org.apache.accumulo.core.clientImpl.ClientInfo;
import org.apache.accumulo.core.clientImpl.DelegationTokenImpl;
import org.apache.accumulo.core.clientImpl.OfflineScanner;
import org.apache.accumulo.core.clientImpl.ScannerImpl;
import org.apache.accumulo.core.clientImpl.Table;
import org.apache.accumulo.core.clientImpl.Tables;
import org.apache.accumulo.core.clientImpl.TabletLocator;
+import org.apache.accumulo.core.conf.ClientProperty;
import org.apache.accumulo.core.data.Key;
import org.apache.accumulo.core.data.Range;
import org.apache.accumulo.core.data.Value;
@@ -146,10 +145,10 @@ public static void setConnectorInfo(Job job, String principal, AuthenticationTok
throws AccumuloSecurityException {
if (token instanceof KerberosToken) {
log.info("Received KerberosToken, attempting to fetch DelegationToken");
- try (AccumuloClient client = Accumulo.newClient()
- .from(Configurator.getClientProperties(CLASS, job.getConfiguration()))
- .as(principal, token).build()) {
- token = client.securityOperations().getDelegationToken(new DelegationTokenConfig());
+ try {
+ org.apache.accumulo.core.client.Instance instance = getInstance(job);
+ org.apache.accumulo.core.client.Connector conn = instance.getConnector(principal, token);
+ token = conn.securityOperations().getDelegationToken(new DelegationTokenConfig());
} catch (Exception e) {
log.warn("Failed to automatically obtain DelegationToken, "
+ "Mappers/Reducers will likely fail to communicate with Accumulo", e);
@@ -500,13 +499,29 @@ protected void setupIterators(TaskAttemptContext context, Scanner scanner, Strin
public void initialize(InputSplit inSplit, TaskAttemptContext attempt) throws IOException {
split = (RangeInputSplit) inSplit;
- log.debug("Initializing input split: " + split);
+ Table.ID tableId = Table.ID.of(split.getTableId());
+ log.debug("Initializing input split: " + split.toString());
- if (context == null) {
- context = new ClientContext(
- Configurator.getClientProperties(CLASS, attempt.getConfiguration()));
+ org.apache.accumulo.core.client.Instance instance = split
+ .getInstance(getClientConfiguration(attempt));
+ if (null == instance) {
+ instance = getInstance(attempt);
+ }
+
+ String principal = split.getPrincipal();
+ if (null == principal) {
+ principal = getPrincipal(attempt);
+ }
+
+ AuthenticationToken token = split.getToken();
+ if (null == token) {
+ token = getAuthenticationToken(attempt);
+ }
+
+ Authorizations authorizations = split.getAuths();
+ if (null == authorizations) {
+ authorizations = getScanAuthorizations(attempt);
}
- Authorizations authorizations = getScanAuthorizations(attempt);
String classLoaderContext = getClassLoaderContext(attempt);
String table = split.getTableName();
@@ -515,7 +530,7 @@ public void initialize(InputSplit inSplit, TaskAttemptContext attempt) throws IO
// but the scanner will use the table id resolved at job setup time
InputTableConfig tableConfig = getInputTableConfig(attempt, split.getTableName());
- log.debug("Creating client with user: " + context.whoami());
+ log.debug("Creating connector with user: " + principal);
log.debug("Creating scanner for table: " + table);
log.debug("Authorizations are: " + authorizations);
@@ -527,7 +542,8 @@ public void initialize(InputSplit inSplit, TaskAttemptContext attempt) throws IO
// Note: BatchScanner will use at most one thread per tablet, currently BatchInputSplit
// will not span tablets
int scanThreads = 1;
- scanner = context.createBatchScanner(split.getTableName(), authorizations, scanThreads);
+ scanner = instance.getConnector(principal, token).createBatchScanner(split.getTableName(),
+ authorizations, scanThreads);
setupIterators(attempt, scanner, split.getTableName(), split);
if (classLoaderContext != null) {
scanner.setClassLoaderContext(classLoaderContext);
@@ -558,12 +574,15 @@ public void initialize(InputSplit inSplit, TaskAttemptContext attempt) throws IO
}
try {
+ org.apache.accumulo.core.client.ClientConfiguration clientConf = getClientConfiguration(
+ attempt);
+ ClientContext context = new ClientContext(ClientConfConverter.toProperties(clientConf));
if (isOffline) {
- scanner = new OfflineScanner(context, Table.ID.of(split.getTableId()), authorizations);
+ scanner = new OfflineScanner(context, tableId, authorizations);
} else {
// Not using public API to create scanner so that we can use table ID
// Table ID is used in case of renames during M/R job
- scanner = new ScannerImpl(context, Table.ID.of(split.getTableId()), authorizations);
+ scanner = new ScannerImpl(context, tableId, authorizations);
}
if (isIsolated) {
log.info("Creating isolated scanner");
@@ -671,13 +690,6 @@ public V getCurrentValue() throws IOException, InterruptedException {
}
}
- Map<String,Map<KeyExtent,List<Range>>> binOfflineTable(JobContext context, Table.ID tableId,
- List<Range> ranges)
- throws TableNotFoundException, AccumuloException, AccumuloSecurityException {
- return Configurator.binOffline(tableId, ranges,
- ClientInfo.from(Configurator.getClientProperties(CLASS, context.getConfiguration())));
- }
-
/**
* Gets the splits of the tables that have been set on the job by reading the metadata table for
* the specified ranges.
@@ -695,8 +707,10 @@ public V getCurrentValue() throws IOException, InterruptedException {
Random random = new SecureRandom();
LinkedList<InputSplit> splits = new LinkedList<>();
Map<String,InputTableConfig> tableConfigs = getInputTableConfigs(context);
- try (AccumuloClient client = Accumulo.newClient()
- .from(Configurator.getClientProperties(CLASS, context.getConfiguration())).build()) {
+ Properties props = ClientConfConverter.toProperties(getClientConfiguration(context));
+ props.setProperty(ClientProperty.AUTH_PRINCIPAL.getKey(), getPrincipal(context));
+ ClientProperty.setAuthenticationToken(props, getAuthenticationToken(context));
+ try (ClientContext client = new ClientContext(props)) {
for (Map.Entry<String,InputTableConfig> tableConfigEntry : tableConfigs.entrySet()) {
String tableName = tableConfigEntry.getKey();
@@ -705,12 +719,15 @@ public V getCurrentValue() throws IOException, InterruptedException {
Table.ID tableId;
// resolve table name to id once, and use id from this point forward
try {
- tableId = Tables.getTableId((ClientContext) client, tableName);
+ tableId = Tables.getTableId(client, tableName);
} catch (TableNotFoundException e) {
throw new IOException(e);
}
- boolean batchScan = Configurator.isBatchScan(CLASS, context.getConfiguration());
+ Authorizations auths = getScanAuthorizations(context);
+
+ boolean batchScan = org.apache.accumulo.core.clientImpl.mapreduce.lib.InputConfigurator
+ .isBatchScan(CLASS, context.getConfiguration());
boolean supportBatchScan = !(tableConfig.isOfflineScan()
|| tableConfig.shouldUseIsolatedScanners() || tableConfig.shouldUseLocalIterators());
if (batchScan && !supportBatchScan)
@@ -734,27 +751,28 @@ public V getCurrentValue() throws IOException, InterruptedException {
TabletLocator tl;
try {
if (tableConfig.isOfflineScan()) {
- binnedRanges = binOfflineTable(context, tableId, ranges);
+ binnedRanges = org.apache.accumulo.core.clientImpl.mapreduce.lib.InputConfigurator
+ .binOffline(tableId, ranges, client);
while (binnedRanges == null) {
// Some tablets were still online, try again
// sleep randomly between 100 and 200 ms
sleepUninterruptibly(100 + random.nextInt(100), TimeUnit.MILLISECONDS);
- binnedRanges = binOfflineTable(context, tableId, ranges);
+ binnedRanges = org.apache.accumulo.core.clientImpl.mapreduce.lib.InputConfigurator
+ .binOffline(tableId, ranges, client);
}
} else {
- tl = TabletLocator.getLocator((ClientContext) client, tableId);
+ tl = TabletLocator.getLocator(client, tableId);
// its possible that the cache could contain complete, but old information about a
// tables tablets... so clear it
tl.invalidateCache();
- while (!tl.binRanges((ClientContext) client, ranges, binnedRanges).isEmpty()) {
+ while (!tl.binRanges(client, ranges, binnedRanges).isEmpty()) {
String tableIdStr = tableId.canonicalID();
- if (!Tables.exists((ClientContext) client, tableId))
+ if (!Tables.exists(client, tableId))
throw new TableDeletedException(tableIdStr);
- if (Tables.getTableState((ClientContext) client, tableId) == TableState.OFFLINE)
- throw new TableOfflineException(
- Tables.getTableOfflineMsg((ClientContext) client, tableId));
+ if (Tables.getTableState(client, tableId) == TableState.OFFLINE)
+ throw new TableOfflineException(Tables.getTableOfflineMsg(client, tableId));
binnedRanges.clear();
log.warn("Unable to locate bins for specified ranges. Retrying.");
// sleep randomly between 100 and 200 ms
diff --git a/core/src/main/java/org/apache/accumulo/core/client/mapreduce/AccumuloOutputFormat.java b/core/src/main/java/org/apache/accumulo/core/client/mapreduce/AccumuloOutputFormat.java
index 82214b393f..3d726e8ec3 100644
--- a/core/src/main/java/org/apache/accumulo/core/client/mapreduce/AccumuloOutputFormat.java
+++ b/core/src/main/java/org/apache/accumulo/core/client/mapreduce/AccumuloOutputFormat.java
@@ -28,6 +28,7 @@
import org.apache.accumulo.core.client.AccumuloSecurityException;
import org.apache.accumulo.core.client.BatchWriter;
import org.apache.accumulo.core.client.BatchWriterConfig;
+import org.apache.accumulo.core.client.Connector;
import org.apache.accumulo.core.client.MultiTableBatchWriter;
import org.apache.accumulo.core.client.MutationsRejectedException;
import org.apache.accumulo.core.client.TableExistsException;
@@ -583,10 +584,11 @@ public void close(TaskAttemptContext attempt) throws IOException, InterruptedExc
public void checkOutputSpecs(JobContext job) throws IOException {
if (!isConnectorInfoSet(job))
throw new IOException("Connector info has not been set.");
- String principal = getPrincipal(job);
- AuthenticationToken token = getAuthenticationToken(job);
- try (AccumuloClient c = Accumulo.newClient()
- .from(Configurator.getClientProperties(CLASS, job.getConfiguration())).build()) {
+ try {
+ // if the instance isn't configured, it will complain here
+ String principal = getPrincipal(job);
+ AuthenticationToken token = getAuthenticationToken(job);
+ Connector c = getInstance(job).getConnector(principal, token);
if (!c.securityOperations().authenticateUser(principal, token))
throw new IOException("Unable to authenticate user");
} catch (AccumuloException | AccumuloSecurityException e) {
diff --git a/core/src/main/java/org/apache/accumulo/core/clientImpl/mapreduce/lib/ConfiguratorBase.java b/core/src/main/java/org/apache/accumulo/core/clientImpl/mapreduce/lib/ConfiguratorBase.java
index cfccdb4837..df63c311eb 100644
--- a/core/src/main/java/org/apache/accumulo/core/clientImpl/mapreduce/lib/ConfiguratorBase.java
+++ b/core/src/main/java/org/apache/accumulo/core/clientImpl/mapreduce/lib/ConfiguratorBase.java
@@ -17,6 +17,7 @@
package org.apache.accumulo.core.clientImpl.mapreduce.lib;
import static com.google.common.base.Preconditions.checkArgument;
+import static java.nio.charset.StandardCharsets.UTF_8;
import static java.util.Objects.requireNonNull;
import java.io.ByteArrayInputStream;
@@ -26,6 +27,7 @@
import java.io.StringWriter;
import java.net.URI;
import java.net.URISyntaxException;
+import java.util.Base64;
import java.util.Properties;
import java.util.Scanner;
@@ -36,8 +38,6 @@
import org.apache.accumulo.core.client.security.tokens.AuthenticationToken;
import org.apache.accumulo.core.client.security.tokens.KerberosToken;
import org.apache.accumulo.core.clientImpl.AuthenticationTokenIdentifier;
-import org.apache.accumulo.core.clientImpl.ClientConfConverter;
-import org.apache.accumulo.core.clientImpl.ClientInfo;
import org.apache.accumulo.core.clientImpl.DelegationTokenImpl;
import org.apache.accumulo.core.conf.ClientProperty;
import org.apache.hadoop.conf.Configuration;
@@ -85,6 +85,15 @@ public String prefix() {
}
}
+ /**
+ * Configuration keys for available {@link Instance} types.
+ *
+ * @since 1.6.0
+ */
+ public static enum InstanceOpts {
+ TYPE, NAME, ZOO_KEEPERS, CLIENT_CONFIG;
+ }
+
public enum ClientOpts {
CLIENT_PROPS, CLIENT_PROPS_FILE
}
@@ -230,12 +239,12 @@ public static Properties getClientProperties(Class<?> implementingClass, Configu
*/
public static void setConnectorInfo(Class<?> implementingClass, Configuration conf,
String principal, AuthenticationToken token) {
+ if (isConnectorInfoSet(implementingClass, conf))
+ throw new IllegalStateException("Connector info for " + implementingClass.getSimpleName()
+ + " can only be set once per job");
+
checkArgument(principal != null, "principal is null");
checkArgument(token != null, "token is null");
- Properties props = getClientProperties(implementingClass, conf);
- props.setProperty(ClientProperty.AUTH_PRINCIPAL.getKey(), principal);
- ClientProperty.setAuthenticationToken(props, token);
- setClientProperties(implementingClass, conf, props);
conf.setBoolean(enumToConfKey(implementingClass, ConnectorInfo.IS_CONFIGURED), true);
}
@@ -306,8 +315,7 @@ public static Boolean isConnectorInfoSet(Class<?> implementingClass, Configurati
* @see #setConnectorInfo(Class, Configuration, String, AuthenticationToken)
*/
public static String getPrincipal(Class<?> implementingClass, Configuration conf) {
- Properties props = getClientProperties(implementingClass, conf);
- return props.getProperty(ClientProperty.AUTH_PRINCIPAL.getKey());
+ return conf.get(enumToConfKey(implementingClass, ConnectorInfo.PRINCIPAL));
}
/**
@@ -321,11 +329,76 @@ public static String getPrincipal(Class<?> implementingClass, Configuration conf
* @return the principal's authentication token
* @since 1.6.0
* @see #setConnectorInfo(Class, Configuration, String, AuthenticationToken)
+ * @see #setConnectorInfo(Class, Configuration, String, String)
*/
public static AuthenticationToken getAuthenticationToken(Class<?> implementingClass,
Configuration conf) {
- Properties props = getClientProperties(implementingClass, conf);
- return ClientProperty.getAuthenticationToken(props);
+ String token = conf.get(enumToConfKey(implementingClass, ConnectorInfo.TOKEN));
+ if (token == null || token.isEmpty())
+ return null;
+ if (token.startsWith(TokenSource.INLINE.prefix())) {
+ String[] args = token.substring(TokenSource.INLINE.prefix().length()).split(":", 2);
+ if (args.length == 2)
+ return org.apache.accumulo.core.client.security.tokens.AuthenticationToken.AuthenticationTokenSerializer
+ .deserialize(args[0], Base64.getDecoder().decode(args[1].getBytes(UTF_8)));
+ } else if (token.startsWith(TokenSource.FILE.prefix())) {
+ String tokenFileName = token.substring(TokenSource.FILE.prefix().length());
+ return getTokenFromFile(conf, getPrincipal(implementingClass, conf), tokenFileName);
+ } else if (token.startsWith(TokenSource.JOB.prefix())) {
+ String[] args = token.substring(TokenSource.JOB.prefix().length()).split(":", 2);
+ if (args.length == 2) {
+ String className = args[0], serviceName = args[1];
+ if (DelegationTokenImpl.class.getName().equals(className)) {
+ return new org.apache.accumulo.core.clientImpl.mapreduce.DelegationTokenStub(serviceName);
+ }
+ }
+ }
+
+ throw new IllegalStateException("Token was not properly serialized into the configuration");
+ }
+
+ /**
+ * Reads from the token file in distributed cache. Currently, the token file stores data separated
+ * by colons e.g. principal:token_class:token
+ *
+ * @param conf
+ * the Hadoop context for the configured job
+ * @return path to the token file as a String
+ * @since 1.6.0
+ * @see #setConnectorInfo(Class, Configuration, String, AuthenticationToken)
+ */
+ public static AuthenticationToken getTokenFromFile(Configuration conf, String principal,
+ String tokenFile) {
+ FSDataInputStream in = null;
+ try {
+ URI[] uris = DistributedCacheHelper.getCacheFiles(conf);
+ Path path = null;
+ for (URI u : uris) {
+ if (u.toString().equals(tokenFile)) {
+ path = new Path(u);
+ }
+ }
+ if (path == null) {
+ throw new IllegalArgumentException(
+ "Couldn't find password file called \"" + tokenFile + "\" in cache.");
+ }
+ FileSystem fs = FileSystem.get(conf);
+ in = fs.open(path);
+ } catch (IOException e) {
+ throw new IllegalArgumentException(
+ "Couldn't open password file called \"" + tokenFile + "\".");
+ }
+ try (java.util.Scanner fileScanner = new java.util.Scanner(in)) {
+ while (fileScanner.hasNextLine()) {
+ org.apache.accumulo.core.clientImpl.Credentials creds = org.apache.accumulo.core.clientImpl.Credentials
+ .deserialize(fileScanner.nextLine());
+ if (principal.equals(creds.getPrincipal())) {
+ return creds.getToken();
+ }
+ }
+ throw new IllegalArgumentException(
+ "Couldn't find token for user \"" + principal + "\" in file \"" + tokenFile + "\"");
+ }
}
/**
@@ -341,14 +414,16 @@ public static AuthenticationToken getAuthenticationToken(Class<?> implementingCl
*/
public static void setZooKeeperInstance(Class<?> implementingClass, Configuration conf,
org.apache.accumulo.core.client.ClientConfiguration clientConfig) {
- Properties props = getClientProperties(implementingClass, conf);
- Properties newProps = ClientConfConverter.toProperties(clientConfig);
- for (Object keyObj : newProps.keySet()) {
- String propKey = (String) keyObj;
- String val = newProps.getProperty(propKey);
- props.setProperty(propKey, val);
+ String key = enumToConfKey(implementingClass, InstanceOpts.TYPE);
+ if (!conf.get(key, "").isEmpty())
+ throw new IllegalStateException(
+ "Instance info can only be set once per job; it has already been configured with "
+ + conf.get(key));
+ conf.set(key, "ZooKeeperInstance");
+ if (clientConfig != null) {
+ conf.set(enumToConfKey(implementingClass, InstanceOpts.CLIENT_CONFIG),
+ clientConfig.serialize());
}
- setClientProperties(implementingClass, conf, props);
}
/**
@@ -361,12 +436,21 @@ public static void setZooKeeperInstance(Class<?> implementingClass, Configuratio
* the Hadoop configuration object to configure
* @return an Accumulo instance
* @since 1.6.0
+ * @see #setZooKeeperInstance(Class, Configuration,
+ * org.apache.accumulo.core.client.ClientConfiguration)
*/
public static org.apache.accumulo.core.client.Instance getInstance(Class<?> implementingClass,
Configuration conf) {
- ClientInfo info = ClientInfo.from(getClientProperties(implementingClass, conf));
- return new org.apache.accumulo.core.client.ZooKeeperInstance(info.getZooKeepers(),
- info.getZooKeepers());
+ String instanceType = conf.get(enumToConfKey(implementingClass, InstanceOpts.TYPE), "");
+
+ if ("ZooKeeperInstance".equals(instanceType)) {
+ return new org.apache.accumulo.core.client.ZooKeeperInstance(
+ getClientConfiguration(implementingClass, conf));
+ } else if (instanceType.isEmpty()) {
+ throw new IllegalStateException(
+ "Instance has not been configured for " + implementingClass.getSimpleName());
+ } else
+ throw new IllegalStateException("Unrecognized instance type " + instanceType);
}
/**
@@ -382,7 +466,23 @@ public static void setZooKeeperInstance(Class<?> implementingClass, Configuratio
*/
public static org.apache.accumulo.core.client.ClientConfiguration getClientConfiguration(
Class<?> implementingClass, Configuration conf) {
- return ClientConfConverter.toClientConf(getClientProperties(implementingClass, conf));
+ String clientConfigString = conf
+ .get(enumToConfKey(implementingClass, InstanceOpts.CLIENT_CONFIG));
+ if (null != clientConfigString) {
+ return org.apache.accumulo.core.client.ClientConfiguration.deserialize(clientConfigString);
+ }
+
+ String instanceName = conf.get(enumToConfKey(implementingClass, InstanceOpts.NAME));
+ String zookeepers = conf.get(enumToConfKey(implementingClass, InstanceOpts.ZOO_KEEPERS));
+ org.apache.accumulo.core.client.ClientConfiguration clientConf = org.apache.accumulo.core.client.ClientConfiguration
+ .loadDefault();
+ if (null != instanceName) {
+ clientConf.withInstance(instanceName);
+ }
+ if (null != zookeepers) {
+ clientConf.withZkHosts(zookeepers);
+ }
+ return clientConf;
}
/**
diff --git a/core/src/main/java/org/apache/accumulo/core/clientImpl/mapreduce/lib/InputConfigurator.java b/core/src/main/java/org/apache/accumulo/core/clientImpl/mapreduce/lib/InputConfigurator.java
index d1915ba69b..6dc4fd6f78 100644
--- a/core/src/main/java/org/apache/accumulo/core/clientImpl/mapreduce/lib/InputConfigurator.java
+++ b/core/src/main/java/org/apache/accumulo/core/clientImpl/mapreduce/lib/InputConfigurator.java
@@ -36,8 +36,6 @@
import java.util.Set;
import java.util.StringTokenizer;
-import org.apache.accumulo.core.client.Accumulo;
-import org.apache.accumulo.core.client.AccumuloClient;
import org.apache.accumulo.core.client.AccumuloException;
import org.apache.accumulo.core.client.AccumuloSecurityException;
import org.apache.accumulo.core.client.BatchScanner;
@@ -49,7 +47,6 @@
import org.apache.accumulo.core.client.TableNotFoundException;
import org.apache.accumulo.core.client.sample.SamplerConfiguration;
import org.apache.accumulo.core.clientImpl.ClientContext;
-import org.apache.accumulo.core.clientImpl.ClientInfo;
import org.apache.accumulo.core.clientImpl.Table;
import org.apache.accumulo.core.clientImpl.Tables;
import org.apache.accumulo.core.data.Key;
@@ -725,22 +722,31 @@ public static void setInputTableConfigs(Class<?> implementingClass, Configuratio
*/
public static void validatePermissions(Class<?> implementingClass, Configuration conf)
throws IOException {
+
Map<String,org.apache.accumulo.core.client.mapreduce.InputTableConfig> inputTableConfigs = getInputTableConfigs(
implementingClass, conf);
- try (AccumuloClient client = Accumulo.newClient()
- .from(getClientProperties(implementingClass, conf)).build()) {
+ if (!isConnectorInfoSet(implementingClass, conf))
+ throw new IOException("Input info has not been set.");
+ String instanceKey = conf.get(enumToConfKey(implementingClass, InstanceOpts.TYPE));
+ if (!"ZooKeeperInstance".equals(instanceKey))
+ throw new IOException("Instance info has not been set.");
+ // validate that we can connect as configured
+ try {
+ String principal = getPrincipal(implementingClass, conf);
+ org.apache.accumulo.core.client.security.tokens.AuthenticationToken token = getAuthenticationToken(
+ implementingClass, conf);
+ org.apache.accumulo.core.client.Connector c = getInstance(implementingClass, conf)
+ .getConnector(principal, token);
+ if (!c.securityOperations().authenticateUser(principal, token))
+ throw new IOException("Unable to authenticate user");
+
if (getInputTableConfigs(implementingClass, conf).size() == 0)
throw new IOException("No table set.");
- String principal = getPrincipal(implementingClass, conf);
- if (principal == null) {
- principal = client.whoami();
- }
-
for (Map.Entry<String,org.apache.accumulo.core.client.mapreduce.InputTableConfig> tableConfig : inputTableConfigs
.entrySet()) {
- if (!client.securityOperations().hasTablePermission(principal, tableConfig.getKey(),
- TablePermission.READ))
+ if (!c.securityOperations().hasTablePermission(getPrincipal(implementingClass, conf),
+ tableConfig.getKey(), TablePermission.READ))
throw new IOException("Unable to access table");
}
for (Map.Entry<String,org.apache.accumulo.core.client.mapreduce.InputTableConfig> tableConfigEntry : inputTableConfigs
@@ -750,7 +756,7 @@ public static void validatePermissions(Class<?> implementingClass, Configuration
if (!tableConfig.shouldUseLocalIterators()) {
if (tableConfig.getIterators() != null) {
for (IteratorSetting iter : tableConfig.getIterators()) {
- if (!client.tableOperations().testClassLoad(tableConfigEntry.getKey(),
+ if (!c.tableOperations().testClassLoad(tableConfigEntry.getKey(),
iter.getIteratorClass(), SortedKeyValueIterator.class.getName()))
throw new AccumuloException("Servers are unable to load " + iter.getIteratorClass()
+ " as a " + SortedKeyValueIterator.class.getName());
@@ -809,102 +815,98 @@ public static void validatePermissions(Class<?> implementingClass, Configuration
}
public static Map<String,Map<KeyExtent,List<Range>>> binOffline(Table.ID tableId,
- List<Range> ranges, ClientInfo clientInfo) throws AccumuloException, TableNotFoundException {
-
- try (ClientContext context = new ClientContext(clientInfo)) {
- Map<String,Map<KeyExtent,List<Range>>> binnedRanges = new HashMap<>();
-
+ List<Range> ranges, ClientContext context) throws AccumuloException, TableNotFoundException {
+ Map<String,Map<KeyExtent,List<Range>>> binnedRanges = new HashMap<>();
+ if (Tables.getTableState(context, tableId) != TableState.OFFLINE) {
+ Tables.clearCache(context);
if (Tables.getTableState(context, tableId) != TableState.OFFLINE) {
- Tables.clearCache(context);
- if (Tables.getTableState(context, tableId) != TableState.OFFLINE) {
- throw new AccumuloException(
- "Table is online tableId:" + tableId + " cannot scan table in offline mode ");
- }
+ throw new AccumuloException(
+ "Table is online tableId:" + tableId + " cannot scan table in offline mode ");
}
+ }
- for (Range range : ranges) {
- Text startRow;
-
- if (range.getStartKey() != null)
- startRow = range.getStartKey().getRow();
- else
- startRow = new Text();
-
- Range metadataRange = new Range(new KeyExtent(tableId, startRow, null).getMetadataEntry(),
- true, null, false);
- Scanner scanner = context.createScanner(MetadataTable.NAME, Authorizations.EMPTY);
- MetadataSchema.TabletsSection.TabletColumnFamily.PREV_ROW_COLUMN.fetch(scanner);
- scanner.fetchColumnFamily(MetadataSchema.TabletsSection.LastLocationColumnFamily.NAME);
- scanner.fetchColumnFamily(MetadataSchema.TabletsSection.CurrentLocationColumnFamily.NAME);
- scanner.fetchColumnFamily(MetadataSchema.TabletsSection.FutureLocationColumnFamily.NAME);
- scanner.setRange(metadataRange);
-
- RowIterator rowIter = new RowIterator(scanner);
- KeyExtent lastExtent = null;
- while (rowIter.hasNext()) {
- Iterator<Map.Entry<Key,Value>> row = rowIter.next();
- String last = "";
- KeyExtent extent = null;
- String location = null;
-
- while (row.hasNext()) {
- Map.Entry<Key,Value> entry = row.next();
- Key key = entry.getKey();
-
- if (key.getColumnFamily()
- .equals(MetadataSchema.TabletsSection.LastLocationColumnFamily.NAME)) {
- last = entry.getValue().toString();
- }
-
- if (key.getColumnFamily()
- .equals(MetadataSchema.TabletsSection.CurrentLocationColumnFamily.NAME)
- || key.getColumnFamily()
- .equals(MetadataSchema.TabletsSection.FutureLocationColumnFamily.NAME)) {
- location = entry.getValue().toString();
- }
+ for (Range range : ranges) {
+ Text startRow;
+
+ if (range.getStartKey() != null)
+ startRow = range.getStartKey().getRow();
+ else
+ startRow = new Text();
+
+ Range metadataRange = new Range(new KeyExtent(tableId, startRow, null).getMetadataEntry(),
+ true, null, false);
+ Scanner scanner = context.createScanner(MetadataTable.NAME, Authorizations.EMPTY);
+ MetadataSchema.TabletsSection.TabletColumnFamily.PREV_ROW_COLUMN.fetch(scanner);
+ scanner.fetchColumnFamily(MetadataSchema.TabletsSection.LastLocationColumnFamily.NAME);
+ scanner.fetchColumnFamily(MetadataSchema.TabletsSection.CurrentLocationColumnFamily.NAME);
+ scanner.fetchColumnFamily(MetadataSchema.TabletsSection.FutureLocationColumnFamily.NAME);
+ scanner.setRange(metadataRange);
+
+ RowIterator rowIter = new RowIterator(scanner);
+ KeyExtent lastExtent = null;
+ while (rowIter.hasNext()) {
+ Iterator<Map.Entry<Key,Value>> row = rowIter.next();
+ String last = "";
+ KeyExtent extent = null;
+ String location = null;
+
+ while (row.hasNext()) {
+ Map.Entry<Key,Value> entry = row.next();
+ Key key = entry.getKey();
+
+ if (key.getColumnFamily()
+ .equals(MetadataSchema.TabletsSection.LastLocationColumnFamily.NAME)) {
+ last = entry.getValue().toString();
+ }
- if (MetadataSchema.TabletsSection.TabletColumnFamily.PREV_ROW_COLUMN.hasColumns(key)) {
- extent = new KeyExtent(key.getRow(), entry.getValue());
- }
+ if (key.getColumnFamily()
+ .equals(MetadataSchema.TabletsSection.CurrentLocationColumnFamily.NAME)
+ || key.getColumnFamily()
+ .equals(MetadataSchema.TabletsSection.FutureLocationColumnFamily.NAME)) {
+ location = entry.getValue().toString();
+ }
+ if (MetadataSchema.TabletsSection.TabletColumnFamily.PREV_ROW_COLUMN.hasColumns(key)) {
+ extent = new KeyExtent(key.getRow(), entry.getValue());
}
- if (location != null)
- return null;
+ }
- if (!extent.getTableId().equals(tableId)) {
- throw new AccumuloException("Saw unexpected table Id " + tableId + " " + extent);
- }
+ if (location != null)
+ return null;
- if (lastExtent != null && !extent.isPreviousExtent(lastExtent)) {
- throw new AccumuloException(" " + lastExtent + " is not previous extent " + extent);
- }
+ if (!extent.getTableId().equals(tableId)) {
+ throw new AccumuloException("Saw unexpected table Id " + tableId + " " + extent);
+ }
- Map<KeyExtent,List<Range>> tabletRanges = binnedRanges.get(last);
- if (tabletRanges == null) {
- tabletRanges = new HashMap<>();
- binnedRanges.put(last, tabletRanges);
- }
+ if (lastExtent != null && !extent.isPreviousExtent(lastExtent)) {
+ throw new AccumuloException(" " + lastExtent + " is not previous extent " + extent);
+ }
- List<Range> rangeList = tabletRanges.get(extent);
- if (rangeList == null) {
- rangeList = new ArrayList<>();
- tabletRanges.put(extent, rangeList);
- }
+ Map<KeyExtent,List<Range>> tabletRanges = binnedRanges.get(last);
+ if (tabletRanges == null) {
+ tabletRanges = new HashMap<>();
+ binnedRanges.put(last, tabletRanges);
+ }
- rangeList.add(range);
+ List<Range> rangeList = tabletRanges.get(extent);
+ if (rangeList == null) {
+ rangeList = new ArrayList<>();
+ tabletRanges.put(extent, rangeList);
+ }
- if (extent.getEndRow() == null
- || range.afterEndKey(new Key(extent.getEndRow()).followingKey(PartialKey.ROW))) {
- break;
- }
+ rangeList.add(range);
- lastExtent = extent;
+ if (extent.getEndRow() == null
+ || range.afterEndKey(new Key(extent.getEndRow()).followingKey(PartialKey.ROW))) {
+ break;
}
+ lastExtent = extent;
}
- return binnedRanges;
+
}
+ return binnedRanges;
}
private static String toBase64(Writable writable) {
diff --git a/test/src/main/java/org/apache/accumulo/test/mapred/TokenFileIT.java b/test/src/main/java/org/apache/accumulo/test/mapred/TokenFileIT.java
index 7e105aa84b..95348f8cbb 100644
--- a/test/src/main/java/org/apache/accumulo/test/mapred/TokenFileIT.java
+++ b/test/src/main/java/org/apache/accumulo/test/mapred/TokenFileIT.java
@@ -32,6 +32,7 @@
import org.apache.accumulo.core.client.BatchWriterConfig;
import org.apache.accumulo.core.client.Scanner;
import org.apache.accumulo.core.clientImpl.ClientInfo;
+import org.apache.accumulo.core.clientImpl.Credentials;
import org.apache.accumulo.core.data.Key;
import org.apache.accumulo.core.data.Mutation;
import org.apache.accumulo.core.data.Value;
@@ -169,10 +170,11 @@ public void testMR() throws Exception {
}
bw.close();
- File tf = folder.newFile("client.properties");
- try (PrintStream out = new PrintStream(tf)) {
- getClientInfo().getProperties().store(out, "Credentials for " + getClass().getName());
- }
+ File tf = folder.newFile("root_test.pw");
+ PrintStream out = new PrintStream(tf);
+ String outString = new Credentials(getAdminPrincipal(), getAdminToken()).serialize();
+ out.println(outString);
+ out.close();
MRTokenFileTester.main(new String[] {tf.getAbsolutePath(), table1, table2});
assertNull(e1);
diff --git a/test/src/main/java/org/apache/accumulo/test/mapreduce/TokenFileIT.java b/test/src/main/java/org/apache/accumulo/test/mapreduce/TokenFileIT.java
index 1e8917c7d5..88931216cf 100644
--- a/test/src/main/java/org/apache/accumulo/test/mapreduce/TokenFileIT.java
+++ b/test/src/main/java/org/apache/accumulo/test/mapreduce/TokenFileIT.java
@@ -32,6 +32,7 @@
import org.apache.accumulo.core.client.BatchWriterConfig;
import org.apache.accumulo.core.client.Scanner;
import org.apache.accumulo.core.clientImpl.ClientInfo;
+import org.apache.accumulo.core.clientImpl.Credentials;
import org.apache.accumulo.core.data.Key;
import org.apache.accumulo.core.data.Mutation;
import org.apache.accumulo.core.data.Value;
@@ -165,10 +166,11 @@ public void testMR() throws Exception {
}
bw.close();
- File tf = folder.newFile("client.properties");
- try (PrintStream out = new PrintStream(tf)) {
- getClientInfo().getProperties().store(out, "Credentials for " + getClass().getName());
- }
+ File tf = folder.newFile("root_test.pw");
+ PrintStream out = new PrintStream(tf);
+ String outString = new Credentials(getAdminPrincipal(), getAdminToken()).serialize();
+ out.println(outString);
+ out.close();
MRTokenFileTester.main(new String[] {tf.getAbsolutePath(), table1, table2});
assertNull(e1);
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services