Skip to content

[improve][schema] Do not print error log with stacktrace for 404 #19130

New issue

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

Sign up for GitHub

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

Already on GitHub? Sign in to your account

Merged
codelipenghui merged 1 commit into from
Jan 5, 2023
Merged

[improve][schema] Do not print error log with stacktrace for 404 #19130

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
7 changes: 7 additions & 0 deletions pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -846,6 +846,13 @@ protected static boolean isRedirectException(Throwable ex) {
== Status.TEMPORARY_REDIRECT.getStatusCode();
}

protected static boolean isNotFoundException(Throwable ex) {
Throwable realCause = FutureUtil.unwrapCompletionException(ex);
return realCause instanceof WebApplicationException
&& ((WebApplicationException) realCause).getResponse().getStatus()
== Status.NOT_FOUND.getStatusCode();
}

protected static String getTopicNotFoundErrorMessage(String topic) {
return String.format("Topic %s not found", topic);
}
Expand Down
5 changes: 5 additions & 0 deletions pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/SchemasResourceBase.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -230,5 +230,10 @@ private CompletableFuture<Void> validateOwnershipAndOperationAsync(boolean autho
.thenCompose(__ -> validateTopicOperationAsync(topicName, operation));
}


protected boolean shouldPrintErrorLog(Throwable ex) {
return !isRedirectException(ex) && !isNotFoundException(ex);
}

