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

[improve][pip] PIP-332: peek messages from topic subscription with offset value #21927

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

Conversation

BalajiLaks
Copy link

PIP: 332

Motivation

Currently we are able to peek any number of messages using the Pulsar Java Admin API.

However, we are only able to view these messages starting from the most recent message. If we want to view the 100th message, we have to peek all the top 100 messages.

We want to display these messages on our UI in a table. When the number of messages to be viewed is large, we want to split them by page and display them on the UI. So, we would like the ability to view any batch of messages instead of all of the top messages.

Modifications

The Pulsar Admin API in Java currently provides a way to peek messages using this

This function includes a default hardcoded offset value of 1. We want to give the user the option to input a different offset value.

The current peekMessages(String topic, String subName, int numMessages) method will remain the same and use offset 1.

We will be overloading this method with offset. It will default to 1 if not specified.

Verifying this change

  • Make sure that the change passes the CI checks.

This change is a trivial rework / code cleanup without any test coverage.

Does this pull request potentially affect one of the following parts:

If the box was checked, please highlight the changes

  • Dependencies (add or upgrade a dependency)
  • The public API
  • The schema
  • The default values of configurations
  • The threading model
  • The binary protocol
  • The REST endpoints
  • The admin CLI options
  • The metrics
  • Anything that affects deployment

Documentation

  • doc
  • doc-required
  • doc-not-needed
  • doc-complete

Matching PR in forked repository

PR in forked repository:

@github-actions github-actions bot added the doc-required Your PR changes impact docs and you will update later. label Jan 18, 2024
@david-streamlio
Copy link
Contributor

Can you include the documentation changes for this in the PR as well?

@dao-jun
Copy link
Member

dao-jun commented Feb 18, 2024

@BalajiLaks It seems that this PIP has been blocked for a long time, please send a VOTE mail to the mail list

* @throws PulsarAdminException
* Unexpected error
*/
List<Message<byte[]>> peekMessages(String topic, String subName, int offset, int numMessages) throws PulsarAdminException;
Copy link
Contributor

Choose a reason for hiding this comment

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

Pulsar does not use the "offset" concept like Kafka or some other messaging systems. So it might be a little confusing here. Instead, you can use messagePosition like the name in

private CompletableFuture<List<Message<byte[]>>> peekNthMessage(String topic, String subName, int messagePosition) {

Copy link
Contributor

Choose a reason for hiding this comment

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

BTW, after introducing these new methods, you can change the original calls to default methods and then you won't need to copy the API docs to this new method.

    /**
     * <original API docs...>
     * @param  offset
     *            Start index to start reading messages.
     * <original API docs...>
     */
    List<Message<byte[]>> peekMessages(String topic, String subName, int startPosition, int numMessages) throws PulsarAdminException;

    default List<Message<byte[]>> peekMessages(String topic, String subName, int numMessages) throws PulsarAdminException {
        return peekMessages(topic, subName, 1, numMessages);
    }

Choose a reason for hiding this comment

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

Thanks for looking into this. The idea of introducing offset as one of the method parameters was to be overwrite offset value of 1 and pass in any number as a startingIndex to fetch numMessages of sorts. This would potentially help in paging while pulling messages from a subscription backlog.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Labels
doc-required Your PR changes impact docs and you will update later. type/PIP
Projects
None yet
Development

Successfully merging this pull request may close these issues.

5 participants