[feat][admin] PIP-415: Support getting message ID by index #24222

liangyepianzhou · 2025-04-27T12:20:30Z

Motivation

we can now obtain the offset of a message by its message id:

Get the message by id using get-message-by-id cmd
Get the index of the message using Message.getIndex()

But we cannot obtain the message id by offset. Then we need to add a new API to get the message id by offset.

Modifications

Add a new http API to retrieve the message ID by offset.
We propose to add a new API to retrieve the message ID by offset, enabling us to cache the mapping between message ID and offset.
This will allow us to use offsets for seek and acknowledgment operations when consuming messages through the standardized API.

Verifying this change

Make sure that the change passes the CI checks.

(Please pick either of the following options)

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

(or)

This change is already covered by existing tests, such as (please describe tests).

(or)

This change added tests and can be verified as follows:

(example:)

Added integration tests for end-to-end deployment with large payloads (10MB)
Extended integration test for recovery after broker failure

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

If the box was checked, please highlight the changes

Documentation

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

Matching PR in forked repository

PR in forked repository:

pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java

StevenLuMT

#24220 PIP has a lot of adjustments.
@liangyepianzhou please ping me after you modify the code and I will review it again.

AuroraTwinkle · 2025-05-29T08:57:54Z

Good jobs

pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java

pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/PersistentTopics.java

pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java

BewareMyPower · 2025-05-30T10:29:30Z

