Skip to content

Commit

Permalink
split to 2 PRsd
Browse files Browse the repository at this point in the history
  • Loading branch information
@poorbarcode
poorbarcode committed Jan 31, 2024
1 parent e1e43be commit 4962c07
Show file tree
Hide file tree
Showing 2 changed files with 21 additions and 19 deletions.
4 changes: 3 additions & 1 deletion pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -5012,7 +5012,9 @@ protected CompletableFuture<Void> internalRemoveSubscribeRate(boolean isGlobal)

protected void handleTopicPolicyException(String methodName, Throwable thr, AsyncResponse asyncResponse) {
Throwable cause = thr.getCause();
if (!isRedirectException(cause) && !isNotFoundException(cause)) {
if (!(cause instanceof WebApplicationException) || !(
((WebApplicationException) cause).getResponse().getStatus() == 307
|| ((WebApplicationException) cause).getResponse().getStatus() == 404)) {
log.error("[{}] Failed to perform {} on topic {}",
clientAppId(), methodName, topicName, cause);
}
Expand Down
36 changes: 18 additions & 18 deletions pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/PersistentTopics.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -119,7 +119,7 @@ public void getList(
internalGetListAsync(Optional.ofNullable(bundle))
.thenAccept(topicList -> asyncResponse.resume(filterSystemTopic(topicList, includeSystemTopic)))
.exceptionally(ex -> {
if (!isRedirectException(ex) && !isNotFoundException(ex)) {
if (!isRedirectException(ex)) {
log.error("[{}] Failed to get topic list {}", clientAppId(), namespaceName, ex);
}
resumeAsyncResponseExceptionally(asyncResponse, ex);
Expand Down Expand Up @@ -150,7 +150,7 @@ public void getPartitionedTopicList(
.thenAccept(partitionedTopicList -> asyncResponse.resume(
filterSystemTopic(partitionedTopicList, includeSystemTopic)))
.exceptionally(ex -> {
if (!isRedirectException(ex) && !isNotFoundException(ex)) {
if (!isRedirectException(ex)) {
log.error("[{}] Failed to get partitioned topic list {}", clientAppId(), namespaceName, ex);
}
resumeAsyncResponseExceptionally(asyncResponse, ex);
Expand Down Expand Up @@ -335,7 +335,7 @@ public void createNonPartitionedTopic(
internalCreateNonPartitionedTopicAsync(authoritative, properties)
.thenAccept(__ -> asyncResponse.resume(Response.noContent().build()))
.exceptionally(ex -> {
if (!isRedirectException(ex) && !isNotFoundException(ex)) {
if (!isRedirectException(ex)) {
log.error("[{}] Failed to create non-partitioned topic {}", clientAppId(), topicName, ex);
}
resumeAsyncResponseExceptionally(asyncResponse, ex);
Expand Down Expand Up @@ -825,7 +825,7 @@ public void updatePartitionedTopic(
asyncResponse.resume(Response.noContent().build());
})
.exceptionally(ex -> {
if (!isRedirectException(ex) && !isNotFoundException(ex)) {
if (!isRedirectException(ex)) {
log.error("[{}][{}] Failed to update partition to {}",
clientAppId(), topicName, numPartitions, ex);
}
Expand Down Expand Up @@ -934,7 +934,7 @@ public void getProperties(
internalGetPropertiesAsync(authoritative)
.thenAccept(asyncResponse::resume)
.exceptionally(ex -> {
if (!isRedirectException(ex) && !isNotFoundException(ex)) {
if (!isRedirectException(ex)) {
log.error("[{}] Failed to get topic {} properties", clientAppId(), topicName, ex);
}
resumeAsyncResponseExceptionally(asyncResponse, ex);
Expand Down Expand Up @@ -970,7 +970,7 @@ public void updateProperties(
internalUpdatePropertiesAsync(authoritative, properties)
.thenAccept(__ -> asyncResponse.resume(Response.noContent().build()))
.exceptionally(ex -> {
if (!isRedirectException(ex) && !isNotFoundException(ex)) {
if (!isRedirectException(ex)) {
log.error("[{}] Failed to update topic {} properties", clientAppId(), topicName, ex);
}
resumeAsyncResponseExceptionally(asyncResponse, ex);
Expand Down Expand Up @@ -1004,7 +1004,7 @@ public void removeProperties(
internalRemovePropertiesAsync(authoritative, key)
.thenAccept(__ -> asyncResponse.resume(Response.noContent().build()))
.exceptionally(ex -> {
if (!isRedirectException(ex) && !isNotFoundException(ex)) {
if (!isRedirectException(ex)) {
log.error("[{}] Failed to remove key {} in properties on topic {}",
clientAppId(), key, topicName, ex);
}
Expand Down Expand Up @@ -1125,7 +1125,7 @@ public void deleteTopic(
} else if (isManagedLedgerNotFoundException(t)) {
ex = new RestException(Response.Status.NOT_FOUND,
getTopicNotFoundErrorMessage(topicName.toString()));
} else if (!isRedirectException(ex) && !isNotFoundException(ex)) {
} else if (!isRedirectException(ex)) {
log.error("[{}] Failed to delete topic {}", clientAppId(), topicName, t);
}
resumeAsyncResponseExceptionally(asyncResponse, ex);
Expand Down Expand Up @@ -1209,7 +1209,7 @@ public void getStats(
.thenAccept(asyncResponse::resume)
.exceptionally(ex -> {
// If the exception is not redirect exception we need to log it.
if (!isRedirectException(ex) && !isNotFoundException(ex)) {
if (!isRedirectException(ex)) {
log.error("[{}] Failed to get stats for {}", clientAppId(), topicName, ex);
}
resumeAsyncResponseExceptionally(asyncResponse, ex);
Expand Down Expand Up @@ -1243,7 +1243,7 @@ public void getInternalStats(
internalGetInternalStatsAsync(authoritative, metadata)
.thenAccept(asyncResponse::resume)
.exceptionally(ex -> {
if (!isRedirectException(ex) && !isNotFoundException(ex)) {
if (!isRedirectException(ex)) {
log.error("[{}] Failed to get internal stats for topic {}", clientAppId(), topicName, ex);
}
resumeAsyncResponseExceptionally(asyncResponse, ex);
Expand Down Expand Up @@ -1892,7 +1892,7 @@ public void peekNthMessage(
internalPeekNthMessageAsync(decode(encodedSubName), messagePosition, authoritative)
.thenAccept(asyncResponse::resume)
.exceptionally(ex -> {
if (!isRedirectException(ex) && !isNotFoundException(ex)) {
if (!isRedirectException(ex)) {
log.error("[{}] Failed to get peek nth message for topic {} subscription {}", clientAppId(),
topicName, decode(encodedSubName), ex);
}
Expand Down Expand Up @@ -1934,7 +1934,7 @@ public void examineMessage(
internalExamineMessageAsync(initialPosition, messagePosition, authoritative)
.thenAccept(asyncResponse::resume)
.exceptionally(ex -> {
if (!isRedirectException(ex) && !isNotFoundException(ex)) {
if (!isRedirectException(ex)) {
log.error("[{}] Failed to examine a specific message on the topic {}", clientAppId(), topicName,
ex);
}
Expand Down Expand Up @@ -1976,7 +1976,7 @@ public void getMessageById(
.thenAccept(asyncResponse::resume)
.exceptionally(ex -> {
// If the exception is not redirect exception we need to log it.
if (!isRedirectException(ex) && !isNotFoundException(ex)) {
if (!isRedirectException(ex)) {
log.error("[{}] Failed to get message with ledgerId {} entryId {} from {}",
clientAppId(), ledgerId, entryId, topicName, ex);
}
Expand Down Expand Up @@ -2020,7 +2020,7 @@ public void getMessageIdByTimestamp(
}
})
.exceptionally(ex -> {
if (!isRedirectException(ex) && !isNotFoundException(ex)) {
if (!isRedirectException(ex)) {
log.error("[{}] Failed to get message ID by timestamp {} from {}",
clientAppId(), timestamp, topicName, ex);
}
Expand Down Expand Up @@ -2055,7 +2055,7 @@ public void getBacklog(
log.warn("[{}] Failed to get topic backlog {}: Namespace does not exist", clientAppId(),
namespaceName);
ex = new RestException(Response.Status.NOT_FOUND, "Namespace does not exist");
} else if (!isRedirectException(ex) && !isNotFoundException(ex)) {
} else if (!isRedirectException(ex)) {
log.error("[{}] Failed to get estimated backlog for topic {}", clientAppId(), encodedTopic, ex);
}
resumeAsyncResponseExceptionally(asyncResponse, ex);
Expand Down Expand Up @@ -3173,7 +3173,7 @@ public void terminate(
internalTerminateAsync(authoritative)
.thenAccept(asyncResponse::resume)
.exceptionally(ex -> {
if (!isRedirectException(ex) && !isNotFoundException(ex)) {
if (!isRedirectException(ex)) {
log.error("[{}] Failed to terminated topic {}", clientAppId(), topicName, ex);
}
resumeAsyncResponseExceptionally(asyncResponse, ex);
Expand Down Expand Up @@ -3269,7 +3269,7 @@ public void compactionStatus(
internalCompactionStatusAsync(authoritative)
.thenAccept(asyncResponse::resume)
.exceptionally(ex -> {
if (!isRedirectException(ex) && !isNotFoundException(ex)) {
if (!isRedirectException(ex)) {
log.error("[{}] Failed to get the status of a compaction operation for the topic {}",
clientAppId(), topicName, ex);
}
Expand Down Expand Up @@ -3408,7 +3408,7 @@ public void trimTopic(
validateTopicName(tenant, namespace, encodedTopic);
internalTrimTopic(asyncResponse, authoritative).exceptionally(ex -> {
// If the exception is not redirect exception we need to log it.
if (!isRedirectException(ex) && !isNotFoundException(ex)) {
if (!isRedirectException(ex)) {
log.error("[{}] Failed to trim topic {}", clientAppId(), topicName, ex);
}
resumeAsyncResponseExceptionally(asyncResponse, ex);
Expand Down

0 comments on commit 4962c07

Please sign in to comment.