Fix lint errors by the clang-analyzer checks

Author: BewareMyPower

.clang-tidy

@@ -16,58 +16,59 @@
 # under the License.
 
 ---
+#bugprone-*,
+#google-*,
+#misc-*,
+#modernize-*,
+#performance-*,
+#portability-*,
+#readability-*,
+#-google-readability-braces-around-statements,
+#-google-readability-namespace-comments,
+#-google-runtime-references,
+#-misc-non-private-member-variables-in-classes,
+#-misc-const-correctness,
+#-misc-include-cleaner,
+#-modernize-return-braced-init-list,
+#-modernize-use-trailing-return-type,
+#-modernize-concat-nested-namespaces,
+#-modernize-use-nodiscard,
+#-modernize-avoid-c-arrays,
+#-modernize-type-traits,
+#-modernize-unary-static-assert,
+#-performance-move-const-arg,
+#-performance-avoid-endl,
+#-performance-enum-size,
+#-readability-braces-around-statements,
+#-readability-identifier-length,
+#-readability-magic-numbers,
+#-readability-named-parameter,
+#-readability-redundant-declaration,
+#-readability-avoid-return-with-void-value,
+#-readability-function-cognitive-complexity,
+#-bugprone-narrowing-conversions,
+#-bugprone-easily-swappable-parameters,
+#-bugprone-inc-dec-in-conditions,
+#-bugprone-implicit-widening-of-multiplication-result,
+#-bugprone-unchecked-optional-access,
+#-bugprone-unused-local-non-trivial-variable,
+#-bugprone-unused-return-value,
+#-bugprone-reserved-identifier,
+#-google-build-using-namespace,
+#-misc-unused-parameters,
+#-modernize-deprecated-headers,
+#-modernize-use-equals-default,
+#-modernize-use-using,
+#-readability-enum-initial-value,
 Checks: >
-  # The following checks are based on https://github.com/googleapis/google-cloud-cpp/blob/main/.clang-tidy
   -*,
-  bugprone-*,
-  google-*,
-  misc-*,
-  modernize-*,
-  performance-*,
-  portability-*,
-  readability-*,
-  -google-readability-braces-around-statements,
-  -google-readability-namespace-comments,
-  -google-runtime-references,
-  -misc-non-private-member-variables-in-classes,
-  -misc-const-correctness,
-  -misc-include-cleaner,
-  -modernize-return-braced-init-list,
-  -modernize-use-trailing-return-type,
-  -modernize-concat-nested-namespaces,
-  -modernize-use-nodiscard,
-  -modernize-avoid-c-arrays,
-  -modernize-type-traits,
-  -modernize-unary-static-assert,
-  -performance-move-const-arg,
-  -performance-avoid-endl,
-  -performance-enum-size,
-  -readability-braces-around-statements,
-  -readability-identifier-length,
-  -readability-magic-numbers,
-  -readability-named-parameter,
-  -readability-redundant-declaration,
-  -readability-avoid-return-with-void-value,
-  -readability-function-cognitive-complexity,
-  -bugprone-narrowing-conversions,
-  -bugprone-easily-swappable-parameters,
-  -bugprone-inc-dec-in-conditions,
-  -bugprone-implicit-widening-of-multiplication-result,
-  -bugprone-unchecked-optional-access,
-  -bugprone-unused-local-non-trivial-variable,
-  -bugprone-unused-return-value,
-  # Here are new excluded checks that are not in the original file
-  -bugprone-reserved-identifier,
-  -google-build-using-namespace,
-  -modernize-deprecated-headers,
-  -modernize-use-using,
-  -readability-enum-initial-value,
+  clang-analyzer-*,
+  -clang-analyzer-security.insecureAPI.rand,
 WarningsAsErrors: '*'
 HeaderFileExtensions:
   - ''
   - h
 ImplementationFileExtensions:
-  # TODO: the *.c files are not being linted because they use a different style
   - cc
 HeaderFilterRegex: '.*'
 FormatStyle:     none
@@ -86,5 +87,6 @@ CheckOptions:
   readability-identifier-naming.FunctionCase: 'camelBack'
   readability-identifier-naming.MemberCase: 'camelBack'
   readability-identifier-naming.PrivateMemberSuffix: '_'
