You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by lh...@apache.org on 2024/04/03 11:55:09 UTC
(pulsar) branch branch-3.2 updated: [fix][broker] Fix invalid condition in logging exceptions (#22412)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-3.2
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/branch-3.2 by this push:
new 5dac34262fa [fix][broker] Fix invalid condition in logging exceptions (#22412)
5dac34262fa is described below
commit 5dac34262fa09bea56e97974e7a5d4be245d47e8
Author: Teet Vaher <89...@users.noreply.github.com>
AuthorDate: Wed Apr 3 14:26:24 2024 +0300
[fix][broker] Fix invalid condition in logging exceptions (#22412)
(cherry picked from commit cd6f53baee7dbe0651e6581cb9bd3570017348c7)
---
.../broker/admin/impl/PersistentTopicsBase.java | 84 +++++++++++-----------
.../broker/admin/v2/NonPersistentTopics.java | 6 +-
.../pulsar/broker/admin/v3/Transactions.java | 12 ++--
3 files changed, 51 insertions(+), 51 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 c156e9d04a6..3eddddf8773 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
@@ -878,7 +878,7 @@ public class PersistentTopicsBase extends AdminResource {
}
}).exceptionally(ex -> {
// If the exception is not redirect exception we need to log it.
- if (!isNot307And404Exception(ex)) {
+ if (isNot307And404Exception(ex)) {
log.error("[{}] Failed to get partitioned metadata while unloading topic {}",
clientAppId(), topicName, ex);
}
@@ -888,7 +888,7 @@ public class PersistentTopicsBase extends AdminResource {
}
}).exceptionally(ex -> {
// If the exception is not redirect exception we need to log it.
- if (!isNot307And404Exception(ex)) {
+ if (isNot307And404Exception(ex)) {
log.error("[{}] Failed to validate the global namespace ownership while unloading topic {}",
clientAppId(), topicName, ex);
}
@@ -1055,7 +1055,7 @@ public class PersistentTopicsBase extends AdminResource {
})
.exceptionally(ex -> {
// If the exception is not redirect exception we need to log it.
- if (!isNot307And404Exception(ex)) {
+ if (isNot307And404Exception(ex)) {
log.error("[{}] Failed to unload topic {}, {}", clientAppId(), topicName, ex);
}
resumeAsyncResponseExceptionally(asyncResponse, ex);
@@ -1075,7 +1075,7 @@ public class PersistentTopicsBase extends AdminResource {
})
.exceptionally(ex -> {
// If the exception is not redirect exception we need to log it.
- if (!isNot307And404Exception(ex)) {
+ if (isNot307And404Exception(ex)) {
log.error("[{}] Failed to unload tc {},{}", clientAppId(),
topicName.getPartitionIndex(), ex);
}
@@ -1177,7 +1177,7 @@ public class PersistentTopicsBase extends AdminResource {
}
}).exceptionally(ex -> {
// If the exception is not redirect exception we need to log it.
- if (!isNot307And404Exception(ex)) {
+ if (isNot307And404Exception(ex)) {
log.error("[{}] Failed to get partitioned topic metadata while get"
+ " subscriptions for topic {}", clientAppId(), topicName, ex);
}
@@ -1187,7 +1187,7 @@ public class PersistentTopicsBase extends AdminResource {
}
}).exceptionally(ex -> {
// If the exception is not redirect exception we need to log it.
- if (!isNot307And404Exception(ex)) {
+ if (isNot307And404Exception(ex)) {
log.error("[{}] Failed to validate the global namespace/topic ownership while get subscriptions"
+ " for topic {}", clientAppId(), topicName, ex);
}
@@ -1196,7 +1196,7 @@ public class PersistentTopicsBase extends AdminResource {
})
).exceptionally(ex -> {
// If the exception is not redirect exception we need to log it.
- if (!isNot307And404Exception(ex)) {
+ if (isNot307And404Exception(ex)) {
log.error("[{}] Failed to get subscriptions for {}", clientAppId(), topicName, ex);
}
resumeAsyncResponseExceptionally(asyncResponse, ex);
@@ -1235,7 +1235,7 @@ public class PersistentTopicsBase extends AdminResource {
.thenAccept(topic -> asyncResponse.resume(new ArrayList<>(topic.getSubscriptions().keys())))
.exceptionally(ex -> {
// If the exception is not redirect exception we need to log it.
- if (!isNot307And404Exception(ex)) {
+ if (isNot307And404Exception(ex)) {
log.error("[{}] Failed to get list of subscriptions for {}", clientAppId(), topicName, ex);
}
resumeAsyncResponseExceptionally(asyncResponse, ex);
@@ -1344,7 +1344,7 @@ public class PersistentTopicsBase extends AdminResource {
}
}).exceptionally(ex -> {
// If the exception is not redirect exception we need to log it.
- if (!isNot307And404Exception(ex)) {
+ if (isNot307And404Exception(ex)) {
log.error("[{}] Failed to get partitioned metadata while get managed info for {}",
clientAppId(), topicName, ex);
}
@@ -1354,7 +1354,7 @@ public class PersistentTopicsBase extends AdminResource {
}
}).exceptionally(ex -> {
// If the exception is not redirect exception we need to log it.
- if (!isNot307And404Exception(ex)) {
+ if (isNot307And404Exception(ex)) {
log.error("[{}] Failed to validate the global namespace ownership while get managed info for {}",
clientAppId(), topicName, ex);
}
@@ -1473,7 +1473,7 @@ public class PersistentTopicsBase extends AdminResource {
});
}).exceptionally(ex -> {
// If the exception is not redirect exception we need to log it.
- if (!isNot307And404Exception(ex)) {
+ if (isNot307And404Exception(ex)) {
log.error("[{}] Failed to get partitioned internal stats for {}", clientAppId(), topicName, ex);
}
resumeAsyncResponseExceptionally(asyncResponse, ex);
@@ -1529,7 +1529,7 @@ public class PersistentTopicsBase extends AdminResource {
});
}).exceptionally(ex -> {
// If the exception is not redirect exception we need to log it.
- if (!isNot307And404Exception(ex)) {
+ if (isNot307And404Exception(ex)) {
log.error("[{}] Failed to get partitioned internal stats for {}", clientAppId(), topicName, ex);
}
resumeAsyncResponseExceptionally(asyncResponse, ex);
@@ -1657,7 +1657,7 @@ public class PersistentTopicsBase extends AdminResource {
}).exceptionally(ex -> {
Throwable cause = ex.getCause();
// If the exception is not redirect exception we need to log it.
- if (!isNot307And404Exception(ex)) {
+ if (isNot307And404Exception(ex)) {
log.error("[{}] Failed to analyze subscription backlog {} {}",
clientAppId(), topicName, subName, cause);
}
@@ -1684,7 +1684,7 @@ public class PersistentTopicsBase extends AdminResource {
}).exceptionally(ex -> {
Throwable cause = ex.getCause();
// If the exception is not redirect exception we need to log it.
- if (!isNot307And404Exception(ex)) {
+ if (isNot307And404Exception(ex)) {
log.error("[{}] Failed to update subscription {} {}", clientAppId(), topicName, subName, cause);
}
asyncResponse.resume(new RestException(cause));
@@ -1713,7 +1713,7 @@ public class PersistentTopicsBase extends AdminResource {
}).exceptionally(ex -> {
Throwable cause = ex.getCause();
// If the exception is not redirect exception we need to log it.
- if (!isNot307And404Exception(ex)) {
+ if (isNot307And404Exception(ex)) {
log.error("[{}] Failed to update subscription {} {}", clientAppId(), topicName, subName, cause);
}
asyncResponse.resume(new RestException(cause));
@@ -1882,7 +1882,7 @@ public class PersistentTopicsBase extends AdminResource {
}
}).exceptionally(ex -> {
// If the exception is not redirect exception we need to log it.
- if (!isNot307And404Exception(ex)) {
+ if (isNot307And404Exception(ex)) {
log.error("[{}] Failed to skip all messages for subscription {} on topic {}",
clientAppId(), subName, topicName, ex);
}
@@ -1926,7 +1926,7 @@ public class PersistentTopicsBase extends AdminResource {
}
}).exceptionally(ex -> {
// If the exception is not redirect exception we need to log it.
- if (!isNot307And404Exception(ex)) {
+ if (isNot307And404Exception(ex)) {
log.error("[{}] Failed to skip all messages for subscription {} on topic {}",
clientAppId(), subName, topicName, ex);
}
@@ -1990,7 +1990,7 @@ public class PersistentTopicsBase extends AdminResource {
})
).exceptionally(ex -> {
// If the exception is not redirect exception we need to log it.
- if (!isNot307And404Exception(ex)) {
+ if (isNot307And404Exception(ex)) {
log.error("[{}] Failed to skip {} messages {} {}", clientAppId(), numMessages, topicName,
subName, ex);
}
@@ -2060,7 +2060,7 @@ public class PersistentTopicsBase extends AdminResource {
)
).exceptionally(ex -> {
// If the exception is not redirect exception we need to log it.
- if (!isNot307And404Exception(ex)) {
+ if (isNot307And404Exception(ex)) {
log.error("[{}] Failed to expire messages for all subscription on topic {}", clientAppId(), topicName,
ex);
}
@@ -2127,7 +2127,7 @@ public class PersistentTopicsBase extends AdminResource {
})
).exceptionally(ex -> {
// If the exception is not redirect exception we need to log it.
- if (!isNot307And404Exception(ex)) {
+ if (isNot307And404Exception(ex)) {
log.error("[{}] Failed to expire messages for all subscription up to {} on {}", clientAppId(),
expireTimeInSeconds, topicName, ex);
}
@@ -2335,7 +2335,7 @@ public class PersistentTopicsBase extends AdminResource {
})).exceptionally(ex -> {
// If the exception is not redirect exception we need to log it.
- if (!isNot307And404Exception(ex)) {
+ if (isNot307And404Exception(ex)) {
log.error("[{}] Failed to create subscription {} on topic {}",
clientAppId(), subscriptionName, topicName, ex);
}
@@ -2345,7 +2345,7 @@ public class PersistentTopicsBase extends AdminResource {
}
}).exceptionally(ex -> {
// If the exception is not redirect exception we need to log it.
- if (!isNot307And404Exception(ex)) {
+ if (isNot307And404Exception(ex)) {
log.error("[{}] Failed to create subscription {} on topic {}",
clientAppId(), subscriptionName, topicName, ex);
}
@@ -2475,7 +2475,7 @@ public class PersistentTopicsBase extends AdminResource {
}
}).exceptionally(ex -> {
// If the exception is not redirect exception we need to log it.
- if (!isNot307And404Exception(ex)) {
+ if (isNot307And404Exception(ex)) {
log.error("[{}] Failed to update subscription {} from topic {}",
clientAppId(), subName, topicName, ex);
}
@@ -2514,7 +2514,7 @@ public class PersistentTopicsBase extends AdminResource {
})
.exceptionally(ex -> {
// If the exception is not redirect exception we need to log it.
- if (!isNot307And404Exception(ex)) {
+ if (isNot307And404Exception(ex)) {
log.error("[{}] Failed to analyze back log of subscription {} from topic {}",
clientAppId(), subName, topicName, ex);
}
@@ -2598,7 +2598,7 @@ public class PersistentTopicsBase extends AdminResource {
}
}).exceptionally(ex -> {
// If the exception is not redirect exception we need to log it.
- if (!isNot307And404Exception(ex)) {
+ if (isNot307And404Exception(ex)) {
log.error("[{}] Failed to update subscription {} from topic {}",
clientAppId(), subName, topicName, ex);
}
@@ -2684,7 +2684,7 @@ public class PersistentTopicsBase extends AdminResource {
});
}).exceptionally(ex -> {
// If the exception is not redirect exception we need to log it.
- if (!isNot307And404Exception(ex)) {
+ if (isNot307And404Exception(ex)) {
log.warn("[{}][{}] Failed to reset cursor on subscription {} to position {}",
clientAppId(), topicName, subName, messageId, ex.getCause());
}
@@ -2693,7 +2693,7 @@ public class PersistentTopicsBase extends AdminResource {
});
}).exceptionally(ex -> {
// If the exception is not redirect exception we need to log it.
- if (!isNot307And404Exception(ex)) {
+ if (isNot307And404Exception(ex)) {
log.warn("[{}][{}] Failed to reset cursor on subscription {} to position {}",
clientAppId(), topicName, subName, messageId, ex.getCause());
}
@@ -3329,7 +3329,7 @@ public class PersistentTopicsBase extends AdminResource {
}
}).exceptionally(ex -> {
// If the exception is not redirect exception we need to log it.
- if (!isNot307And404Exception(ex)) {
+ if (isNot307And404Exception(ex)) {
log.error("[{}] Failed to get backlog size for topic {}", clientAppId(),
topicName, ex);
}
@@ -3337,7 +3337,7 @@ public class PersistentTopicsBase extends AdminResource {
return null;
})).exceptionally(ex -> {
// If the exception is not redirect exception we need to log it.
- if (!isNot307And404Exception(ex)) {
+ if (isNot307And404Exception(ex)) {
log.error("[{}] Failed to validate global namespace ownership "
+ "to get backlog size for topic {}", clientAppId(), topicName, ex);
}
@@ -3875,7 +3875,7 @@ public class PersistentTopicsBase extends AdminResource {
}
}).exceptionally(ex -> {
// If the exception is not redirect exception we need to log it.
- if (!isNot307And404Exception(ex)) {
+ if (isNot307And404Exception(ex)) {
log.error("[{}] Failed to terminate topic {}", clientAppId(), topicName, ex);
}
resumeAsyncResponseExceptionally(asyncResponse, ex);
@@ -3883,7 +3883,7 @@ public class PersistentTopicsBase extends AdminResource {
})
).exceptionally(ex -> {
// If the exception is not redirect exception we need to log it.
- if (!isNot307And404Exception(ex)) {
+ if (isNot307And404Exception(ex)) {
log.error("[{}] Failed to terminate topic {}", clientAppId(), topicName, ex);
}
resumeAsyncResponseExceptionally(asyncResponse, ex);
@@ -3973,7 +3973,7 @@ public class PersistentTopicsBase extends AdminResource {
).exceptionally(ex -> {
Throwable cause = FutureUtil.unwrapCompletionException(ex);
// If the exception is not redirect exception we need to log it.
- if (!isNot307And404Exception(cause)) {
+ if (isNot307And404Exception(cause)) {
if (cause instanceof RestException) {
log.warn("[{}] Failed to expire messages up to {} on {}: {}", clientAppId(), expireTimeInSeconds,
topicName, cause.toString());
@@ -4088,7 +4088,7 @@ public class PersistentTopicsBase extends AdminResource {
messageId, isExcluded, batchIndex);
}).exceptionally(ex -> {
// If the exception is not redirect exception we need to log it.
- if (!isNot307And404Exception(ex)) {
+ if (isNot307And404Exception(ex)) {
log.error("[{}] Failed to expire messages up to {} on subscription {} to position {}",
clientAppId(), topicName, subName, messageId, ex);
}
@@ -4239,7 +4239,7 @@ public class PersistentTopicsBase extends AdminResource {
}
}).exceptionally(ex -> {
// If the exception is not redirect exception we need to log it.
- if (!isNot307And404Exception(ex)) {
+ if (isNot307And404Exception(ex)) {
log.error("[{}] Failed to trigger compaction on topic {}", clientAppId(), topicName, ex);
}
resumeAsyncResponseExceptionally(asyncResponse, ex);
@@ -4248,7 +4248,7 @@ public class PersistentTopicsBase extends AdminResource {
}
}).exceptionally(ex -> {
// If the exception is not redirect exception we need to log it.
- if (!isNot307And404Exception(ex)) {
+ if (isNot307And404Exception(ex)) {
log.error("[{}] Failed to validate global namespace ownership to trigger compaction on topic {}",
clientAppId(), topicName, ex);
}
@@ -4277,7 +4277,7 @@ public class PersistentTopicsBase extends AdminResource {
}
}).exceptionally(ex -> {
// If the exception is not redirect exception we need to log it.
- if (!isNot307And404Exception(ex)) {
+ if (isNot307And404Exception(ex)) {
log.error("[{}] Failed to trigger compaction for {}", clientAppId(), topicName, ex);
}
resumeAsyncResponseExceptionally(asyncResponse, ex);
@@ -4313,7 +4313,7 @@ public class PersistentTopicsBase extends AdminResource {
}
}).exceptionally(ex -> {
// If the exception is not redirect exception we need to log it.
- if (!isNot307And404Exception(ex)) {
+ if (isNot307And404Exception(ex)) {
log.error("[{}] Failed to trigger offload for {}", clientAppId(), topicName, ex);
}
resumeAsyncResponseExceptionally(asyncResponse, ex);
@@ -4330,7 +4330,7 @@ public class PersistentTopicsBase extends AdminResource {
asyncResponse.resume(offloadProcessStatus);
}).exceptionally(ex -> {
// If the exception is not redirect exception we need to log it.
- if (!isNot307And404Exception(ex)) {
+ if (isNot307And404Exception(ex)) {
log.error("[{}] Failed to offload status on topic {}", clientAppId(), topicName, ex);
}
resumeAsyncResponseExceptionally(asyncResponse, ex);
@@ -4607,7 +4607,7 @@ public class PersistentTopicsBase extends AdminResource {
});
}).exceptionally(ex -> {
// If the exception is not redirect exception we need to log it.
- if (!isNot307And404Exception(ex)) {
+ if (isNot307And404Exception(ex)) {
log.error("[{}] Failed to get last messageId {}", clientAppId(), topicName, ex);
}
resumeAsyncResponseExceptionally(asyncResponse, ex);
@@ -5112,7 +5112,7 @@ public class PersistentTopicsBase extends AdminResource {
resultFuture.exceptionally(ex -> {
// If the exception is not redirect exception we need to log it.
- if (!isNot307And404Exception(ex)) {
+ if (isNot307And404Exception(ex)) {
log.warn("[{}] Failed to change replicated subscription status to {} - {} {}", clientAppId(), enabled,
topicName, subName, ex);
}
@@ -5159,7 +5159,7 @@ public class PersistentTopicsBase extends AdminResource {
}
).exceptionally(ex -> {
// If the exception is not redirect exception we need to log it.
- if (!isNot307And404Exception(ex)) {
+ if (isNot307And404Exception(ex)) {
log.error("[{}] Failed to set replicated subscription status on {} {}", clientAppId(),
topicName, subName, ex);
}
@@ -5260,7 +5260,7 @@ public class PersistentTopicsBase extends AdminResource {
}
resultFuture.exceptionally(ex -> {
- if (!isNot307And404Exception(ex)) {
+ if (isNot307And404Exception(ex)) {
log.error("[{}] Failed to get replicated subscription status on {} {}", clientAppId(),
topicName, subName, ex);
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/NonPersistentTopics.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/NonPersistentTopics.java
index a22ad4b242f..7de7d7363c0 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/NonPersistentTopics.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/NonPersistentTopics.java
@@ -132,7 +132,7 @@ public class NonPersistentTopics extends PersistentTopics {
})
.thenAccept(asyncResponse::resume)
.exceptionally(ex -> {
- if (!isNot307And404Exception(ex)) {
+ if (isNot307And404Exception(ex)) {
log.error("[{}] Failed to get internal stats for topic {}", clientAppId(), topicName, ex);
}
resumeAsyncResponseExceptionally(asyncResponse, ex);
@@ -479,7 +479,7 @@ public class NonPersistentTopics extends PersistentTopics {
}
asyncResponse.resume(topicList);
}).exceptionally(ex -> {
- if (!isNot307And404Exception(ex)) {
+ if (isNot307And404Exception(ex)) {
log.error("[{}] Failed to list topics on namespace bundle {}/{}", clientAppId(),
namespaceName, bundleRange, ex);
}
@@ -488,7 +488,7 @@ public class NonPersistentTopics extends PersistentTopics {
});
}
}).exceptionally(ex -> {
- if (!isNot307And404Exception(ex)) {
+ if (isNot307And404Exception(ex)) {
log.error("[{}] Failed to list topics on namespace bundle {}/{}", clientAppId(),
namespaceName, bundleRange, ex);
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v3/Transactions.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v3/Transactions.java
index 667d8ce581e..b2a3ae8cdb9 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v3/Transactions.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v3/Transactions.java
@@ -105,7 +105,7 @@ public class Transactions extends TransactionsBase {
Long.parseLong(leastSigBits))
.thenAccept(asyncResponse::resume)
.exceptionally(ex -> {
- if (!isNot307And404Exception(ex)) {
+ if (isNot307And404Exception(ex)) {
log.error("[{}] Failed to get transaction state in transaction buffer {}",
clientAppId(), topicName, ex);
}
@@ -143,7 +143,7 @@ public class Transactions extends TransactionsBase {
Long.parseLong(leastSigBits), subName)
.thenAccept(asyncResponse::resume)
.exceptionally(ex -> {
- if (!isNot307And404Exception(ex)) {
+ if (isNot307And404Exception(ex)) {
log.error("[{}] Failed to get transaction state in pending ack {}",
clientAppId(), topicName, ex);
}
@@ -181,7 +181,7 @@ public class Transactions extends TransactionsBase {
internalGetTransactionBufferStats(authoritative, lowWaterMarks, segmentStats)
.thenAccept(asyncResponse::resume)
.exceptionally(ex -> {
- if (!isNot307And404Exception(ex)) {
+ if (isNot307And404Exception(ex)) {
log.error("[{}] Failed to get transaction buffer stats in topic {}",
clientAppId(), topicName, ex);
}
@@ -217,7 +217,7 @@ public class Transactions extends TransactionsBase {
internalGetPendingAckStats(authoritative, subName, lowWaterMarks)
.thenAccept(asyncResponse::resume)
.exceptionally(ex -> {
- if (!isNot307And404Exception(ex)) {
+ if (isNot307And404Exception(ex)) {
log.error("[{}] Failed to get transaction pending ack stats in topic {}",
clientAppId(), topicName, ex);
}
@@ -314,7 +314,7 @@ public class Transactions extends TransactionsBase {
internalGetPendingAckInternalStats(authoritative, subName, metadata)
.thenAccept(asyncResponse::resume)
.exceptionally(ex -> {
- if (!isNot307And404Exception(ex)) {
+ if (isNot307And404Exception(ex)) {
log.error("[{}] Failed to get pending ack internal stats {}",
clientAppId(), topicName, ex);
}
@@ -365,7 +365,7 @@ public class Transactions extends TransactionsBase {
internalGetTransactionBufferInternalStats(authoritative, metadata)
.thenAccept(asyncResponse::resume)
.exceptionally(ex -> {
- if (!isNot307And404Exception(ex)) {
+ if (isNot307And404Exception(ex)) {
log.error("[{}] Failed to get transaction buffer internal stats {}",
clientAppId(), topicName, ex);
}