This is an automated email from the ASF dual-hosted git repository.
virajjasani pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/phoenix.git
The following commit(s) were added to refs/heads/master by this push:
new 80c9125514 PHOENIX-7953 Add addPhoenixDependencyJars utility for
MapReduce jobs (#2564)
80c9125514 is described below
commit 80c91255142dbb4cd7f58ea5c1ee8cedee6e97a7
Author: Ujjawal <[email protected]>
AuthorDate: Tue Jul 7 09:16:37 2026 +0530
PHOENIX-7953 Add addPhoenixDependencyJars utility for MapReduce jobs (#2564)
---
.../phoenix/query/ConnectionQueryServicesImpl.java | 6 +--
phoenix-core-server/pom.xml | 12 +++++
.../phoenix/mapreduce/MultiHfileOutputFormat.java | 2 +
.../phoenix/mapreduce/PhoenixSyncTableTool.java | 1 +
.../phoenix/mapreduce/index/IndexScrutinyTool.java | 1 +
.../apache/phoenix/mapreduce/index/IndexTool.java | 1 +
.../phoenix/mapreduce/transform/TransformTool.java | 1 +
.../mapreduce/util/PhoenixMapReduceUtil.java | 61 ++++++++++++++++++++++
.../phoenix/schema/stats/UpdateStatisticsTool.java | 9 +---
.../apache/phoenix/end2end/MetaDataEndPointIT.java | 10 ++--
.../end2end/MetadataGetTableReadLockIT.java | 6 +--
.../phoenix/iterate/PhoenixQueryTimeoutIT.java | 6 ++-
.../mapreduce/util/PhoenixMapReduceUtilTest.java | 50 ++++++++++++++++++
13 files changed, 145 insertions(+), 21 deletions(-)
diff --git
a/phoenix-core-client/src/main/java/org/apache/phoenix/query/ConnectionQueryServicesImpl.java
b/phoenix-core-client/src/main/java/org/apache/phoenix/query/ConnectionQueryServicesImpl.java
index 5a8334ed63..2c27a57fd3 100644
---
a/phoenix-core-client/src/main/java/org/apache/phoenix/query/ConnectionQueryServicesImpl.java
+++
b/phoenix-core-client/src/main/java/org/apache/phoenix/query/ConnectionQueryServicesImpl.java
@@ -5471,14 +5471,14 @@ public class ConnectionQueryServicesImpl extends
DelegateQueryServices
// is safe.
// A second flavor of transient failure is "already running another
snapshot on the same table",
// produced by RPC-level retries. The original snapshot() RPC has already
been accepted by the
- // master and the snapshot procedure is in flight, but the client retries
the call and the master
+ // master and the snapshot procedure is in flight, but the client retries
the call and the
+ // master
// rejects the duplicate. In that case the existing snapshot procedure
will succeed, so we treat
// it as success once the snapshot of that name shows up in
listSnapshots(). If it never appears
// within the polling window we fall through to a normal retry.
final int maxAttempts = 5;
final long backoffMs = 1000L;
- final long alreadyRunningWaitMs =
- TimeUnit.MINUTES.toMillis(2L);
+ final long alreadyRunningWaitMs = TimeUnit.MINUTES.toMillis(2L);
SQLException sqlE = null;
for (int attempt = 1; attempt <= maxAttempts; attempt++) {
sqlE = null;
diff --git a/phoenix-core-server/pom.xml b/phoenix-core-server/pom.xml
index ea4fde4bbd..8eb053363a 100644
--- a/phoenix-core-server/pom.xml
+++ b/phoenix-core-server/pom.xml
@@ -185,6 +185,18 @@
<groupId>org.xerial.snappy</groupId>
<artifactId>snappy-java</artifactId>
</dependency>
+ <dependency>
+ <groupId>com.lmax</groupId>
+ <artifactId>disruptor</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>com.jayway.jsonpath</groupId>
+ <artifactId>json-path</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.mongodb</groupId>
+ <artifactId>bson</artifactId>
+ </dependency>
</dependencies>
<build>
diff --git
a/phoenix-core-server/src/main/java/org/apache/phoenix/mapreduce/MultiHfileOutputFormat.java
b/phoenix-core-server/src/main/java/org/apache/phoenix/mapreduce/MultiHfileOutputFormat.java
index 9847729b11..565a14b9ac 100644
---
a/phoenix-core-server/src/main/java/org/apache/phoenix/mapreduce/MultiHfileOutputFormat.java
+++
b/phoenix-core-server/src/main/java/org/apache/phoenix/mapreduce/MultiHfileOutputFormat.java
@@ -78,6 +78,7 @@ import
org.apache.hadoop.mapreduce.lib.partition.TotalOrderPartitioner;
import org.apache.phoenix.mapreduce.bulkload.TableRowkeyPair;
import org.apache.phoenix.mapreduce.bulkload.TargetTableRef;
import org.apache.phoenix.mapreduce.bulkload.TargetTableRefFunctions;
+import org.apache.phoenix.mapreduce.util.PhoenixMapReduceUtil;
import org.apache.phoenix.util.EnvironmentEdgeManager;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -685,6 +686,7 @@ public class MultiHfileOutputFormat extends
FileOutputFormat<TableRowkeyPair, Ce
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 80eacde25e..09b0e5c4d9 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 @@ public class PhoenixSyncTableTool extends Configured
implements Tool {
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 c9111aa7b8..426e13e9da 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 @@ public class IndexScrutinyTool extends Configured
implements Tool {
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 cc918dc46f..8662dbdec2 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 @@ public class IndexTool extends Configured implements Tool {
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 daca9a0461..2bc12485dc 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 class TransformTool extends Configured implements
Tool {
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 b5f62bdfc5..d20cd1b7fe 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.mapreduce.util.PhoenixConfigurationUtil.SchemaType;
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)}.
+ * <p>
+ * 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 ea3a0bebdd..0d7be097d0 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 static
org.apache.phoenix.query.QueryServicesOptions.DEFAULT_IS_NAMESPACE
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.db.DBInputFormat.NullDBWritable;
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 @@ public class UpdateStatisticsTool extends Configured
implements Tool {
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 66fdc1540b..dfd21bf91c 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 @@ import org.junit.experimental.categories.Category;
* {@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:
*
* <pre>
* testMetadataMetricsOfCreateTable
@@ -54,8 +54,8 @@ import org.junit.experimental.categories.Category;
* </pre>
*
* 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 f2344fc8c9..58c4b8fbc8 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 class MetadataGetTableReadLockIT extends BaseTest {
}
}
- 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<HRegion> regions = utility.getHBaseCluster().getRegions(sysCatalog);
@@ -105,8 +106,7 @@ public class MetadataGetTableReadLockIT extends BaseTest {
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 98724c8dc6..8dd5ea5d7f 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 class PhoenixQueryTimeoutIT extends
ParallelStatsDisabledIT {
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 0000000000..46ff4f3684
--- /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"));
+ }
+}
