storage: Fix hadoop compatibility issues. #TASK-7320

j-coll · j-coll · commit 74dda6731ac7 · 2025-01-17T21:29:19.000Z
diff --git a/opencga-storage/opencga-storage-hadoop/opencga-storage-hadoop-compat/opencga-storage-hadoop-compat-api/src/main/java/org/opencb/opencga/storage/hadoop/HBaseCompatApi.java b/opencga-storage/opencga-storage-hadoop/opencga-storage-hadoop-compat/opencga-storage-hadoop-compat-api/src/main/java/org/opencb/opencga/storage/hadoop/HBaseCompatApi.java
@@ -36,4 +36,6 @@ public static HBaseCompatApi getInstance() {
     public abstract List<ServerName> getServerList(Admin admin) throws IOException;
 
     public abstract byte[][] getTableStartKeys(Admin admin, Table table) throws IOException;
+
+    public abstract boolean isSnappyAvailable();
 }
diff --git a/opencga-storage/opencga-storage-hadoop/opencga-storage-hadoop-compat/opencga-storage-hadoop-compat-hbase2.0/src/main/java/org/opencb/opencga/storage/hadoop/HBaseCompat.java b/opencga-storage/opencga-storage-hadoop/opencga-storage-hadoop-compat/opencga-storage-hadoop-compat-hbase2.0/src/main/java/org/opencb/opencga/storage/hadoop/HBaseCompat.java
@@ -7,6 +7,7 @@
 import org.apache.hadoop.hbase.client.RegionInfo;
 import org.apache.hadoop.hbase.client.Table;
 import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.hadoop.io.compress.SnappyCodec;
 import org.opencb.opencga.storage.hadoop.variant.annotation.phoenix.PhoenixCompat;
 import org.opencb.opencga.storage.hadoop.variant.annotation.phoenix.PhoenixCompatApi;
 
@@ -45,4 +46,9 @@ public byte[][] getTableStartKeys(Admin admin, Table table) throws IOException {
         }
         return startKeys;
     }
+
+    @Override
+    public boolean isSnappyAvailable() {
+        return SnappyCodec.isNativeCodeLoaded();
+    }
 }
diff --git a/opencga-storage/opencga-storage-hadoop/opencga-storage-hadoop-compat/opencga-storage-hadoop-compat-hbase2.2/src/main/java/org/opencb/opencga/storage/hadoop/HBaseCompat.java b/opencga-storage/opencga-storage-hadoop/opencga-storage-hadoop-compat/opencga-storage-hadoop-compat-hbase2.2/src/main/java/org/opencb/opencga/storage/hadoop/HBaseCompat.java
@@ -7,6 +7,7 @@
 import org.apache.hadoop.hbase.client.RegionInfo;
 import org.apache.hadoop.hbase.client.Table;
 import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.hadoop.io.compress.SnappyCodec;
 import org.opencb.opencga.storage.hadoop.variant.annotation.phoenix.PhoenixCompat;
 import org.opencb.opencga.storage.hadoop.variant.annotation.phoenix.PhoenixCompatApi;
 
@@ -45,4 +46,9 @@ public byte[][] getTableStartKeys(Admin admin, Table table) throws IOException {
         }
         return startKeys;
     }
+
+    @Override
+    public boolean isSnappyAvailable() {
+        return SnappyCodec.isNativeCodeLoaded();
+    }
 }
diff --git a/opencga-storage/opencga-storage-hadoop/opencga-storage-hadoop-compat/opencga-storage-hadoop-compat-hbase2.4/pom.xml b/opencga-storage/opencga-storage-hadoop/opencga-storage-hadoop-compat/opencga-storage-hadoop-compat-hbase2.4/pom.xml
@@ -32,7 +32,7 @@
 
     <properties>
         <hbase.version>2.4.17</hbase.version>
-        <hadoop.version>2.10.0</hadoop.version>
+        <hadoop.version>3.3.4</hadoop.version>
         <phoenix.version>5.1.3</phoenix.version>
         <phoenix-thirdparty.version>1.1.0</phoenix-thirdparty.version>
     </properties>
diff --git a/opencga-storage/opencga-storage-hadoop/opencga-storage-hadoop-compat/opencga-storage-hadoop-compat-hbase2.4/src/main/java/org/opencb/opencga/storage/hadoop/HBaseCompat.java b/opencga-storage/opencga-storage-hadoop/opencga-storage-hadoop-compat/opencga-storage-hadoop-compat-hbase2.4/src/main/java/org/opencb/opencga/storage/hadoop/HBaseCompat.java
@@ -38,4 +38,10 @@ public List<ServerName> getServerList(Admin admin) throws IOException {
     public byte[][] getTableStartKeys(Admin admin, Table table) throws IOException {
         return table.getRegionLocator().getStartKeys();
     }
+
+    @Override
+    public boolean isSnappyAvailable() {
+        // [HADOOP-17125] - Using snappy-java in SnappyCodec - 3.3.1, 3.4.0
+        return true;
+    }
 }
