tables
configurePartitioner(job, tablesStartKeys);
TableMapReduceUtil.addDependencyJars(job);
+ PhoenixMapReduceUtil.addPhoenixDependencyJars(job.getConfiguration());
TableMapReduceUtil.initCredentials(job);
}
diff --git a/phoenix-core-server/src/main/java/org/apache/phoenix/mapreduce/PhoenixSyncTableTool.java b/phoenix-core-server/src/main/java/org/apache/phoenix/mapreduce/PhoenixSyncTableTool.java
index 80eacde25e7..09b0e5c4d92 100644
--- a/phoenix-core-server/src/main/java/org/apache/phoenix/mapreduce/PhoenixSyncTableTool.java
+++ b/phoenix-core-server/src/main/java/org/apache/phoenix/mapreduce/PhoenixSyncTableTool.java
@@ -168,6 +168,7 @@ private Job configureAndCreatePhoenixSyncTableJob(PTableType tableType) throws E
job.setJarByClass(PhoenixSyncTableTool.class);
TableMapReduceUtil.initCredentials(job);
TableMapReduceUtil.addDependencyJars(job);
+ PhoenixMapReduceUtil.addPhoenixDependencyJars(job.getConfiguration());
configureInput(job, tableType);
configureOutput(job);
obtainTargetClusterTokens(job);
diff --git a/phoenix-core-server/src/main/java/org/apache/phoenix/mapreduce/index/IndexScrutinyTool.java b/phoenix-core-server/src/main/java/org/apache/phoenix/mapreduce/index/IndexScrutinyTool.java
index c9111aa7b88..426e13e9da1 100644
--- a/phoenix-core-server/src/main/java/org/apache/phoenix/mapreduce/index/IndexScrutinyTool.java
+++ b/phoenix-core-server/src/main/java/org/apache/phoenix/mapreduce/index/IndexScrutinyTool.java
@@ -346,6 +346,7 @@ private Job configureSubmittableJob(Job job, Path outputPath,
job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(Text.class);
TableMapReduceUtil.addDependencyJars(job);
+ PhoenixMapReduceUtil.addPhoenixDependencyJars(job.getConfiguration());
return job;
}
diff --git a/phoenix-core-server/src/main/java/org/apache/phoenix/mapreduce/index/IndexTool.java b/phoenix-core-server/src/main/java/org/apache/phoenix/mapreduce/index/IndexTool.java
index cc918dc46f3..8662dbdec2c 100644
--- a/phoenix-core-server/src/main/java/org/apache/phoenix/mapreduce/index/IndexTool.java
+++ b/phoenix-core-server/src/main/java/org/apache/phoenix/mapreduce/index/IndexTool.java
@@ -807,6 +807,7 @@ private Job configureSubmittableJobUsingDirectApi(Job job) throws Exception {
job.setOutputKeyClass(NullWritable.class);
job.setOutputValueClass(NullWritable.class);
TableMapReduceUtil.addDependencyJars(job);
+ PhoenixMapReduceUtil.addPhoenixDependencyJars(job.getConfiguration());
job.setNumReduceTasks(1);
return job;
}
diff --git a/phoenix-core-server/src/main/java/org/apache/phoenix/mapreduce/transform/TransformTool.java b/phoenix-core-server/src/main/java/org/apache/phoenix/mapreduce/transform/TransformTool.java
index daca9a04616..2bc12485dcd 100644
--- a/phoenix-core-server/src/main/java/org/apache/phoenix/mapreduce/transform/TransformTool.java
+++ b/phoenix-core-server/src/main/java/org/apache/phoenix/mapreduce/transform/TransformTool.java
@@ -641,6 +641,7 @@ public Job configureJob() throws Exception {
job.setOutputKeyClass(NullWritable.class);
job.setOutputValueClass(NullWritable.class);
TableMapReduceUtil.addDependencyJars(job);
+ PhoenixMapReduceUtil.addPhoenixDependencyJars(job.getConfiguration());
job.setReducerClass(PhoenixTransformReducer.class);
diff --git a/phoenix-core-server/src/main/java/org/apache/phoenix/mapreduce/util/PhoenixMapReduceUtil.java b/phoenix-core-server/src/main/java/org/apache/phoenix/mapreduce/util/PhoenixMapReduceUtil.java
index b5f62bdfc55..d20cd1b7fe8 100644
--- a/phoenix-core-server/src/main/java/org/apache/phoenix/mapreduce/util/PhoenixMapReduceUtil.java
+++ b/phoenix-core-server/src/main/java/org/apache/phoenix/mapreduce/util/PhoenixMapReduceUtil.java
@@ -17,14 +17,26 @@
*/
package org.apache.phoenix.mapreduce.util;
+import com.fasterxml.jackson.annotation.JsonAutoDetect;
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.jayway.jsonpath.spi.json.JsonProvider;
+import com.lmax.disruptor.EventFactory;
import java.io.IOException;
import java.sql.Connection;
import java.sql.SQLException;
+import org.antlr.runtime.CharStream;
+import org.apache.commons.csv.CSVFormat;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
+import org.apache.hadoop.hbase.metrics.Gauge;
+import org.apache.hadoop.hbase.metrics.impl.MetricRegistriesImpl;
import org.apache.hadoop.mapreduce.InputFormat;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.db.DBWritable;
+import org.apache.htrace.SpanReceiver;
+import org.apache.phoenix.compat.hbase.CompatUtil;
import org.apache.phoenix.jdbc.PhoenixConnection;
import org.apache.phoenix.mapreduce.PhoenixInputFormat;
import org.apache.phoenix.mapreduce.PhoenixOutputFormat;
@@ -32,18 +44,67 @@
import org.apache.phoenix.schema.PTable;
import org.apache.phoenix.schema.PTableType;
import org.apache.phoenix.util.EnvironmentEdgeManager;
+import org.joda.time.Chronology;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
/**
* Utility class for setting Configuration parameters for the Map Reduce job
*/
public final class PhoenixMapReduceUtil {
+ private static final Logger LOGGER = LoggerFactory.getLogger(PhoenixMapReduceUtil.class);
+
public static final String INVALID_TIME_RANGE_EXCEPTION_MESSAGE = "Invalid time range for table";
private PhoenixMapReduceUtil() {
}
+ /**
+ * Add Phoenix and its dependency jars to the job configuration so they are shipped to YARN
+ * containers via the distributed cache. Analogous to HBase's
+ * {@link TableMapReduceUtil#addHBaseDependencyJars(Configuration)}.
+ *
+ * Callers should still invoke {@link TableMapReduceUtil#addDependencyJars(Job)} for HBase's own
+ * jars before calling this method.
+ */
+ public static void addPhoenixDependencyJars(Configuration conf) throws IOException {
+ TableMapReduceUtil.addDependencyJarsForClasses(conf,
+ // phoenix-core-client
+ PhoenixConnection.class,
+ // phoenix-hbase-compat
+ CompatUtil.class,
+ // phoenix-shaded-guava
+ org.apache.phoenix.thirdparty.com.google.common.collect.Lists.class,
+ // phoenix-shaded-commons-cli
+ org.apache.phoenix.thirdparty.org.apache.commons.cli.Options.class,
+ // joda-time
+ Chronology.class,
+ // antlr-runtime
+ CharStream.class,
+ // htrace-core
+ SpanReceiver.class,
+ // hbase-metrics-api (Gauge)
+ Gauge.class,
+ // hbase-metrics (MetricRegistriesImpl)
+ MetricRegistriesImpl.class,
+ // disruptor
+ EventFactory.class,
+ // jackson-core
+ TypeReference.class,
+ // jackson-databind
+ ObjectMapper.class,
+ // jackson-annotations
+ JsonAutoDetect.class,
+ // commons-csv
+ CSVFormat.class,
+ // json-path
+ JsonProvider.class,
+ // bson
+ org.bson.io.BsonInput.class);
+ }
+
/**
* @param inputClass DBWritable class
* @param tableName Input table name
diff --git a/phoenix-core-server/src/main/java/org/apache/phoenix/schema/stats/UpdateStatisticsTool.java b/phoenix-core-server/src/main/java/org/apache/phoenix/schema/stats/UpdateStatisticsTool.java
index ea3a0bebdd2..0d7be097d0e 100644
--- a/phoenix-core-server/src/main/java/org/apache/phoenix/schema/stats/UpdateStatisticsTool.java
+++ b/phoenix-core-server/src/main/java/org/apache/phoenix/schema/stats/UpdateStatisticsTool.java
@@ -23,15 +23,12 @@
import java.nio.charset.StandardCharsets;
import java.sql.Connection;
-import org.antlr.runtime.CharStream;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Admin;
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
-import org.apache.hadoop.hbase.metrics.Gauge;
-import org.apache.hadoop.hbase.metrics.impl.MetricRegistriesImpl;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.JobPriority;
@@ -40,14 +37,12 @@
import org.apache.hadoop.mapreduce.lib.output.NullOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
-import org.apache.htrace.SpanReceiver;
import org.apache.phoenix.jdbc.PhoenixConnection;
import org.apache.phoenix.mapreduce.util.ConnectionUtil;
import org.apache.phoenix.mapreduce.util.PhoenixConfigurationUtil;
import org.apache.phoenix.mapreduce.util.PhoenixConfigurationUtil.MRJobType;
import org.apache.phoenix.mapreduce.util.PhoenixMapReduceUtil;
import org.apache.phoenix.util.SchemaUtil;
-import org.joda.time.Chronology;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -216,9 +211,7 @@ private void configureJob() throws Exception {
job.setPriority(this.jobPriority);
TableMapReduceUtil.addDependencyJars(job);
- TableMapReduceUtil.addDependencyJarsForClasses(job.getConfiguration(), PhoenixConnection.class,
- Chronology.class, CharStream.class, SpanReceiver.class, Gauge.class,
- MetricRegistriesImpl.class);
+ PhoenixMapReduceUtil.addPhoenixDependencyJars(job.getConfiguration());
LOGGER.info("UpdateStatisticsTool running for: " + tableName + " on snapshot: " + snapshotName
+ " with restore dir: " + restoreDir);
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/MetaDataEndPointIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/MetaDataEndPointIT.java
index 66fdc1540bb..dfd21bf91c7 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/MetaDataEndPointIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/MetaDataEndPointIT.java
@@ -44,9 +44,9 @@
* {@link MetricsMetadataSourceFactory#getMetadataMetricsSource()} singleton, whose counters
* (CREATE_TABLE_COUNT, METADATA_CACHE_ESTIMATED_USED_SIZE, METADATA_CACHE_ADD_COUNT, ...) are
* incremented by every server-side metadata operation in the same JVM. Running this test class in
- * the {@link ParallelStatsDisabledTest} group lets parallel tests in other classes in the same
- * fork mutate those counters between this test's "capture baseline" and "verify after CREATE
- * TABLE" calls, producing intermittent strict-equality assertion failures such as:
+ * the {@link ParallelStatsDisabledTest} group lets parallel tests in other classes in the same fork
+ * mutate those counters between this test's "capture baseline" and "verify after CREATE TABLE"
+ * calls, producing intermittent strict-equality assertion failures such as:
*
*
* testMetadataMetricsOfCreateTable
@@ -54,8 +54,8 @@
*
*
* Categorize as {@link NeedsOwnMiniClusterTest} so failsafe runs this class in its own forked JVM
- * (reuseForks=false), guaranteeing exclusive ownership of the metric singleton for the duration
- * of the suite. Tests within this class still run sequentially in that fork, which is sufficient
+ * (reuseForks=false), guaranteeing exclusive ownership of the metric singleton for the duration of
+ * the suite. Tests within this class still run sequentially in that fork, which is sufficient
* because each individual test captures and verifies its own counter deltas in a single thread.
*/
@Category(NeedsOwnMiniClusterTest.class)
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/MetadataGetTableReadLockIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/MetadataGetTableReadLockIT.java
index f2344fc8c93..58c4b8fbc86 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/MetadataGetTableReadLockIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/MetadataGetTableReadLockIT.java
@@ -97,7 +97,8 @@ public void testBlockedReadDoesNotBlockAnotherRead() throws Exception {
}
}
- private static void waitForCoprocessorOnSystemCatalog(Class coprocessorClass) throws Exception {
+ private static void waitForCoprocessorOnSystemCatalog(Class coprocessorClass)
+ throws Exception {
final TableName sysCatalog = TableName.valueOf("SYSTEM.CATALOG");
utility.waitFor(10000, 100, () -> {
List regions = utility.getHBaseCluster().getRegions(sysCatalog);
@@ -105,8 +106,7 @@ private static void waitForCoprocessorOnSystemCatalog(Class coprocessorClass)
return false;
}
for (HRegion region : regions) {
- if (region.getCoprocessorHost()
- .findCoprocessor(coprocessorClass.getName()) == null) {
+ if (region.getCoprocessorHost().findCoprocessor(coprocessorClass.getName()) == null) {
return false;
}
}
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/iterate/PhoenixQueryTimeoutIT.java b/phoenix-core/src/it/java/org/apache/phoenix/iterate/PhoenixQueryTimeoutIT.java
index 98724c8dc64..8dd5ea5d7f4 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/iterate/PhoenixQueryTimeoutIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/iterate/PhoenixQueryTimeoutIT.java
@@ -201,8 +201,10 @@ public void testScanningResultIteratorQueryTimeoutForPagingWithVeryLowTimeout()
return false;
}
for (HRegion region : regions) {
- if (region.getCoprocessorHost().findCoprocessor(
- DelayedRegionObserver.class.getName()) == null) {
+ if (
+ region.getCoprocessorHost().findCoprocessor(DelayedRegionObserver.class.getName())
+ == null
+ ) {
return false;
}
}
diff --git a/phoenix-core/src/test/java/org/apache/phoenix/mapreduce/util/PhoenixMapReduceUtilTest.java b/phoenix-core/src/test/java/org/apache/phoenix/mapreduce/util/PhoenixMapReduceUtilTest.java
new file mode 100644
index 00000000000..46ff4f36848
--- /dev/null
+++ b/phoenix-core/src/test/java/org/apache/phoenix/mapreduce/util/PhoenixMapReduceUtilTest.java
@@ -0,0 +1,50 @@
+/*
+ * 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.phoenix.mapreduce.util;
+
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+
+import org.apache.hadoop.conf.Configuration;
+import org.junit.Test;
+
+public class PhoenixMapReduceUtilTest {
+
+ @Test
+ public void testAddPhoenixDependencyJars() throws Exception {
+ Configuration conf = new Configuration();
+ PhoenixMapReduceUtil.addPhoenixDependencyJars(conf);
+ String tmpjars = conf.get("tmpjars");
+ assertNotNull("tmpjars should not be null", tmpjars);
+ // Classes loaded from target/classes (same build) won't resolve to a jar,
+ // so we only assert on external dependencies that are actual jars on the classpath.
+ assertTrue("Should contain phoenix-shaded-guava", tmpjars.contains("phoenix-shaded-guava"));
+ assertTrue("Should contain joda-time", tmpjars.contains("joda-time"));
+ assertTrue("Should contain antlr-runtime", tmpjars.contains("antlr-runtime"));
+ assertTrue("Should contain htrace-core", tmpjars.contains("htrace-core"));
+ assertTrue("Should contain hbase-metrics-api", tmpjars.contains("hbase-metrics-api"));
+ assertTrue("Should contain hbase-metrics", tmpjars.contains("hbase-metrics"));
+ assertTrue("Should contain disruptor", tmpjars.contains("disruptor"));
+ assertTrue("Should contain jackson-core", tmpjars.contains("jackson-core"));
+ assertTrue("Should contain jackson-databind", tmpjars.contains("jackson-databind"));
+ assertTrue("Should contain jackson-annotations", tmpjars.contains("jackson-annotations"));
+ assertTrue("Should contain commons-csv", tmpjars.contains("commons-csv"));
+ assertTrue("Should contain json-path", tmpjars.contains("json-path"));
+ assertTrue("Should contain bson", tmpjars.contains("bson"));
+ }
+}