private static final Logger log = LoggerFactory.getLogger(SchemasResourceBase.class);
}
20 changes: 10 additions & 10 deletions pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v1/SchemasResource.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -89,10 +89,10 @@ public void getSchema(
) {
validateTopicName(tenant, cluster, namespace, topic);
getSchemaAsync(authoritative)
.thenApply(schemaAndMetadata -> convertToSchemaResponse(schemaAndMetadata))
.thenApply(this::convertToSchemaResponse)
.thenApply(response::resume)
.exceptionally(ex -> {
if (!isRedirectException(ex)) {
if (shouldPrintErrorLog(ex)) {
log.error("[{}] Failed to get schema for topic {}", clientAppId(), topicName, ex);
}
resumeAsyncResponseExceptionally(response, ex);
Expand Down Expand Up @@ -124,10 +124,10 @@ public void getSchema(
) {
validateTopicName(tenant, cluster, namespace, topic);
getSchemaAsync(authoritative, version)
.thenApply(schemaAndMetadata -> convertToSchemaResponse(schemaAndMetadata))
.thenApply(this::convertToSchemaResponse)
.thenAccept(response::resume)
.exceptionally(ex -> {
if (!isRedirectException(ex)) {
if (shouldPrintErrorLog(ex)) {
log.error("[{}] Failed to get schema for topic {} with version {}",
clientAppId(), topicName, version, ex);
}
Expand Down Expand Up @@ -159,10 +159,10 @@ public void getAllSchemas(
) {
validateTopicName(tenant, cluster, namespace, topic);
getAllSchemasAsync(authoritative)
.thenApply(schemaAndMetadata -> convertToAllVersionsSchemaResponse(schemaAndMetadata))
.thenApply(this::convertToAllVersionsSchemaResponse)
.thenAccept(response::resume)
.exceptionally(ex -> {
if (!isRedirectException(ex)) {
if (shouldPrintErrorLog(ex)) {
log.error("[{}] Failed to get all schemas for topic {}", clientAppId(), topicName, ex);
}
resumeAsyncResponseExceptionally(response, ex);
Expand Down Expand Up @@ -197,7 +197,7 @@ public void deleteSchema(
response.resume(DeleteSchemaResponse.builder().version(getLongSchemaVersion(version)).build());
})
.exceptionally(ex -> {
if (!isRedirectException(ex)) {
if (shouldPrintErrorLog(ex)) {
log.error("[{}] Failed to delete schemas for topic {}", clientAppId(), topicName, ex);
}
resumeAsyncResponseExceptionally(response, ex);
Expand Down Expand Up @@ -252,7 +252,7 @@ public void postSchema(
response.resume(Response.status(422, /* Unprocessable Entity */
root.getMessage()).build());
} else {
if (!isRedirectException(ex)) {
if (shouldPrintErrorLog(ex)) {
log.error("[{}] Failed to post schemas for topic {}", clientAppId(), topicName, root);
}
resumeAsyncResponseExceptionally(response, ex);
Expand Down Expand Up @@ -301,7 +301,7 @@ public void testCompatibility(
.schemaCompatibilityStrategy(pair.getRight().name()).build())
.build()))
.exceptionally(ex -> {
if (!isRedirectException(ex)) {
if (shouldPrintErrorLog(ex)) {
log.error("[{}] Failed to test compatibility for topic {}", clientAppId(), topicName, ex);
}
resumeAsyncResponseExceptionally(response, ex);
Expand Down Expand Up @@ -347,7 +347,7 @@ public void getVersionBySchema(
getVersionBySchemaAsync(payload, authoritative)
.thenAccept(version -> response.resume(LongSchemaVersionResponse.builder().version(version).build()))
.exceptionally(ex -> {
if (!isRedirectException(ex)) {
if (shouldPrintErrorLog(ex)) {
log.error("[{}] Failed to get version by schema for topic {}", clientAppId(), topicName, ex);
}
resumeAsyncResponseExceptionally(response, ex);
Expand Down
20 changes: 10 additions & 10 deletions pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/SchemasResource.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -89,10 +89,10 @@ public void getSchema(
@Suspended final AsyncResponse response) {
validateTopicName(tenant, namespace, topic);
getSchemaAsync(authoritative)
.thenApply(schemaAndMetadata -> convertToSchemaResponse(schemaAndMetadata))
.thenApply(this::convertToSchemaResponse)
.thenApply(response::resume)
.exceptionally(ex -> {
if (!isRedirectException(ex)) {
if (shouldPrintErrorLog(ex)) {
log.error("[{}] Failed to get schema for topic {}", clientAppId(), topicName, ex);
}
resumeAsyncResponseExceptionally(response, ex);
Expand Down Expand Up @@ -122,10 +122,10 @@ public void getSchema(
@Suspended final AsyncResponse response) {
validateTopicName(tenant, namespace, topic);
getSchemaAsync(authoritative, version)
.thenApply(schemaAndMetadata -> convertToSchemaResponse(schemaAndMetadata))
.thenApply(this::convertToSchemaResponse)
.thenAccept(response::resume)
.exceptionally(ex -> {
if (!isRedirectException(ex)) {
if (shouldPrintErrorLog(ex)) {
log.error("[{}] Failed to get schema for topic {} with version {}",
clientAppId(), topicName, version, ex);
}
Expand Down Expand Up @@ -155,10 +155,10 @@ public void getAllSchemas(
@Suspended final AsyncResponse response) {
validateTopicName(tenant, namespace, topic);
getAllSchemasAsync(authoritative)
.thenApply(schemaAndMetadata -> convertToAllVersionsSchemaResponse(schemaAndMetadata))
.thenApply(this::convertToAllVersionsSchemaResponse)
.thenAccept(response::resume)
.exceptionally(ex -> {
if (!isRedirectException(ex)) {
if (shouldPrintErrorLog(ex)) {
log.error("[{}] Failed to get all schemas for topic {}", clientAppId(), topicName, ex);
}
resumeAsyncResponseExceptionally(response, ex);
Expand Down Expand Up @@ -191,7 +191,7 @@ public void deleteSchema(
response.resume(DeleteSchemaResponse.builder().version(getLongSchemaVersion(version)).build());
})
.exceptionally(ex -> {
if (!isRedirectException(ex)) {
if (shouldPrintErrorLog(ex)) {
log.error("[{}] Failed to delete schemas for topic {}", clientAppId(), topicName, ex);
}
resumeAsyncResponseExceptionally(response, ex);
Expand Down Expand Up @@ -238,7 +238,7 @@ public void postSchema(
response.resume(Response.status(422, /* Unprocessable Entity */
root.getMessage()).build());
} else {
if (!isRedirectException(ex)) {
if (shouldPrintErrorLog(ex)) {
log.error("[{}] Failed to post schemas for topic {}", clientAppId(), topicName, root);
}
resumeAsyncResponseExceptionally(response, ex);
Expand Down Expand Up @@ -278,7 +278,7 @@ public void testCompatibility(
.schemaCompatibilityStrategy(pair.getRight().name()).build())
.build()))
.exceptionally(ex -> {
if (!isRedirectException(ex)) {
if (shouldPrintErrorLog(ex)) {
log.error("[{}] Failed to test compatibility for topic {}", clientAppId(), topicName, ex);
}
resumeAsyncResponseExceptionally(response, ex);
Expand Down Expand Up @@ -315,7 +315,7 @@ public void getVersionBySchema(
getVersionBySchemaAsync(payload, authoritative)
.thenAccept(version -> response.resume(LongSchemaVersionResponse.builder().version(version).build()))
.exceptionally(ex -> {
if (!isRedirectException(ex)) {
if (shouldPrintErrorLog(ex)) {
log.error("[{}] Failed to get version by schema for topic {}", clientAppId(), topicName, ex);
}
resumeAsyncResponseExceptionally(response, ex);
Expand Down