[Feature][Python] Implement EventListener to align with Java SDK
- 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
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