Removes use of most deprecated code. (#1126)
diff --git a/modules/command/src/test/java/org/apache/fluo/command/FluoProgramTest.java b/modules/command/src/test/java/org/apache/fluo/command/FluoProgramTest.java
index e3586a4..8fbbf9a 100644
--- a/modules/command/src/test/java/org/apache/fluo/command/FluoProgramTest.java
+++ b/modules/command/src/test/java/org/apache/fluo/command/FluoProgramTest.java
@@ -64,7 +64,7 @@
outPS = System.out;
errPS = System.err;
// This will hide usage and error logs when running tests
- try (PrintStream ps = new PrintStream(new NullOutputStream())) {
+ try (PrintStream ps = new PrintStream(NullOutputStream.NULL_OUTPUT_STREAM)) {
System.setOut(ps);
System.setErr(ps);
}
diff --git a/modules/core/src/main/java/org/apache/fluo/core/client/FluoAdminImpl.java b/modules/core/src/main/java/org/apache/fluo/core/client/FluoAdminImpl.java
index 2930838..92e149b 100644
--- a/modules/core/src/main/java/org/apache/fluo/core/client/FluoAdminImpl.java
+++ b/modules/core/src/main/java/org/apache/fluo/core/client/FluoAdminImpl.java
@@ -282,7 +282,7 @@
final String accumuloInstanceName =
client.properties().getProperty(AccumuloProps.CLIENT_INSTANCE_NAME);
- final String accumuloInstanceID = client.instanceOperations().getInstanceID();
+ final String accumuloInstanceID = client.instanceOperations().getInstanceId().canonical();
final String fluoApplicationID = UUID.randomUUID().toString();
// Create node specified by chroot suffix of Zookeeper connection string (if it doesn't exist)
diff --git a/modules/core/src/main/java/org/apache/fluo/core/impl/Environment.java b/modules/core/src/main/java/org/apache/fluo/core/impl/Environment.java
index 5e4988c..f9a2316 100644
--- a/modules/core/src/main/java/org/apache/fluo/core/impl/Environment.java
+++ b/modules/core/src/main/java/org/apache/fluo/core/impl/Environment.java
@@ -97,9 +97,9 @@
"unexpected accumulo instance name " + instanceName + " != " + accumuloInstance);
}
- if (!client.instanceOperations().getInstanceID().equals(accumuloInstanceID)) {
+ if (!client.instanceOperations().getInstanceId().canonical().equals(accumuloInstanceID)) {
throw new IllegalArgumentException("unexpected accumulo instance id "
- + client.instanceOperations().getInstanceID() + " != " + accumuloInstanceID);
+ + client.instanceOperations().getInstanceId() + " != " + accumuloInstanceID);
}
try {
diff --git a/modules/core/src/main/java/org/apache/fluo/core/util/AccumuloUtil.java b/modules/core/src/main/java/org/apache/fluo/core/util/AccumuloUtil.java
index b6eaeb6..3ccad87 100644
--- a/modules/core/src/main/java/org/apache/fluo/core/util/AccumuloUtil.java
+++ b/modules/core/src/main/java/org/apache/fluo/core/util/AccumuloUtil.java
@@ -15,6 +15,8 @@
package org.apache.fluo.core.util;
+import java.util.Properties;
+
import org.apache.accumulo.core.client.Accumulo;
import org.apache.accumulo.core.client.AccumuloClient;
import org.apache.fluo.api.config.FluoConfiguration;
@@ -32,4 +34,10 @@
return Accumulo.newClient().to(config.getAccumuloInstance(), config.getAccumuloZookeepers())
.as(config.getAccumuloUser(), config.getAccumuloPassword()).build();
}
+
+ public static Properties getClientProps(FluoConfiguration config) {
+ return Accumulo.newClientProperties()
+ .to(config.getAccumuloInstance(), config.getAccumuloZookeepers())
+ .as(config.getAccumuloUser(), config.getAccumuloPassword()).build();
+ }
}
diff --git a/modules/mapreduce/pom.xml b/modules/mapreduce/pom.xml
index 771ea3a..bd8d65f 100644
--- a/modules/mapreduce/pom.xml
+++ b/modules/mapreduce/pom.xml
@@ -26,6 +26,10 @@
<description>This module provides utility code for MapReduce jobs that read from or write to a
Apache Fluo table.</description>
<properties>
+ <!-- The checkstyle config comes from the Fluo parent pom and is outdated.
+ It disallows use of the newer Accumulo mapreduce APIs. Disabling for
+ now so Fluo can use new APIs and build. -->
+ <checkstyle.skip>true</checkstyle.skip>
<!-- accumulo2-maven-plugin requires the accumulo version of thrift -->
<thrift.version>0.17.0</thrift.version>
</properties>
@@ -35,6 +39,10 @@
<artifactId>accumulo-core</artifactId>
</dependency>
<dependency>
+ <groupId>org.apache.accumulo</groupId>
+ <artifactId>accumulo-hadoop-mapreduce</artifactId>
+ </dependency>
+ <dependency>
<groupId>org.apache.fluo</groupId>
<artifactId>fluo-accumulo</artifactId>
</dependency>
diff --git a/modules/mapreduce/src/main/java/org/apache/fluo/mapreduce/FluoEntryInputFormat.java b/modules/mapreduce/src/main/java/org/apache/fluo/mapreduce/FluoEntryInputFormat.java
index 9d58a86..3eb701f 100644
--- a/modules/mapreduce/src/main/java/org/apache/fluo/mapreduce/FluoEntryInputFormat.java
+++ b/modules/mapreduce/src/main/java/org/apache/fluo/mapreduce/FluoEntryInputFormat.java
@@ -23,9 +23,9 @@
import java.util.Iterator;
import java.util.List;
-import org.apache.accumulo.core.client.mapreduce.AccumuloInputFormat;
-import org.apache.accumulo.core.client.mapreduce.RangeInputSplit;
import org.apache.accumulo.core.client.security.tokens.PasswordToken;
+import org.apache.accumulo.hadoop.mapreduce.AccumuloInputFormat;
+import org.apache.accumulo.hadoopImpl.mapred.RangeInputSplit;
import org.apache.fluo.api.config.FluoConfiguration;
import org.apache.fluo.api.config.SimpleConfiguration;
import org.apache.fluo.api.data.Bytes;
@@ -35,6 +35,7 @@
import org.apache.fluo.api.data.Span;
import org.apache.fluo.core.impl.Environment;
import org.apache.fluo.core.impl.TransactionImpl;
+import org.apache.fluo.core.util.AccumuloUtil;
import org.apache.fluo.core.util.SpanUtil;
import org.apache.hadoop.mapreduce.InputFormat;
import org.apache.hadoop.mapreduce.InputSplit;
@@ -154,12 +155,8 @@
conf.getConfiguration().set(PROPS_CONF_KEY,
new String(baos.toByteArray(), StandardCharsets.UTF_8));
- AccumuloInputFormat.setZooKeeperInstance(conf, fconfig.getAccumuloInstance(),
- fconfig.getAccumuloZookeepers());
- AccumuloInputFormat.setConnectorInfo(conf, fconfig.getAccumuloUser(),
- new PasswordToken(fconfig.getAccumuloPassword()));
- AccumuloInputFormat.setInputTableName(conf, env.getTable());
- AccumuloInputFormat.setScanAuthorizations(conf, env.getAuthorizations());
+ AccumuloInputFormat.configure().clientProperties(AccumuloUtil.getClientProps(fconfig))
+ .table(env.getTable()).auths(env.getAuthorizations()).store(conf);
}
} catch (Exception e) {
throw new RuntimeException(e);
diff --git a/modules/mapreduce/src/main/java/org/apache/fluo/mapreduce/FluoKeyValueGenerator.java b/modules/mapreduce/src/main/java/org/apache/fluo/mapreduce/FluoKeyValueGenerator.java
index 158bed9..898eb2f 100644
--- a/modules/mapreduce/src/main/java/org/apache/fluo/mapreduce/FluoKeyValueGenerator.java
+++ b/modules/mapreduce/src/main/java/org/apache/fluo/mapreduce/FluoKeyValueGenerator.java
@@ -19,9 +19,9 @@
import java.util.Collection;
import java.util.Collections;
-import org.apache.accumulo.core.client.mapreduce.AccumuloFileOutputFormat;
import org.apache.accumulo.core.client.summary.SummarizerConfiguration;
import org.apache.accumulo.core.data.Key;
+import org.apache.accumulo.hadoop.mapreduce.AccumuloFileOutputFormat;
import org.apache.fluo.accumulo.summarizer.FluoSummarizer;
import org.apache.fluo.accumulo.util.ColumnType;
import org.apache.fluo.accumulo.values.WriteValue;
diff --git a/modules/mapreduce/src/main/java/org/apache/fluo/mapreduce/FluoMutationGenerator.java b/modules/mapreduce/src/main/java/org/apache/fluo/mapreduce/FluoMutationGenerator.java
index 9293c2c..a07d4a0 100644
--- a/modules/mapreduce/src/main/java/org/apache/fluo/mapreduce/FluoMutationGenerator.java
+++ b/modules/mapreduce/src/main/java/org/apache/fluo/mapreduce/FluoMutationGenerator.java
@@ -18,8 +18,8 @@
import java.nio.charset.StandardCharsets;
import org.apache.accumulo.core.client.BatchWriter;
-import org.apache.accumulo.core.client.mapreduce.AccumuloOutputFormat;
import org.apache.accumulo.core.data.Mutation;
+import org.apache.accumulo.hadoop.mapreduce.AccumuloOutputFormat;
import org.apache.fluo.accumulo.util.ColumnType;
import org.apache.fluo.accumulo.values.WriteValue;
import org.apache.fluo.api.data.Bytes;
diff --git a/modules/mapreduce/src/main/java/org/apache/fluo/mapreduce/FluoRowInputFormat.java b/modules/mapreduce/src/main/java/org/apache/fluo/mapreduce/FluoRowInputFormat.java
index 1c5616d..710e31d 100644
--- a/modules/mapreduce/src/main/java/org/apache/fluo/mapreduce/FluoRowInputFormat.java
+++ b/modules/mapreduce/src/main/java/org/apache/fluo/mapreduce/FluoRowInputFormat.java
@@ -23,9 +23,9 @@
import java.util.Iterator;
import java.util.List;
-import org.apache.accumulo.core.client.mapreduce.AccumuloInputFormat;
-import org.apache.accumulo.core.client.mapreduce.RangeInputSplit;
import org.apache.accumulo.core.client.security.tokens.PasswordToken;
+import org.apache.accumulo.hadoop.mapreduce.AccumuloInputFormat;
+import org.apache.accumulo.hadoopImpl.mapreduce.RangeInputSplit;
import org.apache.fluo.api.client.scanner.ColumnScanner;
import org.apache.fluo.api.config.FluoConfiguration;
import org.apache.fluo.api.config.SimpleConfiguration;
@@ -35,6 +35,7 @@
import org.apache.fluo.api.data.Span;
import org.apache.fluo.core.impl.Environment;
import org.apache.fluo.core.impl.TransactionImpl;
+import org.apache.fluo.core.util.AccumuloUtil;
import org.apache.fluo.core.util.SpanUtil;
import org.apache.hadoop.mapreduce.InputFormat;
import org.apache.hadoop.mapreduce.InputSplit;
@@ -155,12 +156,8 @@
conf.getConfiguration().set(PROPS_CONF_KEY,
new String(baos.toByteArray(), StandardCharsets.UTF_8));
- AccumuloInputFormat.setZooKeeperInstance(conf, fconfig.getAccumuloInstance(),
- fconfig.getAccumuloZookeepers());
- AccumuloInputFormat.setConnectorInfo(conf, fconfig.getAccumuloUser(),
- new PasswordToken(fconfig.getAccumuloPassword()));
- AccumuloInputFormat.setInputTableName(conf, env.getTable());
- AccumuloInputFormat.setScanAuthorizations(conf, env.getAuthorizations());
+ AccumuloInputFormat.configure().clientProperties(AccumuloUtil.getClientProps(fconfig))
+ .table(env.getTable()).auths(env.getAuthorizations()).store(conf);
}
} catch (Exception e) {
throw new RuntimeException(e);
diff --git a/modules/mapreduce/src/test/java/org/apache/fluo/mapreduce/it/FluoFileOutputFormatIT.java b/modules/mapreduce/src/test/java/org/apache/fluo/mapreduce/it/FluoFileOutputFormatIT.java
index aa118a6..5aed608 100644
--- a/modules/mapreduce/src/test/java/org/apache/fluo/mapreduce/it/FluoFileOutputFormatIT.java
+++ b/modules/mapreduce/src/test/java/org/apache/fluo/mapreduce/it/FluoFileOutputFormatIT.java
@@ -20,9 +20,9 @@
import java.io.PrintWriter;
import java.nio.charset.StandardCharsets;
-import org.apache.accumulo.core.client.mapreduce.AccumuloFileOutputFormat;
import org.apache.accumulo.core.data.Key;
import org.apache.accumulo.core.data.Value;
+import org.apache.accumulo.hadoop.mapreduce.AccumuloFileOutputFormat;
import org.apache.fluo.api.data.Column;
import org.apache.fluo.integration.ITBaseImpl;
import org.apache.fluo.integration.TestTransaction;
@@ -96,7 +96,7 @@
Assert.assertTrue(job.waitForCompletion(false));
// bulk import rfiles
- aClient.tableOperations().importDirectory(table, outDir.toString(), failDir.toString(), false);
+ aClient.tableOperations().importDirectory(outDir.toString()).to(table).threads(3).load();
// read and update data using transactions
TestTransaction tx1 = new TestTransaction(env);
diff --git a/pom.xml b/pom.xml
index 5e5c9b4..561ceb0 100644
--- a/pom.xml
+++ b/pom.xml
@@ -123,6 +123,11 @@
</dependency>
<dependency>
<groupId>org.apache.accumulo</groupId>
+ <artifactId>accumulo-hadoop-mapreduce</artifactId>
+ <version>${accumulo.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.accumulo</groupId>
<artifactId>accumulo-minicluster</artifactId>
<version>${accumulo.version}</version>
</dependency>
