Skip to content

[Tech Debt] Avoid using reflection to load implementation classes of resources during the compilation phase #879

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
wenjin272 merged 6 commits into from
Jul 13, 2026
Merged

[Tech Debt] Avoid using reflection to load implementation classes of resources during the compilation phase #879

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
f80c35f
refactor(plan): refactor resource provider creation logic to support …
Jul 6, 2026
091f616
fix wrong commit
Jul 7, 2026
7592456
fix python resource descriptor
Jul 10, 2026
d481836
fix java resource descriptor
Jul 10, 2026
0e5d616
fix CR
Jul 10, 2026
c74a420
fix CR
Jul 11, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 6 additions & 3 deletions api/src/main/java/org/apache/flink/agents/api/resource/ResourceDescriptor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -51,15 +51,18 @@ public class ResourceDescriptor {
* <ul>
* <li><b>For Java resources:</b> The fully qualified Java class name (e.g.,
* "com.example.YourJavaClass"). The {@code module} parameter should be empty or null.
* <li><b>For Python resources (when declaring from Java):</b> The Python class name
* (simple name, not module path, e.g., "YourPythonClass"). The Python module path
* must be specified in the {@code module} parameter (e.g., "your_module.submodule").
* </ul>
*
* @param module The Python module path for cross-platform compatibility. Defaults to empty
* string for Java resources. Example: "your_module.submodule"
* @param initialArguments Additional arguments for resource initialization. Can be null or
* empty map if no initial arguments are needed.
* <ul>
* <li><b>For Python resources (when declaring from Java): put the fully-qualified Python
* class under the pythonClazz key in initialArguments (e.g. ("pythonClazz",
* "your_module.submodule.YourPythonClass")). The clazz argument is unused in this
* case.
* </ul>
*/
@JsonCreator
public ResourceDescriptor(
Expand Down
50 changes: 25 additions & 25 deletions plan/src/main/java/org/apache/flink/agents/plan/AgentPlan.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,7 +35,6 @@
import org.apache.flink.agents.api.resource.ResourceDescriptor;
import org.apache.flink.agents.api.resource.ResourceType;
import org.apache.flink.agents.api.resource.SerializableResource;
import org.apache.flink.agents.api.resource.python.PythonResourceWrapper;
import org.apache.flink.agents.api.skills.SkillSourceSpec;
import org.apache.flink.agents.api.skills.Skills;
import org.apache.flink.agents.api.tools.ToolMetadata;
Expand Down Expand Up @@ -316,31 +315,40 @@ private static ResourceDescriptor requireResourceDescriptor(
}

private void extractResource(ResourceType type, Method method) throws Exception {
extractResource(type, method, null);
extractResource(type, method, null, false);
}

private void extractResource(
ResourceType type,
Method method,
Function<ResourceDescriptor, ResourceDescriptor> descriptorDecorator)
Function<ResourceDescriptor, ResourceDescriptor> descriptorDecorator,
boolean isPython)
throws Exception {
String name = method.getName();
ResourceProvider provider;
ResourceDescriptor descriptor = (ResourceDescriptor) method.invoke(null);

descriptor =
descriptorDecorator != null ? descriptorDecorator.apply(descriptor) : descriptor;

if (PythonResourceWrapper.class.isAssignableFrom(
Class.forName(
descriptor.getClazz(),
true,
Thread.currentThread().getContextClassLoader()))) {
provider = new PythonResourceProvider(name, type, descriptor);
} else {
provider = new JavaResourceProvider(name, type, descriptor);
addResourceProvider(createDescriptorResourceProvider(name, type, descriptor, isPython));
}

private ResourceProvider createDescriptorResourceProvider(
String name, ResourceType type, ResourceDescriptor descriptor) {
return createDescriptorResourceProvider(name, type, descriptor, false);
}

private ResourceProvider createDescriptorResourceProvider(
String name, ResourceType type, ResourceDescriptor descriptor, boolean isPython) {
if (isPython || isPythonResource(descriptor)) {
return new PythonResourceProvider(name, type, descriptor);
}
addResourceProvider(provider);
return new JavaResourceProvider(name, type, descriptor);
}

private boolean isPythonResource(ResourceDescriptor descriptor) {
String pythonClazz = descriptor.getArgument("pythonClazz");

@weiqingy weiqingy Jul 7, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Non-blocking:

The old check identified Python resources by whether clazz was a PythonResourceWrapper, while this now uses the presence of pythonClazz. Since PythonResourceProvider.provide() does not require pythonClazz when module is non-empty, do we want to document the assumption that descriptors reaching this classification path will always carry pythonClazz when they should be treated as Python?

@da-daken da-daken Jul 8, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks for your review! I think we should align on a single rule here. Since cross-language Python resources are required to declare pythonClazz, we probably should not leave other entry points open. Today, descriptors that take those alternate paths (e.g. non-empty module without pythonClazz) are classified as Java-side resources at plan-compilation time and fail at runtime. Tightening or closing those paths feels like a separate follow-up issue rather than something we need to solve in this PR—what do you think?

@weiqingy weiqingy Jul 8, 2026 •
edited
Loading

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Agreed. Making pythonClazz the single rule and closing the alternate paths is cleaner than documenting the assumption, and it belongs in its own change. Nit: those descriptors classify as Java at compile time and only fail at runtime, so sharpening that into a clearer compile-time error would be a good tracker. Otherwise LGTM.

@wenjin272 wenjin272 Jul 10, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Given that normal Java AgentPlan paths don’t appear to produce module != empty && pythonClazz missing (module always be ""), I don’t think we need to preserve that old descriptor shape here. It seems limited to manually constructed 3-arg descriptors, so documenting/tightening that contract separately should be enough.

return pythonClazz != null && !pythonClazz.isEmpty();
}

private void extractTool(Method method) throws Exception {
Expand Down Expand Up @@ -525,7 +533,8 @@ private void extractResourceProvidersFromAgent(Agent agent) throws Exception {
new ResourceDescriptor(
desc.getModule(),
PythonMCPServer.class.getName(),
new HashMap<>(desc.getInitialArguments())));
new HashMap<>(desc.getInitialArguments())),
true);
}
}
}
Expand Down Expand Up @@ -576,17 +585,8 @@ private void extractResourceProvidersFromAgent(Agent agent) throws Exception {
for (Map.Entry<String, Object> kv : entry.getValue().entrySet()) {
ResourceDescriptor descriptor =
requireResourceDescriptor(kv.getKey(), type, kv.getValue());
ResourceProvider provider;
if (PythonResourceWrapper.class.isAssignableFrom(
Class.forName(
descriptor.getClazz(),
true,
Thread.currentThread().getContextClassLoader()))) {
provider = new PythonResourceProvider(kv.getKey(), type, descriptor);
} else {
provider = new JavaResourceProvider(kv.getKey(), type, descriptor);
}
addResourceProvider(provider);
addResourceProvider(
createDescriptorResourceProvider(kv.getKey(), type, descriptor));
}
}
}
Expand Down
64 changes: 64 additions & 0 deletions plan/src/test/java/org/apache/flink/agents/plan/AgentPlanTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,11 +24,13 @@
import org.apache.flink.agents.api.OutputEvent;
import org.apache.flink.agents.api.agents.Agent;
import org.apache.flink.agents.api.annotation.ChatModelSetup;
import org.apache.flink.agents.api.annotation.MCPServer;
import org.apache.flink.agents.api.annotation.Tool;
import org.apache.flink.agents.api.context.RunnerContext;
import org.apache.flink.agents.api.resource.Resource;
import org.apache.flink.agents.api.resource.ResourceContext;
import org.apache.flink.agents.api.resource.ResourceDescriptor;
import org.apache.flink.agents.api.resource.ResourceName;
import org.apache.flink.agents.api.resource.ResourceType;
import org.apache.flink.agents.api.resource.SerializableResource;
import org.apache.flink.agents.api.resource.python.PythonResourceAdapter;
Expand Down Expand Up @@ -505,6 +507,25 @@ public void testAddResourceRegistersEmbeddingModelProvider() throws Exception {
assertThat(provider.getType()).isEqualTo(ResourceType.EMBEDDING_MODEL);
}

