Flink Multi-Version Support

[GreatEugenius]

Background and Goals

Currently, Flink Agents depends on Flink version 1.20.3, while Flink has entered the 2.x era with many new features and improvements. To enable users to leverage more Flink features in their Flink Agents jobs, we plan to support multiple Flink versions, including both Flink 1.20.x and 2.x series, allowing users to choose the appropriate Flink Agents version based on their actual Flink cluster version.

Multi-Version Support Strategy

Java Architecture Design

Flink Agents adopts a layered architecture design, implementing multi-version support through the dist module:

flink-agents
├── api/                          # Unified API layer
├── plan/                         # Execution plan module
├── runtime/                      # Runtime module
├── dist/                         # Version compatibility and packaging module (unified)
│   ├── flink-1.20/               # Flink 1.20.x
│   │   ├── src/                  # Version-specific code
│   │   └── pom.xml               # Complete configuration
│   ├── flink-2.0/                # Flink 2.0.x
│   │   ├── src/                  # Version-specific code
│   │   └── pom.xml               # Complete configuration
│   ├── flink-2.1/                # Flink 2.1.x
│   │   ├── src/                  # Version-specific code
│   │   └── pom.xml               # Complete configuration
│   └── flink-2.2/                # Flink 2.2.x (default)
│       └── pom.xml               # Complete configuration
├── integrations/                 # Integration modules
├── examples/                     # Example code
└── python/                       # Python Implementation

Module Responsibilities

API Module

• Provides unified API interfaces
• Defines core abstractions and interface specifications
• Developed based on Flink 2.2 as the project's default version

Plan Module

• Responsible for generating and optimizing execution plans
• Provides unified plan interfaces
• Developed based on Flink 2.2 as the project's default version

Runtime Module

• Implements core runtime logic
• Provides basic runtime capabilities
• Developed based on Flink 2.2 as the project's default version

Dist Module

Responsibilities:

• Unified responsibility for version compatibility implementation and final release package building
• Encapsulates Flink version-specific implementations (through src directory)
• Handles API differences between versions
• Rewrites or extends version-related core classes (e.g., ActionExecutionOperator)
• Generates standalone uber jars using Maven Shade plugin

Implementation Approach:

• Each submodule (e.g., flink-1.20) contains:
  • src/ directory: stores version-specific compatibility code
  • pom.xml: configures dependencies and packaging rules
• Depends on core modules like api, plan, runtime
• Depends on corresponding Flink framework version (scope as provided)
• Excludes conflicting classes from runtime using Maven Shade plugin, using its own version-specific implementation
• Unpacks and reuses test classes from core modules using maven-dependency-plugin
• Rewrites necessary classes and methods for version differences

Packaging Strategy:

• Non-default versions (e.g., flink-1.20):
  • Includes version-specific code from local src directory
  • Depends on core modules api, plan, runtime
  • Depends on corresponding Flink framework version
• Default version (flink-2.2): directly depends on core modules like api, plan, runtime
• Includes all necessary Integration modules (e.g., ollama)
• Generates deployable complete jar packages
• Naming format: flink-agents-dist-flink-{version}-{project.version}.jar

Integrations Module

• Provides integration capabilities with external systems
• Includes Chat Models (Ollama, etc.), Embedding Models, Vector Stores
• Flink version-independent, can be used by all Flink versions

Python Side

Python is an interpreted language that resolves compatibility issues through runtime Flink version detection. Python packages embed corresponding version Java jar packages and automatically select the appropriate version at runtime.

Version Management Mechanism

We introduce the Flink version management utility class FlinkVersionManager to provide version detection functionality, allowing developers to write different code based on Flink version.

from typing import Optional
from packaging import version

class FlinkVersionManager:
    def __init__(self):
        self._flink_version = None
        self._initialized = False

    def _initialize(self):
        if self._initialized:
            return

        # Try multiple ways to get version
        self._flink_version = self._get_java_flink_version() or self._get_pyflink_version()
        self._initialized = True

    def _get_pyflink_version(self) -> Optional[str]:
        try:
            import pkg_resources
            return pkg_resources.get_distribution("apache-flink").version
        except Exception:
            return None

    @property
    def version(self) -> Optional[str]:
        self._initialize()
        return self._flink_version

    @property
    def major_version(self) -> str | None:
        if not self.version:
            return None

        version_parts = self.version.split('-')[0].split('.')
        if len(version_parts) >= 2:
            return f"{version_parts[0]}.{version_parts[1]}"
        return self.version

    def ge(self, target_version: str) -> bool:
        """Greater than or equal"""
        if not self.version:
            return False
        current = self._normalize_version(self.version)
        target = self._normalize_version(target_version)
        return version.parse(current) >= version.parse(target)

    def lt(self, target_version: str) -> bool:
        """Less than"""
        return not self.ge(target_version)

    def _normalize_version(self, version_str: str) -> str:
        """Normalize version string"""
        # Handle various version formats
        parts = version_str.split('-')[0].split('.')
        if len(parts) == 2:
            parts.append('0')
        return '.'.join(parts[:3])

# Global instance
flink_version_manager = FlinkVersionManager()

Usage Example:

from flink_agents.version import flink_version_manager

if flink_version_manager.ge("2.0.0"):
    # Processing logic for Flink 2.0+ version
    result = call_new_version_api()
else:
    # Processing logic for versions before Flink 2.0
    result = call_old_version_api()