Out of the scope of this PR, I think it would be better to add a method to ManagedLedger for it. Because the customized ML implementation (e.g. StreamNative's Ursa engine) might not depend on the broker entry metadata to get the message's index. It could be another PIP.

[2] rename getMessageIDByIndexAndPartitionID to getMessageIDByIndex

liangyepianzhou · 2025-06-03T03:12:29Z

Out of the scope of this PR, I think it would be better to add a method to ManagedLedger for it. Because the customized ML implementation (e.g. StreamNative's Ursa engine) might not depend on the broker entry metadata to get the message's index. It could be another PIP.

Yes, we should add a new manager ledger interface to support customized ML implementations.

Additionally, the current message index retrieval solution has suboptimal performance. We could improve this by adding a field in LedgerInfo to record the first entry's index, then compute subsequent indexes directly from message IDs.

I've created an issue to track this enhancement.

dao-jun · 2025-06-03T03:14:25Z

Overall LGTM, but the current impl will scan all entire ML, maybe you can optimize it via PIP-404

liangyepianzhou · 2025-06-03T03:29:10Z

Overall LGTM, but the current impl will scan all entire ML, maybe you can optimize it via PIP-404

Interesting—I originally proposed directly adding the first entry's index in LedgerInfo to optimize this issue. While the properties field you added could also serve the same purpose, it currently doesn’t record the first entry's index. We can address this optimization in a future PIP, as mentioned in issue.

pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java

nodece · 2025-06-04T07:45:00Z

pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java

+                            managedLedger)
+                            .thenCompose(firstIndex -> {
+                                if (index < firstIndex) {
+                                    return CompletableFuture.completedFuture(PositionFactory.EARLIEST);


Or consider returning null?

Can you explain why you think null is better than earliest?

Explicit signal that no valid position exists for the given index.

Maybe you can get more context here
https://github.com/apache/pulsar/pull/24220/files#r2090841224

Ok, got.

PositionFactory.EARLIEST name feels odd, as its value (-1, -1) doesn't correspond to a real message and may mislead developers into thinking it's a valid starting point. Returning null would more clearly indicate that the index is out of range and no valid position exists when used as the public value.

It's just my opinion.

Return earliest is OK, but do we pass the index to the clients?
I seem to have some impression, but I'm not sure.

(-1, -1) doesn't correspond to a real message and may mislead developers into thinking it's a valid starting point.

It's already documented, so it's good.

but do we pass the index to the clients?

Do you mean the actual first available index? If so, I think the API could be unnecessarily complicated (a good API does not do multiple things)

pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerEntryMetadataE2ETest.java

Copilot

Pull Request Overview

This PR introduces a new HTTP API to retrieve a Pulsar message ID given its index, enabling caching between message IDs and offsets and providing new CLI and admin functionalities.

Adds a new CLI command and corresponding API methods in the client admin components.
Implements REST endpoint and internal broker logic to support retrieving message IDs by index.
Adds tests to verify the new functionality for both partitioned and non-partitioned topics.

Reviewed Changes

Copilot reviewed 7 out of 7 changed files in this pull request and generated no comments.

Show a summary per file

File	Description
pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdTopics.java	Introduces the new CLI command "get-message-id-by-index".
pulsar-client-tools-test/src/test/java/org/apache/pulsar/admin/cli/PulsarAdminToolTest.java	Adds tests verifying the new CLI command functionality.
pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/TopicsImpl.java	Implements synchronous and asynchronous API methods for getMessageIdByIndex.
pulsar-client-admin-api/src/main/java/org/apache/pulsar/client/admin/Topics.java	Updates the Topics interface with new method signatures and documentation.
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerEntryMetadataE2ETest.java	Adds integration test scenarios for the new message ID by index functionality.
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/PersistentTopics.java	Implements new REST endpoint to support retrieving message ID by index.
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java	Provides the internal broker logic for the getMessageIdByIndex API.

Comments suppressed due to low confidence (1)

pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/PersistentTopics.java:5033

The variable 'topicName' is not defined in this scope. It appears you intended to log the resolved topic name (perhaps 'encodedTopic' or a combination of tenant, namespace, and encodedTopic).

log.error("[{}] Failed to get message id by index for topic {}, partition id {}, index {}", clientAppId(), topicName, index, ex);

dao-jun

Just some small comments, Overall LGTM.
By the way, if you rely on the AppendMessageIndexInterceptor, I'd suggest you need to check whether the interceptor configured before start the broker.

pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java

dao-jun · 2025-06-04T09:58:25Z

pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java

+                            managedLedger)
+                            .thenCompose(firstIndex -> {
+                                if (index < firstIndex) {
+                                    return CompletableFuture.completedFuture(PositionFactory.EARLIEST);


Return earliest is OK, but do we pass the index to the clients?
I seem to have some impression, but I'm not sure.

pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java

pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerEntryMetadataE2ETest.java

nodece · 2025-06-05T03:30:36Z

pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerEntryMetadataE2ETest.java

+                PulsarAdminException.class, NotFoundException.class);
+    }
+
+    private void assertThrowsWithCause(Executable executable,


Suggested change

private void assertThrowsWithCause(Executable executable,

private void assertThrowsWithCause(Runnable executable,

nodece · 2025-06-05T03:31:00Z

pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerEntryMetadataE2ETest.java

@@ -41,6 +46,7 @@
 import org.apache.pulsar.common.util.FutureUtil;
 import org.assertj.core.util.Sets;
 import org.awaitility.Awaitility;
+import org.junit.jupiter.api.function.Executable;


Suggested change

import org.junit.jupiter.api.function.Executable;

github-actions bot added the doc-not-needed Your PR changes do not impact docs label Apr 27, 2025

xiangying added 2 commits April 27, 2025 20:28

[feat][admin] PIP-415: Support getting message ID by offset

7f876b6

Add Cli

75194f7

liangyepianzhou force-pushed the pip-415-impl branch from 035132b to 75194f7 Compare April 27, 2025 12:29

liangyepianzhou self-assigned this Apr 27, 2025

liangyepianzhou marked this pull request as ready for review April 27, 2025 12:35

StevenLuMT reviewed Apr 28, 2025

View reviewed changes

pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java Outdated Show resolved Hide resolved

StevenLuMT reviewed Apr 28, 2025

View reviewed changes

pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java Show resolved Hide resolved

AuroraTwinkle reviewed Apr 28, 2025

View reviewed changes

pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java Outdated Show resolved Hide resolved

pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java Outdated Show resolved Hide resolved

xiangying added 3 commits April 28, 2025 14:28

Change offset to index

f29c42e

Add NPE check after find message

392b975

Check where the broker entry metadata is enabled

33fd340

liangyepianzhou requested review from AuroraTwinkle and StevenLuMT May 6, 2025 03:42

xiangying added 2 commits May 14, 2025 10:41

update

f2e5259

typo

c83f4f1

StevenLuMT approved these changes May 20, 2025

View reviewed changes

Aligned with pip implementation

d4cdceb

liangyepianzhou changed the title ~~[feat][admin] PIP-415: Support getting message ID by offset~~ [feat][admin] PIP-415: Support getting message ID by index May 29, 2025

offset -> index

2070e30

liangyepianzhou requested review from lhotari, Technoboy-, nodece, BewareMyPower, dao-jun and poorbarcode and removed request for AuroraTwinkle May 29, 2025 09:46

BewareMyPower requested changes May 29, 2025

View reviewed changes

pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java Outdated Show resolved Hide resolved

skip messages without index

be588b8

BewareMyPower requested changes May 30, 2025

View reviewed changes

[1]: Add a common method to get index from entry

6f588d9

[2] rename getMessageIDByIndexAndPartitionID to getMessageIDByIndex

liangyepianzhou mentioned this pull request Jun 3, 2025

[Enhancement] Support getting message index in ManagedLedger interface #24374

Open

2 tasks

liangyepianzhou requested a review from BewareMyPower June 3, 2025 06:17

StevenLuMT reviewed Jun 4, 2025

View reviewed changes

pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java Show resolved Hide resolved

StevenLuMT reviewed Jun 4, 2025

View reviewed changes

pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java Show resolved Hide resolved

StevenLuMT reviewed Jun 4, 2025

View reviewed changes

pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java Outdated Show resolved Hide resolved

release entry out of getIndexFromEntry method

7025b1c

nodece reviewed Jun 4, 2025

View reviewed changes

pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java Show resolved Hide resolved

nodece reviewed Jun 4, 2025

View reviewed changes

pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java Outdated Show resolved Hide resolved

nodece reviewed Jun 4, 2025

View reviewed changes

pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerEntryMetadataE2ETest.java Outdated Show resolved Hide resolved

nodece requested a review from Copilot June 4, 2025 07:49

Copilot AI reviewed Jun 4, 2025

View reviewed changes

add test and handle empty topic case

ad9b66b

liangyepianzhou force-pushed the pip-415-impl branch from 192c53f to ad9b66b Compare June 4, 2025 08:14

dao-jun approved these changes Jun 4, 2025

View reviewed changes

remove exception signature

44386f2

nodece reviewed Jun 4, 2025

View reviewed changes

pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerEntryMetadataE2ETest.java Show resolved Hide resolved

improve test

35a5ae5

liangyepianzhou force-pushed the pip-415-impl branch from c531f37 to 35a5ae5 Compare June 4, 2025 11:28

nodece reviewed Jun 5, 2025

View reviewed changes

	private void assertThrowsWithCause(Executable executable,
	private void assertThrowsWithCause(Runnable executable,

[feat][admin] PIP-415: Support getting message ID by index #24222

Are you sure you want to change the base?

[feat][admin] PIP-415: Support getting message ID by index #24222

Uh oh!

Conversation

liangyepianzhou commented Apr 27, 2025

Motivation

Modifications

Verifying this change

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

Documentation

Matching PR in forked repository

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

StevenLuMT left a comment • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

AuroraTwinkle commented May 29, 2025

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

BewareMyPower commented May 30, 2025

Uh oh!

liangyepianzhou commented Jun 3, 2025

Uh oh!

dao-jun commented Jun 3, 2025

Uh oh!

liangyepianzhou commented Jun 3, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull Request Overview

Reviewed Changes

Uh oh!

dao-jun left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

StevenLuMT left a comment •

edited

Loading

liangyepianzhou commented Jun 3, 2025 •

edited

Loading