[Feature][E2E] Add hive3 e2e test case

.github/workflows/backend.yml

@@ -806,7 +806,7 @@ jobs:
       matrix:
         java: [ '8', '11' ]
         os: [ 'ubuntu-latest' ]
-    timeout-minutes: 150
+    timeout-minutes: 180
     steps:
       - uses: actions/checkout@v2
       - name: Set up JDK ${{ matrix.java }}

seatunnel-connectors-v2/connector-hive/src/main/java/org/apache/seatunnel/connectors/seatunnel/hive/storage/StorageFactory.java

@@ -18,13 +18,18 @@
 package org.apache.seatunnel.connectors.seatunnel.hive.storage;
 
 public class StorageFactory {
+
     public static Storage getStorageType(String hiveSdLocation) {
         if (hiveSdLocation.startsWith(StorageType.S3.name().toLowerCase())) {
             return new S3Storage();
         } else if (hiveSdLocation.startsWith(StorageType.OSS.name().toLowerCase())) {
             return new OSSStorage();
         } else if (hiveSdLocation.startsWith(StorageType.COS.name().toLowerCase())) {
             return new COSStorage();
+        } else if (hiveSdLocation.startsWith(StorageType.FILE.name().toLowerCase())) {
+            // Currently used in e2e, When Hive uses local files as storage, "file:" needs to be
+            // replaced with "file:/" to avoid being recognized as HDFS storage.
+            return new HDFSStorage(hiveSdLocation.replace("file:", "file:/"));
         } else {
             return new HDFSStorage(hiveSdLocation);
         }

seatunnel-connectors-v2/connector-hive/src/main/java/org/apache/seatunnel/connectors/seatunnel/hive/storage/StorageType.java

@@ -21,5 +21,6 @@ public enum StorageType {
     S3,
     OSS,
     COS,
+    FILE,
     HDFS
 }

seatunnel-e2e/seatunnel-connector-v2-e2e/connector-hive-e2e/pom.xml

@@ -25,6 +25,10 @@
     <artifactId>connector-hive-e2e</artifactId>
     <name>SeaTunnel : E2E : Connector V2 : Hive</name>
 
+    <properties>
+        <hive.version>3.1.3</hive.version>
+    </properties>
+
     <dependencies>
         <dependency>
             <groupId>org.apache.seatunnel</groupId>
@@ -44,6 +48,28 @@
             <version>${project.version}</version>
             <classifier>optional</classifier>
             <scope>test</scope>
+            <exclusions>
+                <exclusion>
+                    <groupId>log4j</groupId>
+                    <artifactId>log4j</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.hive</groupId>
+            <artifactId>hive-jdbc</artifactId>
+            <version>${hive.version}</version>
+            <scope>test</scope>
+            <exclusions>
+                <exclusion>
+                    <groupId>log4j</groupId>
+                    <artifactId>log4j</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>org.apache.logging.log4j</groupId>
+                    <artifactId>log4j-web</artifactId>
+                </exclusion>
+            </exclusions>
         </dependency>
     </dependencies>
 

seatunnel-e2e/seatunnel-connector-v2-e2e/connector-hive-e2e/src/test/java/org/apache/seatunnel/e2e/connector/hive/HiveContainer.java

@@ -0,0 +1,120 @@
+/*
+ * 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.seatunnel.e2e.connector.hive;
+
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.metastore.HiveMetaStoreClient;
+import org.apache.hadoop.hive.metastore.api.MetaException;
+
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.output.Slf4jLogConsumer;
+import org.testcontainers.containers.wait.strategy.Wait;
+import org.testcontainers.containers.wait.strategy.WaitStrategy;
+import org.testcontainers.utility.DockerImageName;
+import org.testcontainers.utility.DockerLoggerFactory;
+
+import java.sql.Connection;
+import java.sql.Driver;
+import java.sql.SQLException;
+import java.util.Properties;
+
+public class HiveContainer extends GenericContainer<HiveContainer> {
+    public static final String IMAGE = "apache/hive";
+    public static final String DEFAULT_TAG = "3.1.3";
+
+    private static final DockerImageName DEFAULT_IMAGE_NAME = DockerImageName.parse(IMAGE);
+
+    public static final int HIVE_SERVER_PORT = 10000;
+
+    public static final int HMS_PORT = 9083;
+
+    private static final String SERVICE_NAME_ENV = "SERVICE_NAME";
+
+    private static final String DRIVER_CLASS_NAME = "org.apache.hive.jdbc.HiveDriver";
+
+    public HiveContainer(Role role) {
+        super(DEFAULT_IMAGE_NAME.withTag(DEFAULT_TAG));
+        this.addExposedPorts(role.exposePort);
+        this.addEnv(SERVICE_NAME_ENV, role.serviceName);
+        this.setWaitStrategy(role.waitStrategy);
+        this.withLogConsumer(
+                new Slf4jLogConsumer(
+                        DockerLoggerFactory.getLogger(
+                                DEFAULT_IMAGE_NAME.withTag(DEFAULT_TAG).toString())));
+    }
+
+    public static HiveContainer hmsStandalone() {
+        return new HiveContainer(Role.HMS_STANDALONE);
+    }
+
+    public static HiveContainer hiveServer() {
+        return new HiveContainer(Role.HIVE_SERVER_WITH_EMBEDDING_HMS);
+    }
+
+    public String getMetastoreUri() {
+        return String.format("thrift://%s:%s", getHost(), getMappedPort(HMS_PORT));
+    }
+
+    public String getHiveJdbcUri() {
+        return String.format(
+                "jdbc:hive2://%s:%s/default", getHost(), getMappedPort(HIVE_SERVER_PORT));
+    }
+
+    public HiveMetaStoreClient createMetaStoreClient() throws MetaException {
+        HiveConf conf = new HiveConf();
+        conf.set("hive.metastore.uris", getMetastoreUri());
+
+        return new HiveMetaStoreClient(conf);
+    }
+
+    public Connection getConnection()
+            throws ClassNotFoundException, InstantiationException, IllegalAccessException,
+                    SQLException {
+        Driver driver = loadHiveJdbcDriver();
+
+        return driver.connect(getHiveJdbcUri(), getJdbcConnectionConfig());
+    }
+
+    public Driver loadHiveJdbcDriver()
+            throws ClassNotFoundException, InstantiationException, IllegalAccessException {
+        return (Driver) Class.forName(DRIVER_CLASS_NAME).newInstance();
+    }
+
+    public Properties getJdbcConnectionConfig() {
+        Properties props = new Properties();
+
+        return props;
+    }
+
+    public enum Role {
+        HIVE_SERVER_WITH_EMBEDDING_HMS(
+                "hiveserver2", HIVE_SERVER_PORT, Wait.forLogMessage(".*Starting HiveServer2.*", 1)),
+        HMS_STANDALONE(
+                "metastore", HMS_PORT, Wait.forLogMessage(".*Starting Hive Metastore Server.*", 1));
+
+        private final String serviceName;
+        private final int exposePort;
+        private final WaitStrategy waitStrategy;
+
+        Role(String serviceName, int exposePort, WaitStrategy waitStrategy) {
+            this.serviceName = serviceName;
+            this.exposePort = exposePort;
+            this.waitStrategy = waitStrategy;
+        }
+    }
+}