lhotari commented on code in PR #25127:
URL: https://github.com/apache/pulsar/pull/25127#discussion_r2764483189
##########
pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/TopicsImpl.java:
##########
@@ -1591,6 +1613,103 @@ public void failed(Throwable throwable) {
return future;
}
+ @Override
+ public CompletableFuture<AnalyzeSubscriptionBacklogResult>
analyzeSubscriptionBacklogAsync(String topic,
+
String subscriptionName,
+
Optional<MessageId> startPosition,
+
long backlogScanMaxEntries) {
+ return analyzeSubscriptionBacklogAsync(topic, subscriptionName,
startPosition,
+ (backlogResult) -> backlogResult.getEntries() >=
backlogScanMaxEntries);
+ }
+
+ @Override
+ public CompletableFuture<AnalyzeSubscriptionBacklogResult>
analyzeSubscriptionBacklogAsync(String topic,
+ String
subscriptionName, Optional<MessageId> startPosition,
+
Predicate<AnalyzeSubscriptionBacklogResult> terminatePredicate) {
+ final CompletableFuture<AnalyzeSubscriptionBacklogResult> future = new
CompletableFuture<>();
+ AtomicReference<AnalyzeSubscriptionBacklogResult> resultRef = new
AtomicReference<>();
+ int partitionIndex = TopicName.get(topic).getPartitionIndex();
+ AtomicReference<Optional<MessageId>> startPositionRef = new
AtomicReference<>(startPosition);
+
+ Supplier<CompletableFuture<AnalyzeSubscriptionBacklogResult>>
resultSupplier =
+ () -> analyzeSubscriptionBacklogAsync(topic, subscriptionName,
startPositionRef.get());
+ BiConsumer<AnalyzeSubscriptionBacklogResult, Throwable> completeAction
= new BiConsumer<>() {
+ @Override
+ public void accept(AnalyzeSubscriptionBacklogResult currentResult,
Throwable throwable) {
+ if (throwable != null) {
+ future.completeExceptionally(throwable);
+ return;
+ }
+
+ AnalyzeSubscriptionBacklogResult mergedResult =
mergeBacklogResults(currentResult, resultRef.get());
+ resultRef.set(mergedResult);
+ if (!mergedResult.isAborted() ||
terminatePredicate.test(mergedResult)) {
+ future.complete(mergedResult);
+ return;
+ }
+
+ // To avoid infinite loops, we ensure the entry count is
incremented after each loop.
+ // Should never happen.
+ if (currentResult.getEntries() <= 0) {
+ log.warn("[{}][{}] Scanned total entry count is zero or
negative, abort analyze backlog, start "
+ + "position is: {}", topic, subscriptionName,
startPositionRef.get());
+ future.completeExceptionally(
+ new PulsarAdminException("Incorrect total entry
count returned from server"));
Review Comment:
I think that this shouldn't complete exceptionally. I would assume that it's
possible to have a case where there's exactly the requested amount of entries
available on the broker side and it would return the results because of
reaching the max limit of an individual call. On the next call, the request
would return 0 entries.
##########
pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/TopicsImpl.java:
##########
@@ -1591,6 +1613,103 @@ public void failed(Throwable throwable) {
return future;
}
+ @Override
+ public CompletableFuture<AnalyzeSubscriptionBacklogResult>
analyzeSubscriptionBacklogAsync(String topic,
+
String subscriptionName,
+
Optional<MessageId> startPosition,
+
long backlogScanMaxEntries) {
+ return analyzeSubscriptionBacklogAsync(topic, subscriptionName,
startPosition,
+ (backlogResult) -> backlogResult.getEntries() >=
backlogScanMaxEntries);
+ }
+
+ @Override
+ public CompletableFuture<AnalyzeSubscriptionBacklogResult>
analyzeSubscriptionBacklogAsync(String topic,
+ String
subscriptionName, Optional<MessageId> startPosition,
+
Predicate<AnalyzeSubscriptionBacklogResult> terminatePredicate) {
+ final CompletableFuture<AnalyzeSubscriptionBacklogResult> future = new
CompletableFuture<>();
+ AtomicReference<AnalyzeSubscriptionBacklogResult> resultRef = new
AtomicReference<>();
+ int partitionIndex = TopicName.get(topic).getPartitionIndex();
+ AtomicReference<Optional<MessageId>> startPositionRef = new
AtomicReference<>(startPosition);
+
+ Supplier<CompletableFuture<AnalyzeSubscriptionBacklogResult>>
resultSupplier =
+ () -> analyzeSubscriptionBacklogAsync(topic, subscriptionName,
startPositionRef.get());
+ BiConsumer<AnalyzeSubscriptionBacklogResult, Throwable> completeAction
= new BiConsumer<>() {
+ @Override
+ public void accept(AnalyzeSubscriptionBacklogResult currentResult,
Throwable throwable) {
+ if (throwable != null) {
+ future.completeExceptionally(throwable);
+ return;
+ }
+
+ AnalyzeSubscriptionBacklogResult mergedResult =
mergeBacklogResults(currentResult, resultRef.get());
+ resultRef.set(mergedResult);
+ if (!mergedResult.isAborted() ||
terminatePredicate.test(mergedResult)) {
+ future.complete(mergedResult);
+ return;
+ }
+
+ // To avoid infinite loops, we ensure the entry count is
incremented after each loop.
+ // Should never happen.
+ if (currentResult.getEntries() <= 0) {
+ log.warn("[{}][{}] Scanned total entry count is zero or
negative, abort analyze backlog, start "
+ + "position is: {}", topic, subscriptionName,
startPositionRef.get());
+ future.completeExceptionally(
+ new PulsarAdminException("Incorrect total entry
count returned from server"));
+ return;
+ }
+
+ // In analyze-backlog, lastMessageId is null only when: total
entries is 0, with false aborted flag
+ // returned. Should never happen.
+ if (StringUtils.isBlank(mergedResult.getLastMessageId())) {
+ log.warn("[{}][{}] Scanned last message id is blank, abort
analyze backlog, start position is: {}",
+ topic, subscriptionName, startPositionRef.get());
+ future.completeExceptionally(
Review Comment:
If the given message id as input doesn't return any results, this would be
expected. Similar case as mentioned in the previous comment.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
