[Feature] Support cross language unified event.

Search before asking

I searched in the issues and found nothing similar.

Description

Currently, if user want to send/receive custom events, they must define a new event type, like

class MyEvent(Event):
    """User custom event."""
    field1: str
    field2: int

This can be somewhat cumbersome at times.

To simplify the definition of Agent, we want to support user only declare the identifier, like:
ctx.send_event(identifier="MyEvent", field1="test", field2=1)

Are you willing to submit a PR?

I'm willing to submit a PR!

@wenjin272

Hi @wenjin272, Is this issue up for grabs?
If so, I can take this up, I put up a draft to show how I'm thinking of solving it: https://github.com/addu390/flink-agents/pull/4/changes

@addu390

Hi, @addu390, I assigned this ticket to you. Thanks for your work and I'll review your draft PR these days.

@wenjin272

@wenjin272 Thank you, I cleaned up the draft and raised PR: #558
Looking forward to your review and feedback, whenever you get a chance.

@addu390

@addu390 Thanks for volunteering on this. The issue is up for grabs. But before moving to the code implementation, could we get aligned with the design first?

The description of the issue might be too brief and the example ctx.send_event(identifier="MyEvent", field1="test", field2=1) is also the final design and might be misleading.

The purposes of this issue are:

To simplify the definition of custom events. Not need to define the event class before using it.
To support cross-language action orchestration, i.e., sending an event in a java implemented action, that triggers a python implemented action. This currently requires defining the two identical event classes in java and python and the serialization between them.

The rough idea is to represent all events with one Event class, with a few required fields (e.g., type) and a list/map of custom fields. In this way, only this one class needs to be implemented in both java and python, which is done by the framework.

With this unified Event, there could be multiple ways (APIs) of creating and accessing Events. Below are some preliminary ideas. I haven't thought them through carefully yet.

Map properties = new HashMap();
properties.put("key1", "value1");
Event event1 = new Event("CustomEvent", properties);
event1.setProperty("key2", "value2"); // or make Event immutable, all properties must be set on instantiation
ctx.sendEvent(event1);

Event event2 = Event.fromJson("{"type": "CustomEvent", "key1": "value1", "key2": "value2"}"); // validate required fields exist
ctx.sendEvent(event2);

@xintongsong

@xintongsong Thanks for the detailed clarification! I'll close #558, I interpreted the issue differently.
But what you explained makes the end-goal clearer. I'll ask/suggest a few design questions before moving to impl.

Seems like we are half way there? Event has attributes (Map<String, Object>), Java routing uses class-name (String) internally.
An Implementation overview:

Make Event a concrete class (currently abstract) with a type field.
Use event.getType() instead of class name (defaults to class-name).
@Action (In java and python), accepts String along-side class types.
JSON for Serde.
Creation APIs: Builder pattern would be a nice too. In python, must haves would be Kwargs and **my_dict. But all of these are easy to extend to later on.

Questions:

Good to re-use attributes in Event for this purpose? Unless it was meant to be for something else.
Any concerns with using JSON for Serde?

Happy to draft a more detailed design if that would be helpful before starting the implementation.

@addu390

Good to re-use attributes in Event for this purpose? Unless it was meant to be for something else.

Re-using attributes sounds good.

Any concerns with using JSON for Serde?

JSON is perfect.

It would be great to see a more detailed design. Thank you, @addu390

@xintongsong

@xintongsong Here's a slightly more detailed design, along with a draft PR to show-case the impl.

Unified Event Design

I'll start with the summary of the goal again,

Problem

Currently, every custom event requires a dedicated Java or Python class. Cross-language events rely on cloudpickle serialization, which is opaque and Python-specific and cannot define ad-hoc event types without creating new classes in both languages.

End Goal

A single Event class that serves both as a base class for subclassed events and as a direct-use for unified events.
String-based routing: events route by a type string rather than by Java/Python class identity/name.
JSON-based cross-language Serde: unified events serialize as plain JSON, no cloudpickle dependency for custom events.
Backward/Full compatibility: existing subclassed events continue to work unchanged.

Core Design

Concrete Event Class

