[Feature][plan][runtime][java] Introduce Resource and Resource Provider in java

[wenjin272]

Search before asking

• I searched in the issues and found nothing similar.

Description

Introduce Resource and Resource Provider in java

Are you willing to submit a PR?

• I'm willing to submit a PR!

[kitalkuyo-gita]

Resource and Resource Provider Feature Design Proposal

Based on my analysis of the Apache Flink Agents codebase, I've designed a comprehensive proposal for introducing Resource and Resource Provider functionality in Java:

1. Overall Architecture Design

1.1 Core Concept Definitions

Resource: An abstracted manageable resource including compute resources, storage resources, network resources, external service connections, etc. Each resource has lifecycle management (create, use, release).

Resource Provider: A component responsible for creating, configuring, and managing specific types of resources, similar to the existing 1 .

1.2 Module Structure

I recommend extending the existing module structure:

• API Module: Add Resource-related core interfaces
• Runtime Module: Implement resource management runtime components
• Plan Module: Support resource declaration and binding

2. API Module Design

2.1 Core Interface Design

Add the following interfaces in the api/src/main/java/org/apache/flink/agents/api/ directory, maintaining consistency with the existing Event system 2 :

Resource Interface:

// api/src/main/java/org/apache/flink/agents/api/Resource.java
public interface Resource extends AutoCloseable {
    String getId();
    ResourceType getType();
    ResourceStatus getStatus();
    Map<String, Object> getProperties();
    void initialize() throws ResourceException;
    boolean isReady();
}

ResourceProvider Interface:

// api/src/main/java/org/apache/flink/agents/api/ResourceProvider.java
public interface ResourceProvider<T extends Resource> {
    ResourceType getSupportedType();
    T createResource(ResourceConfig config) throws ResourceException;
    void releaseResource(T resource) throws ResourceException;
    boolean isHealthy();
}

2.2 Supporting Types and Enums

// api/src/main/java/org/apache/flink/agents/api/ResourceType.java
public enum ResourceType {
    PYTHON_ENVIRONMENT,
    DATABASE_CONNECTION,
    HTTP_CLIENT,
    FILE_SYSTEM,
    MEMORY_CACHE,
    CUSTOM
}

// api/src/main/java/org/apache/flink/agents/api/ResourceStatus.java
public enum ResourceStatus {
    CREATED,
    INITIALIZING,
    READY,
    IN_USE,
    ERROR,
    DESTROYED
}

3. Runtime Module Design

3.1 Resource Manager

Implement in the runtime/src/main/java/org/apache/flink/agents/runtime/resource/ directory:

ResourceManager:

// runtime/src/main/java/org/apache/flink/agents/runtime/resource/ResourceManager.java
public class ResourceManager {
    private final Map<ResourceType, ResourceProvider<?>> providers;
    private final Map<String, Resource> activeResources;
    private final ResourceConfig globalConfig;
    
    public <T extends Resource> T acquireResource(
        ResourceType type, 
        ResourceConfig config
    ) throws ResourceException;
    
    public void releaseResource(String resourceId) throws ResourceException;
    public ResourceMetrics getResourceMetrics();
}

3.2 Integration with Existing Environment Management

Refactor the existing 3 into a PythonResourceProvider:

// runtime/src/main/java/org/apache/flink/agents/runtime/resource/providers/PythonResourceProvider.java
public class PythonResourceProvider implements ResourceProvider<PythonResource> {
    private final PythonEnvironmentManager environmentManager;
    // Adapt existing PythonEnvironmentManager logic
}

3.3 Resource Context Integration

Extend the existing 4 to support resource management:

// runtime/src/main/java/org/apache/flink/agents/runtime/context/ResourceAwareContext.java
public class ResourceAwareContext extends PythonRunnerContext {
    private final ResourceManager resourceManager;
    private final Map<String, Resource> contextResources;
    
    public <T extends Resource> T getResource(String resourceId, Class<T> type);
    public void bindResource(String name, Resource resource);
}

4. Plan Module Integration

4.1 Resource Declaration Support

Extend the existing 5 to support resource declarations:

// plan/src/main/java/org/apache/flink/agents/plan/ResourceAwareWorkflowPlan.java
public class ResourceAwareWorkflowPlan extends WorkflowPlan {
    private final Map<String, ResourceRequirement> resourceRequirements;
    
    public Map<String, ResourceRequirement> getResourceRequirements();
    public void validateResourceAvailability(ResourceManager manager);
}

4.2 Action Resource Binding

// plan/src/main/java/org/apache/flink/agents/plan/ResourceRequirement.java
public class ResourceRequirement {
    private final ResourceType type;
    private final ResourceConfig config;
    private final String bindingName;
    private final boolean required;
}

5. Implementation Phases

5.1 Phase 1: Core Framework

1. Define core interfaces in API module
2. Implement ResourceManager in Runtime module
3. Create basic resource provider examples

5.2 Phase 2: Existing System Migration

1. Refactor PythonEnvironmentManager to PythonResourceProvider
2. Update PythonRunnerContext to support resource management
3. Maintain backward compatibility

5.3 Phase 3: Plan System Integration

1. Extend WorkflowPlan to support resource declarations
2. Implement resource dependency resolution and validation
3. Support dynamic resource allocation

6. Configuration and Extension Points

6.1 Configuration Management

// api/src/main/java/org/apache/flink/agents/api/ResourceConfig.java
public class ResourceConfig {
    private final Map<String, Object> properties;
    private final Duration timeout;
    private final int retryCount;
    private final ResourceScope scope; // WORKFLOW, ACTION, GLOBAL
}

6.2 Monitoring and Metrics

// runtime/src/main/java/org/apache/flink/agents/runtime/resource/ResourceMetrics.java
public class ResourceMetrics {
    public int getActiveResourceCount(ResourceType type);
    public Duration getAverageAcquisitionTime(ResourceType type);
    public List<ResourceUsageReport> getUsageReports();
}

7. Integration with Existing Event System

Based on the existing Event system 6 , add resource-related events:

// api/src/main/java/org/apache/flink/agents/api/ResourceEvent.java
public abstract class ResourceEvent extends Event {
    private final String resourceId;
    private final ResourceType resourceType;
}

public class ResourceAcquiredEvent extends ResourceEvent { }
public class ResourceReleasedEvent extends ResourceEvent { }
public class ResourceErrorEvent extends ResourceEvent { }