You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by xy...@apache.org on 2024/03/28 02:45:19 UTC
(pulsar-client-cpp) branch branch-3.5 updated (916af95 -> 0ba3d7d)
This is an automated email from the ASF dual-hosted git repository.
xyz pushed a change to branch branch-3.5
in repository https://gitbox.apache.org/repos/asf/pulsar-client-cpp.git
from 916af95 Merge branch 'main' into branch-3.5
new e7793d6 Support customize vcpkg directory when INTEGRATE_VCPKG is ON (#417)
new 0e0ca8d Fix broken wireshark build workflow on macOS (#414)
new 7cfd775 Fix minor issues reported by CodeQL (#421)
new b2ad352 Fix wrong results of hasMessageAvailable and readNext after seeking by timestamp (#422)
new 0ba3d7d Release 3.5.1
The 5 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.github/workflows/ci-pr-validation.yaml | 10 ++++
CMakeLists.txt | 4 +-
README.md | 13 +++++
lib/ConsumerImpl.cc | 41 +++++++++-------
lib/ConsumerImpl.h | 15 +++++-
lib/lz4/lz4.cc | 6 +--
perf/PerfProducer.cc | 2 +-
tests/ReaderTest.cc | 84 ++++++++++++++++++++++++++-------
version.txt | 2 +-
9 files changed, 135 insertions(+), 42 deletions(-)
(pulsar-client-cpp) 04/05: Fix wrong results of hasMessageAvailable and readNext after seeking by timestamp (#422)
Posted by xy...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
xyz pushed a commit to branch branch-3.5
in repository https://gitbox.apache.org/repos/asf/pulsar-client-cpp.git
commit b2ad3525eb762de5b92cc2af92cbe2e02c3e90cd
Author: Yunze Xu <xy...@163.com>
AuthorDate: Thu Mar 28 10:36:12 2024 +0800
Fix wrong results of hasMessageAvailable and readNext after seeking by timestamp (#422)
Fixes https://github.com/apache/pulsar-client-cpp/issues/420
It's a catch-up for https://github.com/apache/pulsar/pull/22363
(cherry picked from commit 27d8cc01d5b7cfce94a7d71259dca5cea83ae01e)
---
lib/ConsumerImpl.cc | 41 ++++++++++++++------------
lib/ConsumerImpl.h | 15 +++++++++-
tests/ReaderTest.cc | 84 ++++++++++++++++++++++++++++++++++++++++++-----------
3 files changed, 104 insertions(+), 36 deletions(-)
diff --git a/lib/ConsumerImpl.cc b/lib/ConsumerImpl.cc
index ebc8518..1a0b0cb 100644
--- a/lib/ConsumerImpl.cc
+++ b/lib/ConsumerImpl.cc
@@ -1050,7 +1050,9 @@ void ConsumerImpl::messageProcessed(Message& msg, bool track) {
*/
void ConsumerImpl::clearReceiveQueue() {
if (duringSeek()) {
- startMessageId_ = seekMessageId_.get();
+ if (!hasSoughtByTimestamp_.load(std::memory_order_acquire)) {
+ startMessageId_ = seekMessageId_.get();
+ }
SeekStatus expected = SeekStatus::COMPLETED;
if (seekStatus_.compare_exchange_strong(expected, SeekStatus::NOT_STARTED)) {
auto seekCallback = seekCallback_.release();
@@ -1476,7 +1478,7 @@ void ConsumerImpl::seekAsync(const MessageId& msgId, ResultCallback callback) {
return;
}
const auto requestId = client->newRequestId();
- seekAsyncInternal(requestId, Commands::newSeek(consumerId_, requestId, msgId), msgId, 0L, callback);
+ seekAsyncInternal(requestId, Commands::newSeek(consumerId_, requestId, msgId), SeekArg{msgId}, callback);
}
void ConsumerImpl::seekAsync(uint64_t timestamp, ResultCallback callback) {
@@ -1495,8 +1497,8 @@ void ConsumerImpl::seekAsync(uint64_t timestamp, ResultCallback callback) {
return;
}
const auto requestId = client->newRequestId();
- seekAsyncInternal(requestId, Commands::newSeek(consumerId_, requestId, timestamp), MessageId::earliest(),
- timestamp, callback);
+ seekAsyncInternal(requestId, Commands::newSeek(consumerId_, requestId, timestamp), SeekArg{timestamp},
+ callback);
}
bool ConsumerImpl::isReadCompacted() { return readCompacted_; }
@@ -1509,7 +1511,7 @@ void ConsumerImpl::hasMessageAvailableAsync(HasMessageAvailableCallback callback
(lastDequedMessageId_ == MessageId::earliest()) &&
(startMessageId_.get().value_or(MessageId::earliest()) == MessageId::latest());
}
- if (compareMarkDeletePosition) {
+ if (compareMarkDeletePosition || hasSoughtByTimestamp_.load(std::memory_order_acquire)) {
auto self = get_shared_this_ptr();
getLastMessageIdAsync([self, callback](Result result, const GetLastMessageIdResponse& response) {
if (result != ResultOk) {
@@ -1518,8 +1520,8 @@ void ConsumerImpl::hasMessageAvailableAsync(HasMessageAvailableCallback callback
}
auto handleResponse = [self, response, callback] {
if (response.hasMarkDeletePosition() && response.getLastMessageId().entryId() >= 0) {
- // We only care about comparing ledger ids and entry ids as mark delete position doesn't
- // have other ids such as batch index
+ // We only care about comparing ledger ids and entry ids as mark delete position
+ // doesn't have other ids such as batch index
auto compareResult = compareLedgerAndEntryId(response.getMarkDeletePosition(),
response.getLastMessageId());
callback(ResultOk, self->config_.isStartMessageIdInclusive() ? compareResult <= 0
@@ -1528,7 +1530,8 @@ void ConsumerImpl::hasMessageAvailableAsync(HasMessageAvailableCallback callback
callback(ResultOk, false);
}
};
- if (self->config_.isStartMessageIdInclusive()) {
+ if (self->config_.isStartMessageIdInclusive() &&
+ !self->hasSoughtByTimestamp_.load(std::memory_order_acquire)) {
self->seekAsync(response.getLastMessageId(), [callback, handleResponse](Result result) {
if (result != ResultOk) {
callback(result, {});
@@ -1644,8 +1647,8 @@ bool ConsumerImpl::isConnected() const { return !getCnx().expired() && state_ ==
uint64_t ConsumerImpl::getNumberOfConnectedConsumer() { return isConnected() ? 1 : 0; }
-void ConsumerImpl::seekAsyncInternal(long requestId, SharedBuffer seek, const MessageId& seekId,
- long timestamp, ResultCallback callback) {
+void ConsumerImpl::seekAsyncInternal(long requestId, SharedBuffer seek, const SeekArg& seekArg,
+ ResultCallback callback) {
ClientConnectionPtr cnx = getCnx().lock();
if (!cnx) {
LOG_ERROR(getName() << " Client Connection not ready for Consumer");
@@ -1655,21 +1658,21 @@ void ConsumerImpl::seekAsyncInternal(long requestId, SharedBuffer seek, const Me
auto expected = SeekStatus::NOT_STARTED;
if (!seekStatus_.compare_exchange_strong(expected, SeekStatus::IN_PROGRESS)) {
- LOG_ERROR(getName() << " attempted to seek (" << seekId << ", " << timestamp << " when the status is "
+ LOG_ERROR(getName() << " attempted to seek " << seekArg << " when the status is "
<< static_cast<int>(expected));
callback(ResultNotAllowedError);
return;
}
const auto originalSeekMessageId = seekMessageId_.get();
- seekMessageId_ = seekId;
- seekStatus_ = SeekStatus::IN_PROGRESS;
- seekCallback_ = std::move(callback);
- if (timestamp > 0) {
- LOG_INFO(getName() << " Seeking subscription to " << timestamp);
+ if (boost::get<uint64_t>(&seekArg)) {
+ hasSoughtByTimestamp_.store(true, std::memory_order_release);
} else {
- LOG_INFO(getName() << " Seeking subscription to " << seekId);
+ seekMessageId_ = *boost::get<MessageId>(&seekArg);
}
+ seekStatus_ = SeekStatus::IN_PROGRESS;
+ seekCallback_ = std::move(callback);
+ LOG_INFO(getName() << " Seeking subscription to " << seekArg);
std::weak_ptr<ConsumerImpl> weakSelf{get_shared_this_ptr()};
@@ -1692,7 +1695,9 @@ void ConsumerImpl::seekAsyncInternal(long requestId, SharedBuffer seek, const Me
// It's during reconnection, complete the seek future after connection is established
seekStatus_ = SeekStatus::COMPLETED;
} else {
- startMessageId_ = seekMessageId_.get();
+ if (!hasSoughtByTimestamp_.load(std::memory_order_acquire)) {
+ startMessageId_ = seekMessageId_.get();
+ }
seekCallback_.release()(result);
}
} else {
diff --git a/lib/ConsumerImpl.h b/lib/ConsumerImpl.h
index 82e323b..35636ad 100644
--- a/lib/ConsumerImpl.h
+++ b/lib/ConsumerImpl.h
@@ -22,6 +22,7 @@
#include <pulsar/Reader.h>
#include <boost/optional.hpp>
+#include <boost/variant.hpp>
#include <functional>
#include <list>
#include <memory>
@@ -201,7 +202,18 @@ class ConsumerImpl : public ConsumerImplBase {
BrokerGetLastMessageIdCallback callback);
void clearReceiveQueue();
- void seekAsyncInternal(long requestId, SharedBuffer seek, const MessageId& seekId, long timestamp,
+ using SeekArg = boost::variant<uint64_t, MessageId>;
+ friend std::ostream& operator<<(std::ostream& os, const SeekArg& seekArg) {
+ auto ptr = boost::get<uint64_t>(&seekArg);
+ if (ptr) {
+ os << *ptr;
+ } else {
+ os << *boost::get<MessageId>(&seekArg);
+ }
+ return os;
+ }
+
+ void seekAsyncInternal(long requestId, SharedBuffer seek, const SeekArg& seekArg,
ResultCallback callback);
void processPossibleToDLQ(const MessageId& messageId, ProcessDLQCallBack cb);
@@ -250,6 +262,7 @@ class ConsumerImpl : public ConsumerImplBase {
Synchronized<ResultCallback> seekCallback_{[](Result) {}};
Synchronized<boost::optional<MessageId>> startMessageId_;
Synchronized<MessageId> seekMessageId_{MessageId::earliest()};
+ std::atomic<bool> hasSoughtByTimestamp_{false};
bool duringSeek() const { return seekStatus_ != SeekStatus::NOT_STARTED; }
diff --git a/tests/ReaderTest.cc b/tests/ReaderTest.cc
index 9d2fe5f..92fdf62 100644
--- a/tests/ReaderTest.cc
+++ b/tests/ReaderTest.cc
@@ -700,9 +700,16 @@ TEST_P(ReaderTest, testReceiveAfterSeek) {
client.close();
}
-TEST(ReaderSeekTest, testSeekForMessageId) {
- Client client(serviceUrl);
+class ReaderSeekTest : public ::testing::TestWithParam<bool> {
+ public:
+ void SetUp() override { client = Client{serviceUrl}; }
+ void TearDown() override { client.close(); }
+
+ Client client{serviceUrl};
+};
+
+TEST_F(ReaderSeekTest, testSeekForMessageId) {
const std::string topic = "test-seek-for-message-id-" + std::to_string(time(nullptr));
Producer producer;
@@ -752,18 +759,24 @@ TEST(ReaderSeekTest, testSeekForMessageId) {
producer.close();
}
-class ReaderSeekTest : public ::testing::TestWithParam<bool> {};
-
-TEST(ReaderSeekTest, testStartAtLatestMessageId) {
- Client client(serviceUrl);
+#define EXPECT_HAS_MESSAGE_AVAILABLE(reader, expected) \
+ { \
+ bool actual; \
+ ASSERT_EQ(ResultOk, reader.hasMessageAvailable(actual)); \
+ EXPECT_EQ(actual, (expected)); \
+ }
+TEST_F(ReaderSeekTest, testStartAtLatestMessageId) {
const std::string topic = "test-seek-latest-message-id-" + std::to_string(time(nullptr));
Producer producer;
ASSERT_EQ(ResultOk, client.createProducer(topic, producer));
MessageId id;
- ASSERT_EQ(ResultOk, producer.send(MessageBuilder().setContent("msg").build(), id));
+ for (int i = 0; i < 10; i++) {
+ ASSERT_EQ(ResultOk,
+ producer.send(MessageBuilder().setContent("msg-" + std::to_string(i)).build(), id));
+ }
Reader readerExclusive;
ASSERT_EQ(ResultOk,
@@ -774,20 +787,24 @@ TEST(ReaderSeekTest, testStartAtLatestMessageId) {
client.createReader(topic, MessageId::latest(),
ReaderConfiguration().setStartMessageIdInclusive(true), readerInclusive));
+ EXPECT_HAS_MESSAGE_AVAILABLE(readerExclusive, false);
+ EXPECT_HAS_MESSAGE_AVAILABLE(readerInclusive, true);
+
Message msg;
- bool hasMsgAvaliable = false;
- readerInclusive.hasMessageAvailable(hasMsgAvaliable);
- ASSERT_TRUE(hasMsgAvaliable);
ASSERT_EQ(ResultOk, readerInclusive.readNext(msg, 3000));
- ASSERT_EQ(ResultTimeout, readerExclusive.readNext(msg, 3000));
+ ASSERT_EQ(msg.getDataAsString(), "msg-9");
+
+ readerInclusive.seek(0L);
+ EXPECT_HAS_MESSAGE_AVAILABLE(readerInclusive, true);
+ ASSERT_EQ(ResultOk, readerInclusive.readNext(msg, 3000));
+ ASSERT_EQ(msg.getDataAsString(), "msg-0");
readerExclusive.close();
readerInclusive.close();
producer.close();
}
-TEST(ReaderTest, testSeekInProgress) {
- Client client(serviceUrl);
+TEST_F(ReaderSeekTest, testSeekInProgress) {
const auto topic = "test-seek-in-progress-" + std::to_string(time(nullptr));
Reader reader;
ASSERT_EQ(ResultOk, client.createReader(topic, MessageId::earliest(), {}, reader));
@@ -798,11 +815,9 @@ TEST(ReaderTest, testSeekInProgress) {
Result result;
promise.getFuture().get(result);
ASSERT_EQ(result, ResultNotAllowedError);
- client.close();
}
TEST_P(ReaderSeekTest, testHasMessageAvailableAfterSeekToEnd) {
- Client client(serviceUrl);
const auto topic = "test-has-message-available-after-seek-to-end-" + std::to_string(time(nullptr));
Producer producer;
ASSERT_EQ(ResultOk, client.createProducer(topic, producer));
@@ -814,7 +829,6 @@ TEST_P(ReaderSeekTest, testHasMessageAvailableAfterSeekToEnd) {
bool hasMessageAvailable;
if (GetParam()) {
- // Test the case when `ConsumerImpl.lastMessageIdInBroker_` has been initialized
ASSERT_EQ(ResultOk, reader.hasMessageAvailable(hasMessageAvailable));
}
@@ -834,8 +848,44 @@ TEST_P(ReaderSeekTest, testHasMessageAvailableAfterSeekToEnd) {
ASSERT_EQ(ResultOk, reader.seek(MessageId::latest()));
ASSERT_EQ(ResultOk, reader.hasMessageAvailable(hasMessageAvailable));
ASSERT_FALSE(hasMessageAvailable);
+}
- client.close();
+TEST_P(ReaderSeekTest, testHasMessageAvailableAfterSeekTimestamp) {
+ using namespace std::chrono;
+ const auto topic = "test-has-message-available-after-seek-timestamp-" + std::to_string(time(nullptr));
+ Producer producer;
+ ASSERT_EQ(ResultOk, client.createProducer(topic, producer));
+ MessageId sentMsgId;
+ const auto timestampBeforeSend =
+ duration_cast<milliseconds>(system_clock::now().time_since_epoch()).count();
+ ASSERT_EQ(ResultOk, producer.send(MessageBuilder().setContent("msg").build(), sentMsgId));
+
+ auto createReader = [this, &topic](Reader& reader, const MessageId& msgId) {
+ ASSERT_EQ(ResultOk, client.createReader(topic, msgId, {}, reader));
+ if (GetParam()) {
+ if (msgId == MessageId::earliest()) {
+ EXPECT_HAS_MESSAGE_AVAILABLE(reader, true);
+ } else {
+ EXPECT_HAS_MESSAGE_AVAILABLE(reader, false);
+ }
+ }
+ };
+
+ std::vector<MessageId> msgIds{MessageId::earliest(), sentMsgId, MessageId::latest()};
+
+ for (auto&& msgId : msgIds) {
+ Reader reader;
+ createReader(reader, msgId);
+ ASSERT_EQ(ResultOk,
+ reader.seek(duration_cast<milliseconds>(system_clock::now().time_since_epoch()).count()));
+ EXPECT_HAS_MESSAGE_AVAILABLE(reader, false);
+ }
+ for (auto&& msgId : msgIds) {
+ Reader reader;
+ createReader(reader, msgId);
+ ASSERT_EQ(ResultOk, reader.seek(timestampBeforeSend));
+ EXPECT_HAS_MESSAGE_AVAILABLE(reader, true);
+ }
}
INSTANTIATE_TEST_SUITE_P(Pulsar, ReaderTest, ::testing::Values(true, false));
(pulsar-client-cpp) 05/05: Release 3.5.1
Posted by xy...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
xyz pushed a commit to branch branch-3.5
in repository https://gitbox.apache.org/repos/asf/pulsar-client-cpp.git
commit 0ba3d7d8e9f7d3fb2f2cc3cdf918bb63c4791afc
Author: Yunze Xu <xy...@163.com>
AuthorDate: Thu Mar 28 10:45:07 2024 +0800
Release 3.5.1
---
version.txt | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/version.txt b/version.txt
index 1545d96..d5c0c99 100644
--- a/version.txt
+++ b/version.txt
@@ -1 +1 @@
-3.5.0
+3.5.1
(pulsar-client-cpp) 01/05: Support customize vcpkg directory when INTEGRATE_VCPKG is ON (#417)
Posted by xy...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
xyz pushed a commit to branch branch-3.5
in repository https://gitbox.apache.org/repos/asf/pulsar-client-cpp.git
commit e7793d6752a6bbfd5f526ecf207bd3738179f29e
Author: Yunze Xu <xy...@163.com>
AuthorDate: Thu Mar 14 20:11:26 2024 +0800
Support customize vcpkg directory when INTEGRATE_VCPKG is ON (#417)
### Motivation
Currently when INTEGRATE_VCPKG is ON, the CMAKE_TOOLCHAIN_FILE variable
is always a subdirectory of `${CMAKE_SOURCE_DIR}/vcpkg`. We can only
customize the vcpkg directory when INTEGRATE_VCPKG is OFF, while the
legacy CMake logic is incompatible with this way.
### Modifications
When INTEGRATE_VCPKG is ON, only set CMAKE_TOOLCHAIN_FILE if it's not
defined. The workflow and README are updated for it.
(cherry picked from commit 821871777e247e1ccbfa323ea0d5136cf0e18711)
---
.github/workflows/ci-pr-validation.yaml | 5 +++++
CMakeLists.txt | 4 +++-
README.md | 13 +++++++++++++
3 files changed, 21 insertions(+), 1 deletion(-)
diff --git a/.github/workflows/ci-pr-validation.yaml b/.github/workflows/ci-pr-validation.yaml
index d9dc2ca..5010411 100644
--- a/.github/workflows/ci-pr-validation.yaml
+++ b/.github/workflows/ci-pr-validation.yaml
@@ -114,6 +114,11 @@ jobs:
cmake . -DINTEGRATE_VCPKG=ON -DBUILD_TESTS=ON -DBUILD_PERF_TOOLS=ON
cmake --build . -j8
+ - name: Verify custom vcpkg installation
+ run: |
+ mv vcpkg /tmp/
+ cmake -B build -DINTEGRATE_VCPKG=ON -DCMAKE_TOOLCHAIN_FILE="/tmp/vcpkg/scripts/buildsystems/vcpkg.cmake"
+
cpp20-build:
name: Build with the C++20 standard
needs: formatting-check
diff --git a/CMakeLists.txt b/CMakeLists.txt
index 662e84a..b004653 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -24,7 +24,9 @@ option(USE_ASIO "Use Asio instead of Boost.Asio" OFF)
option(INTEGRATE_VCPKG "Integrate with Vcpkg" OFF)
if (INTEGRATE_VCPKG)
set(USE_ASIO ON)
- set(CMAKE_TOOLCHAIN_FILE "${CMAKE_SOURCE_DIR}/vcpkg/scripts/buildsystems/vcpkg.cmake")
+ if (NOT CMAKE_TOOLCHAIN_FILE)
+ set(CMAKE_TOOLCHAIN_FILE "${CMAKE_SOURCE_DIR}/vcpkg/scripts/buildsystems/vcpkg.cmake")
+ endif ()
endif ()
option(BUILD_TESTS "Build tests" ON)
diff --git a/README.md b/README.md
index f0c13f0..4c86b63 100644
--- a/README.md
+++ b/README.md
@@ -47,6 +47,8 @@ Since it's integrated with vcpkg, see [vcpkg#README](https://github.com/microsof
### How to build from source
+The simplest way is to clone this project with the vcpkg submodule.
+
```bash
git clone https://github.com/apache/pulsar-client-cpp.git
cd pulsar-client-cpp
@@ -57,6 +59,17 @@ cmake --build build -j8
The 1st step will download vcpkg and then install all dependencies according to the version description in [vcpkg.json](./vcpkg.json). The 2nd step will build the Pulsar C++ libraries under `./build/lib/`, where `./build` is the CMake build directory.
+> You can also add the CMAKE_TOOLCHAIN_FILE option if your system already have vcpkg installed.
+>
+> ```bash
+> git clone https://github.com/apache/pulsar-client-cpp.git
+> cd pulsar-client-cpp
+> # For example, you can install vcpkg in /tmp/vcpkg
+> cd /tmp && git clone https://github.com/microsoft/vcpkg.git && cd -
+> cmake -B build -DINTEGRATE_VCPKG=ON -DCMAKE_TOOLCHAIN_FILE="/tmp/vcpkg/scripts/buildsystems/vcpkg.cmake"
+> cmake --build build -j8
+> ```
+
After the build, the hierarchy of the `build` directory will be:
```
(pulsar-client-cpp) 03/05: Fix minor issues reported by CodeQL (#421)
Posted by xy...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
xyz pushed a commit to branch branch-3.5
in repository https://gitbox.apache.org/repos/asf/pulsar-client-cpp.git
commit 7cfd77506c83f259c69b440082d3b558b60695e4
Author: Matteo Merli <mm...@apache.org>
AuthorDate: Wed Mar 27 02:47:42 2024 -0700
Fix minor issues reported by CodeQL (#421)
(cherry picked from commit 763b85c6c4b9bb648b9f7cf62f9ed09f04f3decb)
---
lib/lz4/lz4.cc | 6 +++---
perf/PerfProducer.cc | 2 +-
2 files changed, 4 insertions(+), 4 deletions(-)
diff --git a/lib/lz4/lz4.cc b/lib/lz4/lz4.cc
index d63b977..2f98fb3 100644
--- a/lib/lz4/lz4.cc
+++ b/lib/lz4/lz4.cc
@@ -1175,9 +1175,9 @@ FORCE_INLINE int LZ4_decompress_generic(
s = *ip++;
length += s;
} while (likely((endOnInput) ? ip < iend - RUN_MASK : 1) && (s == 255));
- if ((safeDecode) && unlikely((size_t)(op + length) < (size_t)(op)))
+ if ((safeDecode) && unlikely(length >= (size_t)(oend - op)))
goto _output_error; /* overflow detection */
- if ((safeDecode) && unlikely((size_t)(ip + length) < (size_t)(ip)))
+ if ((safeDecode) && unlikely(length >= (size_t)(iend - ip)))
goto _output_error; /* overflow detection */
}
@@ -1220,7 +1220,7 @@ FORCE_INLINE int LZ4_decompress_generic(
s = *ip++;
length += s;
} while (s == 255);
- if ((safeDecode) && unlikely((size_t)(op + length) < (size_t)op))
+ if ((safeDecode) && unlikely(length >= (size_t)(oend - op)))
goto _output_error; /* overflow detection */
}
length += MINMATCH;
diff --git a/perf/PerfProducer.cc b/perf/PerfProducer.cc
index aeda8e8..cbfef68 100644
--- a/perf/PerfProducer.cc
+++ b/perf/PerfProducer.cc
@@ -160,7 +160,7 @@ void startPerfProducer(const Arguments& args, pulsar::ProducerConfiguration& pro
limiter = std::make_shared<pulsar::RateLimiter>(args.rate);
}
- producerList.resize(args.numTopics * args.numProducers);
+ producerList.resize((size_t)args.numTopics * args.numProducers);
for (int i = 0; i < args.numTopics; i++) {
std::string topic = (args.numTopics == 1) ? args.topic : args.topic + "-" + std::to_string(i);
LOG_INFO("Adding " << args.numProducers << " producers on topic " << topic);
(pulsar-client-cpp) 02/05: Fix broken wireshark build workflow on macOS (#414)
Posted by xy...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
xyz pushed a commit to branch branch-3.5
in repository https://gitbox.apache.org/repos/asf/pulsar-client-cpp.git
commit 0e0ca8da3e7cabf76adb5d5ae30283f909e0aaa3
Author: Yunze Xu <xy...@163.com>
AuthorDate: Sun Mar 17 23:10:15 2024 +0800
Fix broken wireshark build workflow on macOS (#414)
### Motivation
See https://github.com/apache/pulsar-client-cpp/actions/runs/8276076995/job/22644077705
```
Error: The `brew link` step did not complete successfully
The formula built, but is not symlinked into /usr/local
Could not symlink bin/2to3
```
`brew install` failed because the wireshark dependency depends on
python@3.12 and it failed at `brew link`.
### Modifications
Remove the existing binaries that might conflict.
(cherry picked from commit c513f29fadce86bacaf4f878d8d25064f7560083)
---
.github/workflows/ci-pr-validation.yaml | 5 +++++
1 file changed, 5 insertions(+)
diff --git a/.github/workflows/ci-pr-validation.yaml b/.github/workflows/ci-pr-validation.yaml
index 5010411..9f113b0 100644
--- a/.github/workflows/ci-pr-validation.yaml
+++ b/.github/workflows/ci-pr-validation.yaml
@@ -64,6 +64,11 @@ jobs:
run: |
# See https://github.com/Homebrew/homebrew-core/issues/157142
export HOMEBREW_NO_INSTALLED_DEPENDENTS_CHECK=1
+ cd /usr/local/bin
+ rm -f 2to3* idle3* pydoc* python3*
+ rm -f /usr/local/share/man/man1/python3.1 /usr/local/lib/pkgconfig/python3*
+ cd /usr/local/Frameworks/Python.framework
+ rm -rf Headers Python Resources Versions/Current
brew update
brew install pkg-config wireshark protobuf
- name: Build wireshark plugin