[Core][jdk17] Support jdk17

.github/workflows/schedule_backend.yml

@@ -104,7 +104,7 @@ jobs:
     runs-on: ${{ matrix.os }}
     strategy:
       matrix:
-        java: [ '8', '11' ]
+        java: [ '8', '11' ,'17']
         os: [ 'ubuntu-latest', 'windows-latest' ]
     timeout-minutes: 90
     steps:
@@ -126,7 +126,7 @@ jobs:
     runs-on: ${{ matrix.os }}
     strategy:
       matrix:
-        java: [ '8', '11' ]
+        java: [ '8', '11' ,'17']
         os: [ 'ubuntu-latest' ]
     timeout-minutes: 90
     steps:

.mvn/wrapper/maven-wrapper.properties

@@ -15,5 +15,5 @@
 # specific language governing permissions and limitations
 # under the License.
 
-distributionUrl=https://repo.maven.apache.org/maven2/org/apache/maven/apache-maven/3.8.4/apache-maven-3.8.4-bin.zip
-wrapperUrl=https://repo.maven.apache.org/maven2/org/apache/maven/wrapper/maven-wrapper/3.1.0/maven-wrapper-3.1.0.jar
+distributionUrl=https://repo.maven.apache.org/maven2/org/apache/maven/apache-maven/3.8.6/apache-maven-3.8.6-bin.zip
+wrapperUrl=https://repo.maven.apache.org/maven2/org/apache/maven/wrapper/maven-wrapper/3.2.0/maven-wrapper-3.2.0.jar

config/seatunnel-env.cmd

@@ -18,4 +18,29 @@ REM Home directory of spark distribution.
 if "%SPARK_HOME%" == "" set "SPARK_HOME=C:\Program Files\spark"
 
 REM Home directory of flink distribution.
-if "%FLINK_HOME%" == "" set "FLINK_HOME=C:\Program Files\flink"
+if "%FLINK_HOME%" == "" set "FLINK_HOME=C:\Program Files\flink"
+
+REM Initialize JAVA_OPTS to avoid unbound variable error.
+if not defined JAVA_OPTS (
+    set "JAVA_OPTS="
+)
+
+REM Function to check if the current Java version is 17 or higher.
+for /f "tokens=2 delims==" %%v in ('java -XshowSettings:properties -version 2>&1 ^| findstr "java.version"') do (
+    set "java_version=%%v"
+)
+for /f "tokens=1 delims=." %%v in ("%java_version%") do (
+    set "java_major_version=%%v"
+)
+
+REM Set JAVA_OPTS if Java version is 17 or newer.
+if %java_major_version% geq 17 (
+    if defined JAVA_OPTS (
+        set "JAVA_OPTS=%JAVA_OPTS% --add-exports=java.base/sun.net.util=ALL-UNNAMED --add-exports=java.rmi/sun.rmi.registry=ALL-UNNAMED --add-exports=jdk.compiler/com.sun.tools.javac.api=ALL-UNNAMED --add-exports=jdk.compiler/com.sun.tools.javac.file=ALL-UNNAMED --add-exports=jdk.compiler/com.sun.tools.javac.parser=ALL-UNNAMED --add-exports=jdk.compiler/com.sun.tools.javac.tree=ALL-UNNAMED --add-exports=jdk.compiler/com.sun.tools.javac.util=ALL-UNNAMED --add-exports=java.security.jgss/sun.security.krb5=ALL-UNNAMED --add-opens=java.base/java.lang=ALL-UNNAMED --add-opens=java.base/java.net=ALL-UNNAMED --add-opens=java.base/java.io=ALL-UNNAMED --add-opens=java.base/java.nio=ALL-UNNAMED --add-opens=java.base/sun.nio.ch=ALL-UNNAMED --add-opens=java.base/java.lang.reflect=ALL-UNNAMED --add-opens=java.base/java.text=ALL-UNNAMED --add-opens=java.base/java.time=ALL-UNNAMED --add-opens=java.base/java.util=ALL-UNNAMED --add-opens=java.base/java.util.concurrent=ALL-UNNAMED --add-opens=java.base/java.util.concurrent.atomic=ALL-UNNAMED --add-opens=java.base/java.util.concurrent.locks=ALL-UNNAMED"
+    ) else (
+        set "JAVA_OPTS=--add-exports=java.base/sun.net.util=ALL-UNNAMED --add-exports=java.rmi/sun.rmi.registry=ALL-UNNAMED --add-exports=jdk.compiler/com.sun.tools.javac.api=ALL-UNNAMED --add-exports=jdk.compiler/com.sun.tools.javac.file=ALL-UNNAMED --add-exports=jdk.compiler/com.sun.tools.javac.parser=ALL-UNNAMED --add-exports=jdk.compiler/com.sun.tools.javac.tree=ALL-UNNAMED --add-exports=jdk.compiler/com.sun.tools.javac.util=ALL-UNNAMED --add-exports=java.security.jgss/sun.security.krb5=ALL-UNNAMED --add-opens=java.base/java.lang=ALL-UNNAMED --add-opens=java.base/java.net=ALL-UNNAMED --add-opens=java.base/java.io=ALL-UNNAMED --add-opens=java.base/java.nio=ALL-UNNAMED --add-opens=java.base/sun.nio.ch=ALL-UNNAMED --add-opens=java.base/java.lang.reflect=ALL-UNNAMED --add-opens=java.base/java.text=ALL-UNNAMED --add-opens=java.base/java.time=ALL-UNNAMED --add-opens=java.base/java.util=ALL-UNNAMED --add-opens=java.base/java.util.concurrent=ALL-UNNAMED --add-opens=java.base/java.util.concurrent.atomic=ALL-UNNAMED --add-opens=java.base/java.util.concurrent.locks=ALL-UNNAMED"
+    )
+)
+
+REM Export JAVA_OPTS
+set JAVA_OPTS

config/seatunnel-env.sh

@@ -20,3 +20,26 @@
 SPARK_HOME=${SPARK_HOME:-/opt/spark}
 # Home directory of flink distribution.
 FLINK_HOME=${FLINK_HOME:-/opt/flink}
+
+# Initialize JAVA_OPTS to avoid unbound variable error.
+JAVA_OPTS=${JAVA_OPTS:-}
+
+# Function to check if the current Java version is 17 or higher.
+is_java_17_or_newer() {
+    java_version_output=$(java -version 2>&1)
+    if [[ $java_version_output == *"version \""* ]]; then
+        java_version=$(echo $java_version_output | awk -F '"' '/version/ {print $2}')
+        java_major_version=$(echo $java_version | cut -d'.' -f1)
+        if [[ "$java_major_version" -ge 17 ]]; then
+            return 0
+        fi
+    fi
+    return 1
+}
+
+# Set JAVA_OPTS if Java version is 17 or newer.
+if is_java_17_or_newer; then
+    JAVA_OPTS="$JAVA_OPTS --add-exports=java.base/sun.net.util=ALL-UNNAMED --add-exports=java.rmi/sun.rmi.registry=ALL-UNNAMED --add-exports=jdk.compiler/com.sun.tools.javac.api=ALL-UNNAMED --add-exports=jdk.compiler/com.sun.tools.javac.file=ALL-UNNAMED --add-exports=jdk.compiler/com.sun.tools.javac.parser=ALL-UNNAMED --add-exports=jdk.compiler/com.sun.tools.javac.tree=ALL-UNNAMED --add-exports=jdk.compiler/com.sun.tools.javac.util=ALL-UNNAMED --add-exports=java.security.jgss/sun.security.krb5=ALL-UNNAMED --add-opens=java.base/java.lang=ALL-UNNAMED --add-opens=java.base/java.net=ALL-UNNAMED --add-opens=java.base/java.io=ALL-UNNAMED --add-opens=java.base/java.nio=ALL-UNNAMED --add-opens=java.base/sun.nio.ch=ALL-UNNAMED --add-opens=java.base/java.lang.reflect=ALL-UNNAMED --add-opens=java.base/java.text=ALL-UNNAMED --add-opens=java.base/java.time=ALL-UNNAMED --add-opens=java.base/java.util=ALL-UNNAMED --add-opens=java.base/java.util.concurrent=ALL-UNNAMED --add-opens=java.base/java.util.concurrent.atomic=ALL-UNNAMED --add-opens=java.base/java.util.concurrent.locks=ALL-UNNAMED"
+fi
+
+export JAVA_OPTS

