apache / apache/flink-agents

[Feature][Python] Implement EventListener to align with Java SDK

Open
#687 0 comments 0 reactions 0 assignees View on GitHub
feature fixVersion/0.4.0 priority/major
Dominant language
Java
Stars
452
Forks
167
Avg merge
5d 9h
Merged PRs (30d)
49

Description

### Search before asking

- [x] I searched in the [issues](https://github.com/apache/flink-agents/issues) and found nothing similar.

### Related PR
- #641
- #102

### Description

This feature introduces the `EventListener` interface to enable event-driven monitoring and custom logic extensions within the Flink Agent Python SDK, aligning with the existing Java SDK implementation.

#### Design Considerations

During the initial design phase, I explored two different approaches for registering event listeners:

**Option 1: Decorator-based Registration (Considered)**
I initially evaluated a Pythonic approach where a decorator registers a standalone function as a listener by storing its metadata in an internal registry.
```python
# Information stored in an internal registry
REGISTRY = []

def event_listener(func):
REGISTRY.append({
"module": func.__module__,
"qualname": func.__qualname__
})
return func

@event_listener
def my_handler(context, event):
print(f"Received: {event}")
```

*Pros:* Highly idiomatic and concise for Python developers.
*Cons:* This approach deviates significantly from the class/interface-based structure of the Java SDK, breaking API consistency across different language SDKs.

**Option 2: Class-based Interface (Final Decision)**
To maintain a unified developer experience and ensure structural alignment with the Java SDK, I decided to adopt a **class-based interface design**.

#### Proposed API Usage

The user implements the `EventListener` interface, and the framework dynamically instantiates and triggers it during the event lifecycle.

```python
from flink_agents.api.listener import EventListener
from flink_agents.api.event_context import EventContext
from flink_agents.api.events import Event

class MyCustomListener(EventListener):
def on_event_processed(self, context: EventContext, event: Event) -> None:
print(f"Event processed: {event.type} at {context.timestamp}")

# Registration via environment configuration
agents_env.get_config().set(
AgentConfigOptions.EVENT_LISTENERS,
[str(MyCustomListener)]
)
```

### Are you willing to submit a PR?

- [x] I'm willing to submit a PR!

Contributor guide

Open the contributing guide

Research direction

Start with the proposed Python API modules flink_agents.api.listener, flink_agents.api.event_context, and flink_agents.api.events, then compare their intended structure with the existing Java SDK implementation. Trace how AgentConfigOptions.EVENT_LISTENERS is read and how listeners are instantiated and triggered during the event lifecycle. Done means the class-based EventListener usage shown in the issue works and remains aligned with the Java SDK.

Written by the indexing model from the issue text.

Assessment

Tech stack
java, python
Domain
api, backend
Issue type
Feature
Difficulty
4/5
Estimated time
3-5 days
Activity status
Quiet
Clarity
Mostly clear
Newbie friendliness
48/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.