Merge branch 'bewaremypower/fix-release-ci' into branch-4.2.0

Author: BewareMyPower

.github/workflows/ci-build-binary-artifacts.yaml

@@ -31,7 +31,7 @@ jobs:
 
   package-linux:
     name: Build ${{matrix.pkg.name}} ${{matrix.cpu.platform}}
-    runs-on: ubuntu-22.04
+    runs-on: ${{matrix.cpu.runner}}
     timeout-minutes: 500
 
     strategy:
@@ -42,15 +42,20 @@ jobs:
           - { name: 'Deb', type: 'deb', path: 'pkg/deb/BUILD/DEB' }
           - { name: 'Alpine', type: 'apk', path: 'pkg/apk/build' }
         cpu:
-          - { arch: 'x86_64', platform: 'x86_64' }
-          - { arch: 'aarch64', platform: 'arm64' }
+          - { arch: 'x86_64', platform: 'x86_64', runner: 'ubuntu-22.04' }
+          - { arch: 'aarch64', platform: 'arm64', runner: 'ubuntu-22.04-arm' }
 
     steps:
       - name: checkout
         uses: actions/checkout@v3
 
-      - name: Set up QEMU
-        uses: docker/setup-qemu-action@ce360397dd3f832beb865e1373c09c0e9f86d70a
+      - name: Restore vcpkg binary cache
+        uses: actions/cache@v4
+        with:
+          path: ~/.cache/vcpkg/archives
+          key: vcpkg-linux-${{matrix.pkg.type}}-${{matrix.cpu.platform}}-${{hashFiles('vcpkg.json')}}
+          restore-keys: |
+            vcpkg-linux-${{matrix.pkg.type}}-${{matrix.cpu.platform}}-
 
       - name: Package Pulsar source
         run: build-support/generate-source-archive.sh
@@ -65,8 +70,8 @@ jobs:
           tags: build:latest
           platforms: linux/${{matrix.cpu.platform}}
           build-args: PLATFORM=${{matrix.cpu.arch}}
-          cache-from: type=gha
-          cache-to: type=gha,mode=max
+          cache-from: type=gha,scope=${{matrix.pkg.type}}-${{matrix.cpu.platform}}
+          cache-to: type=gha,mode=max,scope=${{matrix.pkg.type}}-${{matrix.cpu.platform}}
 
       - name: Build packages
         run: pkg/${{matrix.pkg.type}}/docker-build-${{matrix.pkg.type}}-${{matrix.cpu.platform}}.sh build:latest
@@ -211,6 +216,14 @@ jobs:
           fetch-depth: 0
           submodules: recursive
 
+      - name: Restore vcpkg binary cache
+        uses: actions/cache@v4
+        with:
+          path: ~/Library/Caches/vcpkg/archives
+          key: vcpkg-macos-${{ matrix.arch }}-${{ hashFiles('vcpkg.json') }}
+          restore-keys: |
+            vcpkg-macos-${{ matrix.arch }}-
+
       - name: Install dependencies
         run: |
           export ARCH=${{ matrix.arch }}

build-support/download-release-artifacts.py

@@ -47,7 +47,7 @@
     for artifact in data['artifacts']:
         name = artifact['name']
         # Skip debug artifact
-        if name.endswith("-Debug"):
+        if name.endswith("-Debug") or name.find("dockerbuild") >= 0:
             continue
         dest_dir = os.path.join(dest_path, name)
         if name.find("windows") >= 0 and os.path.exists(dest_dir + ".tar.gz"):

include/pulsar/Message.h