The Event class now has a type (defaults to class-name) field and an attributes map:
Unified event: new Event("MyCustomType", Map.of("key", "value")) sets type = "MyCustomType".
Subclassed event: new InputEvent("data"), falls back to the class-name.
EventContext has eventType and eventClass (new field)
- eventType: The routing key. For unified events, this is the user-defined string (e.g., "MyCustomType"). For subclassed events, this is the fully qualified class name (e.g., "org.apache.flink.agents.api.InputEvent").
- eventClass: Necessary for unified events, this is always org.apache.flink.agents.api.Event. For subclassed events, this matches eventType. eventClass tells the deserializer which concrete Java class to instantiate.

Cross-Language Flow

For unified events sent from Python:

User calls ctx.send_event("MyEvent", key="value").
Which constructs Event(type="MyEvent", attributes={"key": "value"}).
The event is serialized to JSON
The JSON string is passed to Java (an existing pattern used in the code base).
Java reconstructs the event with Event.fromJson(json), also validates that the type field is present.
Note: For subclassed events, the existing cloudpickle serialization path is as-is.

Action Usage

@Action(listenEvents = {InputEvent.class})                                       // class-based
@Action(listenEventTypes = {"MyCustomEvent"})                                    // string-based
@Action(listenEvents = {InputEvent.class}, listenEventTypes = {"MyCustomEvent"}) // mixed

@action(InputEvent)              # class-based
@action("MyCustomEvent")         # string-based
@action(InputEvent, "MyEvent")   # mixed

Both resolve to string keys in the actionsByEvent map at plan compilation time:

Class-based entries use the fully qualified class name (e.g., org.apache.flink.agents.api.InputEvent in Java).
String-based entries are passed through as-is (e.g., MyCustomEvent).
At runtime, event.getType() produces the routing key, which is looked up in actionsByEvent to find the matching actions.

The design does seem to get into the weeds, not much to really do a deeper LLD (unless I'm missing something).
As always, put out a draft for the above: https://github.com/addu390/flink-agents/pull/5/changes

@xintongsong

The design looks good to me.

The only point I'd like to discuss is whether backward compatibility needs to be maintained. Since Flink-Agents currently does not guarantee API stability, I think maybe we could simply remove the subclassed events to keep the design concise. WDYT @xintongsong

@wenjin272

@wenjin272 Thanks for the first take on the design, I'll clean-up my draft and raise a PR soon.

The only point I'd like to discuss is whether backward compatibility needs to be maintained

Agree that it would simplify the docs and maintenance. But we would lose type-safety, which is nice to have.

@addu390

Hi, @addu390, thank you for your contribution! The code PR has been merged.

After reviewing the relevant documentation, I found that the introduction to Event needs to be updated and supplemented with additional content before we close this issue. The current introduction to Event is located in docs/content/docs/development/workflow_agent.md, and it still describes the original typed event.

I think we need introduce:

Unified event

# send event in one action
ctx.send_event(Event(type="my_event", attributes={"field1": "test", "field2": 42}))
# consume event in another action
field1: str = event.get_attr("field1")

Json SerDe

input_data = ItemData.model_validate(InputEvent.from_event(event).input)

The field of a non-trivial type loses its type information after JSON serialization and deserialization, requiring users to manually reconstruct the object of the specific type. (This is only required for Python actions, not for Java actions, because event passing between Java actions does not require serialization.)

For subclass event, user need implement from_event method, and do validation and deserialization.

@wenjin272

@wenjin272 I just open a PR for it: https://github.com/apache/flink-agents/pull/651/changes

Navigation Menu

Repository navigation

[Feature] Support cross language unified event. #424

Search before asking

Description

Are you willing to submit a PR?

Unified Event Design

Problem

End Goal

Core Design

Cross-Language Flow

Action Usage

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

[Feature] Support cross language unified event. #424

Description

Search before asking

Description

Are you willing to submit a PR?

Activity

addu390 commented on Mar 3, 2026

wenjin272 commented on Mar 3, 2026

addu390 commented on Mar 3, 2026

xintongsong commented on Mar 3, 2026

addu390 commented on Mar 3, 2026

xintongsong commented on Mar 3, 2026

addu390 commented on Mar 3, 2026

Unified Event Design

Problem

End Goal

Core Design

Cross-Language Flow

Action Usage

wenjin272 commented on Mar 4, 2026

addu390 commented on Mar 4, 2026

wenjin272 commented on May 8, 2026

addu390 commented on May 8, 2026

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions