Merge branch 'apache:main' into fix-531-batch-last-sequence-id

Author: zhanglistar

.github/workflows/ci-pr-validation.yaml

@@ -92,9 +92,11 @@ jobs:
 
       - name: Restore vcpkg installed cache
         uses: actions/cache@v4
+        id: vcpkg-cache
+        if: always() && steps.vcpkg-cache.outputs.cache-hit != 'true'
         with:
           path: build/vcpkg_installed
-          key: vcpkg-${{ runner.os }}-${{ hashFiles('vcpkg.json', 'CMakeLists.txt', 'vcpkg-triplets/**') }}
+          key: vcpkg-${{ runner.os }}-${{ hashFiles('vcpkg.json') }}
           restore-keys: vcpkg-${{ runner.os }}-
 
       - name: Build the project
@@ -126,11 +128,13 @@ jobs:
 
       - name: Restore vcpkg installed cache
         uses: actions/cache@v4
+        id: vcpkg-cache
+        if: always() && steps.vcpkg-cache.outputs.cache-hit != 'true'
         with:
           path: |
             build/vcpkg_installed
             build-boost-asio/vcpkg_installed
-          key: vcpkg-${{ runner.os }}-${{ hashFiles('vcpkg.json', 'CMakeLists.txt', 'vcpkg-triplets/**') }}
+          key: vcpkg-${{ runner.os }}-${{ hashFiles('vcpkg.json') }}
           restore-keys: vcpkg-${{ runner.os }}-
 
       - name: Build core libraries
@@ -168,11 +172,6 @@ jobs:
           cmake -B build -DINTEGRATE_VCPKG=ON -DBUILD_TESTS=ON -DBUILD_PERF_TOOLS=ON
           cmake --build build -j8
 
-      - name: Verify custom vcpkg installation
-        run: |
-          mv vcpkg /tmp/vcpkg-custom
-          cmake -B build-2 -DINTEGRATE_VCPKG=ON -DCMAKE_TOOLCHAIN_FILE="/tmp/vcpkg-custom/scripts/buildsystems/vcpkg.cmake"
-
   cpp20-build:
     name: Build with the C++20 standard
     needs: lint
@@ -229,7 +228,7 @@ jobs:
       - name: Restore vcpkg and its artifacts.
         uses: actions/cache@v4
         id: vcpkg-cache
-        continue-on-error: true
+        if: always() && steps.vcpkg-cache.outputs.cache-hit != 'true'
         with:
           path: |
             ${{ github.workspace }}/vcpkg_installed
@@ -238,7 +237,6 @@ jobs:
           key: ${{ runner.os }}-${{ matrix.triplet }}-vcpkg-${{ hashFiles('vcpkg.json') }}
           restore-keys: |
             ${{ runner.os }}-${{ matrix.triplet }}-vcpkg-
-          save-always: true
 
       - name: Get vcpkg(windows)
         if: ${{ runner.os == 'Windows' && steps.vcpkg-cache.outputs.cache-hit != 'true' }}
@@ -387,9 +385,11 @@ jobs:
 
       - name: Restore vcpkg installed cache
         uses: actions/cache@v4
+        id: vcpkg-cache
+        if: always() && steps.vcpkg-cache.outputs.cache-hit != 'true'
         with:
           path: build-osx/vcpkg_installed
-          key: vcpkg-${{ runner.os }}-arm64-${{ hashFiles('vcpkg.json', 'CMakeLists.txt', 'vcpkg-triplets/**') }}
+          key: vcpkg-${{ runner.os }}-arm64-${{ hashFiles('vcpkg.json') }}
           restore-keys: vcpkg-${{ runner.os }}-arm64-
 
       - name: Build libraries

build-support/run_clang_tidy.sh

@@ -34,8 +34,10 @@ SCRIPT=$(which run-clang-tidy)
 set +e
 nproc
 if [[ $? == 0 ]]; then
-    python3 $SCRIPT -p build -j$(nproc) $(cat files.txt)
+    NUM_THREADS=$(nproc)
 else
-    python3 $SCRIPT -p build -j8 $(cat files.txt)
+    NUM_THREADS=8
 fi
+set -e
+python3 $SCRIPT -p build -j$NUM_THREADS $(cat files.txt)
 rm -f files.txt

lib/ClientConnection.cc

@@ -597,7 +597,7 @@ void ClientConnection::tcpConnectAsync() {
 
     auto weakSelf = weak_from_this();
     resolver_->async_resolve(service_url.host(), std::to_string(service_url.port()),
-                             [weakSelf](const ASIO_ERROR& err, tcp::resolver::results_type results) {
+                             [weakSelf](auto err, const auto& results) {
                                  auto self = weakSelf.lock();
                                  if (self) {
                                      self->handleResolve(err, results);

tests/ReaderTest.cc

@@ -21,9 +21,13 @@
 #include <pulsar/Reader.h>
 #include <time.h>
 
+#include <atomic>
+#include <functional>
 #include <future>
+#include <set>
 #include <string>
 #include <thread>
+#include <vector>
 
 #include "HttpHelper.h"
 #include "PulsarFriend.h"
@@ -110,15 +114,27 @@ TEST_P(ReaderTest, testAsyncRead) {
         ASSERT_EQ(ResultOk, producer.send(msg));
     }
 
+    // readNextAsync callbacks may complete in any order (e.g. with partitioned topic); collect all 10 then
+    // verify set
+    std::string received[10];
+    std::atomic<int> receivedCount{0};
     for (int i = 0; i < 10; i++) {
-        reader.readNextAsync([i](Result result, const Message& msg) {
+        reader.readNextAsync([&](Result result, const Message& msg) {
             ASSERT_EQ(ResultOk, result);
-            std::string content = msg.getDataAsString();
-            std::string expected = "my-message-" + std::to_string(i);
-            ASSERT_EQ(expected, content);
+            int idx = receivedCount.fetch_add(1);
+            if (idx < 10) received[idx] = msg.getDataAsString();
         });
     }
 
+    waitUntil(
+        std::chrono::seconds(5), [&]() { return receivedCount.load() == 10; }, 1000);
+    ASSERT_EQ(10, receivedCount.load()) << "Expected 10 messages";
+
+    std::set<std::string> receivedSet(received, received + 10);
+    for (int i = 0; i < 10; i++) {
+        ASSERT_TRUE(receivedSet.count("my-message-" + std::to_string(i))) << "Missing my-message-" << i;
+    }
+
     waitUntil(
         std::chrono::seconds(5),
         [&]() {

tests/TlsNegotiationTest.cc

@@ -135,7 +135,7 @@ TEST(TlsNegotiationTest, testTls12) {
     // It will fail to create producer because mock server doesn't speak Pulsar,
     // but we only care about the handshake.
     Producer producer;
-    client.createProducerAsync("topic", [](Result, Producer) {});
+    client.createProducerAsync("topic", [](auto, const auto&) {});
 
     // Wait for handshake
     ASSERT_TRUE(handshakeFuture.get());
@@ -170,7 +170,7 @@ TEST(TlsNegotiationTest, testTls13) {
 
     Client client(serviceUrl, config);
 
-    client.createProducerAsync("topic", [](Result, Producer) {});
+    client.createProducerAsync("topic", [](auto, const auto&) {});
 
     ASSERT_TRUE(handshakeFuture.get());