fix ci.

Author: lvyanquan

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/main/java/org/apache/flink/cdc/connectors/paimon/sink/v2/bucket/BucketAssignOperator.java

@@ -189,7 +189,9 @@ public void processElement(StreamRecord<Event> streamRecord) throws Exception {
             SchemaChangeEvent schemaChangeEvent = (SchemaChangeEvent) event;
             Schema schema =
                     SchemaUtils.applySchemaChangeEvent(
-                            schemaMaps.get(schemaChangeEvent.tableId()).getSchema(),
+                            Optional.ofNullable(schemaMaps.get(schemaChangeEvent.tableId()))
+                                    .map(TableSchemaInfo::getSchema)
+                                    .orElse(null),
                             schemaChangeEvent);
             schemaMaps.put(schemaChangeEvent.tableId(), new TableSchemaInfo(schema, zoneId));
             // Broadcast SchemachangeEvent.

flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/MySqlToPaimonE2eITCase.java

@@ -95,6 +95,9 @@ public void before() throws Exception {
                 MountableFile.forHostPath(
                         TestUtils.getResource(getPaimonSQLConnectorResourceName())),
                 sharedVolume.toString() + "/" + getPaimonSQLConnectorResourceName());
+        jobManager.copyFileToContainer(
+                MountableFile.forHostPath(TestUtils.getResource("flink-shade-hadoop.jar")),
+                sharedVolume.toString() + "/flink-shade-hadoop.jar");
     }
 
     @After
@@ -141,11 +144,11 @@ public void testSyncWholeDatabase() throws Exception {
                 database,
                 "products",
                 Arrays.asList(
-                        "101, One, Alice, 3.202, red, {\"key1\":\"value1\"}, null",
-                        "102, Two, Bob, 1.703, white, {\"key2\":\"value2\"}, null",
-                        "103, Three, Cecily, 4.105, red, {\"key3\":\"value3\"}, null",
-                        "104, Four, Derrida, 1.857, white, {\"key4\":\"value4\"}, null",
-                        "105, Five, Evelyn, 5.211, red, {\"K\":\"V\",\"k\":\"v\"}, null",
+                        "101, One, Alice, 3.202, red, {\"key1\": \"value1\"}, null",
+                        "102, Two, Bob, 1.703, white, {\"key2\": \"value2\"}, null",
+                        "103, Three, Cecily, 4.105, red, {\"key3\": \"value3\"}, null",
+                        "104, Four, Derrida, 1.857, white, {\"key4\": \"value4\"}, null",
+                        "105, Five, Evelyn, 5.211, red, {\"K\": \"V\", \"k\": \"v\"}, null",
                         "106, Six, Ferris, 9.813, null, null, null",
                         "107, Seven, Grace, 2.117, null, null, null",
                         "108, Eight, Hesse, 6.819, null, null, null",
@@ -194,10 +197,10 @@ public void testSyncWholeDatabase() throws Exception {
         List<String> recordsInSnapshotPhase =
                 new ArrayList<>(
                         Arrays.asList(
-                                "102, Two, Bob, 1.703, white, {\"key2\":\"value2\"}, null, null, null, null, null, null, null, null, null, null",
-                                "103, Three, Cecily, 4.105, red, {\"key3\":\"value3\"}, null, null, null, null, null, null, null, null, null, null",
-                                "104, Four, Derrida, 1.857, white, {\"key4\":\"value4\"}, null, null, null, null, null, null, null, null, null, null",
-                                "105, Five, Evelyn, 5.211, red, {\"K\":\"V\",\"k\":\"v\"}, null, null, null, null, null, null, null, null, null, null",
+                                "102, Two, Bob, 1.703, white, {\"key2\": \"value2\"}, null, null, null, null, null, null, null, null, null, null",
+                                "103, Three, Cecily, 4.105, red, {\"key3\": \"value3\"}, null, null, null, null, null, null, null, null, null, null",
+                                "104, Four, Derrida, 1.857, white, {\"key4\": \"value4\"}, null, null, null, null, null, null, null, null, null, null",
+                                "105, Five, Evelyn, 5.211, red, {\"K\": \"V\", \"k\": \"v\"}, null, null, null, null, null, null, null, null, null, null",
                                 "106, Six, Fay, 9.813, null, null, null, null, null, null, null, null, null, null, null, null",
                                 "107, Seven, Grace, 5.125, null, null, null, null, null, null, null, null, null, null, null, null",
                                 "108, Eight, Hesse, 6.819, null, null, null, null, null, null, null, null, null, null, null, null",
@@ -282,6 +285,8 @@ private List<String> fetchPaimonTableRows(String warehouse, String database, Str
                         "/opt/flink/bin/sql-client.sh",
                         "--jar",
                         sharedVolume.toString() + "/" + getPaimonSQLConnectorResourceName(),
+                        "--jar",
+                        sharedVolume.toString() + "/flink-shade-hadoop.jar",
                         "-f",
                         containerSqlPath);
         if (result.getExitCode() != 0) {

flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/utils/PipelineTestEnvironment.java

@@ -60,7 +60,6 @@
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.time.Duration;
-import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
@@ -142,14 +141,9 @@ public static List<String> getFlinkVersion() {
     public void before() throws Exception {
         LOG.info("Starting containers...");
         jobManagerConsumer = new ToStringConsumer();
-        List<String> cmds = new ArrayList<>();
-        for (String prop : EXTERNAL_PROPS) {
-            cmds.add(String.format("echo '%s' >> /opt/flink/conf/flink-conf.yaml.tmp", prop));
-        }
-        cmds.add("mv /opt/flink/conf/flink-conf.yaml.tmp /opt/flink/conf/flink-conf.yaml");
-        String preCmd = String.join(" && ", cmds);
         jobManager =
                 new GenericContainer<>(getFlinkDockerImageTag())
+                        .withCommand("jobmanager")
                         .withNetwork(NETWORK)
                         .withNetworkAliases(INTER_CONTAINER_JM_ALIAS)
                         .withExposedPorts(JOB_MANAGER_REST_PORT)
@@ -163,6 +157,7 @@ public void before() throws Exception {
         taskManagerConsumer = new ToStringConsumer();
         taskManager =
                 new GenericContainer<>(getFlinkDockerImageTag())
+                        .withCommand("taskmanager")
                         .withNetwork(NETWORK)
                         .withNetworkAliases(INTER_CONTAINER_TM_ALIAS)
                         .withEnv("FLINK_PROPERTIES", FLINK_PROPERTIES)