+  readability-identifier-naming.ProtectedMemberSuffix: '_'
   readability-identifier-naming.PublicMemberSuffix: ''
 SystemHeaders: false

build-support/run_clang_tidy.sh

@@ -18,6 +18,9 @@
 # under the License.
 #
 
+set -e
+cd `dirname $0`/..
+
 FILES=$(find $PWD/include $PWD/lib $PWD/tests $PWD/examples -name "*.h" -o -name "*.cc" \
     | grep -v "lib\/c\/" | grep -v "lib\/checksum\/" | grep -v "lib\/lz4\/" \
     | grep -v "include\/pulsar\/c\/" | grep -v "tests\/c\/")

lib/AckGroupingTracker.h

@@ -91,16 +91,6 @@ class AckGroupingTracker : public std::enable_shared_from_this<AckGroupingTracke
         callback(ResultOk);
     }
 
-    /**
-     * Flush all the pending grouped ACKs (as flush() does), and stop period ACKs sending.
-     */
-    virtual void close() {}
-
-    /**
-     * Flush all the pending grouped ACKs and send them to the broker.
-     */
-    virtual void flush() {}
-
     /**
      * Flush all the pending grouped ACKs (as flush() does), and clean all records about ACKed
      * messages, such as last cumulative ACKed message ID.

lib/AckGroupingTrackerEnabled.cc

@@ -112,7 +112,7 @@ void AckGroupingTrackerEnabled::addAcknowledgeCumulative(const MessageId& msgId,
     }
 }
 
-void AckGroupingTrackerEnabled::close() {
+AckGroupingTrackerEnabled::~AckGroupingTrackerEnabled() {
     isClosed_ = true;
     this->flush();
     std::lock_guard<std::mutex> lock(this->mutexTimer_);

lib/AckGroupingTrackerEnabled.h

@@ -56,15 +56,14 @@ class AckGroupingTrackerEnabled : public AckGroupingTracker {
         pendingIndividualCallbacks_.reserve(ackGroupingMaxSize);
     }
 
-    virtual ~AckGroupingTrackerEnabled() { this->close(); }
+    ~AckGroupingTrackerEnabled();
 
     void start() override;
     bool isDuplicate(const MessageId& msgId) override;
     void addAcknowledge(const MessageId& msgId, ResultCallback callback) override;
     void addAcknowledgeList(const MessageIdList& msgIds, ResultCallback callback) override;
     void addAcknowledgeCumulative(const MessageId& msgId, ResultCallback callback) override;
-    void close() override;
-    void flush() override;
+    void flush();
     void flushAndClean() override;
 
    protected:

lib/ClientImpl.cc

@@ -735,23 +735,14 @@ void ClientImpl::shutdown() {
     auto producers = producers_.move();
     auto consumers = consumers_.move();
 
-    for (auto&& kv : producers) {
-        ProducerImplBasePtr producer = kv.second.lock();
-        if (producer) {
-            producer->shutdown();
-        }
-    }
-
-    for (auto&& kv : consumers) {
-        ConsumerImplBasePtr consumer = kv.second.lock();
-        if (consumer) {
-            consumer->shutdown();
-        }
-    }
-
-    if (producers.size() + consumers.size() > 0) {
-        LOG_DEBUG(producers.size() << " producers and " << consumers.size()
-                                   << " consumers have been shutdown.");
+    auto numProducers = producers.size();
+    auto numConsumers = consumers.size();
+    // Call the destructors out of the lock to avoid deadlocks
+    producers.clear();
+    consumers.clear();
+
+    if (numProducers + numConsumers > 0) {
+        LOG_DEBUG(numProducers << " producers and " << numConsumers << " consumers have been shutdown.");
     }
     if (!pool_.close()) {
         // pool_ has already been closed. It means shutdown() has been called before.

lib/ConsumerConfigurationImpl.h

@@ -21,48 +21,45 @@
 
 #include <pulsar/ConsumerConfiguration.h>
 
-#include <chrono>
-
 namespace pulsar {
 struct ConsumerConfigurationImpl {
-    SchemaInfo schemaInfo;
     long unAckedMessagesTimeoutMs{0};
     long tickDurationInMs{1000};
-
     long negativeAckRedeliveryDelayMs{60000};
     long ackGroupingTimeMs{100};
     long ackGroupingMaxSize{1000};
-    ConsumerType consumerType{ConsumerExclusive};
-    MessageListener messageListener;
-    bool hasMessageListener{false};
+    long brokerConsumerStatsCacheTimeInMs{30 * 1000L};  // 30 seconds
+    long expireTimeOfIncompleteChunkedMessageMs{60000};
+    SchemaInfo schemaInfo;
     ConsumerEventListenerPtr eventListener;
+    CryptoKeyReaderPtr cryptoKeyReader;
+    InitialPosition subscriptionInitialPosition{InitialPosition::InitialPositionLatest};
+    int patternAutoDiscoveryPeriod{60};
+    RegexSubscriptionMode regexSubscriptionMode{RegexSubscriptionMode::PersistentOnly};
+    int priorityLevel{0};
+    bool hasMessageListener{false};
     bool hasConsumerEventListener{false};
+    bool readCompacted{false};
+    bool replicateSubscriptionStateEnabled{false};
+    bool autoAckOldestChunkedMessageOnQueueFull{false};
+    bool startMessageIdInclusive{false};
+    bool batchIndexAckEnabled{false};
+    bool ackReceiptEnabled{false};
+    bool startPaused{false};
+
+    size_t maxPendingChunkedMessage{10};
+    ConsumerType consumerType{ConsumerExclusive};
+    MessageListener messageListener;
     int receiverQueueSize{1000};
     int maxTotalReceiverQueueSizeAcrossPartitions{50000};
     std::string consumerName;
-    long brokerConsumerStatsCacheTimeInMs{30 * 1000L};  // 30 seconds
-    CryptoKeyReaderPtr cryptoKeyReader;
     ConsumerCryptoFailureAction cryptoFailureAction{ConsumerCryptoFailureAction::FAIL};
-    bool readCompacted{false};
-    InitialPosition subscriptionInitialPosition{InitialPosition::InitialPositionLatest};
     BatchReceivePolicy batchReceivePolicy{};
     DeadLetterPolicy deadLetterPolicy;
-    int patternAutoDiscoveryPeriod{60};
-    RegexSubscriptionMode regexSubscriptionMode{RegexSubscriptionMode::PersistentOnly};
-
-    bool replicateSubscriptionStateEnabled{false};
     std::map<std::string, std::string> properties;
     std::map<std::string, std::string> subscriptionProperties;
-    int priorityLevel{0};
     KeySharedPolicy keySharedPolicy;
-    size_t maxPendingChunkedMessage{10};
-    bool autoAckOldestChunkedMessageOnQueueFull{false};
-    bool startMessageIdInclusive{false};
-    long expireTimeOfIncompleteChunkedMessageMs{60000};
-    bool batchIndexAckEnabled{false};
     std::vector<ConsumerInterceptorPtr> interceptors;
-    bool ackReceiptEnabled{false};
-    bool startPaused{false};
 };
 }  // namespace pulsar
 #endif /* LIB_CONSUMERCONFIGURATIONIMPL_H_ */

