You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "twosom (via GitHub)" <gi...@apache.org> on 2026/04/25 11:20:08 UTC

[PR] [Feature][Java] Add java event listeners config [flink-agents]

twosom opened a new pull request, #641:
URL: https://github.com/apache/flink-agents/pull/641

   <!--
   * Thank you very much for contributing to Flink Agents.
   * Please add the relevant components in the PR title. E.g., [api], [runtime], [java], [python], [hotfix], etc.
   -->
   
   <!-- Please link the PR to the relevant issue(s). Hotfix doesn't need this. -->
   Linked issue: #640
   
   ### Purpose of change
   Added event listeners config in java
   
   <!-- What is the purpose of this change? -->
   
   ### Tests
   ok
   <!-- How is this change verified? -->
   
   ### API
   
   <!-- Does this change touches any public APIs? -->
   
   ### Documentation
   
   <!-- Do not remove this section. Check the proper box only. -->
   
   - [x] `doc-needed` <!-- Your PR changes impact docs -->
   - [ ] `doc-not-needed` <!-- Your PR changes do not impact docs -->
   - [x] `doc-included` <!-- Your PR already contains the necessary documentation updates -->
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [Feature][Java] Add java event listeners config [flink-agents]

Posted by "twosom (via GitHub)" <gi...@apache.org>.
twosom commented on PR #641:
URL: https://github.com/apache/flink-agents/pull/641#issuecomment-4448999352

   Thanks for the review, @wenjin272! 
   I've resolved the merge conflicts and updated the branch. Please take another look.


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [Feature][Java] Add java event listeners config [flink-agents]

Posted by "twosom (via GitHub)" <gi...@apache.org>.
twosom commented on PR #641:
URL: https://github.com/apache/flink-agents/pull/641#issuecomment-4362996250

   > Thanks for your contribution, @twosom. Overall looks good to me. I left two minor comments.
   
   Thanks for the review! @wenjin272 
   I’ve applied both minor suggestions! 
   Please take another look when you have a moment. Thanks!


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [Feature][Java] Add java event listeners config [flink-agents]

Posted by "wenjin272 (via GitHub)" <gi...@apache.org>.
wenjin272 merged PR #641:
URL: https://github.com/apache/flink-agents/pull/641


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [Feature][Java] Add java event listeners config [flink-agents]

Posted by "wenjin272 (via GitHub)" <gi...@apache.org>.
wenjin272 commented on code in PR #641:
URL: https://github.com/apache/flink-agents/pull/641#discussion_r3165903700


##########
runtime/src/main/java/org/apache/flink/agents/runtime/operator/ActionExecutionOperator.java:
##########
@@ -191,7 +194,7 @@ public class ActionExecutionOperator<IN, OUT> extends AbstractStreamOperator<OUT
     private transient ListState<Object> currentProcessingKeysOpState;
 
     private final transient EventLogger eventLogger;
-    private final transient List<EventListener> eventListeners;

Review Comment:
   I think we can keep the `final` to maintain alignment with eventLogger. In `initEventListeners`, We can use `addAll` instead of assigning the entire collection at once.



##########
api/src/main/java/org/apache/flink/agents/api/listener/EventListener.java:
##########
@@ -31,6 +31,9 @@
  * <p>Event listeners are executed synchronously after the main event processing is complete but
  * before the next event is processed. Implementations should be lightweight and avoid blocking
  * operations to prevent impacting agent performance.

Review Comment:
   It appears that the EventListener is actually triggered before the Event is processed in `ActionExecutionOperator`, which is inconsistent with the documentation here. We may need to update either the doc or the invocation location.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [Feature][Java] Add java event listeners config [flink-agents]

Posted by "wenjin272 (via GitHub)" <gi...@apache.org>.
wenjin272 commented on code in PR #641:
URL: https://github.com/apache/flink-agents/pull/641#discussion_r3193814116


##########
runtime/src/main/java/org/apache/flink/agents/runtime/operator/ActionExecutionOperator.java:
##########
@@ -411,6 +442,8 @@ private void processEvent(Object key, Event event) throws Exception {
             // If the event is an InputEvent, we submit a new mail to try processing the actions.
             mailboxExecutor.submit(() -> tryProcessActionTaskForKey(key), "process action task");
         }
+
+        notifyEventProcessed(event);

Review Comment:
   I carefully reviewed the logic of `processEvent`. In this function, only an `ActionTask` is created and submitted. It's actual execution occurs in subsequent mailbox iterations. Therefore, merely moving `notifyEventProcessed` to the end of the function still remains inconsistent with the documentation of `EventListener`. After considering all factors, I think it's simpler to place it at the beginning and update the `EventListener` documentation accordingly.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [Feature][Java] Add java event listeners config [flink-agents]

Posted by "wenjin272 (via GitHub)" <gi...@apache.org>.
wenjin272 commented on code in PR #641:
URL: https://github.com/apache/flink-agents/pull/641#discussion_r3193814116


##########
runtime/src/main/java/org/apache/flink/agents/runtime/operator/ActionExecutionOperator.java:
##########
@@ -411,6 +442,8 @@ private void processEvent(Object key, Event event) throws Exception {
             // If the event is an InputEvent, we submit a new mail to try processing the actions.
             mailboxExecutor.submit(() -> tryProcessActionTaskForKey(key), "process action task");
         }
+
+        notifyEventProcessed(event);

Review Comment:
   Hi, @twosom. I carefully reviewed the logic of `processEvent`. In this function, only an `ActionTask` is created and submitted. It's actual execution occurs in subsequent mailbox iterations. Therefore, merely moving `notifyEventProcessed` to the end of the function still remains inconsistent with the documentation of `EventListener`. After considering all factors, I think it's simpler to place it at the beginning and update the `EventListener` documentation accordingly.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [Feature][Java] Add java event listeners config [flink-agents]

Posted by "twosom (via GitHub)" <gi...@apache.org>.
twosom commented on code in PR #641:
URL: https://github.com/apache/flink-agents/pull/641#discussion_r3208707689


##########
runtime/src/main/java/org/apache/flink/agents/runtime/operator/ActionExecutionOperator.java:
##########
@@ -411,6 +442,8 @@ private void processEvent(Object key, Event event) throws Exception {
             // If the event is an InputEvent, we submit a new mail to try processing the actions.
             mailboxExecutor.submit(() -> tryProcessActionTaskForKey(key), "process action task");
         }
+
+        notifyEventProcessed(event);

Review Comment:
   Hi @wenjin272 
   I've updated the PR as suggested. Could you take another look? Thanks!



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [Feature][Java] Add java event listeners config [flink-agents]

Posted by "twosom (via GitHub)" <gi...@apache.org>.
twosom commented on code in PR #641:
URL: https://github.com/apache/flink-agents/pull/641#discussion_r3208603627


##########
runtime/src/main/java/org/apache/flink/agents/runtime/operator/ActionExecutionOperator.java:
##########
@@ -411,6 +442,8 @@ private void processEvent(Object key, Event event) throws Exception {
             // If the event is an InputEvent, we submit a new mail to try processing the actions.
             mailboxExecutor.submit(() -> tryProcessActionTaskForKey(key), "process action task");
         }
+
+        notifyEventProcessed(event);

Review Comment:
   Thanks for the catch, @wenjin272 
   
   I missed the point that the actual execution happens in subsequent mailbox iterations. I agree that calling it at the beginning and updating the Javadoc is the best way to avoid inconsistency. I'll update the code and the documentation accordingly.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org