From 85d3d7f77377861c5458b9e91826f7cf72cf6941 Mon Sep 17 00:00:00 2001 From: gavingaozhangmin Date: Wed, 21 Dec 2022 15:22:58 +0800 Subject: [PATCH 1/5] internalGetMessageById is not allowed on partitioned topic --- .../admin/impl/PersistentTopicsBase.java | 91 +++++++++++-------- 1 file changed, 51 insertions(+), 40 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java index 3fb551967b94e..a6a7e9fadbb94 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java @@ -2761,51 +2761,62 @@ private PositionImpl calculatePositionAckSet(boolean isExcluded, int batchSize, 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 ret; - if (topicName.isGlobal()) { - ret = validateGlobalNamespaceOwnershipAsync(namespaceName); + getPartitionedTopicMetadataAsync(topicName, authoritative, false) + .thenAccept(partitionMetadata -> { + if (!topicName.isPartitioned() && partitionMetadata.partitions > 0) { + log.warn("[{}] Not supported getMessageById operation on partitioned-topic {}", + clientAppId(), topicName); + asyncResponse.resume(new RestException(Status.METHOD_NOT_ALLOWED, + "GetMessageById is not allowed on partitioned-topic")); } 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)); - } + validateTopicOwnershipAsync(topicName, authoritative) + .thenCompose(__ -> validateTopicOperationAsync(topicName, TopicOperation.PEEK_MESSAGES)) + .thenCompose(__ -> { + CompletableFuture 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)); + } - @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(); - } + @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(); + } + } + } + }, 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); } - } - }, 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; + }); } - resumeAsyncResponseExceptionally(asyncResponse, ex); - return null; }); + } protected CompletableFuture internalGetMessageIdByTimestampAsync(long timestamp, boolean authoritative) { From 9fc2a7d1a7171864ddf35d8672602da2ac15f795 Mon Sep 17 00:00:00 2001 From: gavingaozhangmin Date: Wed, 21 Dec 2022 16:42:25 +0800 Subject: [PATCH 2/5] apply comments --- .../apache/pulsar/broker/admin/impl/PersistentTopicsBase.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java index a6a7e9fadbb94..87745aac3cb8a 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java @@ -2763,7 +2763,7 @@ protected void internalGetMessageById(AsyncResponse asyncResponse, long ledgerId // will redirect if the topic not owned by current broker getPartitionedTopicMetadataAsync(topicName, authoritative, false) .thenAccept(partitionMetadata -> { - if (!topicName.isPartitioned() && partitionMetadata.partitions > 0) { + if (partitionMetadata.partitions > 0) { log.warn("[{}] Not supported getMessageById operation on partitioned-topic {}", clientAppId(), topicName); asyncResponse.resume(new RestException(Status.METHOD_NOT_ALLOWED, From ab8aec98df15f61571af34583ff7f7d0b7a9dc11 Mon Sep 17 00:00:00 2001 From: gavingaozhangmin Date: Wed, 21 Dec 2022 22:24:09 +0800 Subject: [PATCH 3/5] apply comments --- .../admin/impl/PersistentTopicsBase.java | 108 +++++++++--------- 1 file changed, 55 insertions(+), 53 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java index 87745aac3cb8a..6051df5226dfe 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java @@ -2760,63 +2760,65 @@ private PositionImpl calculatePositionAckSet(boolean isExcluded, int batchSize, protected void internalGetMessageById(AsyncResponse asyncResponse, long ledgerId, long entryId, boolean authoritative) { - // will redirect if the topic not owned by current broker - getPartitionedTopicMetadataAsync(topicName, authoritative, false) - .thenAccept(partitionMetadata -> { - if (partitionMetadata.partitions > 0) { - log.warn("[{}] Not supported getMessageById operation on partitioned-topic {}", - clientAppId(), topicName); - asyncResponse.resume(new RestException(Status.METHOD_NOT_ALLOWED, - "GetMessageById is not allowed on partitioned-topic")); - } else { - validateTopicOwnershipAsync(topicName, authoritative) - .thenCompose(__ -> validateTopicOperationAsync(topicName, TopicOperation.PEEK_MESSAGES)) - .thenCompose(__ -> { - CompletableFuture 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)); - } + CompletableFuture ret; + // If the topic name is a partition name, no need to get partition topic metadata again + if (!topicName.isPartitioned()) { + ret = getPartitionedTopicMetadataAsync(topicName, authoritative, false) + .thenCompose(topicMetadata -> { + if (topicMetadata.partitions > 0) { + log.warn("[{}] Not supported getMessageById operation on partitioned-topic {}", + clientAppId(), topicName); + asyncResponse.resume(new RestException(Status.METHOD_NOT_ALLOWED, + "GetMessageById is not allowed on partitioned-topic")); + } + return CompletableFuture.completedFuture(null); + }); + } else { + ret = CompletableFuture.completedFuture(null); + } + CompletableFuture future; + if (topicName.isGlobal()) { + future = ret.thenCompose(__ -> validateGlobalNamespaceOwnershipAsync(namespaceName)); + } else { + future = CompletableFuture.completedFuture(null); + } - @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(); - } - } - } - }, 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); + future.thenCompose(ignore -> validateTopicOwnershipAsync(topicName, authoritative)) + .thenCompose(__ -> validateTopicOperationAsync(topicName, TopicOperation.PEEK_MESSAGES)) + .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)); + } + + @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(); + } } - resumeAsyncResponseExceptionally(asyncResponse, ex); - return null; - }); + } + }, 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; }); - } protected CompletableFuture internalGetMessageIdByTimestampAsync(long timestamp, boolean authoritative) { From 73a55868af27ad2822474414496ca59f27e6cc9f Mon Sep 17 00:00:00 2001 From: gavingaozhangmin Date: Wed, 4 Jan 2023 14:42:19 +0800 Subject: [PATCH 4/5] apply comments --- .../admin/impl/PersistentTopicsBase.java | 100 ++++++++---------- .../broker/admin/v1/PersistentTopics.java | 20 ++-- .../broker/admin/v2/PersistentTopics.java | 20 ++-- 3 files changed, 70 insertions(+), 70 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java index 6051df5226dfe..05c977c2167b4 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java @@ -2758,67 +2758,59 @@ private PositionImpl calculatePositionAckSet(boolean isExcluded, int batchSize, return seekPosition; } - protected void internalGetMessageById(AsyncResponse asyncResponse, long ledgerId, long entryId, - boolean authoritative) { - CompletableFuture ret; - // If the topic name is a partition name, no need to get partition topic metadata again - if (!topicName.isPartitioned()) { - ret = getPartitionedTopicMetadataAsync(topicName, authoritative, false) - .thenCompose(topicMetadata -> { - if (topicMetadata.partitions > 0) { - log.warn("[{}] Not supported getMessageById operation on partitioned-topic {}", - clientAppId(), topicName); - asyncResponse.resume(new RestException(Status.METHOD_NOT_ALLOWED, - "GetMessageById is not allowed on partitioned-topic")); - } - return CompletableFuture.completedFuture(null); - }); - } else { - ret = CompletableFuture.completedFuture(null); - } + protected CompletableFuture internalGetMessageById(long ledgerId, long entryId, boolean authoritative) { CompletableFuture future; if (topicName.isGlobal()) { - future = ret.thenCompose(__ -> validateGlobalNamespaceOwnershipAsync(namespaceName)); + future = validateGlobalNamespaceOwnershipAsync(namespaceName); } else { future = CompletableFuture.completedFuture(null); } + return future.thenCompose(__ -> { + if (!topicName.isPartitioned()) { + return getPartitionedTopicMetadataAsync(topicName, authoritative, false) + .thenCompose(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"); + } + return CompletableFuture.completedFuture(null); + }); + } else { + return CompletableFuture.completedFuture(null); + } + }) + .thenCompose(ignore -> validateTopicOwnershipAsync(topicName, authoritative)) + .thenCompose(__ -> validateTopicOperationAsync(topicName, TopicOperation.PEEK_MESSAGES)) + .thenCompose(__ -> getTopicReferenceAsync(topicName)) + .thenCompose(topic -> { + CompletableFuture 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); + } - future.thenCompose(ignore -> validateTopicOwnershipAsync(topicName, authoritative)) - .thenCompose(__ -> validateTopicOperationAsync(topicName, TopicOperation.PEEK_MESSAGES)) - .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)); - } - - @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(); - } - } + @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 internalGetMessageIdByTimestampAsync(long timestamp, boolean authoritative) { diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v1/PersistentTopics.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v1/PersistentTopics.java index 070260c7bfdcd..fe9ced198f48d 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v1/PersistentTopics.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v1/PersistentTopics.java @@ -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 diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/PersistentTopics.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/PersistentTopics.java index a5e30d2180738..30a4d9a208bd1 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/PersistentTopics.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/PersistentTopics.java @@ -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 From 866109b593fb1b65268a4b30ad0e01fd9b52efb1 Mon Sep 17 00:00:00 2001 From: gavingaozhangmin Date: Wed, 4 Jan 2023 23:23:22 +0800 Subject: [PATCH 5/5] apply comments --- .../pulsar/broker/admin/impl/PersistentTopicsBase.java | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java index 05c977c2167b4..84018f7ef0dbe 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java @@ -2766,19 +2766,19 @@ protected CompletableFuture internalGetMessageById(long ledgerId, long future = CompletableFuture.completedFuture(null); } return future.thenCompose(__ -> { - if (!topicName.isPartitioned()) { + if (topicName.isPartitioned()) { + return CompletableFuture.completedFuture(null); + } else { return getPartitionedTopicMetadataAsync(topicName, authoritative, false) - .thenCompose(topicMetadata -> { + .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"); } - return CompletableFuture.completedFuture(null); }); - } else { - return CompletableFuture.completedFuture(null); + } }) .thenCompose(ignore -> validateTopicOwnershipAsync(topicName, authoritative))