[FLINK-39252][sqlserver] fix: replace Java 9+ APIs with Java 8 compatible code in SQL Server connector

Author: Daishuyuan

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-sqlserver/src/main/java/org/apache/flink/cdc/connectors/sqlserver/factory/SqlServerDataSourceFactory.java

@@ -49,7 +49,6 @@
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
-import java.util.Objects;
 import java.util.Properties;
 import java.util.Set;
 import java.util.stream.Collectors;
@@ -274,7 +273,9 @@ public String identifier() {
 
     private static List<String> getTableList(
             @Nullable List<TableId> tableIdList, Selectors selectors) {
-        return Objects.requireNonNullElse(tableIdList, Collections.<TableId>emptyList()).stream()
+        List<TableId> resolvedTableIdList =
+                tableIdList == null ? Collections.emptyList() : tableIdList;
+        return resolvedTableIdList.stream()
                 .filter(selectors::isMatch)
                 // SQL Server tableList format: schemaName.tableName (without database prefix)
                 // See SqlServerSourceBuilder: "Each identifier is of the form

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-sqlserver/src/test/java/org/apache/flink/cdc/connectors/sqlserver/source/SqlServerOnlineSchemaMigrationITCase.java

@@ -53,7 +53,6 @@
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
-import java.util.Map;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
@@ -187,7 +186,7 @@ void testSchemaMigrationFromScratch() throws Exception {
                         new AlterColumnTypeEvent(
                                 tableId,
                                 Collections.singletonMap("ext", DataTypes.DOUBLE()),
-                                Map.of("ext", DataTypes.INT())),
+                                Collections.singletonMap("ext", DataTypes.INT())),
                         DataChangeEvent.insertEvent(
                                 tableId,
                                 generate(

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-sqlserver/src/test/java/org/apache/flink/cdc/connectors/sqlserver/source/SqlServerPipelineITCaseTest.java

@@ -438,7 +438,7 @@ private List<Event> getSnapshotExpected(TableId tableId) {
                                     108,
                                     BinaryStringData.fromString("jacket"),
                                     BinaryStringData.fromString(
-                                            "water resistent black wind breaker"),
+                                            "water resistant black wind breaker"),
                                     0.1d
                                 })));
         snapshotExpected.add(

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-sqlserver/src/test/java/org/apache/flink/cdc/connectors/sqlserver/source/SqlServerTablePatternMatchingTest.java

@@ -55,17 +55,11 @@ public void before() {
     private void initializePatternTestDatabase() {
         try (Connection connection = getJdbcConnection();
                 Statement statement = connection.createStatement()) {
-            // Drop database if exists
-            statement.execute(
-                    String.format(
-                            "IF EXISTS(select 1 from sys.databases where name = '%s') "
-                                    + "BEGIN ALTER DATABASE [%s] SET SINGLE_USER WITH ROLLBACK IMMEDIATE; "
-                                    + "DROP DATABASE [%s]; END",
-                            DATABASE_NAME, DATABASE_NAME, DATABASE_NAME));
+            dropTestDatabase(connection, DATABASE_NAME);
 
             // Create database
-            statement.execute(String.format("CREATE DATABASE %s;", DATABASE_NAME));
-            statement.execute(String.format("USE %s;", DATABASE_NAME));
+            statement.execute(String.format("CREATE DATABASE [%s];", DATABASE_NAME));
+            statement.execute(String.format("USE [%s];", DATABASE_NAME));
 
             // Wait for SQL Server Agent
             statement.execute("WAITFOR DELAY '00:00:03';");

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-sqlserver-cdc/src/test/java/org/apache/flink/cdc/connectors/sqlserver/SqlServerTestBase.java

@@ -33,9 +33,10 @@
 import org.testcontainers.containers.output.Slf4jLogConsumer;
 import org.testcontainers.lifecycle.Startables;
 
-import java.net.URL;
-import java.nio.file.Files;
-import java.nio.file.Paths;
+import java.io.BufferedReader;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.nio.charset.StandardCharsets;
 import java.sql.Connection;
 import java.sql.DriverManager;
 import java.sql.SQLException;
@@ -92,7 +93,7 @@ protected Connection getJdbcConnection() throws SQLException {
                 MSSQL_SERVER_CONTAINER.getPassword());
     }
 
-    private static void dropTestDatabase(Connection connection, String databaseName)
+    protected static void dropTestDatabase(Connection connection, String databaseName)
             throws SQLException {
         try {
             Awaitility.await("Disabling CDC")
@@ -175,14 +176,19 @@ protected static void disableDbCdc(Connection connection, String name) throws SQ
      */
     protected void initializeSqlServerTable(String sqlFile) {
         final String ddlFile = String.format("ddl/%s.sql", sqlFile);
-        final URL ddlTestFile = SqlServerTestBase.class.getClassLoader().getResource(ddlFile);
+        final InputStream ddlTestFile =
+                SqlServerTestBase.class.getClassLoader().getResourceAsStream(ddlFile);
         Assertions.assertThat(ddlTestFile).withFailMessage("Cannot locate " + ddlFile).isNotNull();
-        try (Connection connection = getJdbcConnection();
+        try (InputStream inputStream = ddlTestFile;
+                BufferedReader reader =
+                        new BufferedReader(
+                                new InputStreamReader(inputStream, StandardCharsets.UTF_8));
+                Connection connection = getJdbcConnection();
                 Statement statement = connection.createStatement()) {
             dropTestDatabase(connection, sqlFile);
             final List<String> statements =
                     Arrays.stream(
-                                    Files.readAllLines(Paths.get(ddlTestFile.toURI())).stream()
+                                    reader.lines()
                                             .map(String::trim)
                                             .filter(x -> !x.startsWith("--") && !x.isEmpty())
                                             .map(