Automatic Java Jar Loading

Python code automatically detects the Flink version at runtime and selects the corresponding jar package from the flink_agents/lib/ directory to load.

@staticmethod
def get_execution_environment(
    env: StreamExecutionEnvironment | None = None,
    t_env: StreamTableEnvironment | None = None,
    **kwargs: Dict[str, Any],
) -> "AgentsExecutionEnvironment":
    if env is None:
        return importlib.import_module(
            "flink_agents.runtime.local_execution_environment"
        ).create_instance(env=env, t_env=t_env, **kwargs)
    else:
        major_version = flink_version_manager.major_version
        if major_version:
            # Determine the version-specific lib directory
            version_dir = f"flink-{major_version}"
            lib_base = files("flink_agents.lib")
            version_lib = lib_base / version_dir

            # Check if version-specific directory exists
            if version_lib.is_dir():
                for jar_file in version_lib.iterdir():
                    if jar_file.is_file() and str(jar_file).endswith('.jar'):
                        env.add_jars(f"file://{jar_file}")
            else:
                err_msg = f"Flink Agents dist JAR for Flink {major_version} not found."
                raise FileNotFoundError(err_msg)

            return importlib.import_module(
                "flink_agents.runtime.remote_execution_environment"
            ).create_instance(env=env, t_env=t_env, **kwargs)
        else:
            err_msg = "Apache Flink is not installed."
            raise ModuleNotFoundError(err_msg)

Design Rationale

Currently, Flink Agents adopts a simplified version management approach on the Python side. This is because:

1. Actual Requirements Assessment: Currently, there are no obvious conflicts or incompatibility issues at the Python API layer across different Flink versions
2. API Reusability: Most Python APIs can be reused across different Flink versions without special handling for each version
3. Avoid Over-engineering: Implementing very specific and complex version control mechanisms now might lead to:
   • Designed features mismatching actual needs
   • Unnecessary maintenance costs
   • Need for redesign when future requirements change

Future Evolution:

We will consider introducing more complex version control mechanisms when:

• Python APIs show obvious incompatibilities across different Flink versions
• More fine-grained version control is needed
• New Flink versions introduce breaking changes

Build Artifacts

Java Artifacts

Generated jar package naming format: flink-agents-dist-flink-{version}-{project.version}.jar

Currently Supported Versions:

dist/
├── flink-1.20/target/
│   └── flink-agents-dist-flink-1.20-0.2-SNAPSHOT.jar    # Flink 1.20.x version
├── flink-2.0/target/
│   └── flink-agents-dist-flink-2.0-0.2-SNAPSHOT.jar     # Flink 2.0.x version
├── flink-2.1/target/
│   └── flink-agents-dist-flink-2.1-0.2-SNAPSHOT.jar     # Flink 2.1.x version
└── flink-2.2/target/
    └── flink-agents-dist-flink-2.2-0.2-SNAPSHOT.jar     # Flink 2.2.x version

Python Artifacts

Generated Python package naming format: flink_agents-{version}-py3-none-any.whl

Example:

python/dist/
├── flink_agents-0.2.dev0-py3-none-any.whl       # Wheel package
└── flink_agents-0.2.dev0.tar.gz                 # Source package

Package Contents:

• Python API code (flink_agents/ directory)

• Embedded Java jar packages (stored in flink_agents/lib/ directory, categorized by version)

  flink_agents/lib/
  ├── flink-1.20/
  │   └── flink-agents-dist-flink-1.20-0.2-SNAPSHOT.jar
  ├── flink-2.0/
  │   └── flink-agents-dist-flink-2.0-0.2-SNAPSHOT.jar
  ├── flink-2.1/
  │   └── flink-agents-dist-flink-2.1-0.2-SNAPSHOT.jar
  └── flink-2.2/
      └── flink-agents-dist-flink-2.2-0.2-SNAPSHOT.jar
  

• Python dependencies (declared via pyproject.toml)

  • apache-flink>=1.20.3,<=2.2.0 (removed to prevent automatic PyFlink installation, requires manual installation by users)
  • pydantic, ollama, openai, anthropic, chromadb, etc. (retained)

Build Commands

# Build all versions of Java and Python packages
./tools/build.sh

# Build Java packages only
./tools/build.sh -j

# Build Python packages only
./tools/build.sh -p

Build Process:

1. Java Build: Execute mvn clean install -DskipTests -B

   • Build all modules (api, plan, runtime, dist, integrations)
   • Generate dist jar packages for each version

2. Python Build:

   • Automatically detect and copy all version jar packages from dist/flink-*/target/ to python/flink_agents/lib/flink-*/
   • Use uv to manage Python dependencies and build
   • Generate Python wheel package containing all version jars

Usage Guide

Choosing the Right Version

Users should choose the corresponding Flink Agents version based on their Flink cluster version.

Java Users

Development Phase: Use unified API module for development

<!-- Development dependency: Use Flink version-independent API -->
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-agents-api</artifactId>
    <version>0.2-SNAPSHOT</version>
    <scope>provided</scope>
</dependency>

Python Users

Python packages automatically include jar packages for all supported versions and select the appropriate implementation at runtime based on the Flink version in the environment.

Installation:

# Install Flink Agents only, without PyFlink by default
pip install flink_agents

# Install Flink Agents along with PyFlink
pip install flink_agents apache-flink==0.2.0