docs/en/connector-v2/sink/Hive.md

@@ -10,7 +10,7 @@ Write data to Hive.
 
 In order to use this connector, You must ensure your spark/flink cluster already integrated hive. The tested hive version is 2.3.9.
 
-If you use SeaTunnel Engine, You need put seatunnel-hadoop3-3.1.4-uber.jar and hive-exec-3.1.3.jar and libfb303-0.9.3.jar in $SEATUNNEL_HOME/lib/ dir.
+If you use SeaTunnel Engine, You need put seatunnel-hadoop3-3.3.4-uber.jar and hive-exec-3.1.3.jar and libfb303-0.9.3.jar in $SEATUNNEL_HOME/lib/ dir.
 :::
 
 ## Key features

docs/en/connector-v2/sink/OssFile.md

@@ -17,7 +17,7 @@
 
 ### For SeaTunnel Zeta Engine
 
-1. You must ensure `seatunnel-hadoop3-3.1.4-uber.jar`, `aliyun-sdk-oss-3.4.1.jar`, `hadoop-aliyun-3.1.4.jar` and `jdom-1.1.jar` in `${SEATUNNEL_HOME}/lib/` dir.
+1. You must ensure `seatunnel-hadoop3-3.3.4-uber.jar`, `aliyun-sdk-oss-3.4.1.jar`, `hadoop-aliyun-3.1.4.jar` and `jdom-1.1.jar` in `${SEATUNNEL_HOME}/lib/` dir.
 
 ## Key features
 

docs/en/connector-v2/source/Hive.md

@@ -10,7 +10,7 @@ Read data from Hive.
 
 In order to use this connector, You must ensure your spark/flink cluster already integrated hive. The tested hive version is 2.3.9.
 
-If you use SeaTunnel Engine, You need put seatunnel-hadoop3-3.1.4-uber.jar and hive-exec-3.1.3.jar and libfb303-0.9.3.jar in $SEATUNNEL_HOME/lib/ dir.
+If you use SeaTunnel Engine, You need put seatunnel-hadoop3-3.3.4-uber.jar and hive-exec-3.1.3.jar and libfb303-0.9.3.jar in $SEATUNNEL_HOME/lib/ dir.
 :::
 
 ## Key features

docs/en/connector-v2/source/OssFile.md

@@ -17,7 +17,7 @@
 
 ### For SeaTunnel Zeta Engine
 
-1. You must ensure `seatunnel-hadoop3-3.1.4-uber.jar`, `aliyun-sdk-oss-3.4.1.jar`, `hadoop-aliyun-3.1.4.jar` and `jdom-1.1.jar` in `${SEATUNNEL_HOME}/lib/` dir.
+1. You must ensure `seatunnel-hadoop3-3.3.4-uber.jar`, `aliyun-sdk-oss-3.4.1.jar`, `hadoop-aliyun-3.1.4.jar` and `jdom-1.1.jar` in `${SEATUNNEL_HOME}/lib/` dir.
 
 ## Key features
 

pom.xml

@@ -65,7 +65,7 @@
         <maven.compiler.source>${java.version}</maven.compiler.source>
         <maven.compiler.target>${java.version}</maven.compiler.target>
 
-        <slf4j.version>1.7.25</slf4j.version>
+        <slf4j.version>1.7.36</slf4j.version>
         <log4j2.version>2.17.1</log4j2.version>
         <log4j2-disruptor.version>3.4.4</log4j2-disruptor.version>
         <log4j.version>1.2.17</log4j.version>