lib/ConsumerImpl.cc

@@ -159,23 +159,23 @@ ConsumerImpl::ConsumerImpl(const ClientImplPtr client, const std::string& topic,
 }
 
 ConsumerImpl::~ConsumerImpl() {
-    LOG_DEBUG(getName() << "~ConsumerImpl");
+    LOG_DEBUG(consumerStr_ << "~ConsumerImpl");
     if (state_ == Ready) {
         // this could happen at least in this condition:
         //      consumer seek, caused reconnection, if consumer close happened before connection ready,
         //      then consumer will not send closeConsumer to Broker side, and caused a leak of consumer in
         //      broker.
-        LOG_WARN(getName() << "Destroyed consumer which was not properly closed");
+        LOG_WARN(consumerStr_ << "Destroyed consumer which was not properly closed");
 
         ClientConnectionPtr cnx = getCnx().lock();
         ClientImplPtr client = client_.lock();
         if (client && cnx) {
             int requestId = client->newRequestId();
             cnx->sendRequestWithId(Commands::newCloseConsumer(consumerId_, requestId), requestId);
             cnx->removeConsumer(consumerId_);
-            LOG_INFO(getName() << "Closed consumer for race condition: " << consumerId_);
+            LOG_INFO(consumerStr_ << "Closed consumer for race condition: " << consumerId_);
         } else {
-            LOG_WARN(getName() << "Client is destroyed and cannot send the CloseConsumer command");
+            LOG_WARN(consumerStr_ << "Client is destroyed and cannot send the CloseConsumer command");
         }
     }
     shutdown();
@@ -1330,9 +1330,7 @@ void ConsumerImpl::closeAsync(ResultCallback originalCallback) {
     incomingMessages_.close();
 
     // Flush pending grouped ACK requests.
-    if (ackGroupingTrackerPtr_) {
-        ackGroupingTrackerPtr_->close();
-    }
+    ackGroupingTrackerPtr_.reset();
     negativeAcksTracker_->close();
 
     ClientConnectionPtr cnx = getCnx().lock();
@@ -1360,9 +1358,7 @@ void ConsumerImpl::closeAsync(ResultCallback originalCallback) {
 const std::string& ConsumerImpl::getName() const { return consumerStr_; }
 
 void ConsumerImpl::shutdown() {
-    if (ackGroupingTrackerPtr_) {
-        ackGroupingTrackerPtr_->close();
-    }
+    ackGroupingTrackerPtr_.reset();
     incomingMessages_.clear();
     possibleSendToDeadLetterTopicMessages_.clear();
     resetCnx();

lib/ConsumerImpl.h

@@ -120,7 +120,7 @@ class ConsumerImpl : public ConsumerImplBase {
     void acknowledgeCumulativeAsync(const MessageId& msgId, ResultCallback callback) override;
     void closeAsync(ResultCallback callback) override;
     void start() override;
-    void shutdown() override;
+    void shutdown();
     bool isClosed() override;
     bool isOpen() override;
     Result pauseMessageListener() override;
@@ -277,9 +277,9 @@ class ConsumerImpl : public ConsumerImplBase {
         ChunkedMessageCtx(const ChunkedMessageCtx&) = delete;
         // Here we don't use =default to be compatible with GCC 4.8
         ChunkedMessageCtx(ChunkedMessageCtx&& rhs) noexcept
-            : totalChunks_(rhs.totalChunks_),
-              chunkedMsgBuffer_(std::move(rhs.chunkedMsgBuffer_)),
-              chunkedMessageIds_(std::move(rhs.chunkedMessageIds_)) {}
+            : totalChunks_(rhs.totalChunks_), chunkedMsgBuffer_(std::move(rhs.chunkedMsgBuffer_)) {
+            std::swap(chunkedMessageIds_, rhs.chunkedMessageIds_);
+        }
 
         bool validateChunkId(int chunkId) const noexcept { return chunkId == numChunks(); }
 
@@ -295,7 +295,11 @@ class ConsumerImpl : public ConsumerImplBase {
 
         const std::vector<MessageId>& getChunkedMessageIds() const noexcept { return chunkedMessageIds_; }
 
-        std::vector<MessageId> moveChunkedMessageIds() noexcept { return std::move(chunkedMessageIds_); }
+        std::vector<MessageId> moveChunkedMessageIds() noexcept {
+            std::vector<MessageId> result;
+            result.swap(chunkedMessageIds_);
+            return result;
+        }
 
         long getReceivedTimeMs() const noexcept { return receivedTimeMs_; }
 

lib/ConsumerImplBase.h

@@ -63,7 +63,6 @@ class ConsumerImplBase : public HandlerBase {
     virtual void acknowledgeCumulativeAsync(const MessageId& msgId, ResultCallback callback) = 0;
     virtual void closeAsync(ResultCallback callback) = 0;
     virtual void start() = 0;
-    virtual void shutdown() = 0;
     virtual bool isClosed() = 0;
     virtual bool isOpen() = 0;
     virtual Result pauseMessageListener() = 0;