Skip to content
New issue

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

#

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

Already on GitHub? # to your account

[improve][admin]internalGetMessageById shouldn't be allowed on partitioned topic #19013

Merged
merged 5 commits into from
Jan 9, 2023
Merged

[improve][admin]internalGetMessageById shouldn't be allowed on partitioned topic #19013

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
85d3d7f
internalGetMessageById is not allowed on partitioned topic
Dec 21, 2022
9fc2a7d
apply comments
Dec 21, 2022
ab8aec9
apply comments
Dec 21, 2022
73a5586
apply comments
Jan 4, 2023
866109b
apply comments
Jan 4, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
97 changes: 51 additions & 46 deletions pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2758,54 +2758,59 @@ private PositionImpl calculatePositionAckSet(boolean isExcluded, int batchSize,
return seekPosition;
}

protected void internalGetMessageById(AsyncResponse asyncResponse, long ledgerId, long entryId,
boolean authoritative) {
// will redirect if the topic not owned by current broker
validateTopicOwnershipAsync(topicName, authoritative)
.thenCompose(__ -> validateTopicOperationAsync(topicName, TopicOperation.PEEK_MESSAGES))
.thenCompose(__ -> {
CompletableFuture<Void> ret;
if (topicName.isGlobal()) {
ret = validateGlobalNamespaceOwnershipAsync(namespaceName);
} else {
ret = CompletableFuture.completedFuture(null);
}
return ret;
})
.thenCompose(__ -> getTopicReferenceAsync(topicName))
.thenAccept(topic -> {
ManagedLedgerImpl ledger =
(ManagedLedgerImpl) ((PersistentTopic) topic).getManagedLedger();
ledger.asyncReadEntry(new PositionImpl(ledgerId, entryId),
new AsyncCallbacks.ReadEntryCallback() {
@Override
public void readEntryFailed(ManagedLedgerException exception,
Object ctx) {
asyncResponse.resume(new RestException(exception));
}
protected CompletableFuture<Response> internalGetMessageById(long ledgerId, long entryId, boolean authoritative) {
CompletableFuture<Void> future;
if (topicName.isGlobal()) {
future = validateGlobalNamespaceOwnershipAsync(namespaceName);
} else {
future = CompletableFuture.completedFuture(null);
}
return future.thenCompose(__ -> {
if (topicName.isPartitioned()) {
return CompletableFuture.completedFuture(null);
} else {
return getPartitionedTopicMetadataAsync(topicName, authoritative, false)
.thenAccept(topicMetadata -> {
if (topicMetadata.partitions > 0) {
log.warn("[{}] Not supported getMessageById operation on partitioned-topic {}",
clientAppId(), topicName);
throw new RestException(Status.METHOD_NOT_ALLOWED,
"GetMessageById is not allowed on partitioned-topic");
}
});

@Override
public void readEntryComplete(Entry entry, Object ctx) {
try {
asyncResponse.resume(generateResponseWithEntry(entry));
} catch (IOException exception) {
asyncResponse.resume(new RestException(exception));
} finally {
if (entry != null) {
entry.release();
}
}
}
})
.thenCompose(ignore -> validateTopicOwnershipAsync(topicName, authoritative))
.thenCompose(__ -> validateTopicOperationAsync(topicName, TopicOperation.PEEK_MESSAGES))
.thenCompose(__ -> getTopicReferenceAsync(topicName))
.thenCompose(topic -> {
CompletableFuture<Response> results = new CompletableFuture<>();
ManagedLedgerImpl ledger =
(ManagedLedgerImpl) ((PersistentTopic) topic).getManagedLedger();
ledger.asyncReadEntry(new PositionImpl(ledgerId, entryId),
new AsyncCallbacks.ReadEntryCallback() {
@Override
public void readEntryFailed(ManagedLedgerException exception,
Object ctx) {
throw new RestException(exception);
}

@Override
public void readEntryComplete(Entry entry, Object ctx) {
try {
results.complete(generateResponseWithEntry(entry));
} catch (IOException exception) {
throw new RestException(exception);
} finally {
if (entry != null) {
entry.release();
}
}, null);
}).exceptionally(ex -> {
// If the exception is not redirect exception we need to log it.
if (!isRedirectException(ex)) {
log.error("[{}] Failed to get message with ledgerId {} entryId {} from {}",
clientAppId(), ledgerId, entryId, topicName, ex);
}
resumeAsyncResponseExceptionally(asyncResponse, ex);
return null;
});
}
}
}, null);
return results;
});
}

protected CompletableFuture<MessageId> internalGetMessageIdByTimestampAsync(long timestamp, boolean authoritative) {
Expand Down
20 changes: 12 additions & 8 deletions pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v1/PersistentTopics.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -834,14 +834,18 @@ public void getMessageByID(@Suspended final AsyncResponse asyncResponse, @PathPa
@PathParam("topic") @Encoded String encodedTopic, @PathParam("ledgerId") Long ledgerId,
@PathParam("entryId") Long entryId,
@QueryParam("authoritative") @DefaultValue("false") boolean authoritative) {
try {
validateTopicName(property, cluster, namespace, encodedTopic);
internalGetMessageById(asyncResponse, ledgerId, entryId, authoritative);
} catch (WebApplicationException wae) {
asyncResponse.resume(wae);
} catch (Exception e) {
asyncResponse.resume(new RestException(e));
}
validateTopicName(property, cluster, namespace, encodedTopic);
internalGetMessageById(ledgerId, entryId, authoritative)
.thenAccept(asyncResponse::resume)
.exceptionally(ex -> {
// If the exception is not redirect exception we need to log it.
if (!isRedirectException(ex)) {
log.error("[{}] Failed to get message with ledgerId {} entryId {} from {}",
clientAppId(), ledgerId, entryId, topicName, ex);
}
resumeAsyncResponseExceptionally(asyncResponse, ex);
return null;
});
}

@GET
Expand Down
20 changes: 12 additions & 8 deletions pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/PersistentTopics.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1941,14 +1941,18 @@ public void getMessageById(
@PathParam("entryId") long entryId,
@ApiParam(value = "Whether leader broker redirected this call to this broker. For internal use.")
@QueryParam("authoritative") @DefaultValue("false") boolean authoritative) {
try {
validateTopicName(tenant, namespace, encodedTopic);
internalGetMessageById(asyncResponse, ledgerId, entryId, authoritative);
} catch (WebApplicationException wae) {
asyncResponse.resume(wae);
} catch (Exception e) {
asyncResponse.resume(new RestException(e));
}
validateTopicName(tenant, namespace, encodedTopic);
internalGetMessageById(ledgerId, entryId, authoritative)
.thenAccept(asyncResponse::resume)
.exceptionally(ex -> {
// If the exception is not redirect exception we need to log it.
if (!isRedirectException(ex)) {
log.error("[{}] Failed to get message with ledgerId {} entryId {} from {}",
clientAppId(), ledgerId, entryId, topicName, ex);
}
resumeAsyncResponseExceptionally(asyncResponse, ex);
return null;
});
}

@GET
Expand Down