You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "liangyepianzhou (via GitHub)" <gi...@apache.org> on 2025/04/27 12:20:33 UTC
[PR] [feat][admin] PIP-415: Support getting message ID by offset [pulsar]
liangyepianzhou opened a new pull request, #24222:
URL: https://github.com/apache/pulsar/pull/24222
PIP:https://github.com/apache/pulsar/pull/24220
### Motivation
we can now obtain the offset of a message by its message id:
1. Get the message by id using `get-message-by-id` cmd
2. 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:
<!-- DO NOT REMOVE THIS SECTION. CHECK THE PROPER BOX ONLY. -->
*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
<!-- DO NOT REMOVE THIS SECTION. CHECK THE PROPER BOX ONLY. -->
- [ ] `doc` <!-- Your PR contains doc changes. -->
- [ ] `doc-required` <!-- Your PR changes impact docs and you will update later -->
- [x] `doc-not-needed` <!-- Your PR changes do not impact docs -->
- [ ] `doc-complete` <!-- Docs have been already added -->
### Matching PR in forked repository
PR in forked repository: <!-- ENTER URL HERE -->
<!--
After opening this PR, the build in apache/pulsar will fail and instructions will
be provided for opening a PR in the PR author's forked repository.
apache/pulsar pull requests should be first tested in your own fork since the
apache/pulsar CI based on GitHub Actions has constrained resources and quota.
GitHub Actions provides separate quota for pull requests that are executed in
a forked repository.
The tests will be run in the forked repository until all PR review comments have
been handled, the tests pass and the PR is approved by a reviewer.
-->
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat][admin] PIP-415: Support getting message ID by index [pulsar]
Posted by "nodece (via GitHub)" <gi...@apache.org>.
nodece commented on code in PR #24222:
URL: https://github.com/apache/pulsar/pull/24222#discussion_r2125890356
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java:
##########
@@ -5489,4 +5489,114 @@ protected CompletableFuture<AutoSubscriptionCreationOverride> internalGetAutoSub
return null;
}));
}
+
+ protected CompletableFuture<MessageId> internalGetMessageIDByIndexAsync(Long index, boolean authoritative) {
+ if (!pulsar().getBrokerService().isBrokerEntryMetadataEnabled()) {
+ return FutureUtil.failedFuture(new RestException(Status.PRECONDITION_FAILED,
+ "GetMessageIDByIndex is not allowed when broker entry metadata is disabled"));
+ }
+ if (index == null || index < 0) {
+ return FutureUtil.failedFuture(new RestException(Status.NOT_FOUND,
+ "Invalid message index: " + index));
+ }
+ int partitionIndex = topicName.getPartitionIndex();
+ CompletableFuture<Void> future = validateTopicOperationAsync(topicName, TopicOperation.PEEK_MESSAGES);
+ return future.thenCompose(__ -> {
+ if (topicName.isGlobal()) {
+ return validateGlobalNamespaceOwnershipAsync(namespaceName);
+ } else {
+ return CompletableFuture.completedFuture(null);
+ }
+ }).thenCompose(__ -> {
+ if (topicName.isPartitioned()) {
+ return CompletableFuture.completedFuture(null);
+ } else {
+ return getPartitionedTopicMetadataAsync(topicName, authoritative, false)
+ .thenAccept(topicMetadata -> {
+ if (topicMetadata.partitions > 0) {
+ log.warn("[{}] Not supported getMessageIdByIndex operation on "
+ + "partitioned-topic {}", clientAppId(), topicName);
+ throw new RestException(Status.METHOD_NOT_ALLOWED,
+ "GetMessageIDByIndex is not allowed on partitioned-topic");
+ }
+ });
+ }
+ }).thenCompose(ignore -> validateTopicOwnershipAsync(topicName, authoritative))
+ .thenCompose(__ -> getTopicReferenceAsync(topicName))
+ .thenCompose(topic -> {
+ if (!(topic instanceof PersistentTopic persistentTopic)) {
+ log.error("[{}] Get message id by index on a non-persistent topic {} is not allowed",
+ clientAppId(), topicName);
+ return FutureUtil.failedFuture(new RestException(Status.METHOD_NOT_ALLOWED,
+ "Get message id by index on a non-persistent topic is not allowed"));
+ }
+ ManagedLedger managedLedger = persistentTopic.getManagedLedger();
+ return findMessageIndexByPosition(
+ PositionFactory.create(managedLedger.getFirstPosition().getLedgerId(), 0),
Review Comment:
`findMessageIndexByPosition(managedLedger.getFirstPosition(), ...)` is better.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat][admin] PIP-415: Support getting message ID by index [pulsar]
Posted by "liangyepianzhou (via GitHub)" <gi...@apache.org>.
liangyepianzhou commented on code in PR #24222:
URL: https://github.com/apache/pulsar/pull/24222#discussion_r2125978050
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerEntryMetadataE2ETest.java:
##########
@@ -420,4 +424,74 @@ public void testManagedLedgerTotalSize() throws Exception {
cursor.close();
}
+
+ @Test
+ public void testGetMessageIdByIndex() throws Exception {
+ // 1. test no partitioned topic
+ final String topicName = newTopicName();
+ admin.topics().createNonPartitionedTopic(topicName);
+ @Cleanup
+ Producer<String> producer = pulsarClient.newProducer(Schema.STRING)
+ .topic(topicName)
+ .enableBatching(false)
+ .create();
+ MessageIdImpl messageId = (MessageIdImpl) producer.send("test");
+ Message<byte[]>
+ message =
+ admin.topics().getMessagesById(topicName, messageId.getLedgerId(), messageId.getEntryId()).get(0);
+ long index = message.getIndex().get();
+ MessageIdImpl messageIdByIndex = (MessageIdImpl) admin.topics().getMessageIdByIndex(topicName, index);
+ Assert.assertEquals(messageIdByIndex, messageId);
+
+ // 2. test partitioned topic
+ final String topicName2 = newTopicName();
+ final String partitionedTopicName = topicName2 + "-partition-" + 0;
+ admin.topics().createPartitionedTopic(topicName2, 10);
+ @Cleanup
+ Producer<String> producer2 = pulsarClient.newProducer(Schema.STRING)
+ .topic(topicName2)
+ .enableBatching(false)
+ .create();
+
+ MessageIdImpl messageId2 = null;
+ for (int i = 0; i < 200; i++) {
+ messageId2 = (MessageIdImpl) producer2.send("test" + i);
+ if (messageId2.getPartitionIndex() == 0) {
+ break;
+ }
+ }
+ Message<byte[]>
+ message2 = admin.topics().getMessagesById(partitionedTopicName,
+ messageId2.getLedgerId(), messageId2.getEntryId()).get(0);
+ long index2 = message2.getIndex().get();
+ // 2.1 test partitioned topic name with partition index
+ MessageIdImpl messageIdByIndex2 =
+ (MessageIdImpl) admin.topics().getMessageIdByIndex(partitionedTopicName, index2);
+ Assert.assertEquals(messageIdByIndex2, messageId2);
+ // 2.2 test partitioned topic name without partition index
+ try {
+ messageIdByIndex2 = (MessageIdImpl) admin.topics().getMessageIdByIndex(topicName2, index2);
+ fail("Should not be able to get messageId by index for partitioned topic without partition index");
+ } catch (PulsarAdminException e) {
Review Comment:
update
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat][admin] PIP-415: Support getting message ID by index [pulsar]
Posted by "AuroraTwinkle (via GitHub)" <gi...@apache.org>.
AuroraTwinkle commented on PR #24222:
URL: https://github.com/apache/pulsar/pull/24222#issuecomment-2918764725
Good jobs
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat][admin] PIP-415: Support getting message ID by index [pulsar]
Posted by "liangyepianzhou (via GitHub)" <gi...@apache.org>.
liangyepianzhou commented on code in PR #24222:
URL: https://github.com/apache/pulsar/pull/24222#discussion_r2128055803
##########
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;
Review Comment:
same with https://github.com/apache/pulsar/pull/24222#discussion_r2128055455
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat][admin] PIP-415: Support getting message ID by index [pulsar]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on code in PR #24222:
URL: https://github.com/apache/pulsar/pull/24222#discussion_r2115590549
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/PersistentTopics.java:
##########
@@ -5015,5 +5015,38 @@ public void removeAutoSubscriptionCreation(
});
}
+ @GET
+ @Path("/{tenant}/{namespace}/{topic}/getMessageIdByIndex")
+ @ApiOperation(hidden = true, value = "Get Message ID by index.",
+ notes = "If the specified index is a system message, "
+ + "it will return the message id of the later message.")
+ @ApiResponses(value = {
+ @ApiResponse(code = 307, message = "Current broker doesn't serve the namespace of this topic"),
+ @ApiResponse(code = 403, message = "Don't have admin permission"),
+ @ApiResponse(code = 404, message = "Namespace or partitioned topic does not exist, "
+ + "or the index is invalid"),
+ @ApiResponse(code = 406, message = "The topic is not a persistent topic"),
+ @ApiResponse(code = 412, message = "The broker is not enable broker entry metadata"),
+ })
+ public void getMessageIDByIndexAndPartitionID(@Suspended final AsyncResponse asyncResponse,
Review Comment:
The `AndPartitionID` suffix in the method seems meaningless (and confusing)?
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java:
##########
@@ -5489,4 +5489,114 @@ protected CompletableFuture<AutoSubscriptionCreationOverride> internalGetAutoSub
return null;
}));
}
+
+ protected CompletableFuture<MessageId> internalGetMessageIDByIndexAsync(Long index, boolean authoritative) {
+ if (!pulsar().getBrokerService().isBrokerEntryMetadataEnabled()) {
+ return FutureUtil.failedFuture(new RestException(Status.PRECONDITION_FAILED,
+ "GetMessageIDByIndex is not allowed when broker entry metadata is disabled"));
+ }
+ if (index == null || index < 0) {
+ return FutureUtil.failedFuture(new RestException(Status.NOT_FOUND,
+ "Invalid message index: " + index));
+ }
+ int partitionIndex = topicName.getPartitionIndex();
+ CompletableFuture<Void> future = validateTopicOperationAsync(topicName, TopicOperation.PEEK_MESSAGES);
+ return future.thenCompose(__ -> {
+ if (topicName.isGlobal()) {
+ return validateGlobalNamespaceOwnershipAsync(namespaceName);
+ } else {
+ return CompletableFuture.completedFuture(null);
+ }
+ }).thenCompose(__ -> {
+ if (topicName.isPartitioned()) {
+ return CompletableFuture.completedFuture(null);
+ } else {
+ return getPartitionedTopicMetadataAsync(topicName, authoritative, false)
+ .thenAccept(topicMetadata -> {
+ if (topicMetadata.partitions > 0) {
+ log.warn("[{}] Not supported getMessageIdByIndex operation on "
+ + "partitioned-topic {}", clientAppId(), topicName);
+ throw new RestException(Status.METHOD_NOT_ALLOWED,
+ "GetMessageIDByIndex is not allowed on partitioned-topic");
+ }
+ });
+ }
+ }).thenCompose(ignore -> validateTopicOwnershipAsync(topicName, authoritative))
+ .thenCompose(__ -> getTopicReferenceAsync(topicName))
+ .thenCompose(topic -> {
+ if (!(topic instanceof PersistentTopic persistentTopic)) {
+ log.error("[{}] Get message id by index on a non-persistent topic {} is not allowed",
+ clientAppId(), topicName);
+ return FutureUtil.failedFuture(new RestException(Status.METHOD_NOT_ALLOWED,
+ "Get message id by index on a non-persistent topic is not allowed"));
+ }
+ ManagedLedger managedLedger = persistentTopic.getManagedLedger();
+ return findMessageIndexByPosition(
+ PositionFactory.create(managedLedger.getFirstPosition().getLedgerId(), 0),
Review Comment:
Why not just pass `managedLedger.getFirstPosition()` here?
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java:
##########
@@ -5489,4 +5489,114 @@ protected CompletableFuture<AutoSubscriptionCreationOverride> internalGetAutoSub
return null;
}));
}
+
+ protected CompletableFuture<MessageId> internalGetMessageIDByIndexAsync(Long index, boolean authoritative) {
+ if (!pulsar().getBrokerService().isBrokerEntryMetadataEnabled()) {
+ return FutureUtil.failedFuture(new RestException(Status.PRECONDITION_FAILED,
+ "GetMessageIDByIndex is not allowed when broker entry metadata is disabled"));
+ }
+ if (index == null || index < 0) {
+ return FutureUtil.failedFuture(new RestException(Status.NOT_FOUND,
+ "Invalid message index: " + index));
+ }
+ int partitionIndex = topicName.getPartitionIndex();
+ CompletableFuture<Void> future = validateTopicOperationAsync(topicName, TopicOperation.PEEK_MESSAGES);
+ return future.thenCompose(__ -> {
+ if (topicName.isGlobal()) {
+ return validateGlobalNamespaceOwnershipAsync(namespaceName);
+ } else {
+ return CompletableFuture.completedFuture(null);
+ }
+ }).thenCompose(__ -> {
+ if (topicName.isPartitioned()) {
+ return CompletableFuture.completedFuture(null);
+ } else {
+ return getPartitionedTopicMetadataAsync(topicName, authoritative, false)
+ .thenAccept(topicMetadata -> {
+ if (topicMetadata.partitions > 0) {
+ log.warn("[{}] Not supported getMessageIdByIndex operation on "
+ + "partitioned-topic {}", clientAppId(), topicName);
+ throw new RestException(Status.METHOD_NOT_ALLOWED,
+ "GetMessageIDByIndex is not allowed on partitioned-topic");
+ }
+ });
+ }
+ }).thenCompose(ignore -> validateTopicOwnershipAsync(topicName, authoritative))
+ .thenCompose(__ -> getTopicReferenceAsync(topicName))
+ .thenCompose(topic -> {
+ if (!(topic instanceof PersistentTopic persistentTopic)) {
+ log.error("[{}] Get message id by index on a non-persistent topic {} is not allowed",
+ clientAppId(), topicName);
+ return FutureUtil.failedFuture(new RestException(Status.METHOD_NOT_ALLOWED,
+ "Get message id by index on a non-persistent topic is not allowed"));
+ }
+ ManagedLedger managedLedger = persistentTopic.getManagedLedger();
+ return findMessageIndexByPosition(
+ PositionFactory.create(managedLedger.getFirstPosition().getLedgerId(), 0),
+ managedLedger)
+ .thenCompose(firstIndex -> {
+ if (index < firstIndex) {
+ return CompletableFuture.completedFuture(PositionFactory.EARLIEST);
+ } else {
+ return managedLedger.asyncFindPosition(entry -> {
+ try {
+ BrokerEntryMetadata brokerEntryMetadata =
+ Commands.parseBrokerEntryMetadataIfExist(entry.getDataBuffer());
+ // Skip messages without index
+ if (brokerEntryMetadata == null) {
+ return true;
+ }
+ return brokerEntryMetadata.getIndex() < index;
+ } catch (Exception e) {
+ log.error("Error deserialize message for message position find", e);
+ } finally {
+ entry.release();
+ }
+ return false;
+ });
+ }
+ }).thenCompose(position -> {
+ Position lastPosition = managedLedger.getLastConfirmedEntry();
+ if (position == null || position.compareTo(lastPosition) > 0) {
+ return FutureUtil.failedFuture(new RestException(Status.NOT_FOUND,
+ "Message not found for index " + index));
+ } else {
+ return CompletableFuture.completedFuture(position);
+ }
+ });
+ }).thenCompose(position -> CompletableFuture.completedFuture(
+ new MessageIdImpl(position.getLedgerId(), position.getEntryId(), partitionIndex)));
+ }
+
+ protected CompletableFuture<Long> findMessageIndexByPosition(Position position, ManagedLedger managedLedger) {
+ CompletableFuture<Long> indexFuture = new CompletableFuture<>();
+ managedLedger.asyncReadEntry(position, new AsyncCallbacks.ReadEntryCallback() {
+ @Override
+ public void readEntryComplete(Entry entry, Object ctx) {
+ BrokerEntryMetadata brokerEntryMetadata =
+ Commands.parseBrokerEntryMetadataIfExist(entry.getDataBuffer());
+ if (brokerEntryMetadata == null) {
+ indexFuture.completeExceptionally(new RestException(Status.PRECONDITION_FAILED,
+ "Broker entry metadata is not present in the message"));
+ } else {
+ long index = brokerEntryMetadata.getIndex();
+ if (index < 0) {
+ indexFuture.completeExceptionally(new RestException(Status.PRECONDITION_FAILED,
+ "Invalid message index: " + index));
+ } else {
+ indexFuture.complete(index);
+ }
+ }
Review Comment:
You must release `entry` here.
I think you can add a common method for both `asyncReadEntry` and `asyncFindPosition` to use. Here is an example:
```java
private static Optional<Long> getIndexAndRelease(Entry entry) throws IOException {
try {
final var brokerEntryMetadata = Commands.parseBrokerEntryMetadataIfExist(entry.getDataBuffer());
if (brokerEntryMetadata != null && brokerEntryMetadata.hasIndex()) {
return Optional.of(brokerEntryMetadata.getIndex());
} else {
return Optional.empty();
}
} catch (Throwable throwable) {
throw new IOException(throwable);
}
}
```
The implementation above forces the caller side to handle the two exceptional cases:
- `Optional.empty()` is returned, which indicates the broker entry metadata does not exist or the index does not exist (your code does not take it into consideration)
- The buffer is corrupted, where `IOException` will be thrown
But you can implement your own method if you care much about the overhead of `Optional` or don't want to use checked exception.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat][admin] PIP-415: Support getting message ID by index [pulsar]
Posted by "liangyepianzhou (via GitHub)" <gi...@apache.org>.
liangyepianzhou commented on code in PR #24222:
URL: https://github.com/apache/pulsar/pull/24222#discussion_r2125960979
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java:
##########
@@ -5489,4 +5489,131 @@ protected CompletableFuture<AutoSubscriptionCreationOverride> internalGetAutoSub
return null;
}));
}
+
+ protected CompletableFuture<MessageId> internalGetMessageIDByIndexAsync(Long index, boolean authoritative) {
+ if (!pulsar().getBrokerService().isBrokerEntryMetadataEnabled()) {
+ return FutureUtil.failedFuture(new RestException(Status.PRECONDITION_FAILED,
+ "GetMessageIDByIndex is not allowed when broker entry metadata is disabled"));
+ }
+ if (index == null || index < 0) {
+ return FutureUtil.failedFuture(new RestException(Status.NOT_FOUND,
+ "Invalid message index: " + index));
+ }
+ int partitionIndex = topicName.getPartitionIndex();
+ CompletableFuture<Void> future = validateTopicOperationAsync(topicName, TopicOperation.PEEK_MESSAGES);
+ return future.thenCompose(__ -> {
+ if (topicName.isGlobal()) {
+ return validateGlobalNamespaceOwnershipAsync(namespaceName);
+ } else {
+ return CompletableFuture.completedFuture(null);
+ }
+ }).thenCompose(__ -> {
+ if (topicName.isPartitioned()) {
+ return CompletableFuture.completedFuture(null);
+ } else {
+ return getPartitionedTopicMetadataAsync(topicName, authoritative, false)
+ .thenAccept(topicMetadata -> {
+ if (topicMetadata.partitions > 0) {
+ log.warn("[{}] Not supported getMessageIdByIndex operation on "
+ + "partitioned-topic {}", clientAppId(), topicName);
+ throw new RestException(Status.METHOD_NOT_ALLOWED,
+ "GetMessageIDByIndex is not allowed on partitioned-topic");
+ }
+ });
+ }
+ }).thenCompose(ignore -> validateTopicOwnershipAsync(topicName, authoritative))
+ .thenCompose(__ -> getTopicReferenceAsync(topicName))
+ .thenCompose(topic -> {
+ if (!(topic instanceof PersistentTopic persistentTopic)) {
+ log.error("[{}] Get message id by index on a non-persistent topic {} is not allowed",
+ clientAppId(), topicName);
+ return FutureUtil.failedFuture(new RestException(Status.METHOD_NOT_ALLOWED,
+ "Get message id by index on a non-persistent topic is not allowed"));
+ }
+ ManagedLedger managedLedger = persistentTopic.getManagedLedger();
+ return findMessageIndexByPosition(
+ PositionFactory.create(managedLedger.getFirstPosition().getLedgerId(), 0),
+ managedLedger)
+ .thenCompose(firstIndex -> {
+ if (index < firstIndex) {
+ return CompletableFuture.completedFuture(PositionFactory.EARLIEST);
Review Comment:
Can you explain why you think null is better than earliest?
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat][admin] PIP-415: Support getting message ID by index [pulsar]
Posted by "dao-jun (via GitHub)" <gi...@apache.org>.
dao-jun commented on code in PR #24222:
URL: https://github.com/apache/pulsar/pull/24222#discussion_r2126207218
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java:
##########
@@ -5489,4 +5489,131 @@ protected CompletableFuture<AutoSubscriptionCreationOverride> internalGetAutoSub
return null;
}));
}
+
+ protected CompletableFuture<MessageId> internalGetMessageIDByIndexAsync(Long index, boolean authoritative) {
+ if (!pulsar().getBrokerService().isBrokerEntryMetadataEnabled()) {
+ return FutureUtil.failedFuture(new RestException(Status.PRECONDITION_FAILED,
+ "GetMessageIDByIndex is not allowed when broker entry metadata is disabled"));
+ }
+ if (index == null || index < 0) {
+ return FutureUtil.failedFuture(new RestException(Status.NOT_FOUND,
+ "Invalid message index: " + index));
+ }
+ int partitionIndex = topicName.getPartitionIndex();
+ CompletableFuture<Void> future = validateTopicOperationAsync(topicName, TopicOperation.PEEK_MESSAGES);
+ return future.thenCompose(__ -> {
+ if (topicName.isGlobal()) {
+ return validateGlobalNamespaceOwnershipAsync(namespaceName);
+ } else {
+ return CompletableFuture.completedFuture(null);
+ }
+ }).thenCompose(__ -> {
+ if (topicName.isPartitioned()) {
+ return CompletableFuture.completedFuture(null);
+ } else {
+ return getPartitionedTopicMetadataAsync(topicName, authoritative, false)
+ .thenAccept(topicMetadata -> {
+ if (topicMetadata.partitions > 0) {
+ log.warn("[{}] Not supported getMessageIdByIndex operation on "
+ + "partitioned-topic {}", clientAppId(), topicName);
+ throw new RestException(Status.METHOD_NOT_ALLOWED,
+ "GetMessageIDByIndex is not allowed on partitioned-topic");
+ }
+ });
+ }
+ }).thenCompose(ignore -> validateTopicOwnershipAsync(topicName, authoritative))
+ .thenCompose(__ -> getTopicReferenceAsync(topicName))
+ .thenCompose(topic -> {
+ if (!(topic instanceof PersistentTopic persistentTopic)) {
+ log.error("[{}] Get message id by index on a non-persistent topic {} is not allowed",
+ clientAppId(), topicName);
+ return FutureUtil.failedFuture(new RestException(Status.METHOD_NOT_ALLOWED,
+ "Get message id by index on a non-persistent topic is not allowed"));
+ }
+ ManagedLedger managedLedger = persistentTopic.getManagedLedger();
+ return findMessageIndexByPosition(
+ PositionFactory.create(managedLedger.getFirstPosition().getLedgerId(), 0),
+ managedLedger)
+ .thenCompose(firstIndex -> {
+ if (index < firstIndex) {
+ return CompletableFuture.completedFuture(PositionFactory.EARLIEST);
Review Comment:
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:
##########
@@ -5489,4 +5489,131 @@ protected CompletableFuture<AutoSubscriptionCreationOverride> internalGetAutoSub
return null;
}));
}
+
+ protected CompletableFuture<MessageId> internalGetMessageIDByIndexAsync(Long index, boolean authoritative) {
+ if (!pulsar().getBrokerService().isBrokerEntryMetadataEnabled()) {
+ return FutureUtil.failedFuture(new RestException(Status.PRECONDITION_FAILED,
+ "GetMessageIDByIndex is not allowed when broker entry metadata is disabled"));
+ }
+ if (index == null || index < 0) {
+ return FutureUtil.failedFuture(new RestException(Status.NOT_FOUND,
+ "Invalid message index: " + index));
+ }
+ int partitionIndex = topicName.getPartitionIndex();
+ CompletableFuture<Void> future = validateTopicOperationAsync(topicName, TopicOperation.PEEK_MESSAGES);
+ return future.thenCompose(__ -> {
+ if (topicName.isGlobal()) {
+ return validateGlobalNamespaceOwnershipAsync(namespaceName);
+ } else {
+ return CompletableFuture.completedFuture(null);
+ }
+ }).thenCompose(__ -> {
+ if (topicName.isPartitioned()) {
+ return CompletableFuture.completedFuture(null);
+ } else {
+ return getPartitionedTopicMetadataAsync(topicName, authoritative, false)
+ .thenAccept(topicMetadata -> {
+ if (topicMetadata.partitions > 0) {
+ log.warn("[{}] Not supported getMessageIdByIndex operation on "
+ + "partitioned-topic {}", clientAppId(), topicName);
+ throw new RestException(Status.METHOD_NOT_ALLOWED,
+ "GetMessageIDByIndex is not allowed on partitioned-topic");
+ }
+ });
+ }
+ }).thenCompose(ignore -> validateTopicOwnershipAsync(topicName, authoritative))
+ .thenCompose(__ -> getTopicReferenceAsync(topicName))
+ .thenCompose(topic -> {
+ if (!(topic instanceof PersistentTopic persistentTopic)) {
+ log.error("[{}] Get message id by index on a non-persistent topic {} is not allowed",
+ clientAppId(), topicName);
+ return FutureUtil.failedFuture(new RestException(Status.METHOD_NOT_ALLOWED,
+ "Get message id by index on a non-persistent topic is not allowed"));
+ }
+ ManagedLedger managedLedger = persistentTopic.getManagedLedger();
+ return findMessageIndexByPosition(
+ PositionFactory.create(managedLedger.getFirstPosition().getLedgerId(), 0),
+ managedLedger)
+ .thenCompose(firstIndex -> {
+ if (index < firstIndex) {
+ return CompletableFuture.completedFuture(PositionFactory.EARLIEST);
+ } else {
+ return managedLedger.asyncFindPosition(entry -> {
+ try {
+ Long messageIndex = getIndexFromEntry(entry);
+ if (messageIndex == null) {
+ return false; // Skip messages without index
+ } else {
+ // If the message index is less than the requested index,
+ // we continue searching
+ return messageIndex < index;
+ }
+ } catch (IOException e) {
+ log.error("Error deserialize message for message position find", e);
+ return false;
Review Comment:
return false is OK, same as KoP
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java:
##########
@@ -5489,4 +5489,131 @@ protected CompletableFuture<AutoSubscriptionCreationOverride> internalGetAutoSub
return null;
}));
}
+
+ protected CompletableFuture<MessageId> internalGetMessageIDByIndexAsync(Long index, boolean authoritative) {
+ if (!pulsar().getBrokerService().isBrokerEntryMetadataEnabled()) {
+ return FutureUtil.failedFuture(new RestException(Status.PRECONDITION_FAILED,
+ "GetMessageIDByIndex is not allowed when broker entry metadata is disabled"));
+ }
+ if (index == null || index < 0) {
+ return FutureUtil.failedFuture(new RestException(Status.NOT_FOUND,
+ "Invalid message index: " + index));
+ }
+ int partitionIndex = topicName.getPartitionIndex();
+ CompletableFuture<Void> future = validateTopicOperationAsync(topicName, TopicOperation.PEEK_MESSAGES);
+ return future.thenCompose(__ -> {
+ if (topicName.isGlobal()) {
+ return validateGlobalNamespaceOwnershipAsync(namespaceName);
+ } else {
+ return CompletableFuture.completedFuture(null);
+ }
+ }).thenCompose(__ -> {
+ if (topicName.isPartitioned()) {
+ return CompletableFuture.completedFuture(null);
+ } else {
+ return getPartitionedTopicMetadataAsync(topicName, authoritative, false)
+ .thenAccept(topicMetadata -> {
+ if (topicMetadata.partitions > 0) {
+ log.warn("[{}] Not supported getMessageIdByIndex operation on "
+ + "partitioned-topic {}", clientAppId(), topicName);
+ throw new RestException(Status.METHOD_NOT_ALLOWED,
+ "GetMessageIDByIndex is not allowed on partitioned-topic");
+ }
+ });
+ }
+ }).thenCompose(ignore -> validateTopicOwnershipAsync(topicName, authoritative))
+ .thenCompose(__ -> getTopicReferenceAsync(topicName))
+ .thenCompose(topic -> {
+ if (!(topic instanceof PersistentTopic persistentTopic)) {
+ log.error("[{}] Get message id by index on a non-persistent topic {} is not allowed",
+ clientAppId(), topicName);
+ return FutureUtil.failedFuture(new RestException(Status.METHOD_NOT_ALLOWED,
+ "Get message id by index on a non-persistent topic is not allowed"));
+ }
+ ManagedLedger managedLedger = persistentTopic.getManagedLedger();
+ return findMessageIndexByPosition(
+ PositionFactory.create(managedLedger.getFirstPosition().getLedgerId(), 0),
+ managedLedger)
+ .thenCompose(firstIndex -> {
+ if (index < firstIndex) {
+ return CompletableFuture.completedFuture(PositionFactory.EARLIEST);
+ } else {
+ return managedLedger.asyncFindPosition(entry -> {
+ try {
+ Long messageIndex = getIndexFromEntry(entry);
+ if (messageIndex == null) {
+ return false; // Skip messages without index
+ } else {
+ // If the message index is less than the requested index,
+ // we continue searching
+ return messageIndex < index;
+ }
+ } catch (IOException e) {
+ log.error("Error deserialize message for message position find", e);
+ return false;
+ } finally {
+ entry.release();
+ }
+ });
+ }
+ }).thenCompose(position -> {
+ Position lastPosition = managedLedger.getLastConfirmedEntry();
+ if (position == null || position.compareTo(lastPosition) > 0) {
+ return FutureUtil.failedFuture(new RestException(Status.NOT_FOUND,
+ "Message not found for index " + index));
+ } else {
+ return CompletableFuture.completedFuture(position);
+ }
+ });
+ }).thenCompose(position -> CompletableFuture.completedFuture(
+ new MessageIdImpl(position.getLedgerId(), position.getEntryId(), partitionIndex)));
+ }
+
+ protected CompletableFuture<Long> findMessageIndexByPosition(Position position, ManagedLedger managedLedger) {
+ CompletableFuture<Long> indexFuture = new CompletableFuture<>();
+ managedLedger.asyncReadEntry(position, new AsyncCallbacks.ReadEntryCallback() {
+ @Override
+ public void readEntryComplete(Entry entry, Object ctx) {
+ try {
+ Long index = getIndexFromEntry(entry);
+ if (index == null) {
+ indexFuture.completeExceptionally(new RestException(Status.PRECONDITION_FAILED,
+ "Broker entry metadata is not present in the message"));
+ } else if (index < 0) {
+ indexFuture.completeExceptionally(new RestException(Status.PRECONDITION_FAILED,
+ "Invalid message index: " + index));
+ } else {
+ indexFuture.complete(index);
+ }
+ } catch (IOException e) {
+ indexFuture.completeExceptionally(new RestException(Status.INTERNAL_SERVER_ERROR,
+ "Failed to get index from entry: " + e.getMessage()));
+ } finally {
+ entry.release();
+ }
+ }
+
+ @Override
+ public void readEntryFailed(ManagedLedgerException exception, Object ctx) {
+ log.error("[{}] Failed to read position {} on topic {}",
+ clientAppId(), position, topicName, exception);
+ indexFuture.completeExceptionally(exception);
+ }
+ }, null);
+ return indexFuture;
+ }
+
+
+ private static Long getIndexFromEntry(Entry entry) throws IOException {
+ try {
+ final var brokerEntryMetadata = Commands.parseBrokerEntryMetadataIfExist(entry.getDataBuffer());
+ if (brokerEntryMetadata != null && brokerEntryMetadata.hasIndex()) {
+ return brokerEntryMetadata.getIndex();
+ } else {
+ return null;
+ }
+ } catch (Throwable throwable) {
+ throw new IOException(throwable);
Review Comment:
It maybe throws MessageConrruptException, no need to wrap it to IOException. agree with @nodece
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat][admin] PIP-415: Support getting message ID by index [pulsar]
Posted by "liangyepianzhou (via GitHub)" <gi...@apache.org>.
liangyepianzhou commented on code in PR #24222:
URL: https://github.com/apache/pulsar/pull/24222#discussion_r2126333375
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java:
##########
@@ -5489,4 +5489,131 @@ protected CompletableFuture<AutoSubscriptionCreationOverride> internalGetAutoSub
return null;
}));
}
+
+ protected CompletableFuture<MessageId> internalGetMessageIDByIndexAsync(Long index, boolean authoritative) {
+ if (!pulsar().getBrokerService().isBrokerEntryMetadataEnabled()) {
+ return FutureUtil.failedFuture(new RestException(Status.PRECONDITION_FAILED,
+ "GetMessageIDByIndex is not allowed when broker entry metadata is disabled"));
+ }
+ if (index == null || index < 0) {
+ return FutureUtil.failedFuture(new RestException(Status.NOT_FOUND,
+ "Invalid message index: " + index));
+ }
+ int partitionIndex = topicName.getPartitionIndex();
+ CompletableFuture<Void> future = validateTopicOperationAsync(topicName, TopicOperation.PEEK_MESSAGES);
+ return future.thenCompose(__ -> {
+ if (topicName.isGlobal()) {
+ return validateGlobalNamespaceOwnershipAsync(namespaceName);
+ } else {
+ return CompletableFuture.completedFuture(null);
+ }
+ }).thenCompose(__ -> {
+ if (topicName.isPartitioned()) {
+ return CompletableFuture.completedFuture(null);
+ } else {
+ return getPartitionedTopicMetadataAsync(topicName, authoritative, false)
+ .thenAccept(topicMetadata -> {
+ if (topicMetadata.partitions > 0) {
+ log.warn("[{}] Not supported getMessageIdByIndex operation on "
+ + "partitioned-topic {}", clientAppId(), topicName);
+ throw new RestException(Status.METHOD_NOT_ALLOWED,
+ "GetMessageIDByIndex is not allowed on partitioned-topic");
+ }
+ });
+ }
+ }).thenCompose(ignore -> validateTopicOwnershipAsync(topicName, authoritative))
+ .thenCompose(__ -> getTopicReferenceAsync(topicName))
+ .thenCompose(topic -> {
+ if (!(topic instanceof PersistentTopic persistentTopic)) {
+ log.error("[{}] Get message id by index on a non-persistent topic {} is not allowed",
+ clientAppId(), topicName);
+ return FutureUtil.failedFuture(new RestException(Status.METHOD_NOT_ALLOWED,
+ "Get message id by index on a non-persistent topic is not allowed"));
+ }
+ ManagedLedger managedLedger = persistentTopic.getManagedLedger();
+ return findMessageIndexByPosition(
+ PositionFactory.create(managedLedger.getFirstPosition().getLedgerId(), 0),
+ managedLedger)
+ .thenCompose(firstIndex -> {
+ if (index < firstIndex) {
+ return CompletableFuture.completedFuture(PositionFactory.EARLIEST);
+ } else {
+ return managedLedger.asyncFindPosition(entry -> {
+ try {
+ Long messageIndex = getIndexFromEntry(entry);
+ if (messageIndex == null) {
+ return false; // Skip messages without index
+ } else {
+ // If the message index is less than the requested index,
+ // we continue searching
+ return messageIndex < index;
+ }
+ } catch (IOException e) {
+ log.error("Error deserialize message for message position find", e);
+ return false;
+ } finally {
+ entry.release();
+ }
+ });
+ }
+ }).thenCompose(position -> {
+ Position lastPosition = managedLedger.getLastConfirmedEntry();
+ if (position == null || position.compareTo(lastPosition) > 0) {
+ return FutureUtil.failedFuture(new RestException(Status.NOT_FOUND,
+ "Message not found for index " + index));
+ } else {
+ return CompletableFuture.completedFuture(position);
+ }
+ });
+ }).thenCompose(position -> CompletableFuture.completedFuture(
+ new MessageIdImpl(position.getLedgerId(), position.getEntryId(), partitionIndex)));
+ }
+
+ protected CompletableFuture<Long> findMessageIndexByPosition(Position position, ManagedLedger managedLedger) {
+ CompletableFuture<Long> indexFuture = new CompletableFuture<>();
+ managedLedger.asyncReadEntry(position, new AsyncCallbacks.ReadEntryCallback() {
+ @Override
+ public void readEntryComplete(Entry entry, Object ctx) {
+ try {
+ Long index = getIndexFromEntry(entry);
+ if (index == null) {
+ indexFuture.completeExceptionally(new RestException(Status.PRECONDITION_FAILED,
+ "Broker entry metadata is not present in the message"));
+ } else if (index < 0) {
+ indexFuture.completeExceptionally(new RestException(Status.PRECONDITION_FAILED,
+ "Invalid message index: " + index));
+ } else {
+ indexFuture.complete(index);
+ }
+ } catch (IOException e) {
+ indexFuture.completeExceptionally(new RestException(Status.INTERNAL_SERVER_ERROR,
+ "Failed to get index from entry: " + e.getMessage()));
+ } finally {
+ entry.release();
+ }
+ }
+
+ @Override
+ public void readEntryFailed(ManagedLedgerException exception, Object ctx) {
+ log.error("[{}] Failed to read position {} on topic {}",
+ clientAppId(), position, topicName, exception);
+ indexFuture.completeExceptionally(exception);
+ }
+ }, null);
+ return indexFuture;
+ }
+
+
+ private static Long getIndexFromEntry(Entry entry) throws IOException {
+ try {
+ final var brokerEntryMetadata = Commands.parseBrokerEntryMetadataIfExist(entry.getDataBuffer());
+ if (brokerEntryMetadata != null && brokerEntryMetadata.hasIndex()) {
+ return brokerEntryMetadata.getIndex();
+ } else {
+ return null;
+ }
+ } catch (Throwable throwable) {
+ throw new IOException(throwable);
Review Comment:
Removed exception signature
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat][admin] PIP-415: Support getting message ID by offset [pulsar]
Posted by "StevenLuMT (via GitHub)" <gi...@apache.org>.
StevenLuMT commented on code in PR #24222:
URL: https://github.com/apache/pulsar/pull/24222#discussion_r2062779057
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java:
##########
@@ -5489,4 +5489,53 @@ protected CompletableFuture<AutoSubscriptionCreationOverride> internalGetAutoSub
return null;
}));
}
+
+ protected CompletableFuture<MessageId> internalGetMessageIDByOffsetAsync(Long offset, boolean authoritative) {
+ int partitionIndex = topicName.getPartitionIndex();
+ CompletableFuture<Void> future = validateTopicOperationAsync(topicName, TopicOperation.PEEK_MESSAGES);
+ return future.thenCompose(__ -> {
+ if (topicName.isGlobal()) {
+ return validateGlobalNamespaceOwnershipAsync(namespaceName);
+ } else {
+ return CompletableFuture.completedFuture(null);
+ }
+ }).thenCompose(__ -> {
+ if (topicName.isPartitioned()) {
+ return CompletableFuture.completedFuture(null);
+ } else {
+ return getPartitionedTopicMetadataAsync(topicName, authoritative, false)
+ .thenAccept(topicMetadata -> {
+ if (topicMetadata.partitions > 0) {
+ log.warn("[{}] Not supported getMessageIdByOffset operation on "
+ + "partitioned-topic {}", clientAppId(), topicName);
+ throw new RestException(Status.METHOD_NOT_ALLOWED,
+ "GetMessageIDByOffset is not allowed on partitioned-topic");
+ }
+ });
+ }
+ }).thenCompose(ignore -> validateTopicOwnershipAsync(topicName, authoritative))
+ .thenCompose(__ -> getTopicReferenceAsync(topicName))
+ .thenCompose(topic -> {
+ if (!(topic instanceof PersistentTopic persistentTopic)) {
+ log.error("[{}] Get message id by offset on a non-persistent topic {} is not allowed",
+ clientAppId(), topicName);
+ return FutureUtil.failedFuture(new RestException(Status.METHOD_NOT_ALLOWED,
+ "Get message id by offset on a non-persistent topic is not allowed"));
+ }
+ ManagedLedger managedLedger = persistentTopic.getManagedLedger();
+ return managedLedger.asyncFindPosition(entry -> {
+ try {
+ BrokerEntryMetadata brokerEntryMetadata =
+ Commands.parseBrokerEntryMetadataIfExist(entry.getDataBuffer());
+ return brokerEntryMetadata.getIndex() < offset;
+ } catch (Exception e) {
Review Comment:
Need to determine whether brokerEntryMetadata is empty?
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat][admin] PIP-415: Support getting message ID by index [pulsar]
Posted by "liangyepianzhou (via GitHub)" <gi...@apache.org>.
liangyepianzhou merged PR #24222:
URL: https://github.com/apache/pulsar/pull/24222
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat][admin] PIP-415: Support getting message ID by index [pulsar]
Posted by "StevenLuMT (via GitHub)" <gi...@apache.org>.
StevenLuMT commented on code in PR #24222:
URL: https://github.com/apache/pulsar/pull/24222#discussion_r2125786313
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java:
##########
@@ -5489,4 +5489,129 @@ protected CompletableFuture<AutoSubscriptionCreationOverride> internalGetAutoSub
return null;
}));
}
+
+ protected CompletableFuture<MessageId> internalGetMessageIDByIndexAsync(Long index, boolean authoritative) {
+ if (!pulsar().getBrokerService().isBrokerEntryMetadataEnabled()) {
+ return FutureUtil.failedFuture(new RestException(Status.PRECONDITION_FAILED,
+ "GetMessageIDByIndex is not allowed when broker entry metadata is disabled"));
+ }
+ if (index == null || index < 0) {
+ return FutureUtil.failedFuture(new RestException(Status.NOT_FOUND,
+ "Invalid message index: " + index));
+ }
+ int partitionIndex = topicName.getPartitionIndex();
+ CompletableFuture<Void> future = validateTopicOperationAsync(topicName, TopicOperation.PEEK_MESSAGES);
+ return future.thenCompose(__ -> {
+ if (topicName.isGlobal()) {
+ return validateGlobalNamespaceOwnershipAsync(namespaceName);
+ } else {
+ return CompletableFuture.completedFuture(null);
+ }
+ }).thenCompose(__ -> {
+ if (topicName.isPartitioned()) {
+ return CompletableFuture.completedFuture(null);
+ } else {
+ return getPartitionedTopicMetadataAsync(topicName, authoritative, false)
+ .thenAccept(topicMetadata -> {
+ if (topicMetadata.partitions > 0) {
+ log.warn("[{}] Not supported getMessageIdByIndex operation on "
+ + "partitioned-topic {}", clientAppId(), topicName);
+ throw new RestException(Status.METHOD_NOT_ALLOWED,
+ "GetMessageIDByIndex is not allowed on partitioned-topic");
+ }
+ });
+ }
+ }).thenCompose(ignore -> validateTopicOwnershipAsync(topicName, authoritative))
+ .thenCompose(__ -> getTopicReferenceAsync(topicName))
+ .thenCompose(topic -> {
+ if (!(topic instanceof PersistentTopic persistentTopic)) {
+ log.error("[{}] Get message id by index on a non-persistent topic {} is not allowed",
+ clientAppId(), topicName);
+ return FutureUtil.failedFuture(new RestException(Status.METHOD_NOT_ALLOWED,
+ "Get message id by index on a non-persistent topic is not allowed"));
+ }
+ ManagedLedger managedLedger = persistentTopic.getManagedLedger();
+ return findMessageIndexByPosition(
+ PositionFactory.create(managedLedger.getFirstPosition().getLedgerId(), 0),
+ managedLedger)
+ .thenCompose(firstIndex -> {
+ if (index < firstIndex) {
+ return CompletableFuture.completedFuture(PositionFactory.EARLIEST);
+ } else {
+ return managedLedger.asyncFindPosition(entry -> {
+ try {
+ Long messageIndex = getIndexAndRelease(entry);
+ if (messageIndex == null) {
+ return false; // Skip messages without index
+ } else {
+ // If the message index is less than the requested index,
+ // we continue searching
+ return messageIndex < index;
+ }
+ } catch (IOException e) {
+ log.error("Error deserialize message for message position find", e);
+ return false;
+ }
+ });
+ }
+ }).thenCompose(position -> {
+ Position lastPosition = managedLedger.getLastConfirmedEntry();
+ if (position == null || position.compareTo(lastPosition) > 0) {
+ return FutureUtil.failedFuture(new RestException(Status.NOT_FOUND,
Review Comment:
position.compareTo(lastPosition) > 0
When does this happen? Is it normal?
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat][admin] PIP-415: Support getting message ID by index [pulsar]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on code in PR #24222:
URL: https://github.com/apache/pulsar/pull/24222#discussion_r2126217163
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java:
##########
@@ -5489,4 +5489,131 @@ protected CompletableFuture<AutoSubscriptionCreationOverride> internalGetAutoSub
return null;
}));
}
+
+ protected CompletableFuture<MessageId> internalGetMessageIDByIndexAsync(Long index, boolean authoritative) {
+ if (!pulsar().getBrokerService().isBrokerEntryMetadataEnabled()) {
+ return FutureUtil.failedFuture(new RestException(Status.PRECONDITION_FAILED,
+ "GetMessageIDByIndex is not allowed when broker entry metadata is disabled"));
+ }
+ if (index == null || index < 0) {
+ return FutureUtil.failedFuture(new RestException(Status.NOT_FOUND,
+ "Invalid message index: " + index));
+ }
+ int partitionIndex = topicName.getPartitionIndex();
+ CompletableFuture<Void> future = validateTopicOperationAsync(topicName, TopicOperation.PEEK_MESSAGES);
+ return future.thenCompose(__ -> {
+ if (topicName.isGlobal()) {
+ return validateGlobalNamespaceOwnershipAsync(namespaceName);
+ } else {
+ return CompletableFuture.completedFuture(null);
+ }
+ }).thenCompose(__ -> {
+ if (topicName.isPartitioned()) {
+ return CompletableFuture.completedFuture(null);
+ } else {
+ return getPartitionedTopicMetadataAsync(topicName, authoritative, false)
+ .thenAccept(topicMetadata -> {
+ if (topicMetadata.partitions > 0) {
+ log.warn("[{}] Not supported getMessageIdByIndex operation on "
+ + "partitioned-topic {}", clientAppId(), topicName);
+ throw new RestException(Status.METHOD_NOT_ALLOWED,
+ "GetMessageIDByIndex is not allowed on partitioned-topic");
+ }
+ });
+ }
+ }).thenCompose(ignore -> validateTopicOwnershipAsync(topicName, authoritative))
+ .thenCompose(__ -> getTopicReferenceAsync(topicName))
+ .thenCompose(topic -> {
+ if (!(topic instanceof PersistentTopic persistentTopic)) {
+ log.error("[{}] Get message id by index on a non-persistent topic {} is not allowed",
+ clientAppId(), topicName);
+ return FutureUtil.failedFuture(new RestException(Status.METHOD_NOT_ALLOWED,
+ "Get message id by index on a non-persistent topic is not allowed"));
+ }
+ ManagedLedger managedLedger = persistentTopic.getManagedLedger();
+ return findMessageIndexByPosition(
+ PositionFactory.create(managedLedger.getFirstPosition().getLedgerId(), 0),
+ managedLedger)
+ .thenCompose(firstIndex -> {
+ if (index < firstIndex) {
+ return CompletableFuture.completedFuture(PositionFactory.EARLIEST);
+ } else {
+ return managedLedger.asyncFindPosition(entry -> {
+ try {
+ Long messageIndex = getIndexFromEntry(entry);
+ if (messageIndex == null) {
+ return false; // Skip messages without index
+ } else {
+ // If the message index is less than the requested index,
+ // we continue searching
+ return messageIndex < index;
+ }
+ } catch (IOException e) {
+ log.error("Error deserialize message for message position find", e);
+ return false;
+ } finally {
+ entry.release();
+ }
+ });
+ }
+ }).thenCompose(position -> {
+ Position lastPosition = managedLedger.getLastConfirmedEntry();
+ if (position == null || position.compareTo(lastPosition) > 0) {
+ return FutureUtil.failedFuture(new RestException(Status.NOT_FOUND,
+ "Message not found for index " + index));
+ } else {
+ return CompletableFuture.completedFuture(position);
+ }
+ });
+ }).thenCompose(position -> CompletableFuture.completedFuture(
+ new MessageIdImpl(position.getLedgerId(), position.getEntryId(), partitionIndex)));
+ }
+
+ protected CompletableFuture<Long> findMessageIndexByPosition(Position position, ManagedLedger managedLedger) {
+ CompletableFuture<Long> indexFuture = new CompletableFuture<>();
+ managedLedger.asyncReadEntry(position, new AsyncCallbacks.ReadEntryCallback() {
+ @Override
+ public void readEntryComplete(Entry entry, Object ctx) {
+ try {
+ Long index = getIndexFromEntry(entry);
+ if (index == null) {
+ indexFuture.completeExceptionally(new RestException(Status.PRECONDITION_FAILED,
+ "Broker entry metadata is not present in the message"));
+ } else if (index < 0) {
+ indexFuture.completeExceptionally(new RestException(Status.PRECONDITION_FAILED,
+ "Invalid message index: " + index));
+ } else {
+ indexFuture.complete(index);
+ }
+ } catch (IOException e) {
+ indexFuture.completeExceptionally(new RestException(Status.INTERNAL_SERVER_ERROR,
+ "Failed to get index from entry: " + e.getMessage()));
+ } finally {
+ entry.release();
+ }
+ }
+
+ @Override
+ public void readEntryFailed(ManagedLedgerException exception, Object ctx) {
+ log.error("[{}] Failed to read position {} on topic {}",
+ clientAppId(), position, topicName, exception);
+ indexFuture.completeExceptionally(exception);
+ }
+ }, null);
+ return indexFuture;
+ }
+
+
+ private static Long getIndexFromEntry(Entry entry) throws IOException {
+ try {
+ final var brokerEntryMetadata = Commands.parseBrokerEntryMetadataIfExist(entry.getDataBuffer());
+ if (brokerEntryMetadata != null && brokerEntryMetadata.hasIndex()) {
+ return brokerEntryMetadata.getIndex();
+ } else {
+ return null;
+ }
+ } catch (Throwable throwable) {
+ throw new IOException(throwable);
Review Comment:
The benefit of checked exceptions is to force the caller to handle the exception, which should be a "known" exception. However, the `IOException` thrown in this method is just a wrapper on an "unknown" exception.
A normal practice of exception handling should be:
- Fail fast or propagate the exception for unknown exceptions, because the caller never knows how to handle it correctly
- Handle the known exception according to the method's API docs
Specifically, for unknown exceptions, if propagating it could bring unexpected result, we should catch `Throwable` and handle the case well, for example,
```java
@RequiredArgsConstructor
public class Main {
private final Function<Integer, Integer> func;
public CompletableFuture<Integer> f(int x) {
final var future = new CompletableFuture<Integer>();
new Thread(() -> future.complete(func.apply(x))).start();
return future;
}
public static void main(String[] args) {
final var main = new Main(x -> 1 / x);
final var result = main.f(0).join();
System.out.println(result);
}
}
```
To prevent the future of `Main#f` from never being completed, we should add a catch clause for it:
```java
try {
future.complete(func.apply(x));
} catch (Throwable throwable) {
future.completeExceptionally(throwable);
}
```
But adding an additional method with exception signature is bad:
```java
public CompletableFuture<Integer> f(int x) {
final var future = new CompletableFuture<Integer>();
new Thread(() -> {
try {
future.complete(apply(x));
} catch (IOException e) {
future.completeExceptionally(e);
}
}).start();
return future;
}
public int apply(int x) throws IOException {
try {
return func.apply(x);
} catch (Throwable throwable) {
throw new IOException(throwable);
}
}
```
In this case, it's better to keep `getIndexFromEntry` simple without the exception signature and catch the `Throwable` when calling it like the suggestion [here](https://github.com/apache/pulsar/pull/24222/files#r2125780470).
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat][admin] PIP-415: Support getting message ID by index [pulsar]
Posted by "liangyepianzhou (via GitHub)" <gi...@apache.org>.
liangyepianzhou commented on code in PR #24222:
URL: https://github.com/apache/pulsar/pull/24222#discussion_r2122515338
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java:
##########
@@ -5489,4 +5489,114 @@ protected CompletableFuture<AutoSubscriptionCreationOverride> internalGetAutoSub
return null;
}));
}
+
+ protected CompletableFuture<MessageId> internalGetMessageIDByIndexAsync(Long index, boolean authoritative) {
+ if (!pulsar().getBrokerService().isBrokerEntryMetadataEnabled()) {
+ return FutureUtil.failedFuture(new RestException(Status.PRECONDITION_FAILED,
+ "GetMessageIDByIndex is not allowed when broker entry metadata is disabled"));
+ }
+ if (index == null || index < 0) {
+ return FutureUtil.failedFuture(new RestException(Status.NOT_FOUND,
+ "Invalid message index: " + index));
+ }
+ int partitionIndex = topicName.getPartitionIndex();
+ CompletableFuture<Void> future = validateTopicOperationAsync(topicName, TopicOperation.PEEK_MESSAGES);
+ return future.thenCompose(__ -> {
+ if (topicName.isGlobal()) {
+ return validateGlobalNamespaceOwnershipAsync(namespaceName);
+ } else {
+ return CompletableFuture.completedFuture(null);
+ }
+ }).thenCompose(__ -> {
+ if (topicName.isPartitioned()) {
+ return CompletableFuture.completedFuture(null);
+ } else {
+ return getPartitionedTopicMetadataAsync(topicName, authoritative, false)
+ .thenAccept(topicMetadata -> {
+ if (topicMetadata.partitions > 0) {
+ log.warn("[{}] Not supported getMessageIdByIndex operation on "
+ + "partitioned-topic {}", clientAppId(), topicName);
+ throw new RestException(Status.METHOD_NOT_ALLOWED,
+ "GetMessageIDByIndex is not allowed on partitioned-topic");
+ }
+ });
+ }
+ }).thenCompose(ignore -> validateTopicOwnershipAsync(topicName, authoritative))
+ .thenCompose(__ -> getTopicReferenceAsync(topicName))
+ .thenCompose(topic -> {
+ if (!(topic instanceof PersistentTopic persistentTopic)) {
+ log.error("[{}] Get message id by index on a non-persistent topic {} is not allowed",
+ clientAppId(), topicName);
+ return FutureUtil.failedFuture(new RestException(Status.METHOD_NOT_ALLOWED,
+ "Get message id by index on a non-persistent topic is not allowed"));
+ }
+ ManagedLedger managedLedger = persistentTopic.getManagedLedger();
+ return findMessageIndexByPosition(
+ PositionFactory.create(managedLedger.getFirstPosition().getLedgerId(), 0),
Review Comment:
IMO, the method `findMessageIndexByPosition` may be reused in the future, so it need a param `position` instead of only managerLedger
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat][admin] PIP-415: Support getting message ID by index [pulsar]
Posted by "nodece (via GitHub)" <gi...@apache.org>.
nodece commented on code in PR #24222:
URL: https://github.com/apache/pulsar/pull/24222#discussion_r2125921367
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerEntryMetadataE2ETest.java:
##########
@@ -420,4 +424,74 @@ public void testManagedLedgerTotalSize() throws Exception {
cursor.close();
}
+
+ @Test
+ public void testGetMessageIdByIndex() throws Exception {
+ // 1. test no partitioned topic
+ final String topicName = newTopicName();
+ admin.topics().createNonPartitionedTopic(topicName);
+ @Cleanup
+ Producer<String> producer = pulsarClient.newProducer(Schema.STRING)
+ .topic(topicName)
+ .enableBatching(false)
+ .create();
+ MessageIdImpl messageId = (MessageIdImpl) producer.send("test");
+ Message<byte[]>
+ message =
+ admin.topics().getMessagesById(topicName, messageId.getLedgerId(), messageId.getEntryId()).get(0);
+ long index = message.getIndex().get();
+ MessageIdImpl messageIdByIndex = (MessageIdImpl) admin.topics().getMessageIdByIndex(topicName, index);
+ Assert.assertEquals(messageIdByIndex, messageId);
+
+ // 2. test partitioned topic
+ final String topicName2 = newTopicName();
+ final String partitionedTopicName = topicName2 + "-partition-" + 0;
+ admin.topics().createPartitionedTopic(topicName2, 10);
+ @Cleanup
+ Producer<String> producer2 = pulsarClient.newProducer(Schema.STRING)
+ .topic(topicName2)
+ .enableBatching(false)
+ .create();
+
+ MessageIdImpl messageId2 = null;
+ for (int i = 0; i < 200; i++) {
+ messageId2 = (MessageIdImpl) producer2.send("test" + i);
+ if (messageId2.getPartitionIndex() == 0) {
+ break;
+ }
+ }
+ Message<byte[]>
+ message2 = admin.topics().getMessagesById(partitionedTopicName,
+ messageId2.getLedgerId(), messageId2.getEntryId()).get(0);
+ long index2 = message2.getIndex().get();
+ // 2.1 test partitioned topic name with partition index
+ MessageIdImpl messageIdByIndex2 =
+ (MessageIdImpl) admin.topics().getMessageIdByIndex(partitionedTopicName, index2);
+ Assert.assertEquals(messageIdByIndex2, messageId2);
+ // 2.2 test partitioned topic name without partition index
+ try {
+ messageIdByIndex2 = (MessageIdImpl) admin.topics().getMessageIdByIndex(topicName2, index2);
+ fail("Should not be able to get messageId by index for partitioned topic without partition index");
+ } catch (PulsarAdminException e) {
Review Comment:
`expectThrows` is a better way.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat][admin] PIP-415: Support getting message ID by offset [pulsar]
Posted by "AuroraTwinkle (via GitHub)" <gi...@apache.org>.
AuroraTwinkle commented on code in PR #24222:
URL: https://github.com/apache/pulsar/pull/24222#discussion_r2063082423
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java:
##########
@@ -5489,4 +5489,53 @@ protected CompletableFuture<AutoSubscriptionCreationOverride> internalGetAutoSub
return null;
}));
}
+
+ protected CompletableFuture<MessageId> internalGetMessageIDByOffsetAsync(Long offset, boolean authoritative) {
Review Comment:
It is better to check whether the index is enabled first, and then call this function to avoid unnecessary searches.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat][admin] PIP-415: Support getting message ID by index [pulsar]
Posted by "liangyepianzhou (via GitHub)" <gi...@apache.org>.
liangyepianzhou commented on PR #24222:
URL: https://github.com/apache/pulsar/pull/24222#issuecomment-2933231905
> 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](https://github.com/apache/pulsar/issues/24374) to track this enhancement.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat][admin] PIP-415: Support getting message ID by index [pulsar]
Posted by "liangyepianzhou (via GitHub)" <gi...@apache.org>.
liangyepianzhou commented on code in PR #24222:
URL: https://github.com/apache/pulsar/pull/24222#discussion_r2125825836
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java:
##########
@@ -5489,4 +5489,129 @@ protected CompletableFuture<AutoSubscriptionCreationOverride> internalGetAutoSub
return null;
}));
}
+
+ protected CompletableFuture<MessageId> internalGetMessageIDByIndexAsync(Long index, boolean authoritative) {
+ if (!pulsar().getBrokerService().isBrokerEntryMetadataEnabled()) {
+ return FutureUtil.failedFuture(new RestException(Status.PRECONDITION_FAILED,
+ "GetMessageIDByIndex is not allowed when broker entry metadata is disabled"));
+ }
+ if (index == null || index < 0) {
+ return FutureUtil.failedFuture(new RestException(Status.NOT_FOUND,
+ "Invalid message index: " + index));
+ }
+ int partitionIndex = topicName.getPartitionIndex();
+ CompletableFuture<Void> future = validateTopicOperationAsync(topicName, TopicOperation.PEEK_MESSAGES);
+ return future.thenCompose(__ -> {
+ if (topicName.isGlobal()) {
+ return validateGlobalNamespaceOwnershipAsync(namespaceName);
+ } else {
+ return CompletableFuture.completedFuture(null);
+ }
+ }).thenCompose(__ -> {
+ if (topicName.isPartitioned()) {
+ return CompletableFuture.completedFuture(null);
+ } else {
+ return getPartitionedTopicMetadataAsync(topicName, authoritative, false)
+ .thenAccept(topicMetadata -> {
+ if (topicMetadata.partitions > 0) {
+ log.warn("[{}] Not supported getMessageIdByIndex operation on "
+ + "partitioned-topic {}", clientAppId(), topicName);
+ throw new RestException(Status.METHOD_NOT_ALLOWED,
+ "GetMessageIDByIndex is not allowed on partitioned-topic");
+ }
+ });
+ }
+ }).thenCompose(ignore -> validateTopicOwnershipAsync(topicName, authoritative))
+ .thenCompose(__ -> getTopicReferenceAsync(topicName))
+ .thenCompose(topic -> {
+ if (!(topic instanceof PersistentTopic persistentTopic)) {
+ log.error("[{}] Get message id by index on a non-persistent topic {} is not allowed",
+ clientAppId(), topicName);
+ return FutureUtil.failedFuture(new RestException(Status.METHOD_NOT_ALLOWED,
+ "Get message id by index on a non-persistent topic is not allowed"));
+ }
+ ManagedLedger managedLedger = persistentTopic.getManagedLedger();
+ return findMessageIndexByPosition(
+ PositionFactory.create(managedLedger.getFirstPosition().getLedgerId(), 0),
+ managedLedger)
+ .thenCompose(firstIndex -> {
+ if (index < firstIndex) {
+ return CompletableFuture.completedFuture(PositionFactory.EARLIEST);
+ } else {
+ return managedLedger.asyncFindPosition(entry -> {
+ try {
+ Long messageIndex = getIndexAndRelease(entry);
+ if (messageIndex == null) {
+ return false; // Skip messages without index
+ } else {
+ // If the message index is less than the requested index,
+ // we continue searching
+ return messageIndex < index;
+ }
+ } catch (IOException e) {
+ log.error("Error deserialize message for message position find", e);
+ return false;
+ }
+ });
+ }
+ }).thenCompose(position -> {
+ Position lastPosition = managedLedger.getLastConfirmedEntry();
+ if (position == null || position.compareTo(lastPosition) > 0) {
+ return FutureUtil.failedFuture(new RestException(Status.NOT_FOUND,
Review Comment:
Occurs when the user passes in incorrect parameters.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat][admin] PIP-415: Support getting message ID by index [pulsar]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on code in PR #24222:
URL: https://github.com/apache/pulsar/pull/24222#discussion_r2125858126
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java:
##########
@@ -5489,4 +5489,114 @@ protected CompletableFuture<AutoSubscriptionCreationOverride> internalGetAutoSub
return null;
}));
}
+
+ protected CompletableFuture<MessageId> internalGetMessageIDByIndexAsync(Long index, boolean authoritative) {
+ if (!pulsar().getBrokerService().isBrokerEntryMetadataEnabled()) {
+ return FutureUtil.failedFuture(new RestException(Status.PRECONDITION_FAILED,
+ "GetMessageIDByIndex is not allowed when broker entry metadata is disabled"));
+ }
+ if (index == null || index < 0) {
+ return FutureUtil.failedFuture(new RestException(Status.NOT_FOUND,
+ "Invalid message index: " + index));
+ }
+ int partitionIndex = topicName.getPartitionIndex();
+ CompletableFuture<Void> future = validateTopicOperationAsync(topicName, TopicOperation.PEEK_MESSAGES);
+ return future.thenCompose(__ -> {
+ if (topicName.isGlobal()) {
+ return validateGlobalNamespaceOwnershipAsync(namespaceName);
+ } else {
+ return CompletableFuture.completedFuture(null);
+ }
+ }).thenCompose(__ -> {
+ if (topicName.isPartitioned()) {
+ return CompletableFuture.completedFuture(null);
+ } else {
+ return getPartitionedTopicMetadataAsync(topicName, authoritative, false)
+ .thenAccept(topicMetadata -> {
+ if (topicMetadata.partitions > 0) {
+ log.warn("[{}] Not supported getMessageIdByIndex operation on "
+ + "partitioned-topic {}", clientAppId(), topicName);
+ throw new RestException(Status.METHOD_NOT_ALLOWED,
+ "GetMessageIDByIndex is not allowed on partitioned-topic");
+ }
+ });
+ }
+ }).thenCompose(ignore -> validateTopicOwnershipAsync(topicName, authoritative))
+ .thenCompose(__ -> getTopicReferenceAsync(topicName))
+ .thenCompose(topic -> {
+ if (!(topic instanceof PersistentTopic persistentTopic)) {
+ log.error("[{}] Get message id by index on a non-persistent topic {} is not allowed",
+ clientAppId(), topicName);
+ return FutureUtil.failedFuture(new RestException(Status.METHOD_NOT_ALLOWED,
+ "Get message id by index on a non-persistent topic is not allowed"));
+ }
+ ManagedLedger managedLedger = persistentTopic.getManagedLedger();
+ return findMessageIndexByPosition(
+ PositionFactory.create(managedLedger.getFirstPosition().getLedgerId(), 0),
Review Comment:
Sorry I've thought that `managedLedger.getFirstPosition()` is the same with `PositionFactory.create(managedLedger.getFirstPosition().getLedgerId(), 0)`, so I asked the question.
But now there is a new question, what if the ledger of the first position is empty (i.e. topic is empty)? In this case, `asyncReadEntry` could fail because the entry does not exist. Could you add a test to show the behavior when `getMessageIdByIndex` is called on an empty topic?
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat][admin] PIP-415: Support getting message ID by index [pulsar]
Posted by "nodece (via GitHub)" <gi...@apache.org>.
nodece commented on code in PR #24222:
URL: https://github.com/apache/pulsar/pull/24222#discussion_r2125901799
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java:
##########
@@ -5489,4 +5489,129 @@ protected CompletableFuture<AutoSubscriptionCreationOverride> internalGetAutoSub
return null;
}));
}
+
+ protected CompletableFuture<MessageId> internalGetMessageIDByIndexAsync(Long index, boolean authoritative) {
+ if (!pulsar().getBrokerService().isBrokerEntryMetadataEnabled()) {
+ return FutureUtil.failedFuture(new RestException(Status.PRECONDITION_FAILED,
+ "GetMessageIDByIndex is not allowed when broker entry metadata is disabled"));
+ }
+ if (index == null || index < 0) {
+ return FutureUtil.failedFuture(new RestException(Status.NOT_FOUND,
+ "Invalid message index: " + index));
+ }
+ int partitionIndex = topicName.getPartitionIndex();
+ CompletableFuture<Void> future = validateTopicOperationAsync(topicName, TopicOperation.PEEK_MESSAGES);
+ return future.thenCompose(__ -> {
+ if (topicName.isGlobal()) {
+ return validateGlobalNamespaceOwnershipAsync(namespaceName);
+ } else {
+ return CompletableFuture.completedFuture(null);
+ }
+ }).thenCompose(__ -> {
+ if (topicName.isPartitioned()) {
+ return CompletableFuture.completedFuture(null);
+ } else {
+ return getPartitionedTopicMetadataAsync(topicName, authoritative, false)
+ .thenAccept(topicMetadata -> {
+ if (topicMetadata.partitions > 0) {
+ log.warn("[{}] Not supported getMessageIdByIndex operation on "
+ + "partitioned-topic {}", clientAppId(), topicName);
+ throw new RestException(Status.METHOD_NOT_ALLOWED,
+ "GetMessageIDByIndex is not allowed on partitioned-topic");
+ }
+ });
+ }
+ }).thenCompose(ignore -> validateTopicOwnershipAsync(topicName, authoritative))
+ .thenCompose(__ -> getTopicReferenceAsync(topicName))
+ .thenCompose(topic -> {
+ if (!(topic instanceof PersistentTopic persistentTopic)) {
+ log.error("[{}] Get message id by index on a non-persistent topic {} is not allowed",
+ clientAppId(), topicName);
+ return FutureUtil.failedFuture(new RestException(Status.METHOD_NOT_ALLOWED,
+ "Get message id by index on a non-persistent topic is not allowed"));
+ }
+ ManagedLedger managedLedger = persistentTopic.getManagedLedger();
+ return findMessageIndexByPosition(
+ PositionFactory.create(managedLedger.getFirstPosition().getLedgerId(), 0),
+ managedLedger)
+ .thenCompose(firstIndex -> {
+ if (index < firstIndex) {
+ return CompletableFuture.completedFuture(PositionFactory.EARLIEST);
+ } else {
+ return managedLedger.asyncFindPosition(entry -> {
+ try {
+ Long messageIndex = getIndexAndRelease(entry);
+ if (messageIndex == null) {
+ return false; // Skip messages without index
+ } else {
+ // If the message index is less than the requested index,
+ // we continue searching
+ return messageIndex < index;
+ }
+ } catch (IOException e) {
+ log.error("Error deserialize message for message position find", e);
+ return false;
+ }
+ });
+ }
+ }).thenCompose(position -> {
+ Position lastPosition = managedLedger.getLastConfirmedEntry();
+ if (position == null || position.compareTo(lastPosition) > 0) {
+ return FutureUtil.failedFuture(new RestException(Status.NOT_FOUND,
Review Comment:
Could you add a test to cover this case?
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat][admin] PIP-415: Support getting message ID by index [pulsar]
Posted by "nodece (via GitHub)" <gi...@apache.org>.
nodece commented on code in PR #24222:
URL: https://github.com/apache/pulsar/pull/24222#discussion_r2128065254
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerEntryMetadataE2ETest.java:
##########
@@ -420,4 +426,93 @@ public void testManagedLedgerTotalSize() throws Exception {
cursor.close();
}
+
+ @Test
+ public void testGetMessageIdByIndex() throws Exception {
+ // 1. test no partitioned topic
+ final String topicName = newTopicName();
+ admin.topics().createNonPartitionedTopic(topicName);
+ @Cleanup
+ Producer<String> producer = pulsarClient.newProducer(Schema.STRING)
+ .topic(topicName)
+ .enableBatching(false)
+ .create();
+ MessageIdImpl messageId = (MessageIdImpl) producer.send("test");
+ Message<byte[]>
+ message =
+ admin.topics().getMessagesById(topicName, messageId.getLedgerId(), messageId.getEntryId()).get(0);
+ long index = message.getIndex().get();
+ MessageIdImpl messageIdByIndex = (MessageIdImpl) admin.topics().getMessageIdByIndex(topicName, index);
+ Assert.assertEquals(messageIdByIndex, messageId);
+
+ // 2. test partitioned topic
+ final String topicName2 = newTopicName();
+ final String partitionedTopicName = topicName2 + "-partition-" + 0;
+ admin.topics().createPartitionedTopic(topicName2, 10);
+ @Cleanup
+ Producer<String> producer2 = pulsarClient.newProducer(Schema.STRING)
+ .topic(topicName2)
+ .enableBatching(false)
+ .create();
+
+ MessageIdImpl messageId2 = null;
+ for (int i = 0; i < 200; i++) {
+ messageId2 = (MessageIdImpl) producer2.send("test" + i);
+ if (messageId2.getPartitionIndex() == 0) {
+ break;
+ }
+ }
+ Message<byte[]>
+ message2 = admin.topics().getMessagesById(partitionedTopicName,
+ messageId2.getLedgerId(), messageId2.getEntryId()).get(0);
+ long index2 = message2.getIndex().get();
+ // 2.1 test partitioned topic name with partition index
+ MessageIdImpl messageIdByIndex2 =
+ (MessageIdImpl) admin.topics().getMessageIdByIndex(partitionedTopicName, index2);
+ Assert.assertEquals(messageIdByIndex2, messageId2);
+ // 2.2 test partitioned topic name without partition index
+ assertThrowsWithCause(() -> admin.topics().getMessageIdByIndex(topicName2, index2),
+ PulsarAdminException.class, NotAllowedException.class);
+
+ // 3. test invalid index
+ assertThrowsWithCause(() -> admin.topics().getMessageIdByIndex(topicName, -1),
+ PulsarAdminException.class, NotFoundException.class);
+
+ assertThrowsWithCause(() -> admin.topics().getMessageIdByIndex(topicName, 100000),
+ PulsarAdminException.class, NotFoundException.class);
+ }
+
+ @Test
+ public void testGetMessageIdByIndexForEmptyTopic() throws PulsarAdminException {
+ final String topicName = newTopicName();
+ admin.topics().createNonPartitionedTopic(topicName);
+
+ assertThrowsWithCause(() -> admin.topics().getMessageIdByIndex(topicName, 0),
+ PulsarAdminException.class, NotFoundException.class);
+ }
+
+ @Test
+ public void testGetMessageIdByIndexOutOfIndex() throws PulsarAdminException, PulsarClientException {
+ final String topicName = newTopicName();
+ admin.topics().createNonPartitionedTopic(topicName);
+ @Cleanup
+ final Producer<String> producer = pulsarClient.newProducer(Schema.STRING)
+ .topic(topicName)
+ .create();
+ for (int i = 0; i < 100; i++) {
+ producer.send("msg-" + i);
+ }
+
+ assertThrowsWithCause(() -> admin.topics().getMessageIdByIndex(topicName, 1000),
+ PulsarAdminException.class, NotFoundException.class);
+ }
+
+ private void assertThrowsWithCause(Executable executable,
Review Comment:
Got, but we should not use the junit API.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat][admin] PIP-415: Support getting message ID by index [pulsar]
Posted by "liangyepianzhou (via GitHub)" <gi...@apache.org>.
liangyepianzhou commented on code in PR #24222:
URL: https://github.com/apache/pulsar/pull/24222#discussion_r2125978583
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java:
##########
@@ -5489,4 +5489,129 @@ protected CompletableFuture<AutoSubscriptionCreationOverride> internalGetAutoSub
return null;
}));
}
+
+ protected CompletableFuture<MessageId> internalGetMessageIDByIndexAsync(Long index, boolean authoritative) {
+ if (!pulsar().getBrokerService().isBrokerEntryMetadataEnabled()) {
+ return FutureUtil.failedFuture(new RestException(Status.PRECONDITION_FAILED,
+ "GetMessageIDByIndex is not allowed when broker entry metadata is disabled"));
+ }
+ if (index == null || index < 0) {
+ return FutureUtil.failedFuture(new RestException(Status.NOT_FOUND,
+ "Invalid message index: " + index));
+ }
+ int partitionIndex = topicName.getPartitionIndex();
+ CompletableFuture<Void> future = validateTopicOperationAsync(topicName, TopicOperation.PEEK_MESSAGES);
+ return future.thenCompose(__ -> {
+ if (topicName.isGlobal()) {
+ return validateGlobalNamespaceOwnershipAsync(namespaceName);
+ } else {
+ return CompletableFuture.completedFuture(null);
+ }
+ }).thenCompose(__ -> {
+ if (topicName.isPartitioned()) {
+ return CompletableFuture.completedFuture(null);
+ } else {
+ return getPartitionedTopicMetadataAsync(topicName, authoritative, false)
+ .thenAccept(topicMetadata -> {
+ if (topicMetadata.partitions > 0) {
+ log.warn("[{}] Not supported getMessageIdByIndex operation on "
+ + "partitioned-topic {}", clientAppId(), topicName);
+ throw new RestException(Status.METHOD_NOT_ALLOWED,
+ "GetMessageIDByIndex is not allowed on partitioned-topic");
+ }
+ });
+ }
+ }).thenCompose(ignore -> validateTopicOwnershipAsync(topicName, authoritative))
+ .thenCompose(__ -> getTopicReferenceAsync(topicName))
+ .thenCompose(topic -> {
+ if (!(topic instanceof PersistentTopic persistentTopic)) {
+ log.error("[{}] Get message id by index on a non-persistent topic {} is not allowed",
+ clientAppId(), topicName);
+ return FutureUtil.failedFuture(new RestException(Status.METHOD_NOT_ALLOWED,
+ "Get message id by index on a non-persistent topic is not allowed"));
+ }
+ ManagedLedger managedLedger = persistentTopic.getManagedLedger();
+ return findMessageIndexByPosition(
+ PositionFactory.create(managedLedger.getFirstPosition().getLedgerId(), 0),
+ managedLedger)
+ .thenCompose(firstIndex -> {
+ if (index < firstIndex) {
+ return CompletableFuture.completedFuture(PositionFactory.EARLIEST);
+ } else {
+ return managedLedger.asyncFindPosition(entry -> {
+ try {
+ Long messageIndex = getIndexAndRelease(entry);
+ if (messageIndex == null) {
+ return false; // Skip messages without index
+ } else {
+ // If the message index is less than the requested index,
+ // we continue searching
+ return messageIndex < index;
+ }
+ } catch (IOException e) {
+ log.error("Error deserialize message for message position find", e);
+ return false;
+ }
+ });
+ }
+ }).thenCompose(position -> {
+ Position lastPosition = managedLedger.getLastConfirmedEntry();
+ if (position == null || position.compareTo(lastPosition) > 0) {
+ return FutureUtil.failedFuture(new RestException(Status.NOT_FOUND,
Review Comment:
done
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat][admin] PIP-415: Support getting message ID by index [pulsar]
Posted by "nodece (via GitHub)" <gi...@apache.org>.
nodece commented on code in PR #24222:
URL: https://github.com/apache/pulsar/pull/24222#discussion_r2127871818
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerEntryMetadataE2ETest.java:
##########
@@ -19,20 +19,25 @@
package org.apache.pulsar.broker.service;
import static org.apache.bookkeeper.mledger.proto.MLDataFormats.ManagedLedgerInfo.LedgerInfo;
+import static org.assertj.core.api.AssertionsForClassTypes.assertThatThrownBy;
Review Comment:
```suggestion
import static org.assertj.core.api.Assertions.assertThatThrownBy;
```
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat][admin] PIP-415: Support getting message ID by index [pulsar]
Posted by "nodece (via GitHub)" <gi...@apache.org>.
nodece commented on code in PR #24222:
URL: https://github.com/apache/pulsar/pull/24222#discussion_r2128065254
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerEntryMetadataE2ETest.java:
##########
@@ -420,4 +426,93 @@ public void testManagedLedgerTotalSize() throws Exception {
cursor.close();
}
+
+ @Test
+ public void testGetMessageIdByIndex() throws Exception {
+ // 1. test no partitioned topic
+ final String topicName = newTopicName();
+ admin.topics().createNonPartitionedTopic(topicName);
+ @Cleanup
+ Producer<String> producer = pulsarClient.newProducer(Schema.STRING)
+ .topic(topicName)
+ .enableBatching(false)
+ .create();
+ MessageIdImpl messageId = (MessageIdImpl) producer.send("test");
+ Message<byte[]>
+ message =
+ admin.topics().getMessagesById(topicName, messageId.getLedgerId(), messageId.getEntryId()).get(0);
+ long index = message.getIndex().get();
+ MessageIdImpl messageIdByIndex = (MessageIdImpl) admin.topics().getMessageIdByIndex(topicName, index);
+ Assert.assertEquals(messageIdByIndex, messageId);
+
+ // 2. test partitioned topic
+ final String topicName2 = newTopicName();
+ final String partitionedTopicName = topicName2 + "-partition-" + 0;
+ admin.topics().createPartitionedTopic(topicName2, 10);
+ @Cleanup
+ Producer<String> producer2 = pulsarClient.newProducer(Schema.STRING)
+ .topic(topicName2)
+ .enableBatching(false)
+ .create();
+
+ MessageIdImpl messageId2 = null;
+ for (int i = 0; i < 200; i++) {
+ messageId2 = (MessageIdImpl) producer2.send("test" + i);
+ if (messageId2.getPartitionIndex() == 0) {
+ break;
+ }
+ }
+ Message<byte[]>
+ message2 = admin.topics().getMessagesById(partitionedTopicName,
+ messageId2.getLedgerId(), messageId2.getEntryId()).get(0);
+ long index2 = message2.getIndex().get();
+ // 2.1 test partitioned topic name with partition index
+ MessageIdImpl messageIdByIndex2 =
+ (MessageIdImpl) admin.topics().getMessageIdByIndex(partitionedTopicName, index2);
+ Assert.assertEquals(messageIdByIndex2, messageId2);
+ // 2.2 test partitioned topic name without partition index
+ assertThrowsWithCause(() -> admin.topics().getMessageIdByIndex(topicName2, index2),
+ PulsarAdminException.class, NotAllowedException.class);
+
+ // 3. test invalid index
+ assertThrowsWithCause(() -> admin.topics().getMessageIdByIndex(topicName, -1),
+ PulsarAdminException.class, NotFoundException.class);
+
+ assertThrowsWithCause(() -> admin.topics().getMessageIdByIndex(topicName, 100000),
+ PulsarAdminException.class, NotFoundException.class);
+ }
+
+ @Test
+ public void testGetMessageIdByIndexForEmptyTopic() throws PulsarAdminException {
+ final String topicName = newTopicName();
+ admin.topics().createNonPartitionedTopic(topicName);
+
+ assertThrowsWithCause(() -> admin.topics().getMessageIdByIndex(topicName, 0),
+ PulsarAdminException.class, NotFoundException.class);
+ }
+
+ @Test
+ public void testGetMessageIdByIndexOutOfIndex() throws PulsarAdminException, PulsarClientException {
+ final String topicName = newTopicName();
+ admin.topics().createNonPartitionedTopic(topicName);
+ @Cleanup
+ final Producer<String> producer = pulsarClient.newProducer(Schema.STRING)
+ .topic(topicName)
+ .create();
+ for (int i = 0; i < 100; i++) {
+ producer.send("msg-" + i);
+ }
+
+ assertThrowsWithCause(() -> admin.topics().getMessageIdByIndex(topicName, 1000),
+ PulsarAdminException.class, NotFoundException.class);
+ }
+
+ private void assertThrowsWithCause(Executable executable,
Review Comment:
Got, but we should use the junit API.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat][admin] PIP-415: Support getting message ID by index [pulsar]
Posted by "dao-jun (via GitHub)" <gi...@apache.org>.
dao-jun commented on PR #24222:
URL: https://github.com/apache/pulsar/pull/24222#issuecomment-2933236186
Overall LGTM, but the current impl will scan all entire ML, maybe you can optimize it via PIP-404
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat][admin] PIP-415: Support getting message ID by index [pulsar]
Posted by "nodece (via GitHub)" <gi...@apache.org>.
nodece commented on code in PR #24222:
URL: https://github.com/apache/pulsar/pull/24222#discussion_r2126333577
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerEntryMetadataE2ETest.java:
##########
@@ -420,4 +427,109 @@ public void testManagedLedgerTotalSize() throws Exception {
cursor.close();
}
+
+ @Test
+ public void testGetMessageIdByIndex() throws Exception {
+ // 1. test no partitioned topic
+ final String topicName = newTopicName();
+ admin.topics().createNonPartitionedTopic(topicName);
+ @Cleanup
+ Producer<String> producer = pulsarClient.newProducer(Schema.STRING)
+ .topic(topicName)
+ .enableBatching(false)
+ .create();
+ MessageIdImpl messageId = (MessageIdImpl) producer.send("test");
+ Message<byte[]>
+ message =
+ admin.topics().getMessagesById(topicName, messageId.getLedgerId(), messageId.getEntryId()).get(0);
+ long index = message.getIndex().get();
+ MessageIdImpl messageIdByIndex = (MessageIdImpl) admin.topics().getMessageIdByIndex(topicName, index);
+ Assert.assertEquals(messageIdByIndex, messageId);
+
+ // 2. test partitioned topic
+ final String topicName2 = newTopicName();
+ final String partitionedTopicName = topicName2 + "-partition-" + 0;
+ admin.topics().createPartitionedTopic(topicName2, 10);
+ @Cleanup
+ Producer<String> producer2 = pulsarClient.newProducer(Schema.STRING)
+ .topic(topicName2)
+ .enableBatching(false)
+ .create();
+
+ MessageIdImpl messageId2 = null;
+ for (int i = 0; i < 200; i++) {
+ messageId2 = (MessageIdImpl) producer2.send("test" + i);
+ if (messageId2.getPartitionIndex() == 0) {
+ break;
+ }
+ }
+ Message<byte[]>
+ message2 = admin.topics().getMessagesById(partitionedTopicName,
+ messageId2.getLedgerId(), messageId2.getEntryId()).get(0);
+ long index2 = message2.getIndex().get();
+ // 2.1 test partitioned topic name with partition index
+ MessageIdImpl messageIdByIndex2 =
+ (MessageIdImpl) admin.topics().getMessageIdByIndex(partitionedTopicName, index2);
+ Assert.assertEquals(messageIdByIndex2, messageId2);
+ // 2.2 test partitioned topic name without partition index
+ PulsarAdminException e = Assertions.assertThrows(PulsarAdminException.class, () ->
+ admin.topics().getMessageIdByIndex(topicName2, index2)
+ );
+ assertExceptionChainContains(e, NotAllowedException.class);
+
+ // 3. test invalid index
+ assertThrowsWithCause(() -> admin.topics().getMessageIdByIndex(topicName, -1),
+ PulsarAdminException.class, NotFoundException.class);
+
+ assertThrowsWithCause(() -> admin.topics().getMessageIdByIndex(topicName, 100000),
+ PulsarAdminException.class, NotFoundException.class);
+ }
+
+ @Test
+ public void testGetMessageIdByIndexForEmptyTopic() throws PulsarAdminException {
+ final String topicName = newTopicName();
+ admin.topics().createNonPartitionedTopic(topicName);
+
+ assertThrowsWithCause(() -> admin.topics().getMessageIdByIndex(topicName, 0),
+ PulsarAdminException.class, NotFoundException.class);
+ }
+
+ @Test
+ public void testGetMessageIdByIndexOutOfIndex() throws PulsarAdminException, PulsarClientException {
+ final String topicName = newTopicName();
+ admin.topics().createNonPartitionedTopic(topicName);
+ @Cleanup
+ final Producer<String> producer = pulsarClient.newProducer(Schema.STRING)
+ .topic(topicName)
+ .create();
+ for (int i = 0; i < 100; i++) {
+ producer.send("msg-" + i);
+ }
+
+ assertThrowsWithCause(() -> admin.topics().getMessageIdByIndex(topicName, 1000),
+ PulsarAdminException.class, NotFoundException.class);
+ }
+
+ private void assertExceptionChainContains(Throwable thrown, Class<? extends Throwable> expectedCauseType) {
+ Throwable cause = thrown;
+ boolean found = false;
+
+ while (cause != null) {
+ if (expectedCauseType.isInstance(cause)) {
+ found = true;
+ break;
+ }
+ cause = cause.getCause();
+ }
+
+ Assert.assertTrue(found);
+ }
+
+ private void assertThrowsWithCause(Executable executable,
Review Comment:
Refactor by the chatgpt:
```
import static org.assertj.core.api.Assertions.*;
private void assertThrowsWithCause(Executable executable,
Class<? extends Throwable> expectedException,
Class<? extends Throwable> expectedCause) {
assertThatThrownBy(executable::execute)
.isInstanceOf(expectedException)
.hasCauseInstanceOf(expectedCause);
}
```
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat][admin] PIP-415: Support getting message ID by index [pulsar]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on code in PR #24222:
URL: https://github.com/apache/pulsar/pull/24222#discussion_r2113925533
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java:
##########
@@ -5489,4 +5489,111 @@ protected CompletableFuture<AutoSubscriptionCreationOverride> internalGetAutoSub
return null;
}));
}
+
+ protected CompletableFuture<MessageId> internalGetMessageIDByIndexAsync(Long index, boolean authoritative) {
+ if (!pulsar().getBrokerService().isBrokerEntryMetadataEnabled()) {
+ return FutureUtil.failedFuture(new RestException(Status.PRECONDITION_FAILED,
+ "GetMessageIDByIndex is not allowed when broker entry metadata is disabled"));
+ }
+ if (index == null || index < 0) {
+ return FutureUtil.failedFuture(new RestException(Status.NOT_FOUND,
+ "Invalid message index: " + index));
+ }
+ int partitionIndex = topicName.getPartitionIndex();
+ CompletableFuture<Void> future = validateTopicOperationAsync(topicName, TopicOperation.PEEK_MESSAGES);
+ return future.thenCompose(__ -> {
+ if (topicName.isGlobal()) {
+ return validateGlobalNamespaceOwnershipAsync(namespaceName);
+ } else {
+ return CompletableFuture.completedFuture(null);
+ }
+ }).thenCompose(__ -> {
+ if (topicName.isPartitioned()) {
+ return CompletableFuture.completedFuture(null);
+ } else {
+ return getPartitionedTopicMetadataAsync(topicName, authoritative, false)
+ .thenAccept(topicMetadata -> {
+ if (topicMetadata.partitions > 0) {
+ log.warn("[{}] Not supported getMessageIdByIndex operation on "
+ + "partitioned-topic {}", clientAppId(), topicName);
+ throw new RestException(Status.METHOD_NOT_ALLOWED,
+ "GetMessageIDByIndex is not allowed on partitioned-topic");
+ }
+ });
+ }
+ }).thenCompose(ignore -> validateTopicOwnershipAsync(topicName, authoritative))
+ .thenCompose(__ -> getTopicReferenceAsync(topicName))
+ .thenCompose(topic -> {
+ if (!(topic instanceof PersistentTopic persistentTopic)) {
+ log.error("[{}] Get message id by index on a non-persistent topic {} is not allowed",
+ clientAppId(), topicName);
+ return FutureUtil.failedFuture(new RestException(Status.METHOD_NOT_ALLOWED,
+ "Get message id by index on a non-persistent topic is not allowed"));
+ }
+ ManagedLedger managedLedger = persistentTopic.getManagedLedger();
+ return findMessageIndexByPosition(
+ PositionFactory.create(managedLedger.getFirstPosition().getLedgerId(), 0),
+ managedLedger)
+ .thenCompose(firstIndex -> {
+ if (index < firstIndex) {
+ return CompletableFuture.completedFuture(PositionFactory.EARLIEST);
+ } else {
+ return managedLedger.asyncFindPosition(entry -> {
+ try {
+ BrokerEntryMetadata brokerEntryMetadata =
+ Commands.parseBrokerEntryMetadataIfExist(entry.getDataBuffer());
+ assert brokerEntryMetadata != null;
Review Comment:
https://github.com/streamnative/kop/pull/916 is an example. It assumes the entries distribution is XXXOOO, so the implementation is:
```java
try {
return peekOffsetFromEntry(entry) < offset;
} catch (MetadataCorruptedException.NoBrokerEntryMetadata ignored) {
return skipMessagesWithoutIndex;
```
The search will come to the right side when `skipMessagesWithoutIndex` is true.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat][admin] PIP-415: Support getting message ID by offset [pulsar]
Posted by "AuroraTwinkle (via GitHub)" <gi...@apache.org>.
AuroraTwinkle commented on code in PR #24222:
URL: https://github.com/apache/pulsar/pull/24222#discussion_r2062805073
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java:
##########
@@ -5489,4 +5489,53 @@ protected CompletableFuture<AutoSubscriptionCreationOverride> internalGetAutoSub
return null;
}));
}
+
+ protected CompletableFuture<MessageId> internalGetMessageIDByOffsetAsync(Long offset, boolean authoritative) {
Review Comment:
what happen if the index not enabled.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java:
##########
@@ -5489,4 +5489,53 @@ protected CompletableFuture<AutoSubscriptionCreationOverride> internalGetAutoSub
return null;
}));
}
+
+ protected CompletableFuture<MessageId> internalGetMessageIDByOffsetAsync(Long offset, boolean authoritative) {
+ int partitionIndex = topicName.getPartitionIndex();
+ CompletableFuture<Void> future = validateTopicOperationAsync(topicName, TopicOperation.PEEK_MESSAGES);
+ return future.thenCompose(__ -> {
+ if (topicName.isGlobal()) {
+ return validateGlobalNamespaceOwnershipAsync(namespaceName);
+ } else {
+ return CompletableFuture.completedFuture(null);
+ }
+ }).thenCompose(__ -> {
+ if (topicName.isPartitioned()) {
+ return CompletableFuture.completedFuture(null);
+ } else {
+ return getPartitionedTopicMetadataAsync(topicName, authoritative, false)
+ .thenAccept(topicMetadata -> {
+ if (topicMetadata.partitions > 0) {
+ log.warn("[{}] Not supported getMessageIdByOffset operation on "
+ + "partitioned-topic {}", clientAppId(), topicName);
+ throw new RestException(Status.METHOD_NOT_ALLOWED,
+ "GetMessageIDByOffset is not allowed on partitioned-topic");
+ }
+ });
+ }
+ }).thenCompose(ignore -> validateTopicOwnershipAsync(topicName, authoritative))
+ .thenCompose(__ -> getTopicReferenceAsync(topicName))
+ .thenCompose(topic -> {
+ if (!(topic instanceof PersistentTopic persistentTopic)) {
+ log.error("[{}] Get message id by offset on a non-persistent topic {} is not allowed",
+ clientAppId(), topicName);
+ return FutureUtil.failedFuture(new RestException(Status.METHOD_NOT_ALLOWED,
+ "Get message id by offset on a non-persistent topic is not allowed"));
+ }
+ ManagedLedger managedLedger = persistentTopic.getManagedLedger();
+ return managedLedger.asyncFindPosition(entry -> {
+ try {
+ BrokerEntryMetadata brokerEntryMetadata =
+ Commands.parseBrokerEntryMetadataIfExist(entry.getDataBuffer());
+ return brokerEntryMetadata.getIndex() < offset;
Review Comment:
maybe `brokerEntryMetadata.getIndex() <= offset`?
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat][admin] PIP-415: Support getting message ID by offset [pulsar]
Posted by "liangyepianzhou (via GitHub)" <gi...@apache.org>.
liangyepianzhou commented on code in PR #24222:
URL: https://github.com/apache/pulsar/pull/24222#discussion_r2062978467
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java:
##########
@@ -5489,4 +5489,53 @@ protected CompletableFuture<AutoSubscriptionCreationOverride> internalGetAutoSub
return null;
}));
}
+
+ protected CompletableFuture<MessageId> internalGetMessageIDByOffsetAsync(Long offset, boolean authoritative) {
+ int partitionIndex = topicName.getPartitionIndex();
+ CompletableFuture<Void> future = validateTopicOperationAsync(topicName, TopicOperation.PEEK_MESSAGES);
+ return future.thenCompose(__ -> {
+ if (topicName.isGlobal()) {
+ return validateGlobalNamespaceOwnershipAsync(namespaceName);
+ } else {
+ return CompletableFuture.completedFuture(null);
+ }
+ }).thenCompose(__ -> {
+ if (topicName.isPartitioned()) {
+ return CompletableFuture.completedFuture(null);
+ } else {
+ return getPartitionedTopicMetadataAsync(topicName, authoritative, false)
+ .thenAccept(topicMetadata -> {
+ if (topicMetadata.partitions > 0) {
+ log.warn("[{}] Not supported getMessageIdByOffset operation on "
+ + "partitioned-topic {}", clientAppId(), topicName);
+ throw new RestException(Status.METHOD_NOT_ALLOWED,
+ "GetMessageIDByOffset is not allowed on partitioned-topic");
+ }
+ });
+ }
+ }).thenCompose(ignore -> validateTopicOwnershipAsync(topicName, authoritative))
+ .thenCompose(__ -> getTopicReferenceAsync(topicName))
+ .thenCompose(topic -> {
+ if (!(topic instanceof PersistentTopic persistentTopic)) {
+ log.error("[{}] Get message id by offset on a non-persistent topic {} is not allowed",
+ clientAppId(), topicName);
+ return FutureUtil.failedFuture(new RestException(Status.METHOD_NOT_ALLOWED,
+ "Get message id by offset on a non-persistent topic is not allowed"));
+ }
+ ManagedLedger managedLedger = persistentTopic.getManagedLedger();
+ return managedLedger.asyncFindPosition(entry -> {
+ try {
+ BrokerEntryMetadata brokerEntryMetadata =
+ Commands.parseBrokerEntryMetadataIfExist(entry.getDataBuffer());
+ return brokerEntryMetadata.getIndex() < offset;
+ } catch (Exception e) {
Review Comment:
IMO, It is enough to catch the exception.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java:
##########
@@ -5489,4 +5489,53 @@ protected CompletableFuture<AutoSubscriptionCreationOverride> internalGetAutoSub
return null;
}));
}
+
+ protected CompletableFuture<MessageId> internalGetMessageIDByOffsetAsync(Long offset, boolean authoritative) {
+ int partitionIndex = topicName.getPartitionIndex();
+ CompletableFuture<Void> future = validateTopicOperationAsync(topicName, TopicOperation.PEEK_MESSAGES);
+ return future.thenCompose(__ -> {
+ if (topicName.isGlobal()) {
+ return validateGlobalNamespaceOwnershipAsync(namespaceName);
+ } else {
+ return CompletableFuture.completedFuture(null);
+ }
+ }).thenCompose(__ -> {
+ if (topicName.isPartitioned()) {
+ return CompletableFuture.completedFuture(null);
+ } else {
+ return getPartitionedTopicMetadataAsync(topicName, authoritative, false)
+ .thenAccept(topicMetadata -> {
+ if (topicMetadata.partitions > 0) {
+ log.warn("[{}] Not supported getMessageIdByOffset operation on "
+ + "partitioned-topic {}", clientAppId(), topicName);
+ throw new RestException(Status.METHOD_NOT_ALLOWED,
+ "GetMessageIDByOffset is not allowed on partitioned-topic");
+ }
+ });
+ }
+ }).thenCompose(ignore -> validateTopicOwnershipAsync(topicName, authoritative))
+ .thenCompose(__ -> getTopicReferenceAsync(topicName))
+ .thenCompose(topic -> {
+ if (!(topic instanceof PersistentTopic persistentTopic)) {
+ log.error("[{}] Get message id by offset on a non-persistent topic {} is not allowed",
+ clientAppId(), topicName);
+ return FutureUtil.failedFuture(new RestException(Status.METHOD_NOT_ALLOWED,
+ "Get message id by offset on a non-persistent topic is not allowed"));
+ }
+ ManagedLedger managedLedger = persistentTopic.getManagedLedger();
+ return managedLedger.asyncFindPosition(entry -> {
+ try {
+ BrokerEntryMetadata brokerEntryMetadata =
+ Commands.parseBrokerEntryMetadataIfExist(entry.getDataBuffer());
+ return brokerEntryMetadata.getIndex() < offset;
+ } catch (Exception e) {
+ log.error("Error deserialize message for message position find", e);
+ } finally {
+ entry.release();
+ }
+ return false;
+ });
+ }).thenCompose(position -> CompletableFuture
+ .completedFuture(new MessageIdImpl(position.getLedgerId(), position.getEntryId(), partitionIndex)));
+ }
Review Comment:
The `managedLedger.asyncFindPosition` maybe return a `null`,so there need to add a NPE check.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java:
##########
@@ -5489,4 +5489,53 @@ protected CompletableFuture<AutoSubscriptionCreationOverride> internalGetAutoSub
return null;
}));
}
+
+ protected CompletableFuture<MessageId> internalGetMessageIDByOffsetAsync(Long offset, boolean authoritative) {
Review Comment:
The api will return a 404 error.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java:
##########
@@ -5489,4 +5489,53 @@ protected CompletableFuture<AutoSubscriptionCreationOverride> internalGetAutoSub
return null;
}));
}
+
+ protected CompletableFuture<MessageId> internalGetMessageIDByOffsetAsync(Long offset, boolean authoritative) {
+ int partitionIndex = topicName.getPartitionIndex();
+ CompletableFuture<Void> future = validateTopicOperationAsync(topicName, TopicOperation.PEEK_MESSAGES);
+ return future.thenCompose(__ -> {
+ if (topicName.isGlobal()) {
+ return validateGlobalNamespaceOwnershipAsync(namespaceName);
+ } else {
+ return CompletableFuture.completedFuture(null);
+ }
+ }).thenCompose(__ -> {
+ if (topicName.isPartitioned()) {
+ return CompletableFuture.completedFuture(null);
+ } else {
+ return getPartitionedTopicMetadataAsync(topicName, authoritative, false)
+ .thenAccept(topicMetadata -> {
+ if (topicMetadata.partitions > 0) {
+ log.warn("[{}] Not supported getMessageIdByOffset operation on "
+ + "partitioned-topic {}", clientAppId(), topicName);
+ throw new RestException(Status.METHOD_NOT_ALLOWED,
+ "GetMessageIDByOffset is not allowed on partitioned-topic");
+ }
+ });
+ }
+ }).thenCompose(ignore -> validateTopicOwnershipAsync(topicName, authoritative))
+ .thenCompose(__ -> getTopicReferenceAsync(topicName))
+ .thenCompose(topic -> {
+ if (!(topic instanceof PersistentTopic persistentTopic)) {
+ log.error("[{}] Get message id by offset on a non-persistent topic {} is not allowed",
+ clientAppId(), topicName);
+ return FutureUtil.failedFuture(new RestException(Status.METHOD_NOT_ALLOWED,
+ "Get message id by offset on a non-persistent topic is not allowed"));
+ }
+ ManagedLedger managedLedger = persistentTopic.getManagedLedger();
+ return managedLedger.asyncFindPosition(entry -> {
+ try {
+ BrokerEntryMetadata brokerEntryMetadata =
+ Commands.parseBrokerEntryMetadataIfExist(entry.getDataBuffer());
+ return brokerEntryMetadata.getIndex() < offset;
Review Comment:
No, we need to get the first message after this offset (inclusive).
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat][admin] PIP-415: Support getting message ID by index [pulsar]
Posted by "liangyepianzhou (via GitHub)" <gi...@apache.org>.
liangyepianzhou commented on code in PR #24222:
URL: https://github.com/apache/pulsar/pull/24222#discussion_r2125959015
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java:
##########
@@ -5489,4 +5489,131 @@ protected CompletableFuture<AutoSubscriptionCreationOverride> internalGetAutoSub
return null;
}));
}
+
+ protected CompletableFuture<MessageId> internalGetMessageIDByIndexAsync(Long index, boolean authoritative) {
+ if (!pulsar().getBrokerService().isBrokerEntryMetadataEnabled()) {
+ return FutureUtil.failedFuture(new RestException(Status.PRECONDITION_FAILED,
+ "GetMessageIDByIndex is not allowed when broker entry metadata is disabled"));
+ }
+ if (index == null || index < 0) {
+ return FutureUtil.failedFuture(new RestException(Status.NOT_FOUND,
+ "Invalid message index: " + index));
+ }
+ int partitionIndex = topicName.getPartitionIndex();
+ CompletableFuture<Void> future = validateTopicOperationAsync(topicName, TopicOperation.PEEK_MESSAGES);
+ return future.thenCompose(__ -> {
+ if (topicName.isGlobal()) {
+ return validateGlobalNamespaceOwnershipAsync(namespaceName);
+ } else {
+ return CompletableFuture.completedFuture(null);
+ }
+ }).thenCompose(__ -> {
+ if (topicName.isPartitioned()) {
+ return CompletableFuture.completedFuture(null);
+ } else {
+ return getPartitionedTopicMetadataAsync(topicName, authoritative, false)
+ .thenAccept(topicMetadata -> {
+ if (topicMetadata.partitions > 0) {
+ log.warn("[{}] Not supported getMessageIdByIndex operation on "
+ + "partitioned-topic {}", clientAppId(), topicName);
+ throw new RestException(Status.METHOD_NOT_ALLOWED,
+ "GetMessageIDByIndex is not allowed on partitioned-topic");
+ }
+ });
+ }
+ }).thenCompose(ignore -> validateTopicOwnershipAsync(topicName, authoritative))
+ .thenCompose(__ -> getTopicReferenceAsync(topicName))
+ .thenCompose(topic -> {
+ if (!(topic instanceof PersistentTopic persistentTopic)) {
+ log.error("[{}] Get message id by index on a non-persistent topic {} is not allowed",
+ clientAppId(), topicName);
+ return FutureUtil.failedFuture(new RestException(Status.METHOD_NOT_ALLOWED,
+ "Get message id by index on a non-persistent topic is not allowed"));
+ }
+ ManagedLedger managedLedger = persistentTopic.getManagedLedger();
+ return findMessageIndexByPosition(
+ PositionFactory.create(managedLedger.getFirstPosition().getLedgerId(), 0),
+ managedLedger)
+ .thenCompose(firstIndex -> {
+ if (index < firstIndex) {
+ return CompletableFuture.completedFuture(PositionFactory.EARLIEST);
+ } else {
+ return managedLedger.asyncFindPosition(entry -> {
+ try {
+ Long messageIndex = getIndexFromEntry(entry);
+ if (messageIndex == null) {
+ return false; // Skip messages without index
+ } else {
+ // If the message index is less than the requested index,
+ // we continue searching
+ return messageIndex < index;
+ }
+ } catch (IOException e) {
+ log.error("Error deserialize message for message position find", e);
+ return false;
Review Comment:
Why not? If a message is corrupted, shouldn't we skip it?
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat][admin] PIP-415: Support getting message ID by index [pulsar]
Posted by "liangyepianzhou (via GitHub)" <gi...@apache.org>.
liangyepianzhou commented on code in PR #24222:
URL: https://github.com/apache/pulsar/pull/24222#discussion_r2126361742
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerEntryMetadataE2ETest.java:
##########
@@ -420,4 +427,109 @@ public void testManagedLedgerTotalSize() throws Exception {
cursor.close();
}
+
+ @Test
+ public void testGetMessageIdByIndex() throws Exception {
+ // 1. test no partitioned topic
+ final String topicName = newTopicName();
+ admin.topics().createNonPartitionedTopic(topicName);
+ @Cleanup
+ Producer<String> producer = pulsarClient.newProducer(Schema.STRING)
+ .topic(topicName)
+ .enableBatching(false)
+ .create();
+ MessageIdImpl messageId = (MessageIdImpl) producer.send("test");
+ Message<byte[]>
+ message =
+ admin.topics().getMessagesById(topicName, messageId.getLedgerId(), messageId.getEntryId()).get(0);
+ long index = message.getIndex().get();
+ MessageIdImpl messageIdByIndex = (MessageIdImpl) admin.topics().getMessageIdByIndex(topicName, index);
+ Assert.assertEquals(messageIdByIndex, messageId);
+
+ // 2. test partitioned topic
+ final String topicName2 = newTopicName();
+ final String partitionedTopicName = topicName2 + "-partition-" + 0;
+ admin.topics().createPartitionedTopic(topicName2, 10);
+ @Cleanup
+ Producer<String> producer2 = pulsarClient.newProducer(Schema.STRING)
+ .topic(topicName2)
+ .enableBatching(false)
+ .create();
+
+ MessageIdImpl messageId2 = null;
+ for (int i = 0; i < 200; i++) {
+ messageId2 = (MessageIdImpl) producer2.send("test" + i);
+ if (messageId2.getPartitionIndex() == 0) {
+ break;
+ }
+ }
+ Message<byte[]>
+ message2 = admin.topics().getMessagesById(partitionedTopicName,
+ messageId2.getLedgerId(), messageId2.getEntryId()).get(0);
+ long index2 = message2.getIndex().get();
+ // 2.1 test partitioned topic name with partition index
+ MessageIdImpl messageIdByIndex2 =
+ (MessageIdImpl) admin.topics().getMessageIdByIndex(partitionedTopicName, index2);
+ Assert.assertEquals(messageIdByIndex2, messageId2);
+ // 2.2 test partitioned topic name without partition index
+ PulsarAdminException e = Assertions.assertThrows(PulsarAdminException.class, () ->
+ admin.topics().getMessageIdByIndex(topicName2, index2)
+ );
+ assertExceptionChainContains(e, NotAllowedException.class);
+
+ // 3. test invalid index
+ assertThrowsWithCause(() -> admin.topics().getMessageIdByIndex(topicName, -1),
+ PulsarAdminException.class, NotFoundException.class);
+
+ assertThrowsWithCause(() -> admin.topics().getMessageIdByIndex(topicName, 100000),
+ PulsarAdminException.class, NotFoundException.class);
+ }
+
+ @Test
+ public void testGetMessageIdByIndexForEmptyTopic() throws PulsarAdminException {
+ final String topicName = newTopicName();
+ admin.topics().createNonPartitionedTopic(topicName);
+
+ assertThrowsWithCause(() -> admin.topics().getMessageIdByIndex(topicName, 0),
+ PulsarAdminException.class, NotFoundException.class);
+ }
+
+ @Test
+ public void testGetMessageIdByIndexOutOfIndex() throws PulsarAdminException, PulsarClientException {
+ final String topicName = newTopicName();
+ admin.topics().createNonPartitionedTopic(topicName);
+ @Cleanup
+ final Producer<String> producer = pulsarClient.newProducer(Schema.STRING)
+ .topic(topicName)
+ .create();
+ for (int i = 0; i < 100; i++) {
+ producer.send("msg-" + i);
+ }
+
+ assertThrowsWithCause(() -> admin.topics().getMessageIdByIndex(topicName, 1000),
+ PulsarAdminException.class, NotFoundException.class);
+ }
+
+ private void assertExceptionChainContains(Throwable thrown, Class<? extends Throwable> expectedCauseType) {
+ Throwable cause = thrown;
+ boolean found = false;
+
+ while (cause != null) {
+ if (expectedCauseType.isInstance(cause)) {
+ found = true;
+ break;
+ }
+ cause = cause.getCause();
+ }
+
+ Assert.assertTrue(found);
+ }
+
+ private void assertThrowsWithCause(Executable executable,
Review Comment:
While ChatGPT's provided code wasn't directly applicable to our scenario, it pointed us in the right direction. After testing, confirmed the following implementation works:
```java
private void assertThrowsWithCause(Executable executable,
Class<? extends Throwable> expectedException,
Class<? extends Throwable> expectedCause) {
assertThatThrownBy(executable::execute)
.isInstanceOf(expectedException)
.getCause() // First layer: ResponseProcessingException
.getCause() // Second layer: NotFoundException
.isInstanceOf(expectedCause);
}
```
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat][admin] PIP-415: Support getting message ID by index [pulsar]
Posted by "nodece (via GitHub)" <gi...@apache.org>.
nodece commented on code in PR #24222:
URL: https://github.com/apache/pulsar/pull/24222#discussion_r2127870098
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerEntryMetadataE2ETest.java:
##########
@@ -420,4 +426,93 @@ public void testManagedLedgerTotalSize() throws Exception {
cursor.close();
}
+
+ @Test
+ public void testGetMessageIdByIndex() throws Exception {
+ // 1. test no partitioned topic
+ final String topicName = newTopicName();
+ admin.topics().createNonPartitionedTopic(topicName);
+ @Cleanup
+ Producer<String> producer = pulsarClient.newProducer(Schema.STRING)
+ .topic(topicName)
+ .enableBatching(false)
+ .create();
+ MessageIdImpl messageId = (MessageIdImpl) producer.send("test");
+ Message<byte[]>
+ message =
+ admin.topics().getMessagesById(topicName, messageId.getLedgerId(), messageId.getEntryId()).get(0);
+ long index = message.getIndex().get();
+ MessageIdImpl messageIdByIndex = (MessageIdImpl) admin.topics().getMessageIdByIndex(topicName, index);
+ Assert.assertEquals(messageIdByIndex, messageId);
+
+ // 2. test partitioned topic
+ final String topicName2 = newTopicName();
+ final String partitionedTopicName = topicName2 + "-partition-" + 0;
+ admin.topics().createPartitionedTopic(topicName2, 10);
+ @Cleanup
+ Producer<String> producer2 = pulsarClient.newProducer(Schema.STRING)
+ .topic(topicName2)
+ .enableBatching(false)
+ .create();
+
+ MessageIdImpl messageId2 = null;
+ for (int i = 0; i < 200; i++) {
+ messageId2 = (MessageIdImpl) producer2.send("test" + i);
+ if (messageId2.getPartitionIndex() == 0) {
+ break;
+ }
+ }
+ Message<byte[]>
+ message2 = admin.topics().getMessagesById(partitionedTopicName,
+ messageId2.getLedgerId(), messageId2.getEntryId()).get(0);
+ long index2 = message2.getIndex().get();
+ // 2.1 test partitioned topic name with partition index
+ MessageIdImpl messageIdByIndex2 =
+ (MessageIdImpl) admin.topics().getMessageIdByIndex(partitionedTopicName, index2);
+ Assert.assertEquals(messageIdByIndex2, messageId2);
+ // 2.2 test partitioned topic name without partition index
+ assertThrowsWithCause(() -> admin.topics().getMessageIdByIndex(topicName2, index2),
+ PulsarAdminException.class, NotAllowedException.class);
+
+ // 3. test invalid index
+ assertThrowsWithCause(() -> admin.topics().getMessageIdByIndex(topicName, -1),
+ PulsarAdminException.class, NotFoundException.class);
+
+ assertThrowsWithCause(() -> admin.topics().getMessageIdByIndex(topicName, 100000),
+ PulsarAdminException.class, NotFoundException.class);
+ }
+
+ @Test
+ public void testGetMessageIdByIndexForEmptyTopic() throws PulsarAdminException {
+ final String topicName = newTopicName();
+ admin.topics().createNonPartitionedTopic(topicName);
+
+ assertThrowsWithCause(() -> admin.topics().getMessageIdByIndex(topicName, 0),
+ PulsarAdminException.class, NotFoundException.class);
+ }
+
+ @Test
+ public void testGetMessageIdByIndexOutOfIndex() throws PulsarAdminException, PulsarClientException {
+ final String topicName = newTopicName();
+ admin.topics().createNonPartitionedTopic(topicName);
+ @Cleanup
+ final Producer<String> producer = pulsarClient.newProducer(Schema.STRING)
+ .topic(topicName)
+ .create();
+ for (int i = 0; i < 100; i++) {
+ producer.send("msg-" + i);
+ }
+
+ assertThrowsWithCause(() -> admin.topics().getMessageIdByIndex(topicName, 1000),
+ PulsarAdminException.class, NotFoundException.class);
+ }
+
+ private void assertThrowsWithCause(Executable executable,
Review Comment:
```suggestion
private void assertThrowsWithCause(Runnable executable,
```
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat][admin] PIP-415: Support getting message ID by index [pulsar]
Posted by "liangyepianzhou (via GitHub)" <gi...@apache.org>.
liangyepianzhou commented on code in PR #24222:
URL: https://github.com/apache/pulsar/pull/24222#discussion_r2125831416
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java:
##########
@@ -5489,4 +5489,129 @@ protected CompletableFuture<AutoSubscriptionCreationOverride> internalGetAutoSub
return null;
}));
}
+
+ protected CompletableFuture<MessageId> internalGetMessageIDByIndexAsync(Long index, boolean authoritative) {
+ if (!pulsar().getBrokerService().isBrokerEntryMetadataEnabled()) {
+ return FutureUtil.failedFuture(new RestException(Status.PRECONDITION_FAILED,
+ "GetMessageIDByIndex is not allowed when broker entry metadata is disabled"));
+ }
+ if (index == null || index < 0) {
+ return FutureUtil.failedFuture(new RestException(Status.NOT_FOUND,
+ "Invalid message index: " + index));
+ }
+ int partitionIndex = topicName.getPartitionIndex();
+ CompletableFuture<Void> future = validateTopicOperationAsync(topicName, TopicOperation.PEEK_MESSAGES);
+ return future.thenCompose(__ -> {
+ if (topicName.isGlobal()) {
+ return validateGlobalNamespaceOwnershipAsync(namespaceName);
+ } else {
+ return CompletableFuture.completedFuture(null);
+ }
+ }).thenCompose(__ -> {
+ if (topicName.isPartitioned()) {
+ return CompletableFuture.completedFuture(null);
+ } else {
+ return getPartitionedTopicMetadataAsync(topicName, authoritative, false)
+ .thenAccept(topicMetadata -> {
+ if (topicMetadata.partitions > 0) {
+ log.warn("[{}] Not supported getMessageIdByIndex operation on "
+ + "partitioned-topic {}", clientAppId(), topicName);
+ throw new RestException(Status.METHOD_NOT_ALLOWED,
+ "GetMessageIDByIndex is not allowed on partitioned-topic");
+ }
+ });
+ }
+ }).thenCompose(ignore -> validateTopicOwnershipAsync(topicName, authoritative))
+ .thenCompose(__ -> getTopicReferenceAsync(topicName))
+ .thenCompose(topic -> {
+ if (!(topic instanceof PersistentTopic persistentTopic)) {
+ log.error("[{}] Get message id by index on a non-persistent topic {} is not allowed",
+ clientAppId(), topicName);
+ return FutureUtil.failedFuture(new RestException(Status.METHOD_NOT_ALLOWED,
+ "Get message id by index on a non-persistent topic is not allowed"));
+ }
+ ManagedLedger managedLedger = persistentTopic.getManagedLedger();
+ return findMessageIndexByPosition(
+ PositionFactory.create(managedLedger.getFirstPosition().getLedgerId(), 0),
+ managedLedger)
+ .thenCompose(firstIndex -> {
+ if (index < firstIndex) {
+ return CompletableFuture.completedFuture(PositionFactory.EARLIEST);
+ } else {
+ return managedLedger.asyncFindPosition(entry -> {
+ try {
+ Long messageIndex = getIndexAndRelease(entry);
+ if (messageIndex == null) {
+ return false; // Skip messages without index
+ } else {
+ // If the message index is less than the requested index,
+ // we continue searching
+ return messageIndex < index;
+ }
+ } catch (IOException e) {
+ log.error("Error deserialize message for message position find", e);
Review Comment:
The `Throwable ` is catch in `getIndexFromEntry`. The `IOException` is enough here.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat][admin] PIP-415: Support getting message ID by index [pulsar]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #24222:
URL: https://github.com/apache/pulsar/pull/24222#issuecomment-2994838090
## [Codecov](https://app.codecov.io/gh/apache/pulsar/pull/24222?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
Attention: Patch coverage is `72.38095%` with `29 lines` in your changes missing coverage. Please review.
> Project coverage is 74.33%. Comparing base [(`bbc6224`)](https://app.codecov.io/gh/apache/pulsar/commit/bbc62245c5ddba1de4b1e7cee4ab49334bc36277?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) to head [(`fc2420f`)](https://app.codecov.io/gh/apache/pulsar/commit/fc2420f8cd9dd003995a84cfb552b8b19028bf3c?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
> Report is 1158 commits behind head on master.
| [Files with missing lines](https://app.codecov.io/gh/apache/pulsar/pull/24222?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Patch % | Lines |
|---|---|---|
| [...pulsar/broker/admin/impl/PersistentTopicsBase.java](https://app.codecov.io/gh/apache/pulsar/pull/24222?src=pr&el=tree&filepath=pulsar-broker%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fbroker%2Fadmin%2Fimpl%2FPersistentTopicsBase.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWJyb2tlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2Jyb2tlci9hZG1pbi9pbXBsL1BlcnNpc3RlbnRUb3BpY3NCYXNlLmphdmE=) | 60.81% | [19 Missing and 10 partials :warning: ](https://app.codecov.io/gh/apache/pulsar/pull/24222?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
<details><summary>Additional details and impacted files</summary>
[](https://app.codecov.io/gh/apache/pulsar/pull/24222?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
```diff
@@ Coverage Diff @@
## master #24222 +/- ##
============================================
+ Coverage 73.57% 74.33% +0.76%
- Complexity 32624 32765 +141
============================================
Files 1877 1868 -9
Lines 139502 145497 +5995
Branches 15299 16643 +1344
============================================
+ Hits 102638 108157 +5519
+ Misses 28908 28789 -119
- Partials 7956 8551 +595
```
| [Flag](https://app.codecov.io/gh/apache/pulsar/pull/24222/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [inttests](https://app.codecov.io/gh/apache/pulsar/pull/24222/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `26.81% <0.00%> (+2.22%)` | :arrow_up: |
| [systests](https://app.codecov.io/gh/apache/pulsar/pull/24222/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `23.33% <0.00%> (-1.00%)` | :arrow_down: |
| [unittests](https://app.codecov.io/gh/apache/pulsar/pull/24222/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `73.82% <72.38%> (+0.97%)` | :arrow_up: |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#carryforward-flags-in-the-pull-request-comment) to find out more.
| [Files with missing lines](https://app.codecov.io/gh/apache/pulsar/pull/24222?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [...pache/pulsar/broker/admin/v2/PersistentTopics.java](https://app.codecov.io/gh/apache/pulsar/pull/24222?src=pr&el=tree&filepath=pulsar-broker%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fbroker%2Fadmin%2Fv2%2FPersistentTopics.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWJyb2tlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2Jyb2tlci9hZG1pbi92Mi9QZXJzaXN0ZW50VG9waWNzLmphdmE=) | `88.62% <100.00%> (+7.68%)` | :arrow_up: |
| [...in/java/org/apache/pulsar/client/admin/Topics.java](https://app.codecov.io/gh/apache/pulsar/pull/24222?src=pr&el=tree&filepath=pulsar-client-admin-api%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fclient%2Fadmin%2FTopics.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWNsaWVudC1hZG1pbi1hcGkvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL3B1bHNhci9jbGllbnQvYWRtaW4vVG9waWNzLmphdmE=) | `80.95% <ø> (+3.45%)` | :arrow_up: |
| [...pache/pulsar/client/admin/internal/TopicsImpl.java](https://app.codecov.io/gh/apache/pulsar/pull/24222?src=pr&el=tree&filepath=pulsar-client-admin%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fclient%2Fadmin%2Finternal%2FTopicsImpl.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWNsaWVudC1hZG1pbi9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2NsaWVudC9hZG1pbi9pbnRlcm5hbC9Ub3BpY3NJbXBsLmphdmE=) | `84.38% <100.00%> (+2.37%)` | :arrow_up: |
| [...in/java/org/apache/pulsar/admin/cli/CmdTopics.java](https://app.codecov.io/gh/apache/pulsar/pull/24222?src=pr&el=tree&filepath=pulsar-client-tools%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fadmin%2Fcli%2FCmdTopics.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWNsaWVudC10b29scy9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2FkbWluL2NsaS9DbWRUb3BpY3MuamF2YQ==) | `80.63% <100.00%> (-0.55%)` | :arrow_down: |
| [...pulsar/broker/admin/impl/PersistentTopicsBase.java](https://app.codecov.io/gh/apache/pulsar/pull/24222?src=pr&el=tree&filepath=pulsar-broker%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fbroker%2Fadmin%2Fimpl%2FPersistentTopicsBase.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWJyb2tlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2Jyb2tlci9hZG1pbi9pbXBsL1BlcnNpc3RlbnRUb3BpY3NCYXNlLmphdmE=) | `70.02% <60.81%> (+4.57%)` | :arrow_up: |
... and [1085 files with indirect coverage changes](https://app.codecov.io/gh/apache/pulsar/pull/24222/indirect-changes?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
</details>
<details><summary> :rocket: New features to boost your workflow: </summary>
- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat][admin] PIP-415: Support getting message ID by index [pulsar]
Posted by "liangyepianzhou (via GitHub)" <gi...@apache.org>.
liangyepianzhou commented on code in PR #24222:
URL: https://github.com/apache/pulsar/pull/24222#discussion_r2126017133
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java:
##########
@@ -5489,4 +5489,131 @@ protected CompletableFuture<AutoSubscriptionCreationOverride> internalGetAutoSub
return null;
}));
}
+
+ protected CompletableFuture<MessageId> internalGetMessageIDByIndexAsync(Long index, boolean authoritative) {
+ if (!pulsar().getBrokerService().isBrokerEntryMetadataEnabled()) {
+ return FutureUtil.failedFuture(new RestException(Status.PRECONDITION_FAILED,
+ "GetMessageIDByIndex is not allowed when broker entry metadata is disabled"));
+ }
+ if (index == null || index < 0) {
+ return FutureUtil.failedFuture(new RestException(Status.NOT_FOUND,
+ "Invalid message index: " + index));
+ }
+ int partitionIndex = topicName.getPartitionIndex();
+ CompletableFuture<Void> future = validateTopicOperationAsync(topicName, TopicOperation.PEEK_MESSAGES);
+ return future.thenCompose(__ -> {
+ if (topicName.isGlobal()) {
+ return validateGlobalNamespaceOwnershipAsync(namespaceName);
+ } else {
+ return CompletableFuture.completedFuture(null);
+ }
+ }).thenCompose(__ -> {
+ if (topicName.isPartitioned()) {
+ return CompletableFuture.completedFuture(null);
+ } else {
+ return getPartitionedTopicMetadataAsync(topicName, authoritative, false)
+ .thenAccept(topicMetadata -> {
+ if (topicMetadata.partitions > 0) {
+ log.warn("[{}] Not supported getMessageIdByIndex operation on "
+ + "partitioned-topic {}", clientAppId(), topicName);
+ throw new RestException(Status.METHOD_NOT_ALLOWED,
+ "GetMessageIDByIndex is not allowed on partitioned-topic");
+ }
+ });
+ }
+ }).thenCompose(ignore -> validateTopicOwnershipAsync(topicName, authoritative))
+ .thenCompose(__ -> getTopicReferenceAsync(topicName))
+ .thenCompose(topic -> {
+ if (!(topic instanceof PersistentTopic persistentTopic)) {
+ log.error("[{}] Get message id by index on a non-persistent topic {} is not allowed",
+ clientAppId(), topicName);
+ return FutureUtil.failedFuture(new RestException(Status.METHOD_NOT_ALLOWED,
+ "Get message id by index on a non-persistent topic is not allowed"));
+ }
+ ManagedLedger managedLedger = persistentTopic.getManagedLedger();
+ return findMessageIndexByPosition(
+ PositionFactory.create(managedLedger.getFirstPosition().getLedgerId(), 0),
+ managedLedger)
+ .thenCompose(firstIndex -> {
+ if (index < firstIndex) {
+ return CompletableFuture.completedFuture(PositionFactory.EARLIEST);
+ } else {
+ return managedLedger.asyncFindPosition(entry -> {
+ try {
+ Long messageIndex = getIndexFromEntry(entry);
+ if (messageIndex == null) {
+ return false; // Skip messages without index
+ } else {
+ // If the message index is less than the requested index,
+ // we continue searching
+ return messageIndex < index;
+ }
+ } catch (IOException e) {
+ log.error("Error deserialize message for message position find", e);
+ return false;
+ } finally {
+ entry.release();
+ }
+ });
+ }
+ }).thenCompose(position -> {
+ Position lastPosition = managedLedger.getLastConfirmedEntry();
+ if (position == null || position.compareTo(lastPosition) > 0) {
+ return FutureUtil.failedFuture(new RestException(Status.NOT_FOUND,
+ "Message not found for index " + index));
+ } else {
+ return CompletableFuture.completedFuture(position);
+ }
+ });
+ }).thenCompose(position -> CompletableFuture.completedFuture(
+ new MessageIdImpl(position.getLedgerId(), position.getEntryId(), partitionIndex)));
+ }
+
+ protected CompletableFuture<Long> findMessageIndexByPosition(Position position, ManagedLedger managedLedger) {
+ CompletableFuture<Long> indexFuture = new CompletableFuture<>();
+ managedLedger.asyncReadEntry(position, new AsyncCallbacks.ReadEntryCallback() {
+ @Override
+ public void readEntryComplete(Entry entry, Object ctx) {
+ try {
+ Long index = getIndexFromEntry(entry);
+ if (index == null) {
+ indexFuture.completeExceptionally(new RestException(Status.PRECONDITION_FAILED,
+ "Broker entry metadata is not present in the message"));
+ } else if (index < 0) {
+ indexFuture.completeExceptionally(new RestException(Status.PRECONDITION_FAILED,
+ "Invalid message index: " + index));
+ } else {
+ indexFuture.complete(index);
+ }
+ } catch (IOException e) {
+ indexFuture.completeExceptionally(new RestException(Status.INTERNAL_SERVER_ERROR,
+ "Failed to get index from entry: " + e.getMessage()));
+ } finally {
+ entry.release();
+ }
+ }
+
+ @Override
+ public void readEntryFailed(ManagedLedgerException exception, Object ctx) {
+ log.error("[{}] Failed to read position {} on topic {}",
+ clientAppId(), position, topicName, exception);
+ indexFuture.completeExceptionally(exception);
+ }
+ }, null);
+ return indexFuture;
+ }
+
+
+ private static Long getIndexFromEntry(Entry entry) throws IOException {
+ try {
+ final var brokerEntryMetadata = Commands.parseBrokerEntryMetadataIfExist(entry.getDataBuffer());
+ if (brokerEntryMetadata != null && brokerEntryMetadata.hasIndex()) {
+ return brokerEntryMetadata.getIndex();
+ } else {
+ return null;
+ }
+ } catch (Throwable throwable) {
+ throw new IOException(throwable);
Review Comment:
`Exception Translation` is a common practice. I think it is reasonable to just raise IOException here.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat][admin] PIP-415: Support getting message ID by index [pulsar]
Posted by "liangyepianzhou (via GitHub)" <gi...@apache.org>.
liangyepianzhou commented on code in PR #24222:
URL: https://github.com/apache/pulsar/pull/24222#discussion_r2125990993
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java:
##########
@@ -5489,4 +5489,114 @@ protected CompletableFuture<AutoSubscriptionCreationOverride> internalGetAutoSub
return null;
}));
}
+
+ protected CompletableFuture<MessageId> internalGetMessageIDByIndexAsync(Long index, boolean authoritative) {
+ if (!pulsar().getBrokerService().isBrokerEntryMetadataEnabled()) {
+ return FutureUtil.failedFuture(new RestException(Status.PRECONDITION_FAILED,
+ "GetMessageIDByIndex is not allowed when broker entry metadata is disabled"));
+ }
+ if (index == null || index < 0) {
+ return FutureUtil.failedFuture(new RestException(Status.NOT_FOUND,
+ "Invalid message index: " + index));
+ }
+ int partitionIndex = topicName.getPartitionIndex();
+ CompletableFuture<Void> future = validateTopicOperationAsync(topicName, TopicOperation.PEEK_MESSAGES);
+ return future.thenCompose(__ -> {
+ if (topicName.isGlobal()) {
+ return validateGlobalNamespaceOwnershipAsync(namespaceName);
+ } else {
+ return CompletableFuture.completedFuture(null);
+ }
+ }).thenCompose(__ -> {
+ if (topicName.isPartitioned()) {
+ return CompletableFuture.completedFuture(null);
+ } else {
+ return getPartitionedTopicMetadataAsync(topicName, authoritative, false)
+ .thenAccept(topicMetadata -> {
+ if (topicMetadata.partitions > 0) {
+ log.warn("[{}] Not supported getMessageIdByIndex operation on "
+ + "partitioned-topic {}", clientAppId(), topicName);
+ throw new RestException(Status.METHOD_NOT_ALLOWED,
+ "GetMessageIDByIndex is not allowed on partitioned-topic");
+ }
+ });
+ }
+ }).thenCompose(ignore -> validateTopicOwnershipAsync(topicName, authoritative))
+ .thenCompose(__ -> getTopicReferenceAsync(topicName))
+ .thenCompose(topic -> {
+ if (!(topic instanceof PersistentTopic persistentTopic)) {
+ log.error("[{}] Get message id by index on a non-persistent topic {} is not allowed",
+ clientAppId(), topicName);
+ return FutureUtil.failedFuture(new RestException(Status.METHOD_NOT_ALLOWED,
+ "Get message id by index on a non-persistent topic is not allowed"));
+ }
+ ManagedLedger managedLedger = persistentTopic.getManagedLedger();
+ return findMessageIndexByPosition(
+ PositionFactory.create(managedLedger.getFirstPosition().getLedgerId(), 0),
Review Comment:
>findMessageIndexByPosition(managedLedger.getFirstPosition(), ...) is better.
@nodece `managedLedger.getFirstPosition()` will return [first ledger Id, -1], regardless of whether the topic has data
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat][admin] PIP-415: Support getting message ID by index [pulsar]
Posted by "StevenLuMT (via GitHub)" <gi...@apache.org>.
StevenLuMT commented on code in PR #24222:
URL: https://github.com/apache/pulsar/pull/24222#discussion_r2125809062
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java:
##########
@@ -5489,4 +5489,129 @@ protected CompletableFuture<AutoSubscriptionCreationOverride> internalGetAutoSub
return null;
}));
}
+
+ protected CompletableFuture<MessageId> internalGetMessageIDByIndexAsync(Long index, boolean authoritative) {
+ if (!pulsar().getBrokerService().isBrokerEntryMetadataEnabled()) {
+ return FutureUtil.failedFuture(new RestException(Status.PRECONDITION_FAILED,
+ "GetMessageIDByIndex is not allowed when broker entry metadata is disabled"));
+ }
+ if (index == null || index < 0) {
+ return FutureUtil.failedFuture(new RestException(Status.NOT_FOUND,
+ "Invalid message index: " + index));
+ }
+ int partitionIndex = topicName.getPartitionIndex();
+ CompletableFuture<Void> future = validateTopicOperationAsync(topicName, TopicOperation.PEEK_MESSAGES);
+ return future.thenCompose(__ -> {
+ if (topicName.isGlobal()) {
+ return validateGlobalNamespaceOwnershipAsync(namespaceName);
+ } else {
+ return CompletableFuture.completedFuture(null);
+ }
+ }).thenCompose(__ -> {
+ if (topicName.isPartitioned()) {
+ return CompletableFuture.completedFuture(null);
+ } else {
+ return getPartitionedTopicMetadataAsync(topicName, authoritative, false)
+ .thenAccept(topicMetadata -> {
+ if (topicMetadata.partitions > 0) {
+ log.warn("[{}] Not supported getMessageIdByIndex operation on "
+ + "partitioned-topic {}", clientAppId(), topicName);
+ throw new RestException(Status.METHOD_NOT_ALLOWED,
+ "GetMessageIDByIndex is not allowed on partitioned-topic");
+ }
+ });
+ }
+ }).thenCompose(ignore -> validateTopicOwnershipAsync(topicName, authoritative))
+ .thenCompose(__ -> getTopicReferenceAsync(topicName))
+ .thenCompose(topic -> {
+ if (!(topic instanceof PersistentTopic persistentTopic)) {
+ log.error("[{}] Get message id by index on a non-persistent topic {} is not allowed",
+ clientAppId(), topicName);
+ return FutureUtil.failedFuture(new RestException(Status.METHOD_NOT_ALLOWED,
+ "Get message id by index on a non-persistent topic is not allowed"));
+ }
+ ManagedLedger managedLedger = persistentTopic.getManagedLedger();
+ return findMessageIndexByPosition(
+ PositionFactory.create(managedLedger.getFirstPosition().getLedgerId(), 0),
+ managedLedger)
+ .thenCompose(firstIndex -> {
+ if (index < firstIndex) {
+ return CompletableFuture.completedFuture(PositionFactory.EARLIEST);
+ } else {
+ return managedLedger.asyncFindPosition(entry -> {
+ try {
+ Long messageIndex = getIndexAndRelease(entry);
+ if (messageIndex == null) {
+ return false; // Skip messages without index
+ } else {
+ // If the message index is less than the requested index,
+ // we continue searching
+ return messageIndex < index;
+ }
+ } catch (IOException e) {
+ log.error("Error deserialize message for message position find", e);
+ return false;
+ }
+ });
+ }
+ }).thenCompose(position -> {
+ Position lastPosition = managedLedger.getLastConfirmedEntry();
+ if (position == null || position.compareTo(lastPosition) > 0) {
+ return FutureUtil.failedFuture(new RestException(Status.NOT_FOUND,
+ "Message not found for index " + index));
+ } else {
+ return CompletableFuture.completedFuture(position);
+ }
+ });
+ }).thenCompose(position -> CompletableFuture.completedFuture(
+ new MessageIdImpl(position.getLedgerId(), position.getEntryId(), partitionIndex)));
+ }
+
+ protected CompletableFuture<Long> findMessageIndexByPosition(Position position, ManagedLedger managedLedger) {
+ CompletableFuture<Long> indexFuture = new CompletableFuture<>();
+ managedLedger.asyncReadEntry(position, new AsyncCallbacks.ReadEntryCallback() {
+ @Override
+ public void readEntryComplete(Entry entry, Object ctx) {
+ try {
+ Long index = getIndexAndRelease(entry);
+ if (index == null) {
+ indexFuture.completeExceptionally(new RestException(Status.PRECONDITION_FAILED,
+ "Broker entry metadata is not present in the message"));
+ } else if (index < 0) {
+ indexFuture.completeExceptionally(new RestException(Status.PRECONDITION_FAILED,
+ "Invalid message index: " + index));
+ } else {
+ indexFuture.complete(index);
+ }
+ } catch (IOException e) {
+ indexFuture.completeExceptionally(new RestException(Status.INTERNAL_SERVER_ERROR,
+ "Failed to get index from entry: " + e.getMessage()));
+ }
+ }
+
+ @Override
+ public void readEntryFailed(ManagedLedgerException exception, Object ctx) {
+ log.error("[{}] Failed to read position {} on topic {}",
+ clientAppId(), position, topicName, exception);
+ indexFuture.completeExceptionally(exception);
+ }
+ }, null);
+ return indexFuture;
+ }
+
+
+ private static Long getIndexAndRelease(Entry entry) throws IOException {
+ try {
Review Comment:
Do not nest release in the method to prevent others from using entry after this method. Please mention the method caller to implement release
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java:
##########
@@ -5489,4 +5489,129 @@ protected CompletableFuture<AutoSubscriptionCreationOverride> internalGetAutoSub
return null;
}));
}
+
+ protected CompletableFuture<MessageId> internalGetMessageIDByIndexAsync(Long index, boolean authoritative) {
+ if (!pulsar().getBrokerService().isBrokerEntryMetadataEnabled()) {
+ return FutureUtil.failedFuture(new RestException(Status.PRECONDITION_FAILED,
+ "GetMessageIDByIndex is not allowed when broker entry metadata is disabled"));
+ }
+ if (index == null || index < 0) {
+ return FutureUtil.failedFuture(new RestException(Status.NOT_FOUND,
+ "Invalid message index: " + index));
+ }
+ int partitionIndex = topicName.getPartitionIndex();
+ CompletableFuture<Void> future = validateTopicOperationAsync(topicName, TopicOperation.PEEK_MESSAGES);
+ return future.thenCompose(__ -> {
+ if (topicName.isGlobal()) {
+ return validateGlobalNamespaceOwnershipAsync(namespaceName);
+ } else {
+ return CompletableFuture.completedFuture(null);
+ }
+ }).thenCompose(__ -> {
+ if (topicName.isPartitioned()) {
+ return CompletableFuture.completedFuture(null);
+ } else {
+ return getPartitionedTopicMetadataAsync(topicName, authoritative, false)
+ .thenAccept(topicMetadata -> {
+ if (topicMetadata.partitions > 0) {
+ log.warn("[{}] Not supported getMessageIdByIndex operation on "
+ + "partitioned-topic {}", clientAppId(), topicName);
+ throw new RestException(Status.METHOD_NOT_ALLOWED,
+ "GetMessageIDByIndex is not allowed on partitioned-topic");
+ }
+ });
+ }
+ }).thenCompose(ignore -> validateTopicOwnershipAsync(topicName, authoritative))
+ .thenCompose(__ -> getTopicReferenceAsync(topicName))
+ .thenCompose(topic -> {
+ if (!(topic instanceof PersistentTopic persistentTopic)) {
+ log.error("[{}] Get message id by index on a non-persistent topic {} is not allowed",
+ clientAppId(), topicName);
+ return FutureUtil.failedFuture(new RestException(Status.METHOD_NOT_ALLOWED,
+ "Get message id by index on a non-persistent topic is not allowed"));
+ }
+ ManagedLedger managedLedger = persistentTopic.getManagedLedger();
+ return findMessageIndexByPosition(
+ PositionFactory.create(managedLedger.getFirstPosition().getLedgerId(), 0),
+ managedLedger)
+ .thenCompose(firstIndex -> {
+ if (index < firstIndex) {
+ return CompletableFuture.completedFuture(PositionFactory.EARLIEST);
+ } else {
+ return managedLedger.asyncFindPosition(entry -> {
+ try {
+ Long messageIndex = getIndexAndRelease(entry);
+ if (messageIndex == null) {
+ return false; // Skip messages without index
+ } else {
+ // If the message index is less than the requested index,
+ // we continue searching
+ return messageIndex < index;
+ }
+ } catch (IOException e) {
+ log.error("Error deserialize message for message position find", e);
+ return false;
+ }
+ });
+ }
+ }).thenCompose(position -> {
+ Position lastPosition = managedLedger.getLastConfirmedEntry();
+ if (position == null || position.compareTo(lastPosition) > 0) {
+ return FutureUtil.failedFuture(new RestException(Status.NOT_FOUND,
+ "Message not found for index " + index));
+ } else {
+ return CompletableFuture.completedFuture(position);
+ }
+ });
+ }).thenCompose(position -> CompletableFuture.completedFuture(
+ new MessageIdImpl(position.getLedgerId(), position.getEntryId(), partitionIndex)));
+ }
+
+ protected CompletableFuture<Long> findMessageIndexByPosition(Position position, ManagedLedger managedLedger) {
+ CompletableFuture<Long> indexFuture = new CompletableFuture<>();
+ managedLedger.asyncReadEntry(position, new AsyncCallbacks.ReadEntryCallback() {
+ @Override
+ public void readEntryComplete(Entry entry, Object ctx) {
+ try {
+ Long index = getIndexAndRelease(entry);
+ if (index == null) {
+ indexFuture.completeExceptionally(new RestException(Status.PRECONDITION_FAILED,
+ "Broker entry metadata is not present in the message"));
+ } else if (index < 0) {
+ indexFuture.completeExceptionally(new RestException(Status.PRECONDITION_FAILED,
+ "Invalid message index: " + index));
+ } else {
+ indexFuture.complete(index);
+ }
+ } catch (IOException e) {
+ indexFuture.completeExceptionally(new RestException(Status.INTERNAL_SERVER_ERROR,
+ "Failed to get index from entry: " + e.getMessage()));
+ }
+ }
+
+ @Override
+ public void readEntryFailed(ManagedLedgerException exception, Object ctx) {
+ log.error("[{}] Failed to read position {} on topic {}",
+ clientAppId(), position, topicName, exception);
+ indexFuture.completeExceptionally(exception);
+ }
+ }, null);
+ return indexFuture;
+ }
+
+
+ private static Long getIndexAndRelease(Entry entry) throws IOException {
+ try {
Review Comment:
Do not nest `release` in the method to prevent others from using entry after this method. Please mention the method caller to implement release
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat][admin] PIP-415: Support getting message ID by index [pulsar]
Posted by "nodece (via GitHub)" <gi...@apache.org>.
nodece commented on code in PR #24222:
URL: https://github.com/apache/pulsar/pull/24222#discussion_r2126179551
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java:
##########
@@ -5489,4 +5489,131 @@ protected CompletableFuture<AutoSubscriptionCreationOverride> internalGetAutoSub
return null;
}));
}
+
+ protected CompletableFuture<MessageId> internalGetMessageIDByIndexAsync(Long index, boolean authoritative) {
+ if (!pulsar().getBrokerService().isBrokerEntryMetadataEnabled()) {
+ return FutureUtil.failedFuture(new RestException(Status.PRECONDITION_FAILED,
+ "GetMessageIDByIndex is not allowed when broker entry metadata is disabled"));
+ }
+ if (index == null || index < 0) {
+ return FutureUtil.failedFuture(new RestException(Status.NOT_FOUND,
+ "Invalid message index: " + index));
+ }
+ int partitionIndex = topicName.getPartitionIndex();
+ CompletableFuture<Void> future = validateTopicOperationAsync(topicName, TopicOperation.PEEK_MESSAGES);
+ return future.thenCompose(__ -> {
+ if (topicName.isGlobal()) {
+ return validateGlobalNamespaceOwnershipAsync(namespaceName);
+ } else {
+ return CompletableFuture.completedFuture(null);
+ }
+ }).thenCompose(__ -> {
+ if (topicName.isPartitioned()) {
+ return CompletableFuture.completedFuture(null);
+ } else {
+ return getPartitionedTopicMetadataAsync(topicName, authoritative, false)
+ .thenAccept(topicMetadata -> {
+ if (topicMetadata.partitions > 0) {
+ log.warn("[{}] Not supported getMessageIdByIndex operation on "
+ + "partitioned-topic {}", clientAppId(), topicName);
+ throw new RestException(Status.METHOD_NOT_ALLOWED,
+ "GetMessageIDByIndex is not allowed on partitioned-topic");
+ }
+ });
+ }
+ }).thenCompose(ignore -> validateTopicOwnershipAsync(topicName, authoritative))
+ .thenCompose(__ -> getTopicReferenceAsync(topicName))
+ .thenCompose(topic -> {
+ if (!(topic instanceof PersistentTopic persistentTopic)) {
+ log.error("[{}] Get message id by index on a non-persistent topic {} is not allowed",
+ clientAppId(), topicName);
+ return FutureUtil.failedFuture(new RestException(Status.METHOD_NOT_ALLOWED,
+ "Get message id by index on a non-persistent topic is not allowed"));
+ }
+ ManagedLedger managedLedger = persistentTopic.getManagedLedger();
+ return findMessageIndexByPosition(
+ PositionFactory.create(managedLedger.getFirstPosition().getLedgerId(), 0),
+ managedLedger)
+ .thenCompose(firstIndex -> {
+ if (index < firstIndex) {
+ return CompletableFuture.completedFuture(PositionFactory.EARLIEST);
Review Comment:
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.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat][admin] PIP-415: Support getting message ID by index [pulsar]
Posted by "liangyepianzhou (via GitHub)" <gi...@apache.org>.
liangyepianzhou commented on code in PR #24222:
URL: https://github.com/apache/pulsar/pull/24222#discussion_r2126133107
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java:
##########
@@ -5489,4 +5489,131 @@ protected CompletableFuture<AutoSubscriptionCreationOverride> internalGetAutoSub
return null;
}));
}
+
+ protected CompletableFuture<MessageId> internalGetMessageIDByIndexAsync(Long index, boolean authoritative) {
+ if (!pulsar().getBrokerService().isBrokerEntryMetadataEnabled()) {
+ return FutureUtil.failedFuture(new RestException(Status.PRECONDITION_FAILED,
+ "GetMessageIDByIndex is not allowed when broker entry metadata is disabled"));
+ }
+ if (index == null || index < 0) {
+ return FutureUtil.failedFuture(new RestException(Status.NOT_FOUND,
+ "Invalid message index: " + index));
+ }
+ int partitionIndex = topicName.getPartitionIndex();
+ CompletableFuture<Void> future = validateTopicOperationAsync(topicName, TopicOperation.PEEK_MESSAGES);
+ return future.thenCompose(__ -> {
+ if (topicName.isGlobal()) {
+ return validateGlobalNamespaceOwnershipAsync(namespaceName);
+ } else {
+ return CompletableFuture.completedFuture(null);
+ }
+ }).thenCompose(__ -> {
+ if (topicName.isPartitioned()) {
+ return CompletableFuture.completedFuture(null);
+ } else {
+ return getPartitionedTopicMetadataAsync(topicName, authoritative, false)
+ .thenAccept(topicMetadata -> {
+ if (topicMetadata.partitions > 0) {
+ log.warn("[{}] Not supported getMessageIdByIndex operation on "
+ + "partitioned-topic {}", clientAppId(), topicName);
+ throw new RestException(Status.METHOD_NOT_ALLOWED,
+ "GetMessageIDByIndex is not allowed on partitioned-topic");
+ }
+ });
+ }
+ }).thenCompose(ignore -> validateTopicOwnershipAsync(topicName, authoritative))
+ .thenCompose(__ -> getTopicReferenceAsync(topicName))
+ .thenCompose(topic -> {
+ if (!(topic instanceof PersistentTopic persistentTopic)) {
+ log.error("[{}] Get message id by index on a non-persistent topic {} is not allowed",
+ clientAppId(), topicName);
+ return FutureUtil.failedFuture(new RestException(Status.METHOD_NOT_ALLOWED,
+ "Get message id by index on a non-persistent topic is not allowed"));
+ }
+ ManagedLedger managedLedger = persistentTopic.getManagedLedger();
+ return findMessageIndexByPosition(
+ PositionFactory.create(managedLedger.getFirstPosition().getLedgerId(), 0),
+ managedLedger)
+ .thenCompose(firstIndex -> {
+ if (index < firstIndex) {
+ return CompletableFuture.completedFuture(PositionFactory.EARLIEST);
Review Comment:
Maybe you can get more context here
https://github.com/apache/pulsar/pull/24220/files#r2090841224
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat][admin] PIP-415: Support getting message ID by index [pulsar]
Posted by "liangyepianzhou (via GitHub)" <gi...@apache.org>.
liangyepianzhou commented on code in PR #24222:
URL: https://github.com/apache/pulsar/pull/24222#discussion_r2115240448
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java:
##########
@@ -5489,4 +5489,111 @@ protected CompletableFuture<AutoSubscriptionCreationOverride> internalGetAutoSub
return null;
}));
}
+
+ protected CompletableFuture<MessageId> internalGetMessageIDByIndexAsync(Long index, boolean authoritative) {
+ if (!pulsar().getBrokerService().isBrokerEntryMetadataEnabled()) {
+ return FutureUtil.failedFuture(new RestException(Status.PRECONDITION_FAILED,
+ "GetMessageIDByIndex is not allowed when broker entry metadata is disabled"));
+ }
+ if (index == null || index < 0) {
+ return FutureUtil.failedFuture(new RestException(Status.NOT_FOUND,
+ "Invalid message index: " + index));
+ }
+ int partitionIndex = topicName.getPartitionIndex();
+ CompletableFuture<Void> future = validateTopicOperationAsync(topicName, TopicOperation.PEEK_MESSAGES);
+ return future.thenCompose(__ -> {
+ if (topicName.isGlobal()) {
+ return validateGlobalNamespaceOwnershipAsync(namespaceName);
+ } else {
+ return CompletableFuture.completedFuture(null);
+ }
+ }).thenCompose(__ -> {
+ if (topicName.isPartitioned()) {
+ return CompletableFuture.completedFuture(null);
+ } else {
+ return getPartitionedTopicMetadataAsync(topicName, authoritative, false)
+ .thenAccept(topicMetadata -> {
+ if (topicMetadata.partitions > 0) {
+ log.warn("[{}] Not supported getMessageIdByIndex operation on "
+ + "partitioned-topic {}", clientAppId(), topicName);
+ throw new RestException(Status.METHOD_NOT_ALLOWED,
+ "GetMessageIDByIndex is not allowed on partitioned-topic");
+ }
+ });
+ }
+ }).thenCompose(ignore -> validateTopicOwnershipAsync(topicName, authoritative))
+ .thenCompose(__ -> getTopicReferenceAsync(topicName))
+ .thenCompose(topic -> {
+ if (!(topic instanceof PersistentTopic persistentTopic)) {
+ log.error("[{}] Get message id by index on a non-persistent topic {} is not allowed",
+ clientAppId(), topicName);
+ return FutureUtil.failedFuture(new RestException(Status.METHOD_NOT_ALLOWED,
+ "Get message id by index on a non-persistent topic is not allowed"));
+ }
+ ManagedLedger managedLedger = persistentTopic.getManagedLedger();
+ return findMessageIndexByPosition(
+ PositionFactory.create(managedLedger.getFirstPosition().getLedgerId(), 0),
+ managedLedger)
+ .thenCompose(firstIndex -> {
+ if (index < firstIndex) {
+ return CompletableFuture.completedFuture(PositionFactory.EARLIEST);
+ } else {
+ return managedLedger.asyncFindPosition(entry -> {
+ try {
+ BrokerEntryMetadata brokerEntryMetadata =
+ Commands.parseBrokerEntryMetadataIfExist(entry.getDataBuffer());
+ assert brokerEntryMetadata != null;
Review Comment:
Good catch! Thanks.
First, we have checked that the broker configuration `isBrokerEntryMetadataEnabled = true`, so the entries distribution should be **OOOOOO** or **XXXOOO**.
Second, for the **XXXOOO** scenario, before the lookup completes, we cannot determine whether the user-provided index points to **X** or **O**. Directly failing the request might not be ideal. For example, if a user's message retention time is set to 7 days or even a month, they would be unable to use this API for an entire month after enabling `isBrokerEntryMetadataEnabled`, even if they want to query the latest message ID by index.
The example in [https://github.com/streamnative/kop/pull/916](https://github.com/streamnative/kop/pull/916) is correct—we can skip messages without an index. Additionally, this option could be set to `true` by default. As you mentioned, adding an extra option would make this API harder to understand.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat][admin] PIP-415: Support getting message ID by index [pulsar]
Posted by "liangyepianzhou (via GitHub)" <gi...@apache.org>.
liangyepianzhou commented on code in PR #24222:
URL: https://github.com/apache/pulsar/pull/24222#discussion_r2122511754
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java:
##########
@@ -5489,4 +5489,114 @@ protected CompletableFuture<AutoSubscriptionCreationOverride> internalGetAutoSub
return null;
}));
}
+
+ protected CompletableFuture<MessageId> internalGetMessageIDByIndexAsync(Long index, boolean authoritative) {
+ if (!pulsar().getBrokerService().isBrokerEntryMetadataEnabled()) {
+ return FutureUtil.failedFuture(new RestException(Status.PRECONDITION_FAILED,
+ "GetMessageIDByIndex is not allowed when broker entry metadata is disabled"));
+ }
+ if (index == null || index < 0) {
+ return FutureUtil.failedFuture(new RestException(Status.NOT_FOUND,
+ "Invalid message index: " + index));
+ }
+ int partitionIndex = topicName.getPartitionIndex();
+ CompletableFuture<Void> future = validateTopicOperationAsync(topicName, TopicOperation.PEEK_MESSAGES);
+ return future.thenCompose(__ -> {
+ if (topicName.isGlobal()) {
+ return validateGlobalNamespaceOwnershipAsync(namespaceName);
+ } else {
+ return CompletableFuture.completedFuture(null);
+ }
+ }).thenCompose(__ -> {
+ if (topicName.isPartitioned()) {
+ return CompletableFuture.completedFuture(null);
+ } else {
+ return getPartitionedTopicMetadataAsync(topicName, authoritative, false)
+ .thenAccept(topicMetadata -> {
+ if (topicMetadata.partitions > 0) {
+ log.warn("[{}] Not supported getMessageIdByIndex operation on "
+ + "partitioned-topic {}", clientAppId(), topicName);
+ throw new RestException(Status.METHOD_NOT_ALLOWED,
+ "GetMessageIDByIndex is not allowed on partitioned-topic");
+ }
+ });
+ }
+ }).thenCompose(ignore -> validateTopicOwnershipAsync(topicName, authoritative))
+ .thenCompose(__ -> getTopicReferenceAsync(topicName))
+ .thenCompose(topic -> {
+ if (!(topic instanceof PersistentTopic persistentTopic)) {
+ log.error("[{}] Get message id by index on a non-persistent topic {} is not allowed",
+ clientAppId(), topicName);
+ return FutureUtil.failedFuture(new RestException(Status.METHOD_NOT_ALLOWED,
+ "Get message id by index on a non-persistent topic is not allowed"));
+ }
+ ManagedLedger managedLedger = persistentTopic.getManagedLedger();
+ return findMessageIndexByPosition(
+ PositionFactory.create(managedLedger.getFirstPosition().getLedgerId(), 0),
+ managedLedger)
+ .thenCompose(firstIndex -> {
+ if (index < firstIndex) {
+ return CompletableFuture.completedFuture(PositionFactory.EARLIEST);
+ } else {
+ return managedLedger.asyncFindPosition(entry -> {
+ try {
+ BrokerEntryMetadata brokerEntryMetadata =
+ Commands.parseBrokerEntryMetadataIfExist(entry.getDataBuffer());
+ // Skip messages without index
+ if (brokerEntryMetadata == null) {
+ return true;
+ }
+ return brokerEntryMetadata.getIndex() < index;
+ } catch (Exception e) {
+ log.error("Error deserialize message for message position find", e);
+ } finally {
+ entry.release();
+ }
+ return false;
+ });
+ }
+ }).thenCompose(position -> {
+ Position lastPosition = managedLedger.getLastConfirmedEntry();
+ if (position == null || position.compareTo(lastPosition) > 0) {
+ return FutureUtil.failedFuture(new RestException(Status.NOT_FOUND,
+ "Message not found for index " + index));
+ } else {
+ return CompletableFuture.completedFuture(position);
+ }
+ });
+ }).thenCompose(position -> CompletableFuture.completedFuture(
+ new MessageIdImpl(position.getLedgerId(), position.getEntryId(), partitionIndex)));
+ }
+
+ protected CompletableFuture<Long> findMessageIndexByPosition(Position position, ManagedLedger managedLedger) {
+ CompletableFuture<Long> indexFuture = new CompletableFuture<>();
+ managedLedger.asyncReadEntry(position, new AsyncCallbacks.ReadEntryCallback() {
+ @Override
+ public void readEntryComplete(Entry entry, Object ctx) {
+ BrokerEntryMetadata brokerEntryMetadata =
+ Commands.parseBrokerEntryMetadataIfExist(entry.getDataBuffer());
+ if (brokerEntryMetadata == null) {
+ indexFuture.completeExceptionally(new RestException(Status.PRECONDITION_FAILED,
+ "Broker entry metadata is not present in the message"));
+ } else {
+ long index = brokerEntryMetadata.getIndex();
+ if (index < 0) {
+ indexFuture.completeExceptionally(new RestException(Status.PRECONDITION_FAILED,
+ "Invalid message index: " + index));
+ } else {
+ indexFuture.complete(index);
+ }
+ }
Review Comment:
Thanks~, I add a common method to handle it.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat][admin] PIP-415: Support getting message ID by index [pulsar]
Posted by "liangyepianzhou (via GitHub)" <gi...@apache.org>.
liangyepianzhou commented on PR #24222:
URL: https://github.com/apache/pulsar/pull/24222#issuecomment-2933265649
> 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 https://github.com/apache/pulsar/issues/24374.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat][admin] PIP-415: Support getting message ID by offset [pulsar]
Posted by "AuroraTwinkle (via GitHub)" <gi...@apache.org>.
AuroraTwinkle commented on code in PR #24222:
URL: https://github.com/apache/pulsar/pull/24222#discussion_r2062804195
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java:
##########
@@ -5489,4 +5489,53 @@ protected CompletableFuture<AutoSubscriptionCreationOverride> internalGetAutoSub
return null;
}));
}
+
+ protected CompletableFuture<MessageId> internalGetMessageIDByOffsetAsync(Long offset, boolean authoritative) {
+ int partitionIndex = topicName.getPartitionIndex();
+ CompletableFuture<Void> future = validateTopicOperationAsync(topicName, TopicOperation.PEEK_MESSAGES);
+ return future.thenCompose(__ -> {
+ if (topicName.isGlobal()) {
+ return validateGlobalNamespaceOwnershipAsync(namespaceName);
+ } else {
+ return CompletableFuture.completedFuture(null);
+ }
+ }).thenCompose(__ -> {
+ if (topicName.isPartitioned()) {
+ return CompletableFuture.completedFuture(null);
+ } else {
+ return getPartitionedTopicMetadataAsync(topicName, authoritative, false)
+ .thenAccept(topicMetadata -> {
+ if (topicMetadata.partitions > 0) {
+ log.warn("[{}] Not supported getMessageIdByOffset operation on "
+ + "partitioned-topic {}", clientAppId(), topicName);
+ throw new RestException(Status.METHOD_NOT_ALLOWED,
+ "GetMessageIDByOffset is not allowed on partitioned-topic");
+ }
+ });
+ }
+ }).thenCompose(ignore -> validateTopicOwnershipAsync(topicName, authoritative))
+ .thenCompose(__ -> getTopicReferenceAsync(topicName))
+ .thenCompose(topic -> {
+ if (!(topic instanceof PersistentTopic persistentTopic)) {
+ log.error("[{}] Get message id by offset on a non-persistent topic {} is not allowed",
+ clientAppId(), topicName);
+ return FutureUtil.failedFuture(new RestException(Status.METHOD_NOT_ALLOWED,
+ "Get message id by offset on a non-persistent topic is not allowed"));
+ }
+ ManagedLedger managedLedger = persistentTopic.getManagedLedger();
+ return managedLedger.asyncFindPosition(entry -> {
+ try {
+ BrokerEntryMetadata brokerEntryMetadata =
+ Commands.parseBrokerEntryMetadataIfExist(entry.getDataBuffer());
+ return brokerEntryMetadata.getIndex() < offset;
Review Comment:
maybe `brokerEntryMetadata.getIndex() >= offset`?
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat][admin] PIP-415: Support getting message ID by index [pulsar]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on code in PR #24222:
URL: https://github.com/apache/pulsar/pull/24222#discussion_r2115277817
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java:
##########
@@ -5489,4 +5489,111 @@ protected CompletableFuture<AutoSubscriptionCreationOverride> internalGetAutoSub
return null;
}));
}
+
+ protected CompletableFuture<MessageId> internalGetMessageIDByIndexAsync(Long index, boolean authoritative) {
+ if (!pulsar().getBrokerService().isBrokerEntryMetadataEnabled()) {
+ return FutureUtil.failedFuture(new RestException(Status.PRECONDITION_FAILED,
+ "GetMessageIDByIndex is not allowed when broker entry metadata is disabled"));
+ }
+ if (index == null || index < 0) {
+ return FutureUtil.failedFuture(new RestException(Status.NOT_FOUND,
+ "Invalid message index: " + index));
+ }
+ int partitionIndex = topicName.getPartitionIndex();
+ CompletableFuture<Void> future = validateTopicOperationAsync(topicName, TopicOperation.PEEK_MESSAGES);
+ return future.thenCompose(__ -> {
+ if (topicName.isGlobal()) {
+ return validateGlobalNamespaceOwnershipAsync(namespaceName);
+ } else {
+ return CompletableFuture.completedFuture(null);
+ }
+ }).thenCompose(__ -> {
+ if (topicName.isPartitioned()) {
+ return CompletableFuture.completedFuture(null);
+ } else {
+ return getPartitionedTopicMetadataAsync(topicName, authoritative, false)
+ .thenAccept(topicMetadata -> {
+ if (topicMetadata.partitions > 0) {
+ log.warn("[{}] Not supported getMessageIdByIndex operation on "
+ + "partitioned-topic {}", clientAppId(), topicName);
+ throw new RestException(Status.METHOD_NOT_ALLOWED,
+ "GetMessageIDByIndex is not allowed on partitioned-topic");
+ }
+ });
+ }
+ }).thenCompose(ignore -> validateTopicOwnershipAsync(topicName, authoritative))
+ .thenCompose(__ -> getTopicReferenceAsync(topicName))
+ .thenCompose(topic -> {
+ if (!(topic instanceof PersistentTopic persistentTopic)) {
+ log.error("[{}] Get message id by index on a non-persistent topic {} is not allowed",
+ clientAppId(), topicName);
+ return FutureUtil.failedFuture(new RestException(Status.METHOD_NOT_ALLOWED,
+ "Get message id by index on a non-persistent topic is not allowed"));
+ }
+ ManagedLedger managedLedger = persistentTopic.getManagedLedger();
+ return findMessageIndexByPosition(
+ PositionFactory.create(managedLedger.getFirstPosition().getLedgerId(), 0),
+ managedLedger)
+ .thenCompose(firstIndex -> {
+ if (index < firstIndex) {
+ return CompletableFuture.completedFuture(PositionFactory.EARLIEST);
+ } else {
+ return managedLedger.asyncFindPosition(entry -> {
+ try {
+ BrokerEntryMetadata brokerEntryMetadata =
+ Commands.parseBrokerEntryMetadataIfExist(entry.getDataBuffer());
+ assert brokerEntryMetadata != null;
Review Comment:
Then return `true` instead of `false` here will be the best solution.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat][admin] PIP-415: Support getting message ID by index [pulsar]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on code in PR #24222:
URL: https://github.com/apache/pulsar/pull/24222#discussion_r2113922366
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java:
##########
@@ -5489,4 +5489,111 @@ protected CompletableFuture<AutoSubscriptionCreationOverride> internalGetAutoSub
return null;
}));
}
+
+ protected CompletableFuture<MessageId> internalGetMessageIDByIndexAsync(Long index, boolean authoritative) {
+ if (!pulsar().getBrokerService().isBrokerEntryMetadataEnabled()) {
+ return FutureUtil.failedFuture(new RestException(Status.PRECONDITION_FAILED,
+ "GetMessageIDByIndex is not allowed when broker entry metadata is disabled"));
+ }
+ if (index == null || index < 0) {
+ return FutureUtil.failedFuture(new RestException(Status.NOT_FOUND,
+ "Invalid message index: " + index));
+ }
+ int partitionIndex = topicName.getPartitionIndex();
+ CompletableFuture<Void> future = validateTopicOperationAsync(topicName, TopicOperation.PEEK_MESSAGES);
+ return future.thenCompose(__ -> {
+ if (topicName.isGlobal()) {
+ return validateGlobalNamespaceOwnershipAsync(namespaceName);
+ } else {
+ return CompletableFuture.completedFuture(null);
+ }
+ }).thenCompose(__ -> {
+ if (topicName.isPartitioned()) {
+ return CompletableFuture.completedFuture(null);
+ } else {
+ return getPartitionedTopicMetadataAsync(topicName, authoritative, false)
+ .thenAccept(topicMetadata -> {
+ if (topicMetadata.partitions > 0) {
+ log.warn("[{}] Not supported getMessageIdByIndex operation on "
+ + "partitioned-topic {}", clientAppId(), topicName);
+ throw new RestException(Status.METHOD_NOT_ALLOWED,
+ "GetMessageIDByIndex is not allowed on partitioned-topic");
+ }
+ });
+ }
+ }).thenCompose(ignore -> validateTopicOwnershipAsync(topicName, authoritative))
+ .thenCompose(__ -> getTopicReferenceAsync(topicName))
+ .thenCompose(topic -> {
+ if (!(topic instanceof PersistentTopic persistentTopic)) {
+ log.error("[{}] Get message id by index on a non-persistent topic {} is not allowed",
+ clientAppId(), topicName);
+ return FutureUtil.failedFuture(new RestException(Status.METHOD_NOT_ALLOWED,
+ "Get message id by index on a non-persistent topic is not allowed"));
+ }
+ ManagedLedger managedLedger = persistentTopic.getManagedLedger();
+ return findMessageIndexByPosition(
+ PositionFactory.create(managedLedger.getFirstPosition().getLedgerId(), 0),
+ managedLedger)
+ .thenCompose(firstIndex -> {
+ if (index < firstIndex) {
+ return CompletableFuture.completedFuture(PositionFactory.EARLIEST);
+ } else {
+ return managedLedger.asyncFindPosition(entry -> {
+ try {
+ BrokerEntryMetadata brokerEntryMetadata =
+ Commands.parseBrokerEntryMetadataIfExist(entry.getDataBuffer());
+ assert brokerEntryMetadata != null;
Review Comment:
`assert` should not be used here. It should only work in tests via the `-ea` option.
In production, if `asyncFindPosition` encounters an entry without `BrokerEntryMetadata`, NPE will be thrown here.
First, many unnecessary error logs will be printed.
Second, returning false does not make sense, it will search to the left side. If the topic includes entries like: XXXOO, where `X` is the entry without index and `O` is the entry with index, eventually 0 will be returned.
Here are some solutions to handle this case:
- Fail the search directly if the topic contains entries without broker entry metadata
- Add an option to control the behavior (return false to left or return true to right). But this option assumes the entries distribution is either XX...OO... or OO...XX....For entries like OXOXOX, it does not work. In addition, an extra option makes this API hard to understand.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat][admin] PIP-415: Support getting message ID by index [pulsar]
Posted by "StevenLuMT (via GitHub)" <gi...@apache.org>.
StevenLuMT commented on code in PR #24222:
URL: https://github.com/apache/pulsar/pull/24222#discussion_r2125780470
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java:
##########
@@ -5489,4 +5489,129 @@ protected CompletableFuture<AutoSubscriptionCreationOverride> internalGetAutoSub
return null;
}));
}
+
+ protected CompletableFuture<MessageId> internalGetMessageIDByIndexAsync(Long index, boolean authoritative) {
+ if (!pulsar().getBrokerService().isBrokerEntryMetadataEnabled()) {
+ return FutureUtil.failedFuture(new RestException(Status.PRECONDITION_FAILED,
+ "GetMessageIDByIndex is not allowed when broker entry metadata is disabled"));
+ }
+ if (index == null || index < 0) {
+ return FutureUtil.failedFuture(new RestException(Status.NOT_FOUND,
+ "Invalid message index: " + index));
+ }
+ int partitionIndex = topicName.getPartitionIndex();
+ CompletableFuture<Void> future = validateTopicOperationAsync(topicName, TopicOperation.PEEK_MESSAGES);
+ return future.thenCompose(__ -> {
+ if (topicName.isGlobal()) {
+ return validateGlobalNamespaceOwnershipAsync(namespaceName);
+ } else {
+ return CompletableFuture.completedFuture(null);
+ }
+ }).thenCompose(__ -> {
+ if (topicName.isPartitioned()) {
+ return CompletableFuture.completedFuture(null);
+ } else {
+ return getPartitionedTopicMetadataAsync(topicName, authoritative, false)
+ .thenAccept(topicMetadata -> {
+ if (topicMetadata.partitions > 0) {
+ log.warn("[{}] Not supported getMessageIdByIndex operation on "
+ + "partitioned-topic {}", clientAppId(), topicName);
+ throw new RestException(Status.METHOD_NOT_ALLOWED,
+ "GetMessageIDByIndex is not allowed on partitioned-topic");
+ }
+ });
+ }
+ }).thenCompose(ignore -> validateTopicOwnershipAsync(topicName, authoritative))
+ .thenCompose(__ -> getTopicReferenceAsync(topicName))
+ .thenCompose(topic -> {
+ if (!(topic instanceof PersistentTopic persistentTopic)) {
+ log.error("[{}] Get message id by index on a non-persistent topic {} is not allowed",
+ clientAppId(), topicName);
+ return FutureUtil.failedFuture(new RestException(Status.METHOD_NOT_ALLOWED,
+ "Get message id by index on a non-persistent topic is not allowed"));
+ }
+ ManagedLedger managedLedger = persistentTopic.getManagedLedger();
+ return findMessageIndexByPosition(
+ PositionFactory.create(managedLedger.getFirstPosition().getLedgerId(), 0),
+ managedLedger)
+ .thenCompose(firstIndex -> {
+ if (index < firstIndex) {
+ return CompletableFuture.completedFuture(PositionFactory.EARLIEST);
+ } else {
+ return managedLedger.asyncFindPosition(entry -> {
+ try {
+ Long messageIndex = getIndexAndRelease(entry);
+ if (messageIndex == null) {
+ return false; // Skip messages without index
+ } else {
+ // If the message index is less than the requested index,
+ // we continue searching
+ return messageIndex < index;
+ }
+ } catch (IOException e) {
+ log.error("Error deserialize message for message position find", e);
Review Comment:
1. When this place is thrown wrong, print out the `entry`
2. Should `IOException` be replaced with `Throwable` ?
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat][admin] PIP-415: Support getting message ID by index [pulsar]
Posted by "liangyepianzhou (via GitHub)" <gi...@apache.org>.
liangyepianzhou commented on code in PR #24222:
URL: https://github.com/apache/pulsar/pull/24222#discussion_r2128058546
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerEntryMetadataE2ETest.java:
##########
@@ -19,20 +19,25 @@
package org.apache.pulsar.broker.service;
import static org.apache.bookkeeper.mledger.proto.MLDataFormats.ManagedLedgerInfo.LedgerInfo;
+import static org.assertj.core.api.AssertionsForClassTypes.assertThatThrownBy;
Review Comment:
The difference between these two imports lies in the fact that they come from different entry point classes of the AssertJ library, but their actual functionality is identical. However, using `Assertions` is indeed the more standard practice.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat][admin] PIP-415: Support getting message ID by index [pulsar]
Posted by "StevenLuMT (via GitHub)" <gi...@apache.org>.
StevenLuMT commented on code in PR #24222:
URL: https://github.com/apache/pulsar/pull/24222#discussion_r2125809062
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java:
##########
@@ -5489,4 +5489,129 @@ protected CompletableFuture<AutoSubscriptionCreationOverride> internalGetAutoSub
return null;
}));
}
+
+ protected CompletableFuture<MessageId> internalGetMessageIDByIndexAsync(Long index, boolean authoritative) {
+ if (!pulsar().getBrokerService().isBrokerEntryMetadataEnabled()) {
+ return FutureUtil.failedFuture(new RestException(Status.PRECONDITION_FAILED,
+ "GetMessageIDByIndex is not allowed when broker entry metadata is disabled"));
+ }
+ if (index == null || index < 0) {
+ return FutureUtil.failedFuture(new RestException(Status.NOT_FOUND,
+ "Invalid message index: " + index));
+ }
+ int partitionIndex = topicName.getPartitionIndex();
+ CompletableFuture<Void> future = validateTopicOperationAsync(topicName, TopicOperation.PEEK_MESSAGES);
+ return future.thenCompose(__ -> {
+ if (topicName.isGlobal()) {
+ return validateGlobalNamespaceOwnershipAsync(namespaceName);
+ } else {
+ return CompletableFuture.completedFuture(null);
+ }
+ }).thenCompose(__ -> {
+ if (topicName.isPartitioned()) {
+ return CompletableFuture.completedFuture(null);
+ } else {
+ return getPartitionedTopicMetadataAsync(topicName, authoritative, false)
+ .thenAccept(topicMetadata -> {
+ if (topicMetadata.partitions > 0) {
+ log.warn("[{}] Not supported getMessageIdByIndex operation on "
+ + "partitioned-topic {}", clientAppId(), topicName);
+ throw new RestException(Status.METHOD_NOT_ALLOWED,
+ "GetMessageIDByIndex is not allowed on partitioned-topic");
+ }
+ });
+ }
+ }).thenCompose(ignore -> validateTopicOwnershipAsync(topicName, authoritative))
+ .thenCompose(__ -> getTopicReferenceAsync(topicName))
+ .thenCompose(topic -> {
+ if (!(topic instanceof PersistentTopic persistentTopic)) {
+ log.error("[{}] Get message id by index on a non-persistent topic {} is not allowed",
+ clientAppId(), topicName);
+ return FutureUtil.failedFuture(new RestException(Status.METHOD_NOT_ALLOWED,
+ "Get message id by index on a non-persistent topic is not allowed"));
+ }
+ ManagedLedger managedLedger = persistentTopic.getManagedLedger();
+ return findMessageIndexByPosition(
+ PositionFactory.create(managedLedger.getFirstPosition().getLedgerId(), 0),
+ managedLedger)
+ .thenCompose(firstIndex -> {
+ if (index < firstIndex) {
+ return CompletableFuture.completedFuture(PositionFactory.EARLIEST);
+ } else {
+ return managedLedger.asyncFindPosition(entry -> {
+ try {
+ Long messageIndex = getIndexAndRelease(entry);
+ if (messageIndex == null) {
+ return false; // Skip messages without index
+ } else {
+ // If the message index is less than the requested index,
+ // we continue searching
+ return messageIndex < index;
+ }
+ } catch (IOException e) {
+ log.error("Error deserialize message for message position find", e);
+ return false;
+ }
+ });
+ }
+ }).thenCompose(position -> {
+ Position lastPosition = managedLedger.getLastConfirmedEntry();
+ if (position == null || position.compareTo(lastPosition) > 0) {
+ return FutureUtil.failedFuture(new RestException(Status.NOT_FOUND,
+ "Message not found for index " + index));
+ } else {
+ return CompletableFuture.completedFuture(position);
+ }
+ });
+ }).thenCompose(position -> CompletableFuture.completedFuture(
+ new MessageIdImpl(position.getLedgerId(), position.getEntryId(), partitionIndex)));
+ }
+
+ protected CompletableFuture<Long> findMessageIndexByPosition(Position position, ManagedLedger managedLedger) {
+ CompletableFuture<Long> indexFuture = new CompletableFuture<>();
+ managedLedger.asyncReadEntry(position, new AsyncCallbacks.ReadEntryCallback() {
+ @Override
+ public void readEntryComplete(Entry entry, Object ctx) {
+ try {
+ Long index = getIndexAndRelease(entry);
+ if (index == null) {
+ indexFuture.completeExceptionally(new RestException(Status.PRECONDITION_FAILED,
+ "Broker entry metadata is not present in the message"));
+ } else if (index < 0) {
+ indexFuture.completeExceptionally(new RestException(Status.PRECONDITION_FAILED,
+ "Invalid message index: " + index));
+ } else {
+ indexFuture.complete(index);
+ }
+ } catch (IOException e) {
+ indexFuture.completeExceptionally(new RestException(Status.INTERNAL_SERVER_ERROR,
+ "Failed to get index from entry: " + e.getMessage()));
+ }
+ }
+
+ @Override
+ public void readEntryFailed(ManagedLedgerException exception, Object ctx) {
+ log.error("[{}] Failed to read position {} on topic {}",
+ clientAppId(), position, topicName, exception);
+ indexFuture.completeExceptionally(exception);
+ }
+ }, null);
+ return indexFuture;
+ }
+
+
+ private static Long getIndexAndRelease(Entry entry) throws IOException {
+ try {
Review Comment:
Do not nest `release` in the method to prevent others from using entry after this method. Please mention the method caller to implement `release`
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat][admin] PIP-415: Support getting message ID by index [pulsar]
Posted by "nodece (via GitHub)" <gi...@apache.org>.
nodece commented on code in PR #24222:
URL: https://github.com/apache/pulsar/pull/24222#discussion_r2125911761
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java:
##########
@@ -5489,4 +5489,131 @@ protected CompletableFuture<AutoSubscriptionCreationOverride> internalGetAutoSub
return null;
}));
}
+
+ protected CompletableFuture<MessageId> internalGetMessageIDByIndexAsync(Long index, boolean authoritative) {
+ if (!pulsar().getBrokerService().isBrokerEntryMetadataEnabled()) {
+ return FutureUtil.failedFuture(new RestException(Status.PRECONDITION_FAILED,
+ "GetMessageIDByIndex is not allowed when broker entry metadata is disabled"));
+ }
+ if (index == null || index < 0) {
+ return FutureUtil.failedFuture(new RestException(Status.NOT_FOUND,
+ "Invalid message index: " + index));
+ }
+ int partitionIndex = topicName.getPartitionIndex();
+ CompletableFuture<Void> future = validateTopicOperationAsync(topicName, TopicOperation.PEEK_MESSAGES);
+ return future.thenCompose(__ -> {
+ if (topicName.isGlobal()) {
+ return validateGlobalNamespaceOwnershipAsync(namespaceName);
+ } else {
+ return CompletableFuture.completedFuture(null);
+ }
+ }).thenCompose(__ -> {
+ if (topicName.isPartitioned()) {
+ return CompletableFuture.completedFuture(null);
+ } else {
+ return getPartitionedTopicMetadataAsync(topicName, authoritative, false)
+ .thenAccept(topicMetadata -> {
+ if (topicMetadata.partitions > 0) {
+ log.warn("[{}] Not supported getMessageIdByIndex operation on "
+ + "partitioned-topic {}", clientAppId(), topicName);
+ throw new RestException(Status.METHOD_NOT_ALLOWED,
+ "GetMessageIDByIndex is not allowed on partitioned-topic");
+ }
+ });
+ }
+ }).thenCompose(ignore -> validateTopicOwnershipAsync(topicName, authoritative))
+ .thenCompose(__ -> getTopicReferenceAsync(topicName))
+ .thenCompose(topic -> {
+ if (!(topic instanceof PersistentTopic persistentTopic)) {
+ log.error("[{}] Get message id by index on a non-persistent topic {} is not allowed",
+ clientAppId(), topicName);
+ return FutureUtil.failedFuture(new RestException(Status.METHOD_NOT_ALLOWED,
+ "Get message id by index on a non-persistent topic is not allowed"));
+ }
+ ManagedLedger managedLedger = persistentTopic.getManagedLedger();
+ return findMessageIndexByPosition(
+ PositionFactory.create(managedLedger.getFirstPosition().getLedgerId(), 0),
+ managedLedger)
+ .thenCompose(firstIndex -> {
+ if (index < firstIndex) {
+ return CompletableFuture.completedFuture(PositionFactory.EARLIEST);
+ } else {
+ return managedLedger.asyncFindPosition(entry -> {
+ try {
+ Long messageIndex = getIndexFromEntry(entry);
+ if (messageIndex == null) {
+ return false; // Skip messages without index
+ } else {
+ // If the message index is less than the requested index,
+ // we continue searching
+ return messageIndex < index;
+ }
+ } catch (IOException e) {
+ log.error("Error deserialize message for message position find", e);
+ return false;
+ } finally {
+ entry.release();
+ }
+ });
+ }
+ }).thenCompose(position -> {
+ Position lastPosition = managedLedger.getLastConfirmedEntry();
+ if (position == null || position.compareTo(lastPosition) > 0) {
+ return FutureUtil.failedFuture(new RestException(Status.NOT_FOUND,
+ "Message not found for index " + index));
+ } else {
+ return CompletableFuture.completedFuture(position);
+ }
+ });
+ }).thenCompose(position -> CompletableFuture.completedFuture(
+ new MessageIdImpl(position.getLedgerId(), position.getEntryId(), partitionIndex)));
+ }
+
+ protected CompletableFuture<Long> findMessageIndexByPosition(Position position, ManagedLedger managedLedger) {
+ CompletableFuture<Long> indexFuture = new CompletableFuture<>();
+ managedLedger.asyncReadEntry(position, new AsyncCallbacks.ReadEntryCallback() {
+ @Override
+ public void readEntryComplete(Entry entry, Object ctx) {
+ try {
+ Long index = getIndexFromEntry(entry);
+ if (index == null) {
+ indexFuture.completeExceptionally(new RestException(Status.PRECONDITION_FAILED,
+ "Broker entry metadata is not present in the message"));
+ } else if (index < 0) {
+ indexFuture.completeExceptionally(new RestException(Status.PRECONDITION_FAILED,
+ "Invalid message index: " + index));
+ } else {
+ indexFuture.complete(index);
+ }
+ } catch (IOException e) {
+ indexFuture.completeExceptionally(new RestException(Status.INTERNAL_SERVER_ERROR,
+ "Failed to get index from entry: " + e.getMessage()));
+ } finally {
+ entry.release();
+ }
+ }
+
+ @Override
+ public void readEntryFailed(ManagedLedgerException exception, Object ctx) {
+ log.error("[{}] Failed to read position {} on topic {}",
+ clientAppId(), position, topicName, exception);
+ indexFuture.completeExceptionally(exception);
+ }
+ }, null);
+ return indexFuture;
+ }
+
+
+ private static Long getIndexFromEntry(Entry entry) throws IOException {
+ try {
+ final var brokerEntryMetadata = Commands.parseBrokerEntryMetadataIfExist(entry.getDataBuffer());
+ if (brokerEntryMetadata != null && brokerEntryMetadata.hasIndex()) {
+ return brokerEntryMetadata.getIndex();
+ } else {
+ return null;
+ }
+ } catch (Throwable throwable) {
+ throw new IOException(throwable);
Review Comment:
This `catch` block appears to be unnecessary since it simply wraps and rethrows the original `Throwable` as an `IOException` without any additional handling.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat][admin] PIP-415: Support getting message ID by index [pulsar]
Posted by "nodece (via GitHub)" <gi...@apache.org>.
nodece commented on code in PR #24222:
URL: https://github.com/apache/pulsar/pull/24222#discussion_r2125913607
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java:
##########
@@ -5489,4 +5489,131 @@ protected CompletableFuture<AutoSubscriptionCreationOverride> internalGetAutoSub
return null;
}));
}
+
+ protected CompletableFuture<MessageId> internalGetMessageIDByIndexAsync(Long index, boolean authoritative) {
+ if (!pulsar().getBrokerService().isBrokerEntryMetadataEnabled()) {
+ return FutureUtil.failedFuture(new RestException(Status.PRECONDITION_FAILED,
+ "GetMessageIDByIndex is not allowed when broker entry metadata is disabled"));
+ }
+ if (index == null || index < 0) {
+ return FutureUtil.failedFuture(new RestException(Status.NOT_FOUND,
+ "Invalid message index: " + index));
+ }
+ int partitionIndex = topicName.getPartitionIndex();
+ CompletableFuture<Void> future = validateTopicOperationAsync(topicName, TopicOperation.PEEK_MESSAGES);
+ return future.thenCompose(__ -> {
+ if (topicName.isGlobal()) {
+ return validateGlobalNamespaceOwnershipAsync(namespaceName);
+ } else {
+ return CompletableFuture.completedFuture(null);
+ }
+ }).thenCompose(__ -> {
+ if (topicName.isPartitioned()) {
+ return CompletableFuture.completedFuture(null);
+ } else {
+ return getPartitionedTopicMetadataAsync(topicName, authoritative, false)
+ .thenAccept(topicMetadata -> {
+ if (topicMetadata.partitions > 0) {
+ log.warn("[{}] Not supported getMessageIdByIndex operation on "
+ + "partitioned-topic {}", clientAppId(), topicName);
+ throw new RestException(Status.METHOD_NOT_ALLOWED,
+ "GetMessageIDByIndex is not allowed on partitioned-topic");
+ }
+ });
+ }
+ }).thenCompose(ignore -> validateTopicOwnershipAsync(topicName, authoritative))
+ .thenCompose(__ -> getTopicReferenceAsync(topicName))
+ .thenCompose(topic -> {
+ if (!(topic instanceof PersistentTopic persistentTopic)) {
+ log.error("[{}] Get message id by index on a non-persistent topic {} is not allowed",
+ clientAppId(), topicName);
+ return FutureUtil.failedFuture(new RestException(Status.METHOD_NOT_ALLOWED,
+ "Get message id by index on a non-persistent topic is not allowed"));
+ }
+ ManagedLedger managedLedger = persistentTopic.getManagedLedger();
+ return findMessageIndexByPosition(
+ PositionFactory.create(managedLedger.getFirstPosition().getLedgerId(), 0),
+ managedLedger)
+ .thenCompose(firstIndex -> {
+ if (index < firstIndex) {
+ return CompletableFuture.completedFuture(PositionFactory.EARLIEST);
Review Comment:
Or consider returning null?
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat][admin] PIP-415: Support getting message ID by index [pulsar]
Posted by "nodece (via GitHub)" <gi...@apache.org>.
nodece commented on code in PR #24222:
URL: https://github.com/apache/pulsar/pull/24222#discussion_r2126096493
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java:
##########
@@ -5489,4 +5489,131 @@ protected CompletableFuture<AutoSubscriptionCreationOverride> internalGetAutoSub
return null;
}));
}
+
+ protected CompletableFuture<MessageId> internalGetMessageIDByIndexAsync(Long index, boolean authoritative) {
+ if (!pulsar().getBrokerService().isBrokerEntryMetadataEnabled()) {
+ return FutureUtil.failedFuture(new RestException(Status.PRECONDITION_FAILED,
+ "GetMessageIDByIndex is not allowed when broker entry metadata is disabled"));
+ }
+ if (index == null || index < 0) {
+ return FutureUtil.failedFuture(new RestException(Status.NOT_FOUND,
+ "Invalid message index: " + index));
+ }
+ int partitionIndex = topicName.getPartitionIndex();
+ CompletableFuture<Void> future = validateTopicOperationAsync(topicName, TopicOperation.PEEK_MESSAGES);
+ return future.thenCompose(__ -> {
+ if (topicName.isGlobal()) {
+ return validateGlobalNamespaceOwnershipAsync(namespaceName);
+ } else {
+ return CompletableFuture.completedFuture(null);
+ }
+ }).thenCompose(__ -> {
+ if (topicName.isPartitioned()) {
+ return CompletableFuture.completedFuture(null);
+ } else {
+ return getPartitionedTopicMetadataAsync(topicName, authoritative, false)
+ .thenAccept(topicMetadata -> {
+ if (topicMetadata.partitions > 0) {
+ log.warn("[{}] Not supported getMessageIdByIndex operation on "
+ + "partitioned-topic {}", clientAppId(), topicName);
+ throw new RestException(Status.METHOD_NOT_ALLOWED,
+ "GetMessageIDByIndex is not allowed on partitioned-topic");
+ }
+ });
+ }
+ }).thenCompose(ignore -> validateTopicOwnershipAsync(topicName, authoritative))
+ .thenCompose(__ -> getTopicReferenceAsync(topicName))
+ .thenCompose(topic -> {
+ if (!(topic instanceof PersistentTopic persistentTopic)) {
+ log.error("[{}] Get message id by index on a non-persistent topic {} is not allowed",
+ clientAppId(), topicName);
+ return FutureUtil.failedFuture(new RestException(Status.METHOD_NOT_ALLOWED,
+ "Get message id by index on a non-persistent topic is not allowed"));
+ }
+ ManagedLedger managedLedger = persistentTopic.getManagedLedger();
+ return findMessageIndexByPosition(
+ PositionFactory.create(managedLedger.getFirstPosition().getLedgerId(), 0),
+ managedLedger)
+ .thenCompose(firstIndex -> {
+ if (index < firstIndex) {
+ return CompletableFuture.completedFuture(PositionFactory.EARLIEST);
+ } else {
+ return managedLedger.asyncFindPosition(entry -> {
+ try {
+ Long messageIndex = getIndexFromEntry(entry);
+ if (messageIndex == null) {
+ return false; // Skip messages without index
+ } else {
+ // If the message index is less than the requested index,
+ // we continue searching
+ return messageIndex < index;
+ }
+ } catch (IOException e) {
+ log.error("Error deserialize message for message position find", e);
+ return false;
+ } finally {
+ entry.release();
+ }
+ });
+ }
+ }).thenCompose(position -> {
+ Position lastPosition = managedLedger.getLastConfirmedEntry();
+ if (position == null || position.compareTo(lastPosition) > 0) {
+ return FutureUtil.failedFuture(new RestException(Status.NOT_FOUND,
+ "Message not found for index " + index));
+ } else {
+ return CompletableFuture.completedFuture(position);
+ }
+ });
+ }).thenCompose(position -> CompletableFuture.completedFuture(
+ new MessageIdImpl(position.getLedgerId(), position.getEntryId(), partitionIndex)));
+ }
+
+ protected CompletableFuture<Long> findMessageIndexByPosition(Position position, ManagedLedger managedLedger) {
+ CompletableFuture<Long> indexFuture = new CompletableFuture<>();
+ managedLedger.asyncReadEntry(position, new AsyncCallbacks.ReadEntryCallback() {
+ @Override
+ public void readEntryComplete(Entry entry, Object ctx) {
+ try {
+ Long index = getIndexFromEntry(entry);
+ if (index == null) {
+ indexFuture.completeExceptionally(new RestException(Status.PRECONDITION_FAILED,
+ "Broker entry metadata is not present in the message"));
+ } else if (index < 0) {
+ indexFuture.completeExceptionally(new RestException(Status.PRECONDITION_FAILED,
+ "Invalid message index: " + index));
+ } else {
+ indexFuture.complete(index);
+ }
+ } catch (IOException e) {
+ indexFuture.completeExceptionally(new RestException(Status.INTERNAL_SERVER_ERROR,
+ "Failed to get index from entry: " + e.getMessage()));
+ } finally {
+ entry.release();
+ }
+ }
+
+ @Override
+ public void readEntryFailed(ManagedLedgerException exception, Object ctx) {
+ log.error("[{}] Failed to read position {} on topic {}",
+ clientAppId(), position, topicName, exception);
+ indexFuture.completeExceptionally(exception);
+ }
+ }, null);
+ return indexFuture;
+ }
+
+
+ private static Long getIndexFromEntry(Entry entry) throws IOException {
+ try {
+ final var brokerEntryMetadata = Commands.parseBrokerEntryMetadataIfExist(entry.getDataBuffer());
+ if (brokerEntryMetadata != null && brokerEntryMetadata.hasIndex()) {
+ return brokerEntryMetadata.getIndex();
+ } else {
+ return null;
+ }
+ } catch (Throwable throwable) {
+ throw new IOException(throwable);
Review Comment:
Wrapping all `Throwable` into an `IOException` here adds an extra layer without clear benefit and may obscure the original exception type. Consider removing the `try-catch` and letting exceptions propagate naturally.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat][admin] PIP-415: Support getting message ID by index [pulsar]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on code in PR #24222:
URL: https://github.com/apache/pulsar/pull/24222#discussion_r2126217163
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java:
##########
@@ -5489,4 +5489,131 @@ protected CompletableFuture<AutoSubscriptionCreationOverride> internalGetAutoSub
return null;
}));
}
+
+ protected CompletableFuture<MessageId> internalGetMessageIDByIndexAsync(Long index, boolean authoritative) {
+ if (!pulsar().getBrokerService().isBrokerEntryMetadataEnabled()) {
+ return FutureUtil.failedFuture(new RestException(Status.PRECONDITION_FAILED,
+ "GetMessageIDByIndex is not allowed when broker entry metadata is disabled"));
+ }
+ if (index == null || index < 0) {
+ return FutureUtil.failedFuture(new RestException(Status.NOT_FOUND,
+ "Invalid message index: " + index));
+ }
+ int partitionIndex = topicName.getPartitionIndex();
+ CompletableFuture<Void> future = validateTopicOperationAsync(topicName, TopicOperation.PEEK_MESSAGES);
+ return future.thenCompose(__ -> {
+ if (topicName.isGlobal()) {
+ return validateGlobalNamespaceOwnershipAsync(namespaceName);
+ } else {
+ return CompletableFuture.completedFuture(null);
+ }
+ }).thenCompose(__ -> {
+ if (topicName.isPartitioned()) {
+ return CompletableFuture.completedFuture(null);
+ } else {
+ return getPartitionedTopicMetadataAsync(topicName, authoritative, false)
+ .thenAccept(topicMetadata -> {
+ if (topicMetadata.partitions > 0) {
+ log.warn("[{}] Not supported getMessageIdByIndex operation on "
+ + "partitioned-topic {}", clientAppId(), topicName);
+ throw new RestException(Status.METHOD_NOT_ALLOWED,
+ "GetMessageIDByIndex is not allowed on partitioned-topic");
+ }
+ });
+ }
+ }).thenCompose(ignore -> validateTopicOwnershipAsync(topicName, authoritative))
+ .thenCompose(__ -> getTopicReferenceAsync(topicName))
+ .thenCompose(topic -> {
+ if (!(topic instanceof PersistentTopic persistentTopic)) {
+ log.error("[{}] Get message id by index on a non-persistent topic {} is not allowed",
+ clientAppId(), topicName);
+ return FutureUtil.failedFuture(new RestException(Status.METHOD_NOT_ALLOWED,
+ "Get message id by index on a non-persistent topic is not allowed"));
+ }
+ ManagedLedger managedLedger = persistentTopic.getManagedLedger();
+ return findMessageIndexByPosition(
+ PositionFactory.create(managedLedger.getFirstPosition().getLedgerId(), 0),
+ managedLedger)
+ .thenCompose(firstIndex -> {
+ if (index < firstIndex) {
+ return CompletableFuture.completedFuture(PositionFactory.EARLIEST);
+ } else {
+ return managedLedger.asyncFindPosition(entry -> {
+ try {
+ Long messageIndex = getIndexFromEntry(entry);
+ if (messageIndex == null) {
+ return false; // Skip messages without index
+ } else {
+ // If the message index is less than the requested index,
+ // we continue searching
+ return messageIndex < index;
+ }
+ } catch (IOException e) {
+ log.error("Error deserialize message for message position find", e);
+ return false;
+ } finally {
+ entry.release();
+ }
+ });
+ }
+ }).thenCompose(position -> {
+ Position lastPosition = managedLedger.getLastConfirmedEntry();
+ if (position == null || position.compareTo(lastPosition) > 0) {
+ return FutureUtil.failedFuture(new RestException(Status.NOT_FOUND,
+ "Message not found for index " + index));
+ } else {
+ return CompletableFuture.completedFuture(position);
+ }
+ });
+ }).thenCompose(position -> CompletableFuture.completedFuture(
+ new MessageIdImpl(position.getLedgerId(), position.getEntryId(), partitionIndex)));
+ }
+
+ protected CompletableFuture<Long> findMessageIndexByPosition(Position position, ManagedLedger managedLedger) {
+ CompletableFuture<Long> indexFuture = new CompletableFuture<>();
+ managedLedger.asyncReadEntry(position, new AsyncCallbacks.ReadEntryCallback() {
+ @Override
+ public void readEntryComplete(Entry entry, Object ctx) {
+ try {
+ Long index = getIndexFromEntry(entry);
+ if (index == null) {
+ indexFuture.completeExceptionally(new RestException(Status.PRECONDITION_FAILED,
+ "Broker entry metadata is not present in the message"));
+ } else if (index < 0) {
+ indexFuture.completeExceptionally(new RestException(Status.PRECONDITION_FAILED,
+ "Invalid message index: " + index));
+ } else {
+ indexFuture.complete(index);
+ }
+ } catch (IOException e) {
+ indexFuture.completeExceptionally(new RestException(Status.INTERNAL_SERVER_ERROR,
+ "Failed to get index from entry: " + e.getMessage()));
+ } finally {
+ entry.release();
+ }
+ }
+
+ @Override
+ public void readEntryFailed(ManagedLedgerException exception, Object ctx) {
+ log.error("[{}] Failed to read position {} on topic {}",
+ clientAppId(), position, topicName, exception);
+ indexFuture.completeExceptionally(exception);
+ }
+ }, null);
+ return indexFuture;
+ }
+
+
+ private static Long getIndexFromEntry(Entry entry) throws IOException {
+ try {
+ final var brokerEntryMetadata = Commands.parseBrokerEntryMetadataIfExist(entry.getDataBuffer());
+ if (brokerEntryMetadata != null && brokerEntryMetadata.hasIndex()) {
+ return brokerEntryMetadata.getIndex();
+ } else {
+ return null;
+ }
+ } catch (Throwable throwable) {
+ throw new IOException(throwable);
Review Comment:
The benefit of checked exceptions is to force the caller to handle the exception, which should be a "known" exception. However, the `IOException` thrown in this method is just a wrapper on an "unknown" exception.
A normal practice of exception handling should be:
- Fail fast or propagate the exception for unknown exceptions, because the caller never knows how to handle it correctly
- Handle the known exception according to the method's API docs
Specifically, for unknown exceptions, if propagating it could bring unexpected result, we should catch `Throwable` and handle the case well, for example,
```java
@RequiredArgsConstructor
public class Main {
private final Function<Integer, Integer> func;
public CompletableFuture<Integer> f(int x) {
final var future = new CompletableFuture<Integer>();
new Thread(() -> future.complete(func.apply(x))).start();
return future;
}
public static void main(String[] args) {
final var main = new Main(x -> 1 / x);
final var result = main.f(0).join();
System.out.println(result);
}
}
```
To avoid the future of `Main#f` is never returned, we should add a catch clause for it:
```java
try {
future.complete(func.apply(x));
} catch (Throwable throwable) {
future.completeExceptionally(throwable);
}
```
But adding an additional method with exception signature is bad:
```java
public CompletableFuture<Integer> f(int x) {
final var future = new CompletableFuture<Integer>();
new Thread(() -> {
try {
future.complete(apply(x));
} catch (IOException e) {
future.completeExceptionally(e);
}
}).start();
return future;
}
public int apply(int x) throws IOException {
try {
return func.apply(x);
} catch (Throwable throwable) {
throw new IOException(throwable);
}
}
```
In this case, it's better to keep `getIndexFromEntry` simple without the exception signature and catch the `Throwable` when calling it like the suggestion [here](https://github.com/apache/pulsar/pull/24222/files#r2125780470).
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat][admin] PIP-415: Support getting message ID by index [pulsar]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on code in PR #24222:
URL: https://github.com/apache/pulsar/pull/24222#discussion_r2126226981
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java:
##########
@@ -5489,4 +5489,131 @@ protected CompletableFuture<AutoSubscriptionCreationOverride> internalGetAutoSub
return null;
}));
}
+
+ protected CompletableFuture<MessageId> internalGetMessageIDByIndexAsync(Long index, boolean authoritative) {
+ if (!pulsar().getBrokerService().isBrokerEntryMetadataEnabled()) {
+ return FutureUtil.failedFuture(new RestException(Status.PRECONDITION_FAILED,
+ "GetMessageIDByIndex is not allowed when broker entry metadata is disabled"));
+ }
+ if (index == null || index < 0) {
+ return FutureUtil.failedFuture(new RestException(Status.NOT_FOUND,
+ "Invalid message index: " + index));
+ }
+ int partitionIndex = topicName.getPartitionIndex();
+ CompletableFuture<Void> future = validateTopicOperationAsync(topicName, TopicOperation.PEEK_MESSAGES);
+ return future.thenCompose(__ -> {
+ if (topicName.isGlobal()) {
+ return validateGlobalNamespaceOwnershipAsync(namespaceName);
+ } else {
+ return CompletableFuture.completedFuture(null);
+ }
+ }).thenCompose(__ -> {
+ if (topicName.isPartitioned()) {
+ return CompletableFuture.completedFuture(null);
+ } else {
+ return getPartitionedTopicMetadataAsync(topicName, authoritative, false)
+ .thenAccept(topicMetadata -> {
+ if (topicMetadata.partitions > 0) {
+ log.warn("[{}] Not supported getMessageIdByIndex operation on "
+ + "partitioned-topic {}", clientAppId(), topicName);
+ throw new RestException(Status.METHOD_NOT_ALLOWED,
+ "GetMessageIDByIndex is not allowed on partitioned-topic");
+ }
+ });
+ }
+ }).thenCompose(ignore -> validateTopicOwnershipAsync(topicName, authoritative))
+ .thenCompose(__ -> getTopicReferenceAsync(topicName))
+ .thenCompose(topic -> {
+ if (!(topic instanceof PersistentTopic persistentTopic)) {
+ log.error("[{}] Get message id by index on a non-persistent topic {} is not allowed",
+ clientAppId(), topicName);
+ return FutureUtil.failedFuture(new RestException(Status.METHOD_NOT_ALLOWED,
+ "Get message id by index on a non-persistent topic is not allowed"));
+ }
+ ManagedLedger managedLedger = persistentTopic.getManagedLedger();
+ return findMessageIndexByPosition(
+ PositionFactory.create(managedLedger.getFirstPosition().getLedgerId(), 0),
+ managedLedger)
+ .thenCompose(firstIndex -> {
+ if (index < firstIndex) {
+ return CompletableFuture.completedFuture(PositionFactory.EARLIEST);
Review Comment:
> (-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)
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat][admin] PIP-415: Support getting message ID by index [pulsar]
Posted by "liangyepianzhou (via GitHub)" <gi...@apache.org>.
liangyepianzhou commented on code in PR #24222:
URL: https://github.com/apache/pulsar/pull/24222#discussion_r2125987992
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java:
##########
@@ -5489,4 +5489,114 @@ protected CompletableFuture<AutoSubscriptionCreationOverride> internalGetAutoSub
return null;
}));
}
+
+ protected CompletableFuture<MessageId> internalGetMessageIDByIndexAsync(Long index, boolean authoritative) {
+ if (!pulsar().getBrokerService().isBrokerEntryMetadataEnabled()) {
+ return FutureUtil.failedFuture(new RestException(Status.PRECONDITION_FAILED,
+ "GetMessageIDByIndex is not allowed when broker entry metadata is disabled"));
+ }
+ if (index == null || index < 0) {
+ return FutureUtil.failedFuture(new RestException(Status.NOT_FOUND,
+ "Invalid message index: " + index));
+ }
+ int partitionIndex = topicName.getPartitionIndex();
+ CompletableFuture<Void> future = validateTopicOperationAsync(topicName, TopicOperation.PEEK_MESSAGES);
+ return future.thenCompose(__ -> {
+ if (topicName.isGlobal()) {
+ return validateGlobalNamespaceOwnershipAsync(namespaceName);
+ } else {
+ return CompletableFuture.completedFuture(null);
+ }
+ }).thenCompose(__ -> {
+ if (topicName.isPartitioned()) {
+ return CompletableFuture.completedFuture(null);
+ } else {
+ return getPartitionedTopicMetadataAsync(topicName, authoritative, false)
+ .thenAccept(topicMetadata -> {
+ if (topicMetadata.partitions > 0) {
+ log.warn("[{}] Not supported getMessageIdByIndex operation on "
+ + "partitioned-topic {}", clientAppId(), topicName);
+ throw new RestException(Status.METHOD_NOT_ALLOWED,
+ "GetMessageIDByIndex is not allowed on partitioned-topic");
+ }
+ });
+ }
+ }).thenCompose(ignore -> validateTopicOwnershipAsync(topicName, authoritative))
+ .thenCompose(__ -> getTopicReferenceAsync(topicName))
+ .thenCompose(topic -> {
+ if (!(topic instanceof PersistentTopic persistentTopic)) {
+ log.error("[{}] Get message id by index on a non-persistent topic {} is not allowed",
+ clientAppId(), topicName);
+ return FutureUtil.failedFuture(new RestException(Status.METHOD_NOT_ALLOWED,
+ "Get message id by index on a non-persistent topic is not allowed"));
+ }
+ ManagedLedger managedLedger = persistentTopic.getManagedLedger();
+ return findMessageIndexByPosition(
+ PositionFactory.create(managedLedger.getFirstPosition().getLedgerId(), 0),
Review Comment:
>Could you add a test to show the behavior when getMessageIdByIndex is called on an empty topic?
Done
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat][admin] PIP-415: Support getting message ID by index [pulsar]
Posted by "nodece (via GitHub)" <gi...@apache.org>.
nodece commented on code in PR #24222:
URL: https://github.com/apache/pulsar/pull/24222#discussion_r2127870417
##########
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;
Review Comment:
```suggestion
```
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat][admin] PIP-415: Support getting message ID by index [pulsar]
Posted by "liangyepianzhou (via GitHub)" <gi...@apache.org>.
liangyepianzhou commented on code in PR #24222:
URL: https://github.com/apache/pulsar/pull/24222#discussion_r2122508243
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/PersistentTopics.java:
##########
@@ -5015,5 +5015,38 @@ public void removeAutoSubscriptionCreation(
});
}
+ @GET
+ @Path("/{tenant}/{namespace}/{topic}/getMessageIdByIndex")
+ @ApiOperation(hidden = true, value = "Get Message ID by index.",
+ notes = "If the specified index is a system message, "
+ + "it will return the message id of the later message.")
+ @ApiResponses(value = {
+ @ApiResponse(code = 307, message = "Current broker doesn't serve the namespace of this topic"),
+ @ApiResponse(code = 403, message = "Don't have admin permission"),
+ @ApiResponse(code = 404, message = "Namespace or partitioned topic does not exist, "
+ + "or the index is invalid"),
+ @ApiResponse(code = 406, message = "The topic is not a persistent topic"),
+ @ApiResponse(code = 412, message = "The broker is not enable broker entry metadata"),
+ })
+ public void getMessageIDByIndexAndPartitionID(@Suspended final AsyncResponse asyncResponse,
Review Comment:
You are right! The suffix is incorrect.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat][admin] PIP-415: Support getting message ID by index [pulsar]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on PR #24222:
URL: https://github.com/apache/pulsar/pull/24222#issuecomment-2921951865
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.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat][admin] PIP-415: Support getting message ID by index [pulsar]
Posted by "StevenLuMT (via GitHub)" <gi...@apache.org>.
StevenLuMT commented on PR #24222:
URL: https://github.com/apache/pulsar/pull/24222#issuecomment-3508922679
@liangyepianzhou
https://lists.apache.org/thread/j9vxmmsw1r85l27v3rgp9tcnllk48wyb
Please add two labels, and cherry-pick PIP-415 into `branch-4.0` and `branch-4.1`.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat][admin] PIP-415: Support getting message ID by index [pulsar]
Posted by "StevenLuMT (via GitHub)" <gi...@apache.org>.
StevenLuMT commented on code in PR #24222:
URL: https://github.com/apache/pulsar/pull/24222#discussion_r2125786313
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java:
##########
@@ -5489,4 +5489,129 @@ protected CompletableFuture<AutoSubscriptionCreationOverride> internalGetAutoSub
return null;
}));
}
+
+ protected CompletableFuture<MessageId> internalGetMessageIDByIndexAsync(Long index, boolean authoritative) {
+ if (!pulsar().getBrokerService().isBrokerEntryMetadataEnabled()) {
+ return FutureUtil.failedFuture(new RestException(Status.PRECONDITION_FAILED,
+ "GetMessageIDByIndex is not allowed when broker entry metadata is disabled"));
+ }
+ if (index == null || index < 0) {
+ return FutureUtil.failedFuture(new RestException(Status.NOT_FOUND,
+ "Invalid message index: " + index));
+ }
+ int partitionIndex = topicName.getPartitionIndex();
+ CompletableFuture<Void> future = validateTopicOperationAsync(topicName, TopicOperation.PEEK_MESSAGES);
+ return future.thenCompose(__ -> {
+ if (topicName.isGlobal()) {
+ return validateGlobalNamespaceOwnershipAsync(namespaceName);
+ } else {
+ return CompletableFuture.completedFuture(null);
+ }
+ }).thenCompose(__ -> {
+ if (topicName.isPartitioned()) {
+ return CompletableFuture.completedFuture(null);
+ } else {
+ return getPartitionedTopicMetadataAsync(topicName, authoritative, false)
+ .thenAccept(topicMetadata -> {
+ if (topicMetadata.partitions > 0) {
+ log.warn("[{}] Not supported getMessageIdByIndex operation on "
+ + "partitioned-topic {}", clientAppId(), topicName);
+ throw new RestException(Status.METHOD_NOT_ALLOWED,
+ "GetMessageIDByIndex is not allowed on partitioned-topic");
+ }
+ });
+ }
+ }).thenCompose(ignore -> validateTopicOwnershipAsync(topicName, authoritative))
+ .thenCompose(__ -> getTopicReferenceAsync(topicName))
+ .thenCompose(topic -> {
+ if (!(topic instanceof PersistentTopic persistentTopic)) {
+ log.error("[{}] Get message id by index on a non-persistent topic {} is not allowed",
+ clientAppId(), topicName);
+ return FutureUtil.failedFuture(new RestException(Status.METHOD_NOT_ALLOWED,
+ "Get message id by index on a non-persistent topic is not allowed"));
+ }
+ ManagedLedger managedLedger = persistentTopic.getManagedLedger();
+ return findMessageIndexByPosition(
+ PositionFactory.create(managedLedger.getFirstPosition().getLedgerId(), 0),
+ managedLedger)
+ .thenCompose(firstIndex -> {
+ if (index < firstIndex) {
+ return CompletableFuture.completedFuture(PositionFactory.EARLIEST);
+ } else {
+ return managedLedger.asyncFindPosition(entry -> {
+ try {
+ Long messageIndex = getIndexAndRelease(entry);
+ if (messageIndex == null) {
+ return false; // Skip messages without index
+ } else {
+ // If the message index is less than the requested index,
+ // we continue searching
+ return messageIndex < index;
+ }
+ } catch (IOException e) {
+ log.error("Error deserialize message for message position find", e);
+ return false;
+ }
+ });
+ }
+ }).thenCompose(position -> {
+ Position lastPosition = managedLedger.getLastConfirmedEntry();
+ if (position == null || position.compareTo(lastPosition) > 0) {
+ return FutureUtil.failedFuture(new RestException(Status.NOT_FOUND,
Review Comment:
`position.compareTo(lastPosition) > 0`
When does this happen? Is it normal?
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat][admin] PIP-415: Support getting message ID by index [pulsar]
Posted by "nodece (via GitHub)" <gi...@apache.org>.
nodece commented on code in PR #24222:
URL: https://github.com/apache/pulsar/pull/24222#discussion_r2126090113
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java:
##########
@@ -5489,4 +5489,131 @@ protected CompletableFuture<AutoSubscriptionCreationOverride> internalGetAutoSub
return null;
}));
}
+
+ protected CompletableFuture<MessageId> internalGetMessageIDByIndexAsync(Long index, boolean authoritative) {
+ if (!pulsar().getBrokerService().isBrokerEntryMetadataEnabled()) {
+ return FutureUtil.failedFuture(new RestException(Status.PRECONDITION_FAILED,
+ "GetMessageIDByIndex is not allowed when broker entry metadata is disabled"));
+ }
+ if (index == null || index < 0) {
+ return FutureUtil.failedFuture(new RestException(Status.NOT_FOUND,
+ "Invalid message index: " + index));
+ }
+ int partitionIndex = topicName.getPartitionIndex();
+ CompletableFuture<Void> future = validateTopicOperationAsync(topicName, TopicOperation.PEEK_MESSAGES);
+ return future.thenCompose(__ -> {
+ if (topicName.isGlobal()) {
+ return validateGlobalNamespaceOwnershipAsync(namespaceName);
+ } else {
+ return CompletableFuture.completedFuture(null);
+ }
+ }).thenCompose(__ -> {
+ if (topicName.isPartitioned()) {
+ return CompletableFuture.completedFuture(null);
+ } else {
+ return getPartitionedTopicMetadataAsync(topicName, authoritative, false)
+ .thenAccept(topicMetadata -> {
+ if (topicMetadata.partitions > 0) {
+ log.warn("[{}] Not supported getMessageIdByIndex operation on "
+ + "partitioned-topic {}", clientAppId(), topicName);
+ throw new RestException(Status.METHOD_NOT_ALLOWED,
+ "GetMessageIDByIndex is not allowed on partitioned-topic");
+ }
+ });
+ }
+ }).thenCompose(ignore -> validateTopicOwnershipAsync(topicName, authoritative))
+ .thenCompose(__ -> getTopicReferenceAsync(topicName))
+ .thenCompose(topic -> {
+ if (!(topic instanceof PersistentTopic persistentTopic)) {
+ log.error("[{}] Get message id by index on a non-persistent topic {} is not allowed",
+ clientAppId(), topicName);
+ return FutureUtil.failedFuture(new RestException(Status.METHOD_NOT_ALLOWED,
+ "Get message id by index on a non-persistent topic is not allowed"));
+ }
+ ManagedLedger managedLedger = persistentTopic.getManagedLedger();
+ return findMessageIndexByPosition(
+ PositionFactory.create(managedLedger.getFirstPosition().getLedgerId(), 0),
+ managedLedger)
+ .thenCompose(firstIndex -> {
+ if (index < firstIndex) {
+ return CompletableFuture.completedFuture(PositionFactory.EARLIEST);
Review Comment:
Explicit signal that no valid position exists for the given index.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat][admin] PIP-415: Support getting message ID by index [pulsar]
Posted by "liangyepianzhou (via GitHub)" <gi...@apache.org>.
liangyepianzhou commented on code in PR #24222:
URL: https://github.com/apache/pulsar/pull/24222#discussion_r2125818071
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java:
##########
@@ -5489,4 +5489,129 @@ protected CompletableFuture<AutoSubscriptionCreationOverride> internalGetAutoSub
return null;
}));
}
+
+ protected CompletableFuture<MessageId> internalGetMessageIDByIndexAsync(Long index, boolean authoritative) {
+ if (!pulsar().getBrokerService().isBrokerEntryMetadataEnabled()) {
+ return FutureUtil.failedFuture(new RestException(Status.PRECONDITION_FAILED,
+ "GetMessageIDByIndex is not allowed when broker entry metadata is disabled"));
+ }
+ if (index == null || index < 0) {
+ return FutureUtil.failedFuture(new RestException(Status.NOT_FOUND,
+ "Invalid message index: " + index));
+ }
+ int partitionIndex = topicName.getPartitionIndex();
+ CompletableFuture<Void> future = validateTopicOperationAsync(topicName, TopicOperation.PEEK_MESSAGES);
+ return future.thenCompose(__ -> {
+ if (topicName.isGlobal()) {
+ return validateGlobalNamespaceOwnershipAsync(namespaceName);
+ } else {
+ return CompletableFuture.completedFuture(null);
+ }
+ }).thenCompose(__ -> {
+ if (topicName.isPartitioned()) {
+ return CompletableFuture.completedFuture(null);
+ } else {
+ return getPartitionedTopicMetadataAsync(topicName, authoritative, false)
+ .thenAccept(topicMetadata -> {
+ if (topicMetadata.partitions > 0) {
+ log.warn("[{}] Not supported getMessageIdByIndex operation on "
+ + "partitioned-topic {}", clientAppId(), topicName);
+ throw new RestException(Status.METHOD_NOT_ALLOWED,
+ "GetMessageIDByIndex is not allowed on partitioned-topic");
+ }
+ });
+ }
+ }).thenCompose(ignore -> validateTopicOwnershipAsync(topicName, authoritative))
+ .thenCompose(__ -> getTopicReferenceAsync(topicName))
+ .thenCompose(topic -> {
+ if (!(topic instanceof PersistentTopic persistentTopic)) {
+ log.error("[{}] Get message id by index on a non-persistent topic {} is not allowed",
+ clientAppId(), topicName);
+ return FutureUtil.failedFuture(new RestException(Status.METHOD_NOT_ALLOWED,
+ "Get message id by index on a non-persistent topic is not allowed"));
+ }
+ ManagedLedger managedLedger = persistentTopic.getManagedLedger();
+ return findMessageIndexByPosition(
+ PositionFactory.create(managedLedger.getFirstPosition().getLedgerId(), 0),
+ managedLedger)
+ .thenCompose(firstIndex -> {
+ if (index < firstIndex) {
+ return CompletableFuture.completedFuture(PositionFactory.EARLIEST);
+ } else {
+ return managedLedger.asyncFindPosition(entry -> {
+ try {
+ Long messageIndex = getIndexAndRelease(entry);
+ if (messageIndex == null) {
+ return false; // Skip messages without index
+ } else {
+ // If the message index is less than the requested index,
+ // we continue searching
+ return messageIndex < index;
+ }
+ } catch (IOException e) {
+ log.error("Error deserialize message for message position find", e);
+ return false;
+ }
+ });
+ }
+ }).thenCompose(position -> {
+ Position lastPosition = managedLedger.getLastConfirmedEntry();
+ if (position == null || position.compareTo(lastPosition) > 0) {
+ return FutureUtil.failedFuture(new RestException(Status.NOT_FOUND,
+ "Message not found for index " + index));
+ } else {
+ return CompletableFuture.completedFuture(position);
+ }
+ });
+ }).thenCompose(position -> CompletableFuture.completedFuture(
+ new MessageIdImpl(position.getLedgerId(), position.getEntryId(), partitionIndex)));
+ }
+
+ protected CompletableFuture<Long> findMessageIndexByPosition(Position position, ManagedLedger managedLedger) {
+ CompletableFuture<Long> indexFuture = new CompletableFuture<>();
+ managedLedger.asyncReadEntry(position, new AsyncCallbacks.ReadEntryCallback() {
+ @Override
+ public void readEntryComplete(Entry entry, Object ctx) {
+ try {
+ Long index = getIndexAndRelease(entry);
+ if (index == null) {
+ indexFuture.completeExceptionally(new RestException(Status.PRECONDITION_FAILED,
+ "Broker entry metadata is not present in the message"));
+ } else if (index < 0) {
+ indexFuture.completeExceptionally(new RestException(Status.PRECONDITION_FAILED,
+ "Invalid message index: " + index));
+ } else {
+ indexFuture.complete(index);
+ }
+ } catch (IOException e) {
+ indexFuture.completeExceptionally(new RestException(Status.INTERNAL_SERVER_ERROR,
+ "Failed to get index from entry: " + e.getMessage()));
+ }
+ }
+
+ @Override
+ public void readEntryFailed(ManagedLedgerException exception, Object ctx) {
+ log.error("[{}] Failed to read position {} on topic {}",
+ clientAppId(), position, topicName, exception);
+ indexFuture.completeExceptionally(exception);
+ }
+ }, null);
+ return indexFuture;
+ }
+
+
+ private static Long getIndexAndRelease(Entry entry) throws IOException {
+ try {
Review Comment:
done
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat][admin] PIP-415: Support getting message ID by index [pulsar]
Posted by "nodece (via GitHub)" <gi...@apache.org>.
nodece commented on code in PR #24222:
URL: https://github.com/apache/pulsar/pull/24222#discussion_r2125895005
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java:
##########
@@ -5489,4 +5489,131 @@ protected CompletableFuture<AutoSubscriptionCreationOverride> internalGetAutoSub
return null;
}));
}
+
+ protected CompletableFuture<MessageId> internalGetMessageIDByIndexAsync(Long index, boolean authoritative) {
+ if (!pulsar().getBrokerService().isBrokerEntryMetadataEnabled()) {
+ return FutureUtil.failedFuture(new RestException(Status.PRECONDITION_FAILED,
+ "GetMessageIDByIndex is not allowed when broker entry metadata is disabled"));
+ }
+ if (index == null || index < 0) {
+ return FutureUtil.failedFuture(new RestException(Status.NOT_FOUND,
+ "Invalid message index: " + index));
+ }
+ int partitionIndex = topicName.getPartitionIndex();
+ CompletableFuture<Void> future = validateTopicOperationAsync(topicName, TopicOperation.PEEK_MESSAGES);
+ return future.thenCompose(__ -> {
+ if (topicName.isGlobal()) {
+ return validateGlobalNamespaceOwnershipAsync(namespaceName);
+ } else {
+ return CompletableFuture.completedFuture(null);
+ }
+ }).thenCompose(__ -> {
+ if (topicName.isPartitioned()) {
+ return CompletableFuture.completedFuture(null);
+ } else {
+ return getPartitionedTopicMetadataAsync(topicName, authoritative, false)
+ .thenAccept(topicMetadata -> {
+ if (topicMetadata.partitions > 0) {
+ log.warn("[{}] Not supported getMessageIdByIndex operation on "
+ + "partitioned-topic {}", clientAppId(), topicName);
+ throw new RestException(Status.METHOD_NOT_ALLOWED,
+ "GetMessageIDByIndex is not allowed on partitioned-topic");
+ }
+ });
+ }
+ }).thenCompose(ignore -> validateTopicOwnershipAsync(topicName, authoritative))
+ .thenCompose(__ -> getTopicReferenceAsync(topicName))
+ .thenCompose(topic -> {
+ if (!(topic instanceof PersistentTopic persistentTopic)) {
+ log.error("[{}] Get message id by index on a non-persistent topic {} is not allowed",
+ clientAppId(), topicName);
+ return FutureUtil.failedFuture(new RestException(Status.METHOD_NOT_ALLOWED,
+ "Get message id by index on a non-persistent topic is not allowed"));
+ }
+ ManagedLedger managedLedger = persistentTopic.getManagedLedger();
+ return findMessageIndexByPosition(
+ PositionFactory.create(managedLedger.getFirstPosition().getLedgerId(), 0),
+ managedLedger)
+ .thenCompose(firstIndex -> {
+ if (index < firstIndex) {
+ return CompletableFuture.completedFuture(PositionFactory.EARLIEST);
+ } else {
+ return managedLedger.asyncFindPosition(entry -> {
+ try {
+ Long messageIndex = getIndexFromEntry(entry);
+ if (messageIndex == null) {
+ return false; // Skip messages without index
+ } else {
+ // If the message index is less than the requested index,
+ // we continue searching
+ return messageIndex < index;
+ }
+ } catch (IOException e) {
+ log.error("Error deserialize message for message position find", e);
+ return false;
Review Comment:
It looks like you should not ignore this exception.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat][admin] PIP-415: Support getting message ID by index [pulsar]
Posted by "liangyepianzhou (via GitHub)" <gi...@apache.org>.
liangyepianzhou commented on code in PR #24222:
URL: https://github.com/apache/pulsar/pull/24222#discussion_r2125990993
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java:
##########
@@ -5489,4 +5489,114 @@ protected CompletableFuture<AutoSubscriptionCreationOverride> internalGetAutoSub
return null;
}));
}
+
+ protected CompletableFuture<MessageId> internalGetMessageIDByIndexAsync(Long index, boolean authoritative) {
+ if (!pulsar().getBrokerService().isBrokerEntryMetadataEnabled()) {
+ return FutureUtil.failedFuture(new RestException(Status.PRECONDITION_FAILED,
+ "GetMessageIDByIndex is not allowed when broker entry metadata is disabled"));
+ }
+ if (index == null || index < 0) {
+ return FutureUtil.failedFuture(new RestException(Status.NOT_FOUND,
+ "Invalid message index: " + index));
+ }
+ int partitionIndex = topicName.getPartitionIndex();
+ CompletableFuture<Void> future = validateTopicOperationAsync(topicName, TopicOperation.PEEK_MESSAGES);
+ return future.thenCompose(__ -> {
+ if (topicName.isGlobal()) {
+ return validateGlobalNamespaceOwnershipAsync(namespaceName);
+ } else {
+ return CompletableFuture.completedFuture(null);
+ }
+ }).thenCompose(__ -> {
+ if (topicName.isPartitioned()) {
+ return CompletableFuture.completedFuture(null);
+ } else {
+ return getPartitionedTopicMetadataAsync(topicName, authoritative, false)
+ .thenAccept(topicMetadata -> {
+ if (topicMetadata.partitions > 0) {
+ log.warn("[{}] Not supported getMessageIdByIndex operation on "
+ + "partitioned-topic {}", clientAppId(), topicName);
+ throw new RestException(Status.METHOD_NOT_ALLOWED,
+ "GetMessageIDByIndex is not allowed on partitioned-topic");
+ }
+ });
+ }
+ }).thenCompose(ignore -> validateTopicOwnershipAsync(topicName, authoritative))
+ .thenCompose(__ -> getTopicReferenceAsync(topicName))
+ .thenCompose(topic -> {
+ if (!(topic instanceof PersistentTopic persistentTopic)) {
+ log.error("[{}] Get message id by index on a non-persistent topic {} is not allowed",
+ clientAppId(), topicName);
+ return FutureUtil.failedFuture(new RestException(Status.METHOD_NOT_ALLOWED,
+ "Get message id by index on a non-persistent topic is not allowed"));
+ }
+ ManagedLedger managedLedger = persistentTopic.getManagedLedger();
+ return findMessageIndexByPosition(
+ PositionFactory.create(managedLedger.getFirstPosition().getLedgerId(), 0),
Review Comment:
>findMessageIndexByPosition(managedLedger.getFirstPosition(), ...) is better.
@nodece `managedLedger.getFirstPosition()` will return [first ledger Id, -1]
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat][admin] PIP-415: Support getting message ID by index [pulsar]
Posted by "liangyepianzhou (via GitHub)" <gi...@apache.org>.
liangyepianzhou commented on code in PR #24222:
URL: https://github.com/apache/pulsar/pull/24222#discussion_r2126361742
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerEntryMetadataE2ETest.java:
##########
@@ -420,4 +427,109 @@ public void testManagedLedgerTotalSize() throws Exception {
cursor.close();
}
+
+ @Test
+ public void testGetMessageIdByIndex() throws Exception {
+ // 1. test no partitioned topic
+ final String topicName = newTopicName();
+ admin.topics().createNonPartitionedTopic(topicName);
+ @Cleanup
+ Producer<String> producer = pulsarClient.newProducer(Schema.STRING)
+ .topic(topicName)
+ .enableBatching(false)
+ .create();
+ MessageIdImpl messageId = (MessageIdImpl) producer.send("test");
+ Message<byte[]>
+ message =
+ admin.topics().getMessagesById(topicName, messageId.getLedgerId(), messageId.getEntryId()).get(0);
+ long index = message.getIndex().get();
+ MessageIdImpl messageIdByIndex = (MessageIdImpl) admin.topics().getMessageIdByIndex(topicName, index);
+ Assert.assertEquals(messageIdByIndex, messageId);
+
+ // 2. test partitioned topic
+ final String topicName2 = newTopicName();
+ final String partitionedTopicName = topicName2 + "-partition-" + 0;
+ admin.topics().createPartitionedTopic(topicName2, 10);
+ @Cleanup
+ Producer<String> producer2 = pulsarClient.newProducer(Schema.STRING)
+ .topic(topicName2)
+ .enableBatching(false)
+ .create();
+
+ MessageIdImpl messageId2 = null;
+ for (int i = 0; i < 200; i++) {
+ messageId2 = (MessageIdImpl) producer2.send("test" + i);
+ if (messageId2.getPartitionIndex() == 0) {
+ break;
+ }
+ }
+ Message<byte[]>
+ message2 = admin.topics().getMessagesById(partitionedTopicName,
+ messageId2.getLedgerId(), messageId2.getEntryId()).get(0);
+ long index2 = message2.getIndex().get();
+ // 2.1 test partitioned topic name with partition index
+ MessageIdImpl messageIdByIndex2 =
+ (MessageIdImpl) admin.topics().getMessageIdByIndex(partitionedTopicName, index2);
+ Assert.assertEquals(messageIdByIndex2, messageId2);
+ // 2.2 test partitioned topic name without partition index
+ PulsarAdminException e = Assertions.assertThrows(PulsarAdminException.class, () ->
+ admin.topics().getMessageIdByIndex(topicName2, index2)
+ );
+ assertExceptionChainContains(e, NotAllowedException.class);
+
+ // 3. test invalid index
+ assertThrowsWithCause(() -> admin.topics().getMessageIdByIndex(topicName, -1),
+ PulsarAdminException.class, NotFoundException.class);
+
+ assertThrowsWithCause(() -> admin.topics().getMessageIdByIndex(topicName, 100000),
+ PulsarAdminException.class, NotFoundException.class);
+ }
+
+ @Test
+ public void testGetMessageIdByIndexForEmptyTopic() throws PulsarAdminException {
+ final String topicName = newTopicName();
+ admin.topics().createNonPartitionedTopic(topicName);
+
+ assertThrowsWithCause(() -> admin.topics().getMessageIdByIndex(topicName, 0),
+ PulsarAdminException.class, NotFoundException.class);
+ }
+
+ @Test
+ public void testGetMessageIdByIndexOutOfIndex() throws PulsarAdminException, PulsarClientException {
+ final String topicName = newTopicName();
+ admin.topics().createNonPartitionedTopic(topicName);
+ @Cleanup
+ final Producer<String> producer = pulsarClient.newProducer(Schema.STRING)
+ .topic(topicName)
+ .create();
+ for (int i = 0; i < 100; i++) {
+ producer.send("msg-" + i);
+ }
+
+ assertThrowsWithCause(() -> admin.topics().getMessageIdByIndex(topicName, 1000),
+ PulsarAdminException.class, NotFoundException.class);
+ }
+
+ private void assertExceptionChainContains(Throwable thrown, Class<? extends Throwable> expectedCauseType) {
+ Throwable cause = thrown;
+ boolean found = false;
+
+ while (cause != null) {
+ if (expectedCauseType.isInstance(cause)) {
+ found = true;
+ break;
+ }
+ cause = cause.getCause();
+ }
+
+ Assert.assertTrue(found);
+ }
+
+ private void assertThrowsWithCause(Executable executable,
Review Comment:
While ChatGPT's provided code wasn't directly applicable to our scenario, it pointed us in the right direction. After testing, confirmed the following implementation works:
```java
private void assertThrowsWithCause(Executable executable,
Class<? extends Throwable> expectedException,
Class<? extends Throwable> expectedCause) {
assertThatThrownBy(executable::execute)
.isInstanceOf(expectedException)
.getCause() // First layer: ResponseProcessingException
.getCause() // Second layer: NotFoundException
.isInstanceOf(expectedCause);
}
```
or
```java
private void assertThrowsWithCause(Executable executable,
Class<? extends Throwable> expectedException,
Class<? extends Throwable> expectedCause) {
assertThatThrownBy(executable::execute)
.isInstanceOf(expectedException)
.hasRootCauseInstanceOf(expectedCause);
}
```
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat][admin] PIP-415: Support getting message ID by index [pulsar]
Posted by "liangyepianzhou (via GitHub)" <gi...@apache.org>.
liangyepianzhou commented on code in PR #24222:
URL: https://github.com/apache/pulsar/pull/24222#discussion_r2128055455
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerEntryMetadataE2ETest.java:
##########
@@ -420,4 +426,93 @@ public void testManagedLedgerTotalSize() throws Exception {
cursor.close();
}
+
+ @Test
+ public void testGetMessageIdByIndex() throws Exception {
+ // 1. test no partitioned topic
+ final String topicName = newTopicName();
+ admin.topics().createNonPartitionedTopic(topicName);
+ @Cleanup
+ Producer<String> producer = pulsarClient.newProducer(Schema.STRING)
+ .topic(topicName)
+ .enableBatching(false)
+ .create();
+ MessageIdImpl messageId = (MessageIdImpl) producer.send("test");
+ Message<byte[]>
+ message =
+ admin.topics().getMessagesById(topicName, messageId.getLedgerId(), messageId.getEntryId()).get(0);
+ long index = message.getIndex().get();
+ MessageIdImpl messageIdByIndex = (MessageIdImpl) admin.topics().getMessageIdByIndex(topicName, index);
+ Assert.assertEquals(messageIdByIndex, messageId);
+
+ // 2. test partitioned topic
+ final String topicName2 = newTopicName();
+ final String partitionedTopicName = topicName2 + "-partition-" + 0;
+ admin.topics().createPartitionedTopic(topicName2, 10);
+ @Cleanup
+ Producer<String> producer2 = pulsarClient.newProducer(Schema.STRING)
+ .topic(topicName2)
+ .enableBatching(false)
+ .create();
+
+ MessageIdImpl messageId2 = null;
+ for (int i = 0; i < 200; i++) {
+ messageId2 = (MessageIdImpl) producer2.send("test" + i);
+ if (messageId2.getPartitionIndex() == 0) {
+ break;
+ }
+ }
+ Message<byte[]>
+ message2 = admin.topics().getMessagesById(partitionedTopicName,
+ messageId2.getLedgerId(), messageId2.getEntryId()).get(0);
+ long index2 = message2.getIndex().get();
+ // 2.1 test partitioned topic name with partition index
+ MessageIdImpl messageIdByIndex2 =
+ (MessageIdImpl) admin.topics().getMessageIdByIndex(partitionedTopicName, index2);
+ Assert.assertEquals(messageIdByIndex2, messageId2);
+ // 2.2 test partitioned topic name without partition index
+ assertThrowsWithCause(() -> admin.topics().getMessageIdByIndex(topicName2, index2),
+ PulsarAdminException.class, NotAllowedException.class);
+
+ // 3. test invalid index
+ assertThrowsWithCause(() -> admin.topics().getMessageIdByIndex(topicName, -1),
+ PulsarAdminException.class, NotFoundException.class);
+
+ assertThrowsWithCause(() -> admin.topics().getMessageIdByIndex(topicName, 100000),
+ PulsarAdminException.class, NotFoundException.class);
+ }
+
+ @Test
+ public void testGetMessageIdByIndexForEmptyTopic() throws PulsarAdminException {
+ final String topicName = newTopicName();
+ admin.topics().createNonPartitionedTopic(topicName);
+
+ assertThrowsWithCause(() -> admin.topics().getMessageIdByIndex(topicName, 0),
+ PulsarAdminException.class, NotFoundException.class);
+ }
+
+ @Test
+ public void testGetMessageIdByIndexOutOfIndex() throws PulsarAdminException, PulsarClientException {
+ final String topicName = newTopicName();
+ admin.topics().createNonPartitionedTopic(topicName);
+ @Cleanup
+ final Producer<String> producer = pulsarClient.newProducer(Schema.STRING)
+ .topic(topicName)
+ .create();
+ for (int i = 0; i < 100; i++) {
+ producer.send("msg-" + i);
+ }
+
+ assertThrowsWithCause(() -> admin.topics().getMessageIdByIndex(topicName, 1000),
+ PulsarAdminException.class, NotFoundException.class);
+ }
+
+ private void assertThrowsWithCause(Executable executable,
Review Comment:
The `run` method of `Runnable` does not throw checked exceptions, while our code throws `PulsarAdminException` (a checked exception).
Therefore, the code will fail at compile time because we are throwing a checked exception in the `run` method of `Runnable` (inside a lambda expression), whereas the `run` method of `Runnable` does not declare any exceptions.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat][admin] PIP-415: Support getting message ID by offset [pulsar]
Posted by "StevenLuMT (via GitHub)" <gi...@apache.org>.
StevenLuMT commented on code in PR #24222:
URL: https://github.com/apache/pulsar/pull/24222#discussion_r2062783225
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java:
##########
@@ -5489,4 +5489,53 @@ protected CompletableFuture<AutoSubscriptionCreationOverride> internalGetAutoSub
return null;
}));
}
+
+ protected CompletableFuture<MessageId> internalGetMessageIDByOffsetAsync(Long offset, boolean authoritative) {
+ int partitionIndex = topicName.getPartitionIndex();
+ CompletableFuture<Void> future = validateTopicOperationAsync(topicName, TopicOperation.PEEK_MESSAGES);
+ return future.thenCompose(__ -> {
+ if (topicName.isGlobal()) {
+ return validateGlobalNamespaceOwnershipAsync(namespaceName);
+ } else {
+ return CompletableFuture.completedFuture(null);
+ }
+ }).thenCompose(__ -> {
+ if (topicName.isPartitioned()) {
+ return CompletableFuture.completedFuture(null);
+ } else {
+ return getPartitionedTopicMetadataAsync(topicName, authoritative, false)
+ .thenAccept(topicMetadata -> {
+ if (topicMetadata.partitions > 0) {
+ log.warn("[{}] Not supported getMessageIdByOffset operation on "
+ + "partitioned-topic {}", clientAppId(), topicName);
+ throw new RestException(Status.METHOD_NOT_ALLOWED,
+ "GetMessageIDByOffset is not allowed on partitioned-topic");
+ }
+ });
+ }
+ }).thenCompose(ignore -> validateTopicOwnershipAsync(topicName, authoritative))
+ .thenCompose(__ -> getTopicReferenceAsync(topicName))
+ .thenCompose(topic -> {
+ if (!(topic instanceof PersistentTopic persistentTopic)) {
+ log.error("[{}] Get message id by offset on a non-persistent topic {} is not allowed",
+ clientAppId(), topicName);
+ return FutureUtil.failedFuture(new RestException(Status.METHOD_NOT_ALLOWED,
+ "Get message id by offset on a non-persistent topic is not allowed"));
+ }
+ ManagedLedger managedLedger = persistentTopic.getManagedLedger();
+ return managedLedger.asyncFindPosition(entry -> {
+ try {
+ BrokerEntryMetadata brokerEntryMetadata =
+ Commands.parseBrokerEntryMetadataIfExist(entry.getDataBuffer());
+ return brokerEntryMetadata.getIndex() < offset;
+ } catch (Exception e) {
+ log.error("Error deserialize message for message position find", e);
+ } finally {
+ entry.release();
+ }
+ return false;
+ });
+ }).thenCompose(position -> CompletableFuture
+ .completedFuture(new MessageIdImpl(position.getLedgerId(), position.getEntryId(), partitionIndex)));
+ }
Review Comment:
`managedLedger.asyncFindPosition` throws an exception,
```
thenCompose(position -> CompletableFuture
.completedFuture(new MessageIdImpl(position.getLedgerId(), position.getEntryId(), partitionIndex)));
```
Will it still process and return normally? Should I change to `whenComplete` to receive the exception and do some special processing
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat][admin] PIP-415: Support getting message ID by offset [pulsar]
Posted by "AuroraTwinkle (via GitHub)" <gi...@apache.org>.
AuroraTwinkle commented on code in PR #24222:
URL: https://github.com/apache/pulsar/pull/24222#discussion_r2062804195
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java:
##########
@@ -5489,4 +5489,53 @@ protected CompletableFuture<AutoSubscriptionCreationOverride> internalGetAutoSub
return null;
}));
}
+
+ protected CompletableFuture<MessageId> internalGetMessageIDByOffsetAsync(Long offset, boolean authoritative) {
+ int partitionIndex = topicName.getPartitionIndex();
+ CompletableFuture<Void> future = validateTopicOperationAsync(topicName, TopicOperation.PEEK_MESSAGES);
+ return future.thenCompose(__ -> {
+ if (topicName.isGlobal()) {
+ return validateGlobalNamespaceOwnershipAsync(namespaceName);
+ } else {
+ return CompletableFuture.completedFuture(null);
+ }
+ }).thenCompose(__ -> {
+ if (topicName.isPartitioned()) {
+ return CompletableFuture.completedFuture(null);
+ } else {
+ return getPartitionedTopicMetadataAsync(topicName, authoritative, false)
+ .thenAccept(topicMetadata -> {
+ if (topicMetadata.partitions > 0) {
+ log.warn("[{}] Not supported getMessageIdByOffset operation on "
+ + "partitioned-topic {}", clientAppId(), topicName);
+ throw new RestException(Status.METHOD_NOT_ALLOWED,
+ "GetMessageIDByOffset is not allowed on partitioned-topic");
+ }
+ });
+ }
+ }).thenCompose(ignore -> validateTopicOwnershipAsync(topicName, authoritative))
+ .thenCompose(__ -> getTopicReferenceAsync(topicName))
+ .thenCompose(topic -> {
+ if (!(topic instanceof PersistentTopic persistentTopic)) {
+ log.error("[{}] Get message id by offset on a non-persistent topic {} is not allowed",
+ clientAppId(), topicName);
+ return FutureUtil.failedFuture(new RestException(Status.METHOD_NOT_ALLOWED,
+ "Get message id by offset on a non-persistent topic is not allowed"));
+ }
+ ManagedLedger managedLedger = persistentTopic.getManagedLedger();
+ return managedLedger.asyncFindPosition(entry -> {
+ try {
+ BrokerEntryMetadata brokerEntryMetadata =
+ Commands.parseBrokerEntryMetadataIfExist(entry.getDataBuffer());
+ return brokerEntryMetadata.getIndex() < offset;
Review Comment:
maybe `brokerEntryMetadata.getIndex() <= offset`?
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org