This is an automated email from the ASF dual-hosted git repository.
yuxiqian pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink-cdc.git
The following commit(s) were added to refs/heads/master by this push:
new b54be00c3 [hotfix][runtime/python] Fix Python signature parsing
concurrency issue (#4552)
b54be00c3 is described below
commit b54be00c3c593287f17985b914ea9ae02e467429
Author: yuxiqian <[email protected]>
AuthorDate: Mon Sep 21 17:13:46 2026 +0800
[hotfix][runtime/python] Fix Python signature parsing concurrency issue
(#4552)
---
.../pipeline/tests/stage2/PythonUdfE2eITCase.java | 93 ++++++++++++++++++++++
.../flink/cdc/python/utils/PythonUdfSignature.java | 12 ++-
.../cdc/python/utils/PythonUdfSignatureTest.java | 54 +++++++++++++
3 files changed, 156 insertions(+), 3 deletions(-)
diff --git
a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage2/PythonUdfE2eITCase.java
b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage2/PythonUdfE2eITCase.java
index 873d518cc..e1853f034 100644
---
a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage2/PythonUdfE2eITCase.java
+++
b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage2/PythonUdfE2eITCase.java
@@ -37,6 +37,7 @@ import javax.annotation.Nullable;
import java.time.Duration;
import java.util.Arrays;
import java.util.List;
+import java.util.concurrent.TimeoutException;
import java.util.function.Function;
import java.util.stream.Collectors;
@@ -106,6 +107,98 @@ class PythonUdfE2eITCase extends PipelineTestEnvironment {
"DataChangeEvent{tableId=%s.USERS, before=[], after=[4,
[email protected], 84], op=INSERT, meta=()}");
}
+ @Test
+ void testConcurrentLargeSourceSignatureParsing() throws Exception {
+ submitPipelineJob(buildSignaturePipelineJob());
+ waitUntilJobRunning(Duration.ofMinutes(5));
+
+ String tableId = pythonUdfTestDatabase.getDatabaseName() + ".USERS";
+ waitUntilEventOrTaskManagerStops(
+ "CreateTableEvent{tableId="
+ + tableId
+ + ", schema=columns={`ID` INT NOT NULL,`RESULT_ONE`
STRING,"
+ + "`RESULT_TWO` STRING,`RESULT_THREE` STRING},
primaryKeys=ID, options=()}");
+ waitUntilEventOrTaskManagerStops(
+ "DataChangeEvent{tableId="
+ + tableId
+ + ", before=[], after=[1, [email protected],
[email protected], "
+ + "[email protected]], op=INSERT, meta=()}");
+ }
+
+ private void waitUntilEventOrTaskManagerStops(String event) throws
Exception {
+ long deadline = System.currentTimeMillis() +
EVENT_WAITING_TIMEOUT.toMillis();
+ while (System.currentTimeMillis() < deadline) {
+ String taskManagerLog = taskManagerConsumer.toUtf8String();
+ if (taskManagerLog.contains(event)) {
+ return;
+ }
+ if (!taskManager.isRunning()) {
+ throw new AssertionError(
+ "TaskManager exited while waiting for event: "
+ + event
+ + "\nTaskManager log:\n"
+ + taskManagerLog);
+ }
+ Thread.sleep(1000);
+ }
+ throw new TimeoutException("Timed out waiting for event: " + event);
+ }
+
+ private String buildSignaturePipelineJob() {
+ StringBuilder padding = new StringBuilder();
+ for (int i = 0; i < 8192; i++) {
+ padding.append(" None\n");
+ }
+ return String.format(
+ "source:\n"
+ + " type: mysql\n"
+ + " hostname: %s\n"
+ + " port: 3306\n"
+ + " username: %s\n"
+ + " password: %s\n"
+ + " scan.startup.mode: earliest-offset\n"
+ + " tables: %s.USERS\n"
+ + " server-id: 5600-5610\n"
+ + " server-time-zone: UTC\n"
+ + "\n"
+ + "sink:\n"
+ + " type: values\n"
+ + "\n"
+ + "transform:\n"
+ + " - source-table: %s.USERS\n"
+ + " projection: ID, first_identity(EMAIL) AS
RESULT_ONE, "
+ + "second_identity(EMAIL) AS RESULT_TWO, "
+ + "third_identity(EMAIL) AS RESULT_THREE\n"
+ + " filter: ID = 1\n"
+ + "\n"
+ + "pipeline:\n"
+ + " parallelism: 4\n"
+ + " transform.async-execution.enabled: false\n"
+ + " user-defined-function:\n"
+ + "%s"
+ + "%s"
+ + "%s",
+ INTER_CONTAINER_MYSQL_ALIAS,
+ MYSQL_TEST_USER,
+ MYSQL_TEST_PASSWORD,
+ pythonUdfTestDatabase.getDatabaseName(),
+ pythonUdfTestDatabase.getDatabaseName(),
+ signatureUdf("first_identity", padding.toString()),
+ signatureUdf("second_identity", padding.toString()),
+ signatureUdf("third_identity", padding.toString()));
+ }
+
+ private static String signatureUdf(String name, String padding) {
+ return String.format(
+ " - name: %s\n"
+ + " python-code: |\n"
+ + "%s"
+ + " def eval(value: str) -> str:\n"
+ + " return value\n"
+ + " python-executable: %s\n",
+ name, padding, CONTAINER_PYTHON_EXECUTABLE);
+ }
+
private String buildJobYaml(
String projection, @Nullable String filter, List<UdfEntry>
userDefinedFunctions) {
StringBuilder transform =
diff --git
a/flink-cdc-python/src/main/java/org/apache/flink/cdc/python/utils/PythonUdfSignature.java
b/flink-cdc-python/src/main/java/org/apache/flink/cdc/python/utils/PythonUdfSignature.java
index a9a3976aa..ce5de219c 100644
---
a/flink-cdc-python/src/main/java/org/apache/flink/cdc/python/utils/PythonUdfSignature.java
+++
b/flink-cdc-python/src/main/java/org/apache/flink/cdc/python/utils/PythonUdfSignature.java
@@ -61,14 +61,20 @@ public final class PythonUdfSignature {
}
private static String parseReturnAnnotation(String source, String
pythonExec) {
- PythonInterpreterConfig pemjaConfig =
-
PythonInterpreterConfig.newBuilder().setPythonExec(pythonExec).build();
- try (PythonInterpreter parser = new PythonInterpreter(pemjaConfig)) {
+ try (PythonInterpreter parser =
+ new PythonInterpreter(createInterpreterConfig(pythonExec))) {
parser.exec(loadSignatureScript());
return (String) parser.invoke("eval_return_type", source);
}
}
+ static PythonInterpreterConfig createInterpreterConfig(String pythonExec) {
+ return PythonInterpreterConfig.newBuilder()
+ .setPythonExec(pythonExec)
+ .setExcType(PythonInterpreterConfig.ExecType.SUB_INTERPRETER)
+ .build();
+ }
+
private static String loadSignatureScript() {
try (InputStream in =
PythonUdfSignature.class.getResourceAsStream(SIGNATURE_PARSER_RESOURCE)) {
diff --git
a/flink-cdc-python/src/test/java/org/apache/flink/cdc/python/utils/PythonUdfSignatureTest.java
b/flink-cdc-python/src/test/java/org/apache/flink/cdc/python/utils/PythonUdfSignatureTest.java
index 090f8e2ac..bec97c7db 100644
---
a/flink-cdc-python/src/test/java/org/apache/flink/cdc/python/utils/PythonUdfSignatureTest.java
+++
b/flink-cdc-python/src/test/java/org/apache/flink/cdc/python/utils/PythonUdfSignatureTest.java
@@ -21,6 +21,15 @@ import org.apache.flink.cdc.common.types.DataTypes;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
+import pemja.core.PythonInterpreterConfig;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.CyclicBarrier;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
@@ -33,6 +42,51 @@ class PythonUdfSignatureTest {
PemjaTestSupport.requirePemja();
}
+ @Test
+ void usesSubInterpreterForSignatureParsing() {
+ assertThat(
+
PythonUdfSignature.createInterpreterConfig(PemjaTestSupport.PYTHON_EXEC)
+ .getExecType())
+ .isEqualTo(PythonInterpreterConfig.ExecType.SUB_INTERPRETER);
+ }
+
+ @Test
+ void resolvesLargeSourcesConcurrently() throws Exception {
+ int threadCount = 2;
+ int rounds = 8;
+ String source =
+ "PADDING = '"
+ + "x".repeat(256 * 1024)
+ + "'\n"
+ + "def eval(value: bytes) -> bytes:\n"
+ + " return value\n";
+ ExecutorService executor = Executors.newFixedThreadPool(threadCount);
+ CyclicBarrier barrier = new CyclicBarrier(threadCount);
+ List<Future<?>> futures = new ArrayList<>();
+ try {
+ for (int thread = 0; thread < threadCount; thread++) {
+ futures.add(
+ executor.submit(
+ () -> {
+ barrier.await();
+ for (int round = 0; round < rounds;
round++) {
+ assertThat(
+
PythonUdfSignature.parseReturnType(
+ source,
+
PemjaTestSupport.PYTHON_EXEC))
+ .isEqualTo(DataTypes.BYTES());
+ }
+ return null;
+ }));
+ }
+ for (Future<?> future : futures) {
+ future.get(30, TimeUnit.SECONDS);
+ }
+ } finally {
+ executor.shutdownNow();
+ }
+ }
+
@Test
void resolvesEvenWhenOtherFunctionsArePresent() {
String source =
