PHOENIX-7953 Add addPhoenixDependencyJars utility for MapReduce jobs

phoenix-core-client/src/main/java/org/apache/phoenix/query/ConnectionQueryServicesImpl.java

@@ -5471,14 +5471,14 @@ private void createSnapshot(String snapshotName, String tableName) throws SQLExc
     // 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;

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>

phoenix-core-server/src/main/java/org/apache/phoenix/mapreduce/MultiHfileOutputFormat.java

@@ -78,6 +78,7 @@
 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 static void configureIncrementalLoad(Job job, List<TargetTableRef> tables
 
     configurePartitioner(job, tablesStartKeys);
     TableMapReduceUtil.addDependencyJars(job);
+    PhoenixMapReduceUtil.addPhoenixDependencyJars(job.getConfiguration());
     TableMapReduceUtil.initCredentials(job);
 
   }

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);

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;
     }
 

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;
     }

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);
 

phoenix-core-server/src/main/java/org/apache/phoenix/mapreduce/util/PhoenixMapReduceUtil.java

@@ -17,33 +17,94 @@
  */
 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;
 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

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);

phoenix-core/src/it/java/org/apache/phoenix/end2end/MetaDataEndPointIT.java

@@ -44,18 +44,18 @@
  * {@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
  *     expected:&lt;232498&gt; but was:&lt;240491&gt;
  * </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)

phoenix-core/src/it/java/org/apache/phoenix/end2end/MetadataGetTableReadLockIT.java

@@ -97,16 +97,16 @@ 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<HRegion> regions = utility.getHBaseCluster().getRegions(sysCatalog);
       if (regions.isEmpty()) {
         return false;
       }
       for (HRegion region : regions) {
-        if (region.getCoprocessorHost()
-            .findCoprocessor(coprocessorClass.getName()) == null) {
+        if (region.getCoprocessorHost().findCoprocessor(coprocessorClass.getName()) == null) {
           return false;
         }
       }

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;
           }
         }

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"));
+  }
+}