[INLONG-9009][Sort] Add HBase source and sink connector on flink 1.15

inlong-distribution/src/main/assemblies/sort-connectors-v1.15.xml

@@ -75,5 +75,13 @@
             </includes>
             <fileMode>0644</fileMode>
         </fileSet>
+        <fileSet>
+            <directory>../inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/hbase/target</directory>
+            <outputDirectory>inlong-sort/connectors</outputDirectory>
+            <includes>
+                <include>sort-connector-hbase-v1.15-${project.version}.jar</include>
+            </includes>
+            <fileMode>0644</fileMode>
+        </fileSet>
     </fileSets>
 </assembly>

inlong-sort/sort-end-to-end-tests/sort-end-to-end-tests-v1.15/pom.xml

@@ -138,6 +138,17 @@
             <artifactId>clickhouse-jdbc</artifactId>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>junit</groupId>
+            <artifactId>junit</artifactId>
+            <version>${junit.version}</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.hadoop</groupId>
+            <artifactId>hadoop-common</artifactId>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 
     <build>
@@ -213,6 +224,17 @@
                         </goals>
                         <phase>pre-integration-test</phase>
                     </execution>
+                    <execution>
+                        <id>store-classpath-in-target-for-tests</id>
+                        <goals>
+                            <goal>build-classpath</goal>
+                        </goals>
+                        <phase>package</phase>
+                        <configuration>
+                            <outputFile>${project.build.directory}/hadoop.classpath</outputFile>
+                            <excludeGroupIds>org.apache.flink</excludeGroupIds>
+                        </configuration>
+                    </execution>
                 </executions>
             </plugin>
 

inlong-sort/sort-end-to-end-tests/sort-end-to-end-tests-v1.15/src/test/java/org/apache/inlong/sort/tests/utils/HBaseContainer.java

@@ -0,0 +1,109 @@
+/*
+ * 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.inlong.sort.tests.utils;
+
+import com.github.dockerjava.api.command.InspectContainerResponse;
+import org.testcontainers.containers.Container;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.images.builder.ImageFromDockerfile;
+
+import java.util.Arrays;
+import java.util.stream.Collectors;
+
+/** Standalone containerized HBase instance that builds the image on the fly. */
+@SuppressWarnings("rawtypes")
+public class HBaseContainer extends GenericContainer<HBaseContainer> {
+
+    private static final String HBASE_BIN = "/opt/hbase/bin";
+    private static final int MAX_RETRIES = 3;
+
+    public HBaseContainer(String hbaseVersion) {
+        super(getImageFromDockerfile(hbaseVersion));
+    }
+
+    private static ImageFromDockerfile getImageFromDockerfile(String hbaseVersion) {
+        return new ImageFromDockerfile()
+                .withDockerfileFromBuilder(
+                        builder -> builder.from("adoptopenjdk/openjdk8")
+                                .env("HBASE_VERSION", hbaseVersion)
+                                .run(
+                                        "export INITRD=no"
+                                                + " && export HBASE_DIST=\"http://archive.apache.org/dist/hbase\""
+                                                + " && apt-get update -y"
+                                                + " && apt-get install -y --no-install-recommends curl"
+                                                + " && cd /opt"
+                                                + " && curl -SL $HBASE_DIST/$HBASE_VERSION/hbase-$HBASE_VERSION-bin.tar.gz"
+                                                + " | tar -x -z && mv hbase-${HBASE_VERSION} hbase")
+                                .expose(2181)
+                                .cmd(
+                                        "/bin/sh",
+                                        "-c",
+                                        String.format(
+                                                "nohup %s/start-hbase.sh & sleep infinity",
+                                                HBASE_BIN)));
+    }
+
+    @Override
+    protected void containerIsStarted(InspectContainerResponse containerInfo) {
+        ExecResult res = null;
+        for (int i = 0; i < MAX_RETRIES; i++) {
+            try {
+                res = execCmd("scan 'hbase:meta'");
+                if (res.getStdout().contains("hbase:namespace")) {
+                    return;
+                }
+                Thread.sleep(5000L);
+            } catch (Exception e) {
+                throw new RuntimeException("Failed to verify if container is started.", e);
+            }
+        }
+        throw new IllegalStateException("Failed to start HBase properly:\n" + res);
+    }
+
+    public Container.ExecResult createTable(String table, String... colFamilies) throws Exception {
+        String createCmd =
+                String.format("create '%s',", table)
+                        + Arrays.stream(colFamilies)
+                                .map(cf -> String.format("{NAME=>'%s'}", cf))
+                                .collect(Collectors.joining(","));
+
+        return execCmd(createCmd);
+    }
+
+    public Container.ExecResult putData(
+            String table, String rowKey, String colFamily, String colQualifier, String val)
+            throws Exception {
+        String putCmd =
+                String.format(
+                        "put '%s','%s','%s:%s','%s'", table, rowKey, colFamily, colQualifier, val);
+
+        return execCmd(putCmd);
+    }
+
+    public Container.ExecResult scanTable(String table) throws Exception {
+        String scanCmd = String.format("scan '%s'", table);
+
+        return execCmd(scanCmd);
+    }
+
+    private Container.ExecResult execCmd(String cmd) throws Exception {
+        String hbaseShellCmd = String.format("echo \"%s\" | %s/hbase shell", cmd, HBASE_BIN);
+
+        return execInContainer("sh", "-c", hbaseShellCmd);
+    }
+}

inlong-sort/sort-end-to-end-tests/sort-end-to-end-tests-v1.15/src/test/resources/flinkSql/hbase_e2e.sql

@@ -0,0 +1,33 @@
+CREATE TABLE MyHBaseSource (
+ rowkey STRING,
+ family1 ROW<f1c1 STRING>,
+ family2 ROW<f2c1 STRING, f2c2 STRING>
+) WITH (
+ 'connector' = 'hbase-2.2-inlong',
+ 'table-name' = 'sourceTable',
+ 'zookeeper.quorum' = 'hbase1:2181'
+);
+
+CREATE TABLE MyHBaseSink
+(
+    rowkey STRING,
+    family1 ROW<f1c1 STRING>,
+    family2 ROW<f2c1 STRING, f2c2 STRING>
+) WITH (
+      'connector' = 'hbase-2.2-inlong',
+      'table-name' = 'sinkTable',
+      'zookeeper.quorum' = 'hbase2:2181',
+      'sink.buffer-flush.max-rows' = '1',
+      'sink.buffer-flush.interval' = '2s'
+      );
+
+INSERT INTO MyHBaseSink
+SELECT rowkey,
+       ROW(a),
+       ROW(b, c)
+FROM (
+    SELECT rowkey,
+           REGEXP_REPLACE(family1.f1c1, 'v', 'value') as a,
+           family2.f2c1 as b,
+           family2.f2c2 as c
+    FROM MyHBaseSource) source;