@@ -91,7 +91,7 @@
         <skip.pmd.check>false</skip.pmd.check>
         <maven.deploy.skip>false</maven.deploy.skip>
         <maven.javadoc.skip>false</maven.javadoc.skip>
-        <maven-surefire-plugin.version>2.22.2</maven-surefire-plugin.version>
+        <maven-surefire-plugin.version>3.3.1</maven-surefire-plugin.version>
         <maven-failsafe-plugin.version>2.22.2</maven-failsafe-plugin.version>
         <nexus-staging-maven-plugin.version>1.6.8</nexus-staging-maven-plugin.version>
         <maven-source-plugin.version>3.0.1</maven-source-plugin.version>
@@ -133,6 +133,8 @@
         <spotless.version>2.29.0</spotless.version>
         <jsqlparser.version>4.5</jsqlparser.version>
         <json-path.version>2.7.0</json-path.version>
+        <okhttp.version>4.12.0</okhttp.version>
+        <surefire.argLine />
         <!-- Option args -->
         <skipUT>false</skipUT>
         <skipIT>true</skipIT>
@@ -470,7 +472,7 @@
 
             <dependency>
                 <groupId>org.apache.seatunnel</groupId>
-                <artifactId>seatunnel-hadoop3-3.1.4-uber</artifactId>
+                <artifactId>seatunnel-hadoop3-3.3.4-uber</artifactId>
                 <version>${project.version}</version>
                 <classifier>optional</classifier>
                 <scope>provided</scope>
@@ -585,6 +587,7 @@
                             <classpathDependencyExclude>ch.qos.logback:logback-core</classpathDependencyExclude>
                             <classpathDependencyExclude>org.apache.logging.log4j:log4j-to-slf4j</classpathDependencyExclude>
                         </classpathDependencyExcludes>
+                        <argLine>${surefire.argLine}</argLine>
                     </configuration>
                 </plugin>
                 <plugin>
@@ -937,6 +940,52 @@
                 <activeByDefault>false</activeByDefault>
             </activation>
         </profile>
+        <profile>
+            <id>jdk8</id>
+            <activation>
+                <jdk>1.8</jdk>
+            </activation>
+            <properties>
+                <surefire.argLine />
+            </properties>
+        </profile>
+        <profile>
+            <id>jdk11</id>
+            <activation>
+                <jdk>11</jdk>
+            </activation>
+            <properties>
+                <surefire.argLine />
+            </properties>
+        </profile>
+        <profile>
+            <id>jdk17</id>
+            <activation>
+                <jdk>[17,)</jdk>
+            </activation>
+            <properties>
+                <surefire.argLine>--add-exports java.base/sun.net.util=ALL-UNNAMED
+                    --add-exports java.rmi/sun.rmi.registry=ALL-UNNAMED
+                    --add-exports jdk.compiler/com.sun.tools.javac.api=ALL-UNNAMED
+                    --add-exports jdk.compiler/com.sun.tools.javac.file=ALL-UNNAMED
+                    --add-exports jdk.compiler/com.sun.tools.javac.parser=ALL-UNNAMED
+                    --add-exports jdk.compiler/com.sun.tools.javac.tree=ALL-UNNAMED
+                    --add-exports jdk.compiler/com.sun.tools.javac.util=ALL-UNNAMED
+                    --add-exports java.security.jgss/sun.security.krb5=ALL-UNNAMED
+                    --add-opens java.base/java.lang=ALL-UNNAMED
+                    --add-opens java.base/java.net=ALL-UNNAMED
+                    --add-opens java.base/java.io=ALL-UNNAMED
+                    --add-opens java.base/java.nio=ALL-UNNAMED
+                    --add-opens java.base/sun.nio.ch=ALL-UNNAMED
+                    --add-opens java.base/java.lang.reflect=ALL-UNNAMED
+                    --add-opens java.base/java.text=ALL-UNNAMED
+                    --add-opens java.base/java.time=ALL-UNNAMED
+                    --add-opens java.base/java.util=ALL-UNNAMED
+                    --add-opens java.base/java.util.concurrent=ALL-UNNAMED
+                    --add-opens java.base/java.util.concurrent.atomic=ALL-UNNAMED
+                    --add-opens java.base/java.util.concurrent.locks=ALL-UNNAMED</surefire.argLine>
+            </properties>
+        </profile>
     </profiles>
 
 </project>