diff --git a/opencga-storage/opencga-storage-hadoop/opencga-storage-hadoop-core/pom.xml b/opencga-storage/opencga-storage-hadoop/opencga-storage-hadoop-core/pom.xml
@@ -448,6 +448,9 @@
                                 <ignoredUsedUndeclaredDependency>
                                     org.apache.tephra:tephra-core
                                 </ignoredUsedUndeclaredDependency>
+                                <ignoredUsedUndeclaredDependency>
+                                    org.apache.tephra:tephra-core-shaded
+                                </ignoredUsedUndeclaredDependency>
                                 <ignoredUsedUndeclaredDependency>
                                     com.lmax:disruptor
                                 </ignoredUsedUndeclaredDependency>
diff --git a/opencga-storage/opencga-storage-hadoop/opencga-storage-hadoop-core/src/main/java/org/opencb/opencga/storage/hadoop/variant/io/VariantExporterDriver.java b/opencga-storage/opencga-storage-hadoop/opencga-storage-hadoop-core/src/main/java/org/opencb/opencga/storage/hadoop/variant/io/VariantExporterDriver.java
@@ -9,7 +9,6 @@
 import org.apache.hadoop.io.compress.CompressionCodec;
 import org.apache.hadoop.io.compress.DeflateCodec;
 import org.apache.hadoop.io.compress.GzipCodec;
-import org.apache.hadoop.io.compress.SnappyCodec;
 import org.apache.hadoop.mapred.JobContext;
 import org.apache.hadoop.mapreduce.*;
 import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
@@ -24,6 +23,7 @@
 import org.opencb.biodata.models.variant.avro.GeneCancerAssociation;
 import org.opencb.biodata.models.variant.avro.VariantAvro;
 import org.opencb.opencga.storage.core.variant.io.VariantWriterFactory;
+import org.opencb.opencga.storage.hadoop.HBaseCompat;
 import org.opencb.opencga.storage.hadoop.variant.AbstractVariantsTableDriver;
 import org.opencb.opencga.storage.hadoop.variant.mr.VariantFileOutputFormat;
 import org.opencb.opencga.storage.hadoop.variant.mr.VariantLocusKey;
@@ -144,7 +144,7 @@ protected void setupJob(Job job) throws IOException {
                     LazyOutputFormat.setOutputFormatClass(job, TextOutputFormat.class);
                     outputFormatClass = LazyOutputFormat.class;
                 }
