fix ci

Author: yuxiqian

flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/regular/SchemaCoordinator.java

@@ -336,8 +336,8 @@ private void applySchemaChange(int sourceSubTaskId) {
                             LOG.info(
                                     "SchemaOperator {} has not submitted schema change request yet. Waiting...",
                                     sourceSubTaskId),
-                    Duration.ofMillis(100),
-                    rpcTimeout);
+                    rpcTimeout,
+                    Duration.ofMillis(100));
         } catch (TimeoutException e) {
             throw new RuntimeException(
                     "Timeout waiting for schema change request from SchemaOperator.", e);

flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/operators/sink/DataSinkOperatorWithSchemaEvolveTest.java

@@ -87,16 +87,12 @@ private void processSchemaChangeEvent(
         // Create the flush event to process before the schema change event
         FlushEvent flushEvent = createFlushEvent(tableId, event);
 
-        // Send schema change request to coordinator
-        schemaOperatorHarness.requestSchemaChangeEvent(tableId, event);
-
         // Send flush event to SinkWriterOperator
         dataSinkWriterOperator.processElement(new StreamRecord<>(flushEvent));
 
-        // Wait for coordinator to complete the schema change and get the finished schema change
-        // events
+        // Send schema change request to coordinator
         SchemaChangeResponse schemaEvolveResponse =
-                schemaOperatorHarness.requestSchemaChangeResult(tableId, event);
+                schemaOperatorHarness.requestSchemaChangeEvent(tableId, event);
         List<SchemaChangeEvent> finishedSchemaChangeEvents =
                 schemaEvolveResponse.getAppliedSchemaChangeEvents();
 

flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/testutils/operators/RegularEventOperatorTestHarness.java

@@ -23,6 +23,7 @@
 import org.apache.flink.cdc.common.event.SchemaChangeEventType;
 import org.apache.flink.cdc.common.event.SchemaChangeEventTypeFamily;
 import org.apache.flink.cdc.common.event.TableId;
+import org.apache.flink.cdc.common.pipeline.PipelineOptions;
 import org.apache.flink.cdc.common.pipeline.SchemaChangeBehavior;
 import org.apache.flink.cdc.common.schema.Schema;
 import org.apache.flink.cdc.runtime.operators.schema.common.CoordinationResponseUtils;
@@ -91,6 +92,9 @@ public class RegularEventOperatorTestHarness<OP extends AbstractStreamOperator<E
 
     public static final OperatorID SINK_OPERATOR_ID = new OperatorID(15214L, 15514L);
 
+    private static final Duration DEFAULT_RPC_TIMEOUT =
+            PipelineOptions.DEFAULT_SCHEMA_OPERATOR_RPC_TIMEOUT;
+
     private final OP operator;
     private final int numOutputs;
     private final SchemaCoordinator schemaRegistry;
@@ -130,7 +134,7 @@ RegularEventOperatorTestHarness<OP, E> with(OP operator, int numOutputs) {
                 operator,
                 numOutputs,
                 null,
-                null,
+                DEFAULT_RPC_TIMEOUT,
                 SchemaChangeBehavior.EVOLVE,
                 Arrays.stream(SchemaChangeEventTypeFamily.ALL).collect(Collectors.toSet()),
                 Collections.emptySet());
@@ -143,7 +147,7 @@ RegularEventOperatorTestHarness<OP, E> withDuration(
                 operator,
                 numOutputs,
                 evolveDuration,
-                null,
+                DEFAULT_RPC_TIMEOUT,
                 SchemaChangeBehavior.EVOLVE,
                 Arrays.stream(SchemaChangeEventTypeFamily.ALL).collect(Collectors.toSet()),
                 Collections.emptySet());
@@ -159,7 +163,7 @@ RegularEventOperatorTestHarness<OP, E> withDurationAndBehavior(
                 operator,
                 numOutputs,
                 evolveDuration,
-                null,
+                DEFAULT_RPC_TIMEOUT,
                 behavior,
                 Arrays.stream(SchemaChangeEventTypeFamily.ALL).collect(Collectors.toSet()),
                 Collections.emptySet());
@@ -176,7 +180,7 @@ RegularEventOperatorTestHarness<OP, E> withDurationAndFineGrainedBehavior(
                 operator,
                 numOutputs,
                 evolveDuration,
-                null,
+                DEFAULT_RPC_TIMEOUT,
                 behavior,
                 enabledEventTypes,
                 Collections.emptySet());
@@ -195,7 +199,7 @@ RegularEventOperatorTestHarness<OP, E> withDurationAndFineGrainedBehaviorWithErr
                 operator,
                 numOutputs,
                 evolveDuration,
-                null,
+                DEFAULT_RPC_TIMEOUT,
                 behavior,
                 enabledEventTypes,
                 errorOnEventTypes);