seatunnel-connectors-v2/connector-amazondynamodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/amazondynamodb/serialize/DefaultSeaTunnelRowDeserializer.java

@@ -23,8 +23,10 @@
 import org.apache.seatunnel.api.table.type.SeaTunnelRow;
 import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
 import org.apache.seatunnel.common.exception.CommonError;
+import org.apache.seatunnel.common.utils.DateTimeUtils;
+import org.apache.seatunnel.common.utils.DateUtils;
 
-import lombok.AllArgsConstructor;
+import lombok.RequiredArgsConstructor;
 import software.amazon.awssdk.core.SdkBytes;
 import software.amazon.awssdk.services.dynamodb.model.AttributeValue;
 
@@ -34,16 +36,29 @@
 import java.time.LocalDate;
 import java.time.LocalDateTime;
 import java.time.LocalTime;
+import java.time.format.DateTimeFormatter;
+import java.time.format.DateTimeFormatterBuilder;
+import java.time.temporal.ChronoField;
+import java.time.temporal.TemporalAccessor;
+import java.time.temporal.TemporalQueries;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 
-@AllArgsConstructor
+@RequiredArgsConstructor
 public class DefaultSeaTunnelRowDeserializer implements SeaTunnelRowDeserializer {
 
     private final SeaTunnelRowType typeInfo;
 
+    public static DateTimeFormatter TIME_FORMAT =
+            new DateTimeFormatterBuilder()
+                    .appendPattern("HH:mm:ss")
+                    .appendFraction(ChronoField.NANO_OF_SECOND, 0, 9, true)
+                    .toFormatter();
+
+    public Map<String, DateTimeFormatter> fieldFormatterMap = new HashMap<>();
+
     @Override
     public SeaTunnelRow deserialize(Map<String, AttributeValue> item) {
         SeaTunnelDataType<?>[] seaTunnelDataTypes = typeInfo.getFieldTypes();
@@ -63,7 +78,9 @@ private List<Object> convertRow(
     }
 
     private Object convert(
-            String field, SeaTunnelDataType<?> seaTunnelDataType, AttributeValue attributeValue) {
+            String fieldName,
+            SeaTunnelDataType<?> seaTunnelDataType,
+            AttributeValue attributeValue) {
         if (attributeValue.type().equals(AttributeValue.Type.NUL)) {
             return null;
         }
@@ -90,11 +107,26 @@ private Object convert(
             case STRING:
                 return attributeValue.s();
             case TIME:
-                return LocalTime.parse(attributeValue.s());
+                return TIME_FORMAT.parse(attributeValue.s());
             case DATE:
-                return LocalDate.parse(attributeValue.s());
+                DateTimeFormatter dateFormatter = fieldFormatterMap.get(fieldName);
+                if (dateFormatter == null) {
+                    dateFormatter = DateUtils.matchDateFormatter(attributeValue.s());
+                    fieldFormatterMap.put(fieldName, dateFormatter);
+                }
+
+                return dateFormatter.parse(attributeValue.s()).query(TemporalQueries.localDate());
             case TIMESTAMP:
-                return LocalDateTime.parse(attributeValue.s());
+                DateTimeFormatter dateTimeFormatter = fieldFormatterMap.get(fieldName);
+                if (dateTimeFormatter == null) {
+                    dateTimeFormatter = DateTimeUtils.matchDateTimeFormatter(attributeValue.s());
+                    fieldFormatterMap.put(fieldName, dateTimeFormatter);
+                }
+
+                TemporalAccessor parsedTimestamp = dateTimeFormatter.parse(attributeValue.s());
+                LocalTime localTime = parsedTimestamp.query(TemporalQueries.localTime());
+                LocalDate localDate = parsedTimestamp.query(TemporalQueries.localDate());
+                return LocalDateTime.of(localDate, localTime);
             case BYTES:
                 return attributeValue.b().asByteArray();
             case MAP:
@@ -106,7 +138,7 @@ private Object convert(
                                     seatunnelMap.put(
                                             s,
                                             convert(
-                                                    field,
+                                                    fieldName,
                                                     ((MapType) seaTunnelDataType).getValueType(),
                                                     attributeValueInfo));
                                 });
@@ -126,7 +158,7 @@ private Object convert(
                                 array,
                                 index,
                                 convert(
-                                        field,
+                                        fieldName,
                                         ((ArrayType<?, ?>) seaTunnelDataType).getElementType(),
                                         datas.get(index)));
                     }
@@ -149,7 +181,7 @@ private Object convert(
                 return array;
             default:
                 throw CommonError.convertToSeaTunnelTypeError(
-                        "AmazonDynamodb", seaTunnelDataType.getSqlType().toString(), field);
+                        "AmazonDynamodb", seaTunnelDataType.getSqlType().toString(), fieldName);
         }
     }
 }