-                if (SnappyCodec.isNativeCodeLoaded()) {
+                if (HBaseCompat.getInstance().isSnappyAvailable()) {
                     FileOutputFormat.setCompressOutput(job, true);
                     // FIXME: SnappyCodec might not be available in client side
 //                      FileOutputFormat.setOutputCompressorClass(job, SnappyCodec.class);
diff --git a/opencga-storage/opencga-storage-hadoop/opencga-storage-hadoop-core/src/main/java/org/opencb/opencga/storage/hadoop/variant/mr/StreamVariantDriver.java b/opencga-storage/opencga-storage-hadoop/opencga-storage-hadoop-core/src/main/java/org/opencb/opencga/storage/hadoop/variant/mr/StreamVariantDriver.java
@@ -5,7 +5,6 @@
 import org.apache.hadoop.io.compress.CompressionCodec;
 import org.apache.hadoop.io.compress.DeflateCodec;
 import org.apache.hadoop.io.compress.GzipCodec;
-import org.apache.hadoop.io.compress.SnappyCodec;
 import org.apache.hadoop.mapred.JobContext;
 import org.apache.hadoop.mapreduce.Job;
 import org.apache.hadoop.mapreduce.OutputFormat;
@@ -17,6 +16,7 @@
 import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
 import org.apache.hadoop.util.Tool;
 import org.opencb.opencga.storage.core.variant.io.VariantWriterFactory;
+import org.opencb.opencga.storage.hadoop.HBaseCompat;
 import org.opencb.opencga.storage.hadoop.utils.ValueOnlyTextOutputFormat;
 import org.opencb.opencga.storage.hadoop.variant.io.VariantDriver;
 import org.slf4j.Logger;
@@ -164,7 +164,7 @@ protected void setupJob(Job job) throws IOException {
         outputFormatClass = LazyOutputFormat.class;
 
         job.setOutputFormatClass(ValueOnlyTextOutputFormat.class);
-        if (SnappyCodec.isNativeCodeLoaded()) {
+        if (HBaseCompat.getInstance().isSnappyAvailable()) {
             FileOutputFormat.setCompressOutput(job, true);
             // FIXME: SnappyCodec might not be available in client side
 //                    FileOutputFormat.setOutputCompressorClass(job, SnappyCodec.class);
diff --git a/opencga-storage/opencga-storage-hadoop/opencga-storage-hadoop-core/src/main/java/org/opencb/opencga/storage/hadoop/variant/mr/VariantFileOutputFormat.java b/opencga-storage/opencga-storage-hadoop/opencga-storage-hadoop-core/src/main/java/org/opencb/opencga/storage/hadoop/variant/mr/VariantFileOutputFormat.java
@@ -20,7 +20,6 @@
 import org.apache.hadoop.fs.FSDataOutputStream;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.fs.azurebfs.services.AbfsOutputStream;
 import org.apache.hadoop.io.NullWritable;
 import org.apache.hadoop.io.compress.CompressionCodec;
 import org.apache.hadoop.io.compress.GzipCodec;
@@ -54,6 +53,15 @@
  */
 public class VariantFileOutputFormat extends FileOutputFormat<Variant, NullWritable> {
 
+    private static Class<?> abfsOutputStreamClass;
+
+    static {
+        try {
+            abfsOutputStreamClass = Class.forName("org.apache.hadoop.fs.azurebfs.services.AbfsOutputStream");
+        } catch (ClassNotFoundException e) {
+            abfsOutputStreamClass = null;
+        }
+    }
 
     public static final String VARIANT_OUTPUT_FORMAT = "variant.output_format";
 
@@ -74,7 +82,7 @@ public RecordWriter<Variant, NullWritable> getRecordWriter(TaskAttemptContext jo
         FileSystem fs = file.getFileSystem(conf);
         FSDataOutputStream fsOs = fs.create(file, false);
         OutputStream out;
-        if (fsOs.getWrappedStream() instanceof AbfsOutputStream) {
+        if (isAbfsOutputStream(fsOs)) {
             // Disable flush on ABFS. See HADOOP-16548
             out = new FilterOutputStream(fsOs) {
                 @Override
@@ -92,6 +100,10 @@ public void flush() throws IOException {
         return new VariantRecordWriter(configureWriter(job, countingOut), countingOut);
     }
 
+    private static boolean isAbfsOutputStream(FSDataOutputStream fsOs) {
+        return abfsOutputStreamClass != null && abfsOutputStreamClass.isInstance(fsOs.getWrappedStream());
+    }
+
     private DataWriter<Variant> configureWriter(final TaskAttemptContext job, OutputStream fileOut) throws IOException {
 //        job.getCounter(VcfDataWriter.class.getName(), "failed").increment(0); // init
         final Configuration conf = job.getConfiguration();

Original file line number	Diff line number	Diff line change
`@@ -36,4 +36,6 @@ public static HBaseCompatApi getInstance() {`
`36`	`36`	`public abstract List<ServerName> getServerList(Admin admin) throws IOException;`
`37`	`37`
`38`	`38`	`public abstract byte[][] getTableStartKeys(Admin admin, Table table) throws IOException;`
	`39`	`+`
	`40`	`+ public abstract boolean isSnappyAvailable();`
`39`	`41`	`}`
Original file line number	Diff line number	Diff line change
`@@ -7,6 +7,7 @@`
`7`	`7`	`import org.apache.hadoop.hbase.client.RegionInfo;`
`8`	`8`	`import org.apache.hadoop.hbase.client.Table;`
`9`	`9`	`import org.apache.hadoop.hbase.util.Bytes;`
	`10`	`+import org.apache.hadoop.io.compress.SnappyCodec;`
`10`	`11`	`import org.opencb.opencga.storage.hadoop.variant.annotation.phoenix.PhoenixCompat;`
`11`	`12`	`import org.opencb.opencga.storage.hadoop.variant.annotation.phoenix.PhoenixCompatApi;`
`12`	`13`
`@@ -45,4 +46,9 @@ public byte[][] getTableStartKeys(Admin admin, Table table) throws IOException {`
`45`	`46`	`}`
`46`	`47`	`return startKeys;`
`47`	`48`	`}`
	`49`	`+`
	`50`	`+ @Override`
	`51`	`+ public boolean isSnappyAvailable() {`
	`52`	`+ return SnappyCodec.isNativeCodeLoaded();`
	`53`	`+ }`
`48`	`54`	`}`
Original file line number	Diff line number	Diff line change
`@@ -38,4 +38,10 @@ public List<ServerName> getServerList(Admin admin) throws IOException {`
`38`	`38`	`public byte[][] getTableStartKeys(Admin admin, Table table) throws IOException {`
`39`	`39`	`return table.getRegionLocator().getStartKeys();`
`40`	`40`	`}`
	`41`	`+`
	`42`	`+ @Override`
	`43`	`+ public boolean isSnappyAvailable() {`
	`44`	`+ // [HADOOP-17125] - Using snappy-java in SnappyCodec - 3.3.1, 3.4.0`
	`45`	`+ return true;`
	`46`	`+ }`
`41`	`47`	`}`