@@ -220,6 +220,14 @@ class PULSAR_PUBLIC Message {
      */
     const std::string& getProducerName() const noexcept;
 
+    /**
+     * Get the source cluster from which the message was replicated.
+     *
+     * @return the optional pointer to the source cluster name if the message was replicated, the pointer is
+     * valid as the Message instance is alive
+     */
+    std::optional<const std::string*> getReplicatedFrom() const;
+
     /**
      * @return the optional encryption context that is present when the message is encrypted, the pointer is
      * valid as the Message instance is alive

include/pulsar/c/message.h

@@ -230,6 +230,15 @@ PULSAR_PUBLIC void pulsar_message_set_schema_version(pulsar_message_t *message,
  */
 PULSAR_PUBLIC const char *pulsar_message_get_producer_name(pulsar_message_t *message);
 
+/**
+ * Get the source cluster from which the message was replicated.
+ *
+ * The pointer points to internal storage owned by the message wrapper, so the caller should not free it.
+ *
+ * @return the source cluster name, or NULL if the message is not replicated
+ */
+PULSAR_PUBLIC const char *pulsar_message_get_replicated_from(pulsar_message_t *message);
+
 /**
  * Check if the message has a null value.
  *

lib/ConsumerImpl.cc

@@ -1117,14 +1117,6 @@ Result ConsumerImpl::receiveHelper(Message& msg, int timeout) {
         if (state_ != Ready) {
             return ResultAlreadyClosed;
         }
-        auto cnx = getCnx().lock();
-        if (cnx) {
-            LOG_WARN(getName() << " Receive timeout after " << timeout << " ms, connection: "
-                               << cnx->cnxString() << ", queue size: " << incomingMessages_.size());
-        } else {
-            LOG_WARN(getName() << " Receive timeout after " << timeout
-                               << " ms, no connection, queue size: " << incomingMessages_.size());
-        }
         return ResultTimeout;
     }
 }

lib/Message.cc

@@ -239,6 +239,13 @@ const std::string& Message::getProducerName() const noexcept {
     return impl_->metadata.producer_name();
 }
 
+std::optional<const std::string*> Message::getReplicatedFrom() const {
+    if (!impl_ || !impl_->metadata.has_replicated_from()) {
+        return std::nullopt;
+    }
+    return &impl_->metadata.replicated_from();
+}
+
 std::optional<const EncryptionContext*> Message::getEncryptionContext() const {
     if (!impl_ || !impl_->encryptionContext_.has_value()) {
         return std::nullopt;

lib/MultiTopicsConsumerImpl.cc

@@ -603,14 +603,6 @@ Result MultiTopicsConsumerImpl::receive(Message& msg, int timeout) {
         if (state_ != Ready) {
             return ResultAlreadyClosed;
         }
-        auto cnx = getCnx().lock();
-        if (cnx) {
-            LOG_WARN(getName() << " Receive timeout after " << timeout << " ms, connection: "
-                               << cnx->cnxString() << ", queue size: " << incomingMessages_.size());
-        } else {
-            LOG_WARN(getName() << " Receive timeout after " << timeout
-                               << " ms, no connection, queue size: " << incomingMessages_.size());
-        }
         return ResultTimeout;
     }
 }

lib/ProducerImpl.cc

@@ -873,6 +873,9 @@ void ProducerImpl::handleSendTimeout(const ASIO_ERROR& err) {
     }
 
     lock.unlock();
+    if (pendingMessages.empty()) {
+        return;
+    }
     auto cnx = getCnx().lock();
     if (cnx) {
         LOG_WARN(getName() << "Send timeout due to queueing delay, connection: " << cnx->cnxString()

lib/c/c_Message.cc

@@ -151,4 +151,9 @@ const char *pulsar_message_get_producer_name(pulsar_message_t *message) {
     return message->message.getProducerName().c_str();
 }
 
+const char *pulsar_message_get_replicated_from(pulsar_message_t *message) {
+    const auto replicatedFrom = message->message.getReplicatedFrom();
+    return replicatedFrom ? replicatedFrom.value()->c_str() : nullptr;
+}
+
 int pulsar_message_has_null_value(pulsar_message_t *message) { return message->message.hasNullValue(); }

pkg/apk/docker-build-apk-arm64.sh

@@ -24,7 +24,9 @@ ROOT_DIR=$(git rev-parse --show-toplevel)
 
 IMAGE_NAME=${1:-apachepulsar/pulsar-build:alpine-3.16-arm64}
 
+mkdir -p $HOME/.cache/vcpkg/archives
 docker run -v $ROOT_DIR:/pulsar-client-cpp \
+        -v $HOME/.cache/vcpkg/archives:/root/.cache/vcpkg/archives \
         --env PLATFORM=aarch64 \
         $IMAGE_NAME \
         /pulsar-client-cpp/pkg/apk/build-apk.sh