@Test
public void testAddResourceDescriptorWithPythonClazzUsesPythonResourceProvider()
throws Exception {
Agent agent = new Agent();
ResourceDescriptor descriptor =
ResourceDescriptor.Builder.newBuilder(String.class.getName())
.addInitialArgument("pythonClazz", "test.module.EmbeddingClazz")
.build();
agent.addResource("myEmbedding", ResourceType.EMBEDDING_MODEL, descriptor);

AgentPlan plan = new AgentPlan(agent);

ResourceProvider provider =
plan.getResourceProviders().get(ResourceType.EMBEDDING_MODEL).get("myEmbedding");
assertThat(provider).isInstanceOf(PythonResourceProvider.class);
assertThat(((PythonResourceProvider) provider).getDescriptor().getClazz())
.isEqualTo(String.class.getName());
}

@Test
public void testAddResourceRegistersVectorStoreJavaProvider() throws Exception {
Agent agent = new Agent();
Expand Down Expand Up @@ -535,6 +556,49 @@ public void testAddResourceRejectsNonDescriptorForUnsupportedType() {
Assertions.assertThrows(IllegalStateException.class, () -> new AgentPlan(agent));
}

/** Test agent with descriptor-based Python resource. */
public static class TestAgentWithDescriptorPythonResource extends Agent {
@ChatModelSetup
public static ResourceDescriptor pythonChatModelByDescriptor() {
return ResourceDescriptor.Builder.newBuilder(String.class.getName())
.addInitialArgument("pythonClazz", "test.module.TestClazz")
.build();
}
}

@Test
public void testDescriptorWithPythonClazzUsesPythonResourceProvider() throws Exception {
AgentPlan plan = new AgentPlan(new TestAgentWithDescriptorPythonResource());

ResourceProvider provider =
plan.getResourceProviders()
.get(ResourceType.CHAT_MODEL)
.get("pythonChatModelByDescriptor");
assertThat(provider).isInstanceOf(PythonResourceProvider.class);
assertThat(((PythonResourceProvider) provider).getDescriptor().getClazz())
.isEqualTo(String.class.getName());
}

/** Test agent with explicit Python MCP server declaration. */
public static class TestAgentWithPythonMCPServer extends Agent {
@MCPServer(lang = "python")
public static ResourceDescriptor testMcpServer() {
return ResourceDescriptor.Builder.newBuilder(ResourceName.MCP_SERVER)
.addInitialArgument("endpoint", "http://127.0.0.1:8000/mcp")
.addInitialArgument("timeout", 30)
.build();
}
}

@Test
public void testPythonMCPServerUsesPythonResourceProviderAtCompileTime() throws Exception {
AgentPlan plan = new AgentPlan(new TestAgentWithPythonMCPServer());

ResourceProvider provider =
plan.getResourceProviders().get(ResourceType.MCP_SERVER).get("testMcpServer");
assertThat(provider).isInstanceOf(PythonResourceProvider.class);
}

@Test
public void testAddResourceMCPServerRejectedWithGuidance() {
Agent agent = new Agent();
Expand Down
Loading