1 /***********************************************************************
2  * Copyright (c) 2013-2025 General Atomics Integrated Intelligence, Inc.
3  * All rights reserved. This program and the accompanying materials
4  * are made available under the terms of the Apache License, Version 2.0
5  * which accompanies this distribution and is available at
6  * https://www.apache.org/licenses/LICENSE-2.0
7  ***********************************************************************/
8 
9 package org.locationtech.geomesa.hbase.utils
10 
11 import org.apache.hadoop.conf.Configuration
12 import org.apache.hadoop.fs.Path
13 import org.apache.hadoop.hbase.client.{Admin, HBaseAdmin}
14 import org.apache.hadoop.hbase.io.compress.Compression.Algorithm
15 import org.apache.hadoop.hbase.io.encoding.DataBlockEncoding
16 import org.apache.hadoop.hbase.regionserver.BloomType
17 import org.apache.hadoop.hbase.{Coprocessor, NamespaceDescriptor, TableName}
18 
19 import java.lang.reflect.InvocationTargetException
20 import scala.util.Try
21 
22 /**
23   * Reflection wrapper for method signature differences in the HBase API
24   */
25 object HBaseVersions {
26 
27   /**
28    * Create a new table
29    *
30    * @param admin admin connection to hbase
31    * @param name table name
32    * @param colFamilies column families
33    * @param bloom bloom filter
34    * @param compression compression
35    * @param encoding data block encoding
36    * @param coprocessor coprocessor class and optional jar path
37    * @param splits initial table splits (empty for no splits)
38    */
39   def createTableAsync(
40     admin: Admin,
41     name: TableName,
42     colFamilies: Seq[Array[Byte]],
43     bloom: Option[BloomType],
44     compression: Option[Algorithm],
45     encoding: Option[DataBlockEncoding],
46     inMemory: Option[Boolean],
47     coprocessor: Option[(String, Option[Path])],
48     splits: Seq[Array[Byte]]): Unit = {
49 
50     if (name.getNamespaceAsString != null &&
51         name.getNamespaceAsString != NamespaceDescriptor.DEFAULT_NAMESPACE_NAME_STR &&
52         name.getNamespaceAsString != NamespaceDescriptor.SYSTEM_NAMESPACE_NAME_STR &&
53         Try(Option(admin.getNamespaceDescriptor(name.getNamespaceAsString))).getOrElse(None).isEmpty) {
54       admin.createNamespace(NamespaceDescriptor.create(name.getNamespaceAsString).build())
55     }
56 
57     val descriptor = hTableDescriptorClass.getConstructor(classOf[TableName]).newInstance(name).asInstanceOf[AnyRef]
58 
59     colFamilies.foreach { k =>
60       val column = hColumnDescriptorClass.getConstructor(classOf[Array[Byte]]).newInstance(k)
61       bloom.foreach(_setBloomFilterType(column, _))
62       compression.foreach(_setCompressionType(column, _))
63       encoding.foreach(_setDataBlockEncoding(column, _))
64       inMemory.foreach(_setInMemory(column, _))
65       _addFamily(descriptor, column)
66     }
67 
68     coprocessor.foreach { case (clas, path) =>
69       _addCoprocessor(descriptor, clas, path.orNull, Coprocessor.PRIORITY_USER, null)
70     }
71 
72     _createTableAsync(admin, descriptor, if (splits.isEmpty) { null } else { splits.toArray })
73   }
74 
75   /**
76    * Disable a table asynchronously
77    *
78    * @param admin admin hbase connection
79    * @param table table to disable
80    */
81   def disableTableAsync(admin: Admin, table: TableName): Unit = _disableTableAsync(admin, table)
82 
83   /**
84    * Checks whether HBase is available, and throws an exception if not
85    *
86    * @param conf HBase configuration
87    */
88   def checkAvailable(conf: Configuration): Unit = _available(conf)
89 
90   private lazy val hTableDescriptorClass = Class.forName("org.apache.hadoop.hbase.HTableDescriptor")
91 
92   private lazy val hColumnDescriptorClass = Class.forName("org.apache.hadoop.hbase.HColumnDescriptor")
93 
94   private lazy val _setBloomFilterType: (Any, BloomType) => Unit =
95     findMethod(hColumnDescriptorClass, "setBloomFilterType", classOf[BloomType])
96 
97   private lazy val _setCompressionType: (Any, Algorithm) => Unit =
98     findMethod(hColumnDescriptorClass, "setCompressionType", classOf[Algorithm])
99 
100   private lazy val _setDataBlockEncoding: (Any, DataBlockEncoding) => Unit =
101     findMethod(hColumnDescriptorClass, "setDataBlockEncoding", classOf[DataBlockEncoding])
102 
103   private lazy val _setInMemory: (Any, Boolean) => Unit =
104     findMethod(hColumnDescriptorClass, "setInMemory", classOf[Boolean])
105 
106   // HBase 1.3 signature: public HTableDescriptor addFamily(final HColumnDescriptor family)
107   // CDH 5.12 signature: public void addFamily(final HColumnDescriptor family)
108   private lazy val _addFamily: (Any, Any) => Unit =
109     findMethod(hTableDescriptorClass, "addFamily", hColumnDescriptorClass)
110 
111   private lazy val _disableTableAsync: (Admin, TableName) => Unit =
112     findMethod(classOf[Admin], "disableTableAsync", classOf[TableName])
113 
114   // HBase 1.3 signature:
115   // public HTableDescriptor addCoprocessor(String className, Path jarFilePath, int priority, final Map<String, String> kvs)
116   // CDH 5.12 signature:
117   // public void addCoprocessor(String className, Path jarFilePath, int priority, final Map<String, String> kvs)
118   private lazy val _addCoprocessor: (Any, String, Path, Int, java.util.Map[String, String]) => Unit = {
119     val methods = hTableDescriptorClass.getMethods
120     val method = methods.find(m => m.getName == "addCoprocessor" && m.getParameterCount == 4).getOrElse {
121       throw new NoSuchMethodException("Couldn't find HTableDescriptor.addCoprocessor method")
122     }
123     val parameterTypes = method.getParameterTypes.asInstanceOf[Array[AnyRef]]
124     val expected = Array[AnyRef](classOf[String], classOf[Path], classOf[Int], classOf[java.util.Map[String, String]])
125     if (java.util.Arrays.equals(parameterTypes, expected)) {
126       (descriptor, className, jarFilePath, priority, kvs) => {
127         try { method.invoke(descriptor, className, jarFilePath, Int.box(priority), kvs)  } catch {
128           case e: InvocationTargetException => throw e.getCause
129         }
130       }
131 
132     } else {
133       throw new NoSuchMethodException(
134         s"Couldn't find HTableDescriptor.addCoprocessor method with correct parameters: $method")
135     }
136   }
137 
138   private lazy val _createTableAsync: (Admin, AnyRef, Array[Array[Byte]]) => Unit = {
139     val methods = classOf[Admin].getMethods
140     val method = methods.find(m => m.getName == "createTableAsync" && m.getParameterCount == 2).getOrElse {
141       throw new NoSuchMethodException("Couldn't find Admin.createTableAsync method")
142     }
143     val parameterTypes = method.getParameterTypes
144     if (parameterTypes.lengthCompare(2) == 0
145         && parameterTypes.head.isAssignableFrom(hTableDescriptorClass)
146         && parameterTypes.last == classOf[Array[Array[Byte]]]) {
147       (admin, descriptor, splits) => {
148         try { method.invoke(admin, descriptor, splits) } catch {
149           case e: InvocationTargetException => throw e.getCause
150         }
151       }
152     } else {
153       throw new NoSuchMethodException(s"Couldn't find Admin.createTableAsync method with correct parameters: $method")
154     }
155   }
156 
157   private lazy val _available: Configuration => Unit = {
158     val names = Seq("available", "checkHBaseAvailable")
159     val method = classOf[HBaseAdmin].getMethods.find(m => names.contains(m.getName)).getOrElse {
160       throw new NoSuchMethodException("Couldn't find HBaseAdmin.available method")
161     }
162     if (method.getParameterCount != 1 || method.getParameterTypes.head != classOf[Configuration]) {
163       throw new NoSuchMethodException(s"Couldn't find HBaseAdmin.available method with correct parameters: $method")
164     }
165     conf => {
166       try { method.invoke(null, conf) } catch {
167         case e: InvocationTargetException => throw e.getCause
168       }
169     }
170   }
171 
172   private def findMethod(clas: Class[_], name: String, param: Class[_]): (Any, Any) => Unit = {
173     val method = clas.getMethods.find(_.getName == name).getOrElse {
174       throw new NoSuchMethodException(s"Couldn't find ${clas.getSimpleName}.$name method")
175     }
176     val parameterTypes = method.getParameterTypes
177     if (parameterTypes.length == 1 && parameterTypes.head == param) {
178       (obj, p) => {
179         try { method.invoke(obj, p.asInstanceOf[AnyRef]) } catch {
180           case e: InvocationTargetException => throw e.getCause
181         }
182       }
183     } else {
184       throw new NoSuchMethodException(
185         s"Couldn't find ${clas.getSimpleName}.$name method with correct parameters: $method")
186     }
187   }
188 }