seatunnel-connectors-v2/connector-amazondynamodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/amazondynamodb/serialize/DefaultSeaTunnelRowSerializer.java

@@ -22,13 +22,20 @@
 import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
 import org.apache.seatunnel.api.table.type.SeaTunnelRow;
 import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
+import org.apache.seatunnel.api.table.type.SqlType;
 import org.apache.seatunnel.common.exception.CommonError;
+import org.apache.seatunnel.common.utils.DateTimeUtils;
+import org.apache.seatunnel.common.utils.DateUtils;
+import org.apache.seatunnel.common.utils.TimeUtils;
 import org.apache.seatunnel.connectors.seatunnel.amazondynamodb.config.AmazonDynamoDBSourceOptions;
 
 import software.amazon.awssdk.core.SdkBytes;
 import software.amazon.awssdk.services.dynamodb.model.AttributeValue;
 import software.amazon.awssdk.services.dynamodb.model.PutItemRequest;
 
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.time.LocalTime;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.HashMap;
@@ -43,6 +50,12 @@ public class DefaultSeaTunnelRowSerializer implements SeaTunnelRowSerializer {
     private final AmazonDynamoDBSourceOptions amazondynamodbSourceOptions;
     private final List<AttributeValue.Type> measurementsType;
 
+    private final DateTimeUtils.Formatter dateTimeFormatter =
+            DateTimeUtils.Formatter.YYYY_MM_DD_HH_MM_SS_SSSSSS;
+    private final DateUtils.Formatter dateFormatter = DateUtils.Formatter.YYYY_MM_DD;
+
+    private final TimeUtils.Formatter timeFormatter = TimeUtils.Formatter.HH_MM_SS;
+
     public DefaultSeaTunnelRowSerializer(
             SeaTunnelRowType seaTunnelRowType,
             AmazonDynamoDBSourceOptions amazondynamodbSourceOptions) {
@@ -125,6 +138,19 @@ private AttributeValue convertItem(
                         .n(Integer.toString(((Number) value).intValue()))
                         .build();
             case S:
+                if (seaTunnelDataType.getSqlType().equals(SqlType.DATE)) {
+                    return AttributeValue.builder()
+                            .s(DateUtils.toString((LocalDate) value, dateFormatter))
+                            .build();
+                } else if (seaTunnelDataType.getSqlType().equals(SqlType.TIME)) {
+                    return AttributeValue.builder()
+                            .s(TimeUtils.toString((LocalTime) value, timeFormatter))
+                            .build();
+                } else if (seaTunnelDataType.getSqlType().equals(SqlType.TIMESTAMP)) {
+                    return AttributeValue.builder()
+                            .s(DateTimeUtils.toString((LocalDateTime) value, dateTimeFormatter))
+                            .build();
+                }
                 return AttributeValue.builder().s(String.valueOf(value)).build();
             case BOOL:
                 return AttributeValue.builder().bool((Boolean) value).build();