fix: Event consumer should stop on input_required #566
Conversation
Summary of ChangesHello @cmtoan, I'm Gemini Code Assist1! I'm currently reviewing this pull request and will post my feedback shortly. In the meantime, here's a summary to help you and other reviewers quickly get up to speed! This pull request addresses a critical issue in the event consumption mechanism where the system failed to halt event processing when a task entered an 'input required' state. By modifying the event consumer to recognize this state as a termination condition, the change ensures proper flow control and consistency with other language implementations, preventing unnecessary resource usage and potential logical errors. Highlights
Using Gemini Code AssistThe full guide for Gemini Code Assist can be found on our documentation page, here are some quick tips. Invoking Gemini You can request assistance from Gemini at any point by creating a comment using either
Customization To customize Gemini Code Assist for GitHub experience, repository maintainers can create a configuration file and/or provide a custom code review style guide (such as PEP-8 for Python) by creating and adding files to a Limitations & Feedback Gemini Code Assist may make mistakes. Please leave feedback on any instances where its feedback is incorrect or counter productive. You can react with 👍 and 👎 on @gemini-code-assist comments. If you're interested in giving your feedback about your experience with Gemini Code Assist for Github and other Google products, sign up here. You can also get AI-powered code generation, chat, as well as code reviews directly in the IDE at no cost with the Gemini Code Assist IDE Extension. Footnotes
|
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request correctly implements the requirement for the EventConsumer to stop processing events when a task with the INPUT_REQUIRED state is encountered. The change is isolated and a new helper method is introduced for the logic. A corresponding unit test has been added to verify the new behavior. My review includes suggestions to improve the clarity of the new method's name and to address code duplication in the test suite for better long-term maintainability.
| isFinalEvent = true; | ||
| } else if (event instanceof Task task) { | ||
| isFinalEvent = task.status().state().isFinal(); | ||
| isFinalEvent = isFinalOrInterruptedTask(task); |
There was a problem hiding this comment.
To match the suggested rename of the helper method for clarity, this call should be updated.
| isFinalEvent = isFinalOrInterruptedTask(task); | |
| isFinalEvent = isStreamTerminatingTask(task); |
| private boolean isFinalOrInterruptedTask(Task task) { | ||
| return task.status().state().isFinal() || task.status().state() == TaskState.INPUT_REQUIRED; | ||
| } |
There was a problem hiding this comment.
The method name isFinalOrInterruptedTask can be misleading, as INPUT_REQUIRED is not a final state for a task, but rather a state that should terminate the event stream. A more descriptive name like isStreamTerminatingTask would improve clarity. For even better readability, you could also consider storing task.status().state() in a local variable.
| private boolean isFinalOrInterruptedTask(Task task) { | |
| return task.status().state().isFinal() || task.status().state() == TaskState.INPUT_REQUIRED; | |
| } | |
| private boolean isStreamTerminatingTask(Task task) { | |
| return task.status().state().isFinal() || task.status().state() == TaskState.INPUT_REQUIRED; | |
| } |
| publisher.subscribe(new Flow.Subscriber<>() { | ||
| private Flow.Subscription subscription; | ||
|
|
||
| @Override | ||
| public void onSubscribe(Flow.Subscription subscription) { | ||
| this.subscription = subscription; | ||
| subscription.request(1); | ||
| } | ||
|
|
||
| @Override | ||
| public void onNext(EventQueueItem item) { | ||
| receivedEvents.add(item.getEvent()); | ||
| subscription.request(1); | ||
| } | ||
|
|
||
| @Override | ||
| public void onError(Throwable throwable) { | ||
| error.set(throwable); | ||
| } | ||
|
|
||
| @Override | ||
| public void onComplete() { | ||
| subscription.cancel(); | ||
| } | ||
| }); |
There was a problem hiding this comment.
This subscriber implementation is duplicated across several tests in this file (e.g., testConsumeAllMultipleEvents, testConsumeMessageEvents). To improve maintainability and reduce boilerplate code, consider extracting this logic into a private helper method. The helper could take a Flow.Publisher and return the collected events, making each test more concise and focused on its specific assertions.
|
There are flaky tests in QuarkusA2ARestTest. |
1 participant
Description
Event consumer should stop on input_required task as fixed in the python version a2aproject/a2a-python#167