Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Add API of PullConsumer #342

Open
wants to merge 2 commits into
base: master
Choose a base branch
from
Open

Conversation

aaron-ai
Copy link
Member

@aaron-ai aaron-ai commented Jan 16, 2023

Similar to the LitePullConsumer provided in RocketMQ 4.0, RocketMQ 5.0 will also provide a brand-new API that can achieve the same functionality, providing higher controllability and flexibility to meet more diverse requirements for message processing.

Fixes #341

@codecov-commenter
Copy link

codecov-commenter commented Jan 16, 2023

Codecov Report

Merging #342 (c6db59e) into master (1184903) will decrease coverage by 0.27%.
The diff coverage is n/a.

@@             Coverage Diff              @@
##             master     #342      +/-   ##
============================================
- Coverage     34.36%   34.08%   -0.28%     
+ Complexity      660      656       -4     
============================================
  Files           220      220              
  Lines         11450    11450              
  Branches        277      277              
============================================
- Hits           3935     3903      -32     
- Misses         7261     7294      +33     
+ Partials        254      253       -1     
Flag Coverage Δ
java 61.54% <ø> (-0.35%) ⬇️

Flags with carried forward coverage won't be shown. Click here to find out more.

Impacted Files Coverage Δ
golang/client.go 36.51% <0.00%> (-4.42%) ⬇️
...e/rocketmq/client/java/impl/ClientManagerImpl.java 78.81% <0.00%> (-3.45%) ⬇️
...g/apache/rocketmq/client/java/impl/ClientImpl.java 47.30% <0.00%> (-1.91%) ⬇️

📣 We’re building smart automated test selection to slash your CI/CD build times. Learn more

@aaron-ai aaron-ai force-pushed the add_pull_api branch 2 times, most recently from ece52b0 to 20f53e1 Compare January 16, 2023 05:43
@aaron-ai aaron-ai changed the title Add pull API Add API of PullConsumer Jan 17, 2023
@github-actions
Copy link

This PR is stale because it has been open for 30 days with no activity. It will be closed in 3 days if no further activity occurs. If you wish not to mark it as stale, please leave a comment in this PR.

@github-actions github-actions bot added the stale Pull request is stale label Feb 17, 2023
@aaron-ai aaron-ai added no stale This will never be considered stale and removed stale Pull request is stale labels Feb 17, 2023
@zhouxinyu zhouxinyu self-requested a review February 17, 2023 06:34
@socutes
Copy link

socutes commented Feb 19, 2023

In the scenario of stream processing, there are several methods to obtain offset information that are often used, and it is recommended to add them:
1.Gets the current consumption progress point. This is often used when the application crashes, data checks, and when obtaining the next consumption point. eg: position()
2.Gets the earliest offset of the topic. eg: beginningOffsets()
3.Gets the offset at the end of the topic. eg: endOffsets()
@zhouxinyu @aaron-ai

@zhouxinyu zhouxinyu added no stale This will never be considered stale and removed no stale This will never be considered stale labels Feb 20, 2023
@aaron-ai
Copy link
Member Author

In the scenario of stream processing, there are several methods to obtain offset information that are often used, and it is recommended to add them: 1.Gets the current consumption progress point. This is often used when the application crashes, data checks, and when obtaining the next consumption point. eg: position() 2.Gets the earliest offset of the topic. eg: beginningOffsets() 3.Gets the offset at the end of the topic. eg: endOffsets() @zhouxinyu @aaron-ai

Make sense. At the same time, would it be more appropriate to place these methods in the Ops related interface rather than the pull consumer?

@zhouxinyu
Copy link
Member

@socutes I noticed that we already have some methods support seek to begin or end of a specifc message queue. These methods provide query and seek semantic, do we really need the query-only methods?

/**
* Commit offset manually.
*/
void commit() throws ClientException;
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The semantics of commit are unclear, for example, for a single queue client, the pull progress is 200, and the poll function returns 100 at this time, we don't know the consumption progress. If we commit 200 will lose the message, commit 100 is not correct either. For users, it is not the case that the previous batch of messages needs to be consumed before taking the next batch. So the meaning here is auto commit. For stream frameworks, it is usually expected that commit and ckpt are atomic, and I think commit(mq, offset) should also be added as a manual interface.

* @param timeout the maximum time to block.
* @return list of fetched messages.
*/
List<MessageView> poll(Duration timeout);
Copy link
Member Author

@aaron-ai aaron-ai Mar 6, 2023

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Should we also provide an asynchronous interface for PullConsumer#poll? If we only offer a synchronous interface, it means that each call will occupy one user thread. Typically, the poll interface is called for multiple queues, so occupying one thread may not have a significant impact. However, an asynchronous interface can provide greater flexibility, such as forming a future chain with the user's own methods, reducing unnecessary threads when multiple clients exist in one process, and potentially offering other benefits.

We welcome everyone to join the discussion.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Labels
no stale This will never be considered stale
Projects
None yet
Development

Successfully merging this pull request may close these issues.

Add API of PullConsumer
5 participants