You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by xt...@apache.org on 2025/12/17 03:33:02 UTC
(flink-agents) branch release-0.1 updated: [runtime] Remove userCodeClassLoader field for ActionTask to fix can't be serialized to state. (#379)
This is an automated email from the ASF dual-hosted git repository.
xtsong pushed a commit to branch release-0.1
in repository https://gitbox.apache.org/repos/asf/flink-agents.git
The following commit(s) were added to refs/heads/release-0.1 by this push:
new 2a7c2cb [runtime] Remove userCodeClassLoader field for ActionTask to fix can't be serialized to state. (#379)
2a7c2cb is described below
commit 2a7c2cb55fcd961655966cbe6b2014469ba31082
Author: Wenjin Xie <16...@users.noreply.github.com>
AuthorDate: Wed Dec 17 11:29:33 2025 +0800
[runtime] Remove userCodeClassLoader field for ActionTask to fix can't be serialized to state. (#379)
---
.../flink/agents/runtime/operator/ActionExecutionOperator.java | 6 +++---
.../java/org/apache/flink/agents/runtime/operator/ActionTask.java | 2 +-
.../org/apache/flink/agents/runtime/operator/JavaActionTask.java | 8 ++------
.../flink/agents/runtime/python/operator/PythonActionTask.java | 4 ++--
.../agents/runtime/python/operator/PythonGeneratorActionTask.java | 2 +-
.../agents/runtime/{RescalingITCase.java => RescalingTest.java} | 4 ++--
6 files changed, 11 insertions(+), 15 deletions(-)
diff --git a/runtime/src/main/java/org/apache/flink/agents/runtime/operator/ActionExecutionOperator.java b/runtime/src/main/java/org/apache/flink/agents/runtime/operator/ActionExecutionOperator.java
index 95b991b..d00e8d8 100644
--- a/runtime/src/main/java/org/apache/flink/agents/runtime/operator/ActionExecutionOperator.java
+++ b/runtime/src/main/java/org/apache/flink/agents/runtime/operator/ActionExecutionOperator.java
@@ -405,7 +405,8 @@ public class ActionExecutionOperator<IN, OUT> extends AbstractStreamOperator<OUT
}
} else {
maybeInitActionState(key, sequenceNumber, actionTask.action, actionTask.event);
- ActionTask.ActionTaskResult actionTaskResult = actionTask.invoke();
+ ActionTask.ActionTaskResult actionTaskResult =
+ actionTask.invoke(getRuntimeContext().getUserCodeClassLoader());
// We remove the RunnerContext of the action task from the map after it is finished. The
// RunnerContext will be added later if the action task has a generated action task,
@@ -638,8 +639,7 @@ public class ActionExecutionOperator<IN, OUT> extends AbstractStreamOperator<OUT
private ActionTask createActionTask(Object key, Action action, Event event) {
if (action.getExec() instanceof JavaFunction) {
- return new JavaActionTask(
- key, event, action, getRuntimeContext().getUserCodeClassLoader());
+ return new JavaActionTask(key, event, action);
} else if (action.getExec() instanceof PythonFunction) {
return new PythonActionTask(key, event, action);
} else {
diff --git a/runtime/src/main/java/org/apache/flink/agents/runtime/operator/ActionTask.java b/runtime/src/main/java/org/apache/flink/agents/runtime/operator/ActionTask.java
index d8a30b3..34f7850 100644
--- a/runtime/src/main/java/org/apache/flink/agents/runtime/operator/ActionTask.java
+++ b/runtime/src/main/java/org/apache/flink/agents/runtime/operator/ActionTask.java
@@ -87,7 +87,7 @@ public abstract class ActionTask {
}
/** Invokes the action task. */
- public abstract ActionTaskResult invoke() throws Exception;
+ public abstract ActionTaskResult invoke(ClassLoader userCodeClassLoader) throws Exception;
public class ActionTaskResult {
private final boolean finished;
diff --git a/runtime/src/main/java/org/apache/flink/agents/runtime/operator/JavaActionTask.java b/runtime/src/main/java/org/apache/flink/agents/runtime/operator/JavaActionTask.java
index 5898758..9fc641b 100644
--- a/runtime/src/main/java/org/apache/flink/agents/runtime/operator/JavaActionTask.java
+++ b/runtime/src/main/java/org/apache/flink/agents/runtime/operator/JavaActionTask.java
@@ -30,17 +30,13 @@ import static org.apache.flink.util.Preconditions.checkState;
* action task will be invoked only once.
*/
public class JavaActionTask extends ActionTask {
-
- private final ClassLoader userCodeClassLoader;
-
- public JavaActionTask(Object key, Event event, Action action, ClassLoader userCodeClassLoader) {
+ public JavaActionTask(Object key, Event event, Action action) {
super(key, event, action);
checkState(action.getExec() instanceof JavaFunction);
- this.userCodeClassLoader = userCodeClassLoader;
}
@Override
- public ActionTaskResult invoke() throws Exception {
+ public ActionTaskResult invoke(ClassLoader userCodeClassLoader) throws Exception {
LOG.debug(
"Try execute java action {} for event {} with key {}.",
action.getName(),
diff --git a/runtime/src/main/java/org/apache/flink/agents/runtime/python/operator/PythonActionTask.java b/runtime/src/main/java/org/apache/flink/agents/runtime/python/operator/PythonActionTask.java
index c3e1f84..89b39c1 100644
--- a/runtime/src/main/java/org/apache/flink/agents/runtime/python/operator/PythonActionTask.java
+++ b/runtime/src/main/java/org/apache/flink/agents/runtime/python/operator/PythonActionTask.java
@@ -43,7 +43,7 @@ public class PythonActionTask extends ActionTask {
"Python action only accept python event, but got " + event);
}
- public ActionTaskResult invoke() throws Exception {
+ public ActionTaskResult invoke(ClassLoader userCodeClassLoader) throws Exception {
LOG.debug(
"Try execute python action {} for event {} with key {}.",
action.getName(),
@@ -64,7 +64,7 @@ public class PythonActionTask extends ActionTask {
ActionTask tempGeneratedActionTask =
new PythonGeneratorActionTask(key, event, action, pythonGeneratorRef);
tempGeneratedActionTask.setRunnerContext(runnerContext);
- return tempGeneratedActionTask.invoke();
+ return tempGeneratedActionTask.invoke(userCodeClassLoader);
}
return new ActionTaskResult(
true, runnerContext.drainEvents(event.getSourceTimestamp()), null);
diff --git a/runtime/src/main/java/org/apache/flink/agents/runtime/python/operator/PythonGeneratorActionTask.java b/runtime/src/main/java/org/apache/flink/agents/runtime/python/operator/PythonGeneratorActionTask.java
index 4119fcd..96afa19 100644
--- a/runtime/src/main/java/org/apache/flink/agents/runtime/python/operator/PythonGeneratorActionTask.java
+++ b/runtime/src/main/java/org/apache/flink/agents/runtime/python/operator/PythonGeneratorActionTask.java
@@ -32,7 +32,7 @@ public class PythonGeneratorActionTask extends PythonActionTask {
}
@Override
- public ActionTaskResult invoke() {
+ public ActionTaskResult invoke(ClassLoader userCodeClassLoader) {
LOG.debug(
"Try execute python generator action {} for event {} with key {}.",
action.getName(),
diff --git a/runtime/src/test/java/org/apache/flink/agents/runtime/RescalingITCase.java b/runtime/src/test/java/org/apache/flink/agents/runtime/RescalingTest.java
similarity index 99%
rename from runtime/src/test/java/org/apache/flink/agents/runtime/RescalingITCase.java
rename to runtime/src/test/java/org/apache/flink/agents/runtime/RescalingTest.java
index cb2e255..dca3ed2 100644
--- a/runtime/src/test/java/org/apache/flink/agents/runtime/RescalingITCase.java
+++ b/runtime/src/test/java/org/apache/flink/agents/runtime/RescalingTest.java
@@ -89,7 +89,7 @@ import static org.junit.Assert.assertTrue;
/** This test case is derived from an existing test in Flink. */
@RunWith(Parameterized.class)
-public class RescalingITCase extends TestLogger {
+public class RescalingTest extends TestLogger {
@ClassRule
public static final TestExecutorResource<ScheduledExecutorService> EXECUTOR_RESOURCE =
@@ -104,7 +104,7 @@ public class RescalingITCase extends TestLogger {
return Arrays.asList(new Object[][] {{"hashmap"}, {"rocksdb"}});
}
- public RescalingITCase(String backend) {
+ public RescalingTest(String backend) {
this.backend = backend;
}