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:18 UTC
(pulsar) branch branch-2.10 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-2.10
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/branch-2.10 by this push:
new 71ce55baf16 [fix][broker] Fix invalid condition in logging exceptions (#22412)
71ce55baf16 is described below
commit 71ce55baf165666981f88f757d9f6b17207fcb6e
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 | 80 +++++++++++-----------
.../broker/admin/v2/NonPersistentTopics.java | 4 +-
.../pulsar/broker/admin/v3/Transactions.java | 8 +--
3 files changed, 46 insertions(+), 46 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 6e537900f39..3e3603a2e3c 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
@@ -840,7 +840,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);
}
@@ -850,7 +850,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);
}
@@ -1059,7 +1059,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);
@@ -1079,7 +1079,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);
}
@@ -1205,7 +1205,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);
}
@@ -1215,7 +1215,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);
}
@@ -1224,7 +1224,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);
@@ -1263,7 +1263,7 @@ public class PersistentTopicsBase extends AdminResource {
.thenAccept(topic -> asyncResponse.resume(Lists.newArrayList(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);
@@ -1375,7 +1375,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);
}
@@ -1385,7 +1385,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);
}
@@ -1487,7 +1487,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);
@@ -1542,7 +1542,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);
@@ -1656,7 +1656,7 @@ public class PersistentTopicsBase extends AdminResource {
"Subscription has active connected consumers"));
} else {
// 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));
@@ -1851,7 +1851,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);
}
@@ -1894,7 +1894,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);
}
@@ -1956,7 +1956,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);
}
@@ -2023,7 +2023,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);
}
@@ -2086,7 +2086,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);
}
@@ -2292,7 +2292,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);
}
@@ -2302,7 +2302,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);
}
@@ -2432,7 +2432,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);
}
@@ -2513,7 +2513,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());
}
@@ -2522,7 +2522,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());
}
@@ -3070,7 +3070,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);
}
@@ -3080,7 +3080,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);
}
resumeAsyncResponseExceptionally(asyncResponse, ex);
@@ -3088,7 +3088,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 get backlog size for topic "
+ "{}", clientAppId(), topicName, ex);
}
@@ -3600,7 +3600,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);
@@ -3608,7 +3608,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);
@@ -3689,7 +3689,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 up to {} on {}", clientAppId(),
expireTimeInSeconds, topicName, ex);
}
@@ -3799,7 +3799,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 up to {} on subscription {} to position {}",
clientAppId(), topicName, subName, messageId, ex);
}
@@ -3955,7 +3955,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);
@@ -3964,7 +3964,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);
}
@@ -3993,7 +3993,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);
@@ -4030,7 +4030,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);
@@ -4047,7 +4047,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);
@@ -4520,7 +4520,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);
@@ -4943,7 +4943,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);
}
@@ -4988,7 +4988,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);
}
@@ -5090,7 +5090,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 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 b75ec765796..1ae3ee2c628 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
@@ -498,7 +498,7 @@ public class NonPersistentTopics extends PersistentTopics {
}
asyncResponse.resume(topicList);
} catch (Exception e) {
- if (!isNot307And404Exception(e)) {
+ if (isNot307And404Exception(e)) {
log.error("[{}] Failed to list topics on namespace bundle {}/{}", clientAppId(),
namespaceName, bundleRange, e);
}
@@ -506,7 +506,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 a0faeb65459..b2597f40ff1 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
@@ -90,7 +90,7 @@ public class Transactions extends TransactionsBase {
Long.parseLong(leastSigBits))
.thenAccept(stat -> asyncResponse.resume(stat))
.exceptionally(ex -> {
- if (!isNot307And404Exception(ex)) {
+ if (isNot307And404Exception(ex)) {
log.error("[{}] Failed to get transaction state in transaction buffer {}",
clientAppId(), topicName, ex);
}
@@ -128,7 +128,7 @@ public class Transactions extends TransactionsBase {
Long.parseLong(leastSigBits), subName)
.thenAccept(stat -> asyncResponse.resume(stat))
.exceptionally(ex -> {
- if (!isNot307And404Exception(ex)) {
+ if (isNot307And404Exception(ex)) {
log.error("[{}] Failed to get transaction state in pending ack {}",
clientAppId(), topicName, ex);
}
@@ -162,7 +162,7 @@ public class Transactions extends TransactionsBase {
internalGetTransactionBufferStats(authoritative)
.thenAccept(stat -> asyncResponse.resume(stat))
.exceptionally(ex -> {
- if (!isNot307And404Exception(ex)) {
+ if (isNot307And404Exception(ex)) {
log.error("[{}] Failed to get transaction buffer stats in topic {}",
clientAppId(), topicName, ex);
}
@@ -197,7 +197,7 @@ public class Transactions extends TransactionsBase {
internalGetPendingAckStats(authoritative, subName)
.thenAccept(stats -> asyncResponse.resume(stats))
.exceptionally(ex -> {
- if (!isNot307And404Exception(ex)) {
+ if (isNot307And404Exception(ex)) {
log.error("[{}] Failed to get transaction pending ack stats in topic {}",
clientAppId(), topicName, ex);
}