From 1f7fdb86c409c1486d160528137f10ce07dcf3b2 Mon Sep 17 00:00:00 2001 From: Yunze Xu Date: Wed, 9 Nov 2022 12:43:23 +0800 Subject: [PATCH 01/13] Fix MessageId::getDataAsString() crashed with MSVC debug config (#108) Fixes https://github.com/apache/pulsar-client-cpp/issues/107 ### Motivation The `MessageId::getDataAsString()` API returns a `std::string` to the application side. In most cases it's not an issue. However, when building Windows DLLs with `LINK_STATIC=ON`, the library will be built with `/MTd` or `/MT` option to link 3rd party dependencies statically. In this case, the DLL and the application have different C runtime libraries that allocate or deallocate memory. The returned `std::string` object is allocated inside the DLL, while it will be destroyed in the application. The destruction could crash because the application C runtime cannot find the heap address from the C runtime in DLL. ### Modifications For MSVC debug build, change the API to return a const reference to `std::string`. Then the original `std::string` object will be deallocated inside the DLL. --- include/pulsar/Message.h | 7 +++++++ lib/Message.cc | 8 ++++++++ 2 files changed, 15 insertions(+) diff --git a/include/pulsar/Message.h b/include/pulsar/Message.h index a778660a..b7b3fdd0 100644 --- a/include/pulsar/Message.h +++ b/include/pulsar/Message.h @@ -90,8 +90,15 @@ class PULSAR_PUBLIC Message { * Get string representation of the message * * @return the string representation of the message payload + * + * NOTE: For MSVC with debug mode, return a thread local std::string object to avoid memory allocation + * across DLLs and applications, which could lead to a crash. */ +#if defined(_MSC_VER) && !defined(NDEBUG) + const std::string& getDataAsString() const; +#else std::string getDataAsString() const; +#endif /** * Get key value message. diff --git a/lib/Message.cc b/lib/Message.cc index 84f203f7..46aeb473 100644 --- a/lib/Message.cc +++ b/lib/Message.cc @@ -54,7 +54,15 @@ const void* Message::getData() const { return impl_->payload.data(); } std::size_t Message::getLength() const { return impl_->payload.readableBytes(); } +#if defined(_MSC_VER) && !defined(NDEBUG) +const std::string& Message::getDataAsString() const { + thread_local std::string value; + value = std::string{static_cast(getData()), getLength()}; + return value; +} +#else std::string Message::getDataAsString() const { return std::string((const char*)getData(), getLength()); } +#endif Message::Message() : impl_() {} From 5a27ba33578b9e10e45244a92091556097e16f7c Mon Sep 17 00:00:00 2001 From: Baodi Shi Date: Thu, 10 Nov 2022 10:37:01 +0800 Subject: [PATCH 02/13] [feat] Support WaitForExclusive producer access mode. (#109) --- include/pulsar/ProducerConfiguration.h | 7 ++++- lib/ClientConnection.cc | 37 ++++++++++++++---------- lib/ClientConnection.h | 1 + lib/HandlerBase.cc | 1 + lib/HandlerBase.h | 3 +- lib/ProducerImpl.cc | 13 ++++++++- tests/ProducerTest.cc | 39 +++++++++++++++++++++++++- 7 files changed, 82 insertions(+), 19 deletions(-) diff --git a/include/pulsar/ProducerConfiguration.h b/include/pulsar/ProducerConfiguration.h index 39ecbe09..873e1383 100644 --- a/include/pulsar/ProducerConfiguration.h +++ b/include/pulsar/ProducerConfiguration.h @@ -89,7 +89,12 @@ class PULSAR_PUBLIC ProducerConfiguration { /** * Require exclusive access for producer. Fail immediately if there's already a producer connected. */ - Exclusive = 1 + Exclusive = 1, + + /** + * Producer creation is pending until it can acquire exclusive access. + */ + WaitForExclusive = 2 }; ProducerConfiguration(); diff --git a/lib/ClientConnection.cc b/lib/ClientConnection.cc index b3df8310..154f0c5d 100644 --- a/lib/ClientConnection.cc +++ b/lib/ClientConnection.cc @@ -1063,22 +1063,29 @@ void ClientConnection::handleIncomingCommand(BaseCommand& incomingCmd) { PendingRequestsMap::iterator it = pendingRequests_.find(producerSuccess.request_id()); if (it != pendingRequests_.end()) { PendingRequestData requestData = it->second; - pendingRequests_.erase(it); - lock.unlock(); - - ResponseData data; - data.producerName = producerSuccess.producer_name(); - data.lastSequenceId = producerSuccess.last_sequence_id(); - if (producerSuccess.has_schema_version()) { - data.schemaVersion = producerSuccess.schema_version(); - } - if (producerSuccess.has_topic_epoch()) { - data.topicEpoch = Optional::of(producerSuccess.topic_epoch()); + if (!producerSuccess.producer_ready()) { + LOG_INFO(cnxString_ << " Producer " << producerSuccess.producer_name() + << " has been queued up at broker. req_id: " + << producerSuccess.request_id()); + requestData.hasGotResponse->store(true); + lock.unlock(); } else { - data.topicEpoch = Optional::empty(); + pendingRequests_.erase(it); + lock.unlock(); + ResponseData data; + data.producerName = producerSuccess.producer_name(); + data.lastSequenceId = producerSuccess.last_sequence_id(); + if (producerSuccess.has_schema_version()) { + data.schemaVersion = producerSuccess.schema_version(); + } + if (producerSuccess.has_topic_epoch()) { + data.topicEpoch = Optional::of(producerSuccess.topic_epoch()); + } else { + data.topicEpoch = Optional::empty(); + } + requestData.promise.setValue(data); + requestData.timer->cancel(); } - requestData.promise.setValue(data); - requestData.timer->cancel(); } break; } @@ -1481,7 +1488,7 @@ Future ClientConnection::sendRequestWithId(SharedBuffer cm void ClientConnection::handleRequestTimeout(const boost::system::error_code& ec, PendingRequestData pendingRequestData) { - if (!ec) { + if (!ec && !pendingRequestData.hasGotResponse->load()) { pendingRequestData.promise.setFailed(ResultTimeout); } } diff --git a/lib/ClientConnection.h b/lib/ClientConnection.h index ad5c3adf..a07e2cd4 100644 --- a/lib/ClientConnection.h +++ b/lib/ClientConnection.h @@ -172,6 +172,7 @@ class PULSAR_PUBLIC ClientConnection : public std::enable_shared_from_this promise; DeadlineTimerPtr timer; + std::shared_ptr hasGotResponse{std::make_shared(false)}; }; struct LookupRequestData { diff --git a/lib/HandlerBase.cc b/lib/HandlerBase.cc index 0989eacc..1e13fb1c 100644 --- a/lib/HandlerBase.cc +++ b/lib/HandlerBase.cc @@ -130,6 +130,7 @@ void HandlerBase::handleDisconnection(Result result, ClientConnectionWeakPtr con case NotStarted: case Closing: case Closed: + case Producer_Fenced: case Failed: LOG_DEBUG(handler->getName() << "Ignoring connection closed event since the handler is not used anymore"); diff --git a/lib/HandlerBase.h b/lib/HandlerBase.h index 4a5df5c7..6faaebd8 100644 --- a/lib/HandlerBase.h +++ b/lib/HandlerBase.h @@ -118,7 +118,8 @@ class HandlerBase { Ready, Closing, Closed, - Failed + Failed, + Producer_Fenced }; std::atomic state_; diff --git a/lib/ProducerImpl.cc b/lib/ProducerImpl.cc index f3e61204..23b2d96a 100644 --- a/lib/ProducerImpl.cc +++ b/lib/ProducerImpl.cc @@ -236,7 +236,15 @@ void ProducerImpl::handleCreateProducer(const ClientConnectionPtr& cnx, Result r cnx->sendRequestWithId(Commands::newCloseProducer(producerId_, requestId), requestId); } - if (producerCreatedPromise_.isComplete()) { + if (result == ResultProducerFenced) { + state_ = Producer_Fenced; + failPendingMessages(result, true); + auto client = client_.lock(); + if (client) { + client->cleanupProducer(this); + } + producerCreatedPromise_.setFailed(result); + } else if (producerCreatedPromise_.isComplete()) { if (result == ResultProducerBlockedQuotaExceededException) { LOG_WARN(getName() << "Backlog is exceeded on topic. Sending exception to producer"); failPendingMessages(ResultProducerBlockedQuotaExceededException, true); @@ -378,6 +386,9 @@ bool ProducerImpl::isValidProducerState(const SendCallback& callback) const { case HandlerBase::Closed: callback(ResultAlreadyClosed, {}); return false; + case HandlerBase::Producer_Fenced: + callback(ResultProducerFenced, {}); + return false; case HandlerBase::NotStarted: case HandlerBase::Failed: default: diff --git a/tests/ProducerTest.cc b/tests/ProducerTest.cc index 77a79e1a..a0b1e7e2 100644 --- a/tests/ProducerTest.cc +++ b/tests/ProducerTest.cc @@ -275,7 +275,8 @@ TEST(ProducerTest, testChunkingMaxMessageSize) { TEST(ProducerTest, testExclusiveProducer) { Client client(serviceUrl); - std::string topicName = "persistent://public/default/testExclusiveProducer"; + std::string topicName = + "persistent://public/default/testExclusiveProducer" + std::to_string(time(nullptr)); Producer producer1; ProducerConfiguration producerConfiguration1; @@ -296,6 +297,42 @@ TEST(ProducerTest, testExclusiveProducer) { ASSERT_EQ(ResultProducerBusy, client.createProducer(topicName, producerConfiguration3, producer3)); } +TEST(ProducerTest, testWaitForExclusiveProducer) { + Client client(serviceUrl); + + std::string topicName = + "persistent://public/default/testWaitForExclusiveProducer" + std::to_string(time(nullptr)); + + Producer producer1; + ProducerConfiguration producerConfiguration1; + producerConfiguration1.setProducerName("p-name-1"); + producerConfiguration1.setAccessMode(ProducerConfiguration::Exclusive); + + ASSERT_EQ(ResultOk, client.createProducer(topicName, producerConfiguration1, producer1)); + + ASSERT_EQ(ResultOk, producer1.send(MessageBuilder().setContent("content").build())); + + Producer producer2; + ProducerConfiguration producerConfiguration2; + producerConfiguration2.setProducerName("p-name-2"); + producerConfiguration2.setAccessMode(ProducerConfiguration::WaitForExclusive); + + Latch latch(1); + client.createProducerAsync(topicName, producerConfiguration2, + [&latch, &producer2](Result res, Producer producer) { + ASSERT_EQ(ResultOk, res); + latch.countdown(); + producer2 = producer; + }); + + // when p1 close, p2 success created. + producer1.close(); + latch.wait(); + ASSERT_EQ(ResultOk, producer2.send(MessageBuilder().setContent("content").build())); + + producer2.close(); +} + TEST_P(ProducerTest, testFlushNoBatch) { Client client(serviceUrl); From ad79becf0814a992ad493bc625ba0a489900451f Mon Sep 17 00:00:00 2001 From: Zike Yang Date: Thu, 10 Nov 2022 20:51:40 +0800 Subject: [PATCH 03/13] [fix] Fix wrong behavior when removing the chunkedMessageCtx (#110) Fixes #104 ### Motivation Currently, the consumer ack the last message when the chunked messages exceed maxPendingChunkMessages. This is wrong behavior. This may lead to unexpected data loss. This PR also fixes serval issues related to maxPendingChunkedMessages: https://github.com/apache/pulsar-client-cpp/blob/1f7fdb86c409c1486d160528137f10ce07dcf3b2/lib/ConsumerImpl.cc#L387-L407 In the current logic, there are two `putIfAbsent` operations here, and they are confusing. If a new chunk message is received, it will be added to the chunkedMessageCache. But if the size of the cache reaches the maxPendingChunkedMessages, it will remove at least 1 ctx from the cache due to `chunkedMessageCache_.size() - maxPendingChunkedMessage_ + 1`. But the message is then put into the cache again. This can lead to unnecessary ctx buffer memory allocations. Here are some key point of this issue: image ### Modifications * Fix consumer acked the wrong message when pending chunked messages exceed maxPendingChunkMessages * Fix wrong behavior when remove the ctx from the chunkedMessageCache. --- lib/ConsumerImpl.cc | 43 ++++++++++----------- lib/ConsumerImpl.h | 1 + tests/MessageChunkingTest.cc | 73 ++++++++++++++++++++++++++++++++++++ 3 files changed, 94 insertions(+), 23 deletions(-) diff --git a/lib/ConsumerImpl.cc b/lib/ConsumerImpl.cc index 09667294..17d95a71 100644 --- a/lib/ConsumerImpl.cc +++ b/lib/ConsumerImpl.cc @@ -322,6 +322,19 @@ void ConsumerImpl::unsubscribeAsync(ResultCallback originalCallback) { } } +void ConsumerImpl::discardChunkMessages(std::string uuid, MessageId messageId, bool autoAck) { + if (autoAck) { + doAcknowledgeIndividual(messageId, [uuid, messageId](Result result) { + if (result != ResultOk) { + LOG_WARN("Failed to acknowledge discarded chunk, uuid: " << uuid + << ", messageId: " << messageId); + } + }); + } else { + trackMessage(messageId); + } +} + void ConsumerImpl::triggerCheckExpiredChunkedTimer() { checkExpiredChunkedTimer_->expires_from_now( boost::posix_time::milliseconds(expireTimeOfIncompleteChunkedMessageMs_)); @@ -347,12 +360,7 @@ void ConsumerImpl::triggerCheckExpiredChunkedTimer() { } for (const MessageId& msgId : ctx.getChunkedMessageIds()) { LOG_INFO("Removing expired chunk messages: uuid: " << uuid << ", messageId: " << msgId); - doAcknowledgeIndividual(msgId, [uuid, msgId](Result result) { - if (result != ResultOk) { - LOG_WARN("Failed to acknowledge discarded chunk, uuid: " - << uuid << ", messageId: " << msgId); - } - }); + discardChunkMessages(uuid, msgId, true); } return true; }); @@ -383,29 +391,18 @@ Optional ConsumerImpl::processMessageChunk(const SharedBuffer& pay auto it = chunkedMessageCache_.find(uuid); - if (chunkId == 0) { - if (it == chunkedMessageCache_.end()) { - it = chunkedMessageCache_.putIfAbsent( - uuid, ChunkedMessageCtx{metadata.num_chunks_from_msg(), metadata.total_chunk_msg_size()}); - } + if (chunkId == 0 && it == chunkedMessageCache_.end()) { if (maxPendingChunkedMessage_ > 0 && chunkedMessageCache_.size() >= maxPendingChunkedMessage_) { chunkedMessageCache_.removeOldestValues( chunkedMessageCache_.size() - maxPendingChunkedMessage_ + 1, - [this, messageId](const std::string& uuid, const ChunkedMessageCtx& ctx) { - if (autoAckOldestChunkedMessageOnQueueFull_) { - doAcknowledgeIndividual(messageId, [uuid, messageId](Result result) { - if (result != ResultOk) { - LOG_WARN("Failed to acknowledge discarded chunk, uuid: " - << uuid << ", messageId: " << messageId); - } - }); - } else { - trackMessage(messageId); + [this](const std::string& uuid, const ChunkedMessageCtx& ctx) { + for (const MessageId& msgId : ctx.getChunkedMessageIds()) { + discardChunkMessages(uuid, msgId, autoAckOldestChunkedMessageOnQueueFull_); } }); - it = chunkedMessageCache_.putIfAbsent( - uuid, ChunkedMessageCtx{metadata.num_chunks_from_msg(), metadata.total_chunk_msg_size()}); } + it = chunkedMessageCache_.putIfAbsent( + uuid, ChunkedMessageCtx{metadata.num_chunks_from_msg(), metadata.total_chunk_msg_size()}); } auto& chunkedMsgCtx = it->second; diff --git a/lib/ConsumerImpl.h b/lib/ConsumerImpl.h index 9ba65773..b0a24d4c 100644 --- a/lib/ConsumerImpl.h +++ b/lib/ConsumerImpl.h @@ -307,6 +307,7 @@ class ConsumerImpl : public ConsumerImplBase { std::atomic_bool expireChunkMessageTaskScheduled_{false}; void triggerCheckExpiredChunkedTimer(); + void discardChunkMessages(std::string uuid, MessageId messageId, bool autoAck); /** * Process a chunk. If the chunk is the last chunk of a message, concatenate all buffered chunks into the diff --git a/tests/MessageChunkingTest.cc b/tests/MessageChunkingTest.cc index 8675886f..5818bde7 100644 --- a/tests/MessageChunkingTest.cc +++ b/tests/MessageChunkingTest.cc @@ -180,6 +180,79 @@ TEST_P(MessageChunkingTest, testExpireIncompleteChunkMessage) { consumer.close(); } +TEST_P(MessageChunkingTest, testMaxPendingChunkMessages) { + if (toString(GetParam()) != "None") { + return; + } + const std::string topic = "MessageChunkingTest-testMaxPendingChunkMessages-" + toString(GetParam()) + + std::to_string(time(nullptr)); + Consumer consumer; + ConsumerConfiguration consumerConf; + consumerConf.setMaxPendingChunkedMessage(1); + consumerConf.setAutoAckOldestChunkedMessageOnQueueFull(true); + createConsumer(topic, consumer, consumerConf); + Producer producer; + createProducer(topic, producer); + + auto msg = MessageBuilder().setContent("chunk-0-0|").build(); + auto& metadata = PulsarFriend::getMessageMetadata(msg); + metadata.set_num_chunks_from_msg(2); + metadata.set_chunk_id(0); + metadata.set_uuid("0"); + metadata.set_total_chunk_msg_size(100); + + producer.send(msg); + + auto msg2 = MessageBuilder().setContent("chunk-1-0|").build(); + auto& metadata2 = PulsarFriend::getMessageMetadata(msg2); + metadata2.set_num_chunks_from_msg(2); + metadata2.set_uuid("1"); + metadata2.set_chunk_id(0); + metadata2.set_total_chunk_msg_size(100); + + producer.send(msg2); + + auto msg3 = MessageBuilder().setContent("chunk-1-1|").build(); + auto& metadata3 = PulsarFriend::getMessageMetadata(msg3); + metadata3.set_num_chunks_from_msg(2); + metadata3.set_uuid("1"); + metadata3.set_chunk_id(1); + metadata3.set_total_chunk_msg_size(100); + + producer.send(msg3); + + Message receivedMsg; + ASSERT_EQ(ResultOk, consumer.receive(receivedMsg, 3000)); + ASSERT_EQ(receivedMsg.getDataAsString(), "chunk-1-0|chunk-1-1|"); + + consumer.redeliverUnacknowledgedMessages(); + + // The consumer may acknowledge the wrong message(the latest message) in the old version of codes. This + // test case ensure that it should not happen again. + Message receivedMsg2; + ASSERT_EQ(ResultOk, consumer.receive(receivedMsg2, 3000)); + ASSERT_EQ(receivedMsg2.getDataAsString(), "chunk-1-0|chunk-1-1|"); + + consumer.acknowledge(receivedMsg2); + + consumer.redeliverUnacknowledgedMessages(); + auto msg4 = MessageBuilder().setContent("chunk-0-1|").build(); + auto& metadata4 = PulsarFriend::getMessageMetadata(msg4); + metadata4.set_num_chunks_from_msg(2); + metadata4.set_uuid("0"); + metadata4.set_chunk_id(1); + metadata4.set_total_chunk_msg_size(100); + + producer.send(msg4); + + // This ensures that the message chunk-0-0 was acknowledged successfully. So we cannot receive it anymore. + Message receivedMsg3; + consumer.receive(receivedMsg3, 3000); + + producer.close(); + consumer.close(); +} + // The CI env is Ubuntu 16.04, the gtest-dev version is 1.8.0 that doesn't have INSTANTIATE_TEST_SUITE_P INSTANTIATE_TEST_CASE_P(Pulsar, MessageChunkingTest, ::testing::Values(CompressionNone, CompressionLZ4, CompressionZLib, CompressionZSTD, From d396c905a040b5f14ebcd5eec003bd78b3b8ec50 Mon Sep 17 00:00:00 2001 From: A Date: Mon, 14 Nov 2022 10:16:58 +0700 Subject: [PATCH 04/13] [Improve][Build] Use pulsar client as a submodule for C++ projects (#115) Fixes #100 ### Motivation If your project builds through CMake and you try to add pulsar-client in some directory for build with whole project (i.e. third-party submodule) you can't do this because pulsar-client CMake uses CMAKE_*_DIR for configure, and this variable always related to top-level project. ### Modifications Change CMAKE_(SOURCES|BINARY)_DIR to PROJECT_*_DIR for library and tests --- CMakeLists.txt | 46 ++++++++++++++++++++++---------------------- tests/CMakeLists.txt | 2 +- 2 files changed, 24 insertions(+), 24 deletions(-) diff --git a/CMakeLists.txt b/CMakeLists.txt index 28be829e..bbad70fc 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -20,13 +20,13 @@ cmake_minimum_required(VERSION 3.4) project (pulsar-cpp) -set(CMAKE_MODULE_PATH ${CMAKE_MODULE_PATH} "${CMAKE_SOURCE_DIR}/cmake_modules") +set(CMAKE_MODULE_PATH ${CMAKE_MODULE_PATH} "${PROJECT_SOURCE_DIR}/cmake_modules") -execute_process(COMMAND cat ${CMAKE_SOURCE_DIR}/version.txt OUTPUT_STRIP_TRAILING_WHITESPACE +execute_process(COMMAND cat ${PROJECT_SOURCE_DIR}/version.txt OUTPUT_STRIP_TRAILING_WHITESPACE OUTPUT_VARIABLE PULSAR_CLIENT_VERSION) message(STATUS "Pulsar Client version: ${PULSAR_CLIENT_VERSION}") -execute_process(COMMAND ${CMAKE_SOURCE_DIR}/build-support/gen-pulsar-version-macro.py OUTPUT_STRIP_TRAILING_WHITESPACE +execute_process(COMMAND ${PROJECT_SOURCE_DIR}/build-support/gen-pulsar-version-macro.py OUTPUT_STRIP_TRAILING_WHITESPACE OUTPUT_VARIABLE PULSAR_CLIENT_VERSION_MACRO) message(STATUS "Pulsar Client version macro: ${PULSAR_CLIENT_VERSION_MACRO}") @@ -36,11 +36,11 @@ configure_file(templates/Version.h.in include/pulsar/Version.h @ONLY) option(LINK_STATIC "Link against static libraries" OFF) if (VCPKG_TRIPLET) message(STATUS "Use vcpkg, triplet is ${VCPKG_TRIPLET}") - set(CMAKE_PREFIX_PATH "${CMAKE_SOURCE_DIR}/vcpkg_installed/${VCPKG_TRIPLET}") + set(CMAKE_PREFIX_PATH "${PROJECT_SOURCE_DIR}/vcpkg_installed/${VCPKG_TRIPLET}") message(STATUS "Use CMAKE_PREFIX_PATH: ${CMAKE_PREFIX_PATH}") set(PROTOC_PATH "${CMAKE_PREFIX_PATH}/tools/protobuf/protoc") message(STATUS "Use protoc: ${PROTOC_PATH}") - set(VCPKG_DEBUG_ROOT "${CMAKE_SOURCE_DIR}/vcpkg_installed/${VCPKG_TRIPLET}/debug") + set(VCPKG_DEBUG_ROOT "${PROJECT_SOURCE_DIR}/vcpkg_installed/${VCPKG_TRIPLET}/debug") if (CMAKE_BUILD_TYPE STREQUAL "Debug") set(ZLIB_ROOT ${VCPKG_DEBUG_ROOT}) set(OPENSSL_ROOT_DIR ${VCPKG_DEBUG_ROOT}) @@ -311,13 +311,13 @@ MESSAGE(STATUS "HAS_SNAPPY: ${HAS_SNAPPY}") set(ADDITIONAL_LIBRARIES $ENV{PULSAR_ADDITIONAL_LIBRARIES}) link_directories( $ENV{PULSAR_ADDITIONAL_LIBRARY_PATH} ) -set(AUTOGEN_DIR ${CMAKE_BINARY_DIR}/generated) +set(AUTOGEN_DIR ${PROJECT_BINARY_DIR}/generated) file(MAKE_DIRECTORY ${AUTOGEN_DIR}) include_directories( - ${CMAKE_SOURCE_DIR} - ${CMAKE_SOURCE_DIR}/include - ${CMAKE_BINARY_DIR}/include + ${PROJECT_SOURCE_DIR} + ${PROJECT_SOURCE_DIR}/include + ${PROJECT_BINARY_DIR}/include ${AUTOGEN_DIR} ${Boost_INCLUDE_DIRS} ${OPENSSL_INCLUDE_DIR} @@ -391,7 +391,7 @@ if(NOT APPLE AND NOT MSVC) set(COMMON_LIBS ${COMMON_LIBS} rt) endif () -link_directories(${CMAKE_BINARY_DIR}/lib) +link_directories(${PROJECT_BINARY_DIR}/lib) set(LIB_NAME $ENV{PULSAR_LIBRARY_NAME}) if (NOT LIB_NAME) @@ -421,26 +421,26 @@ if (BUILD_WIRESHARK) endif() find_package(ClangTools) -set(BUILD_SUPPORT_DIR "${CMAKE_SOURCE_DIR}/build-support") +set(BUILD_SUPPORT_DIR "${PROJECT_SOURCE_DIR}/build-support") add_custom_target(format ${BUILD_SUPPORT_DIR}/run_clang_format.py ${CLANG_FORMAT_BIN} 0 ${BUILD_SUPPORT_DIR}/clang_format_exclusions.txt - ${CMAKE_SOURCE_DIR}/lib - ${CMAKE_SOURCE_DIR}/perf - ${CMAKE_SOURCE_DIR}/examples - ${CMAKE_SOURCE_DIR}/tests - ${CMAKE_SOURCE_DIR}/include - ${CMAKE_SOURCE_DIR}/wireshark) + ${PROJECT_SOURCE_DIR}/lib + ${PROJECT_SOURCE_DIR}/perf + ${PROJECT_SOURCE_DIR}/examples + ${PROJECT_SOURCE_DIR}/tests + ${PROJECT_SOURCE_DIR}/include + ${PROJECT_SOURCE_DIR}/wireshark) # `make check-format` option (for CI test) add_custom_target(check-format ${BUILD_SUPPORT_DIR}/run_clang_format.py ${CLANG_FORMAT_BIN} 1 ${BUILD_SUPPORT_DIR}/clang_format_exclusions.txt - ${CMAKE_SOURCE_DIR}/lib - ${CMAKE_SOURCE_DIR}/perf - ${CMAKE_SOURCE_DIR}/examples - ${CMAKE_SOURCE_DIR}/tests - ${CMAKE_SOURCE_DIR}/include - ${CMAKE_SOURCE_DIR}/wireshark) + ${PROJECT_SOURCE_DIR}/lib + ${PROJECT_SOURCE_DIR}/perf + ${PROJECT_SOURCE_DIR}/examples + ${PROJECT_SOURCE_DIR}/tests + ${PROJECT_SOURCE_DIR}/include + ${PROJECT_SOURCE_DIR}/wireshark) diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt index cb0510dc..3ce579e5 100644 --- a/tests/CMakeLists.txt +++ b/tests/CMakeLists.txt @@ -56,6 +56,6 @@ file(GLOB TEST_SOURCES *.cc c/*.cc) add_executable(pulsar-tests ${TEST_SOURCES} ${PROTO_SOURCES}) -target_include_directories(pulsar-tests PRIVATE ${CMAKE_SOURCE_DIR}/lib ${AUTOGEN_DIR}/lib) +target_include_directories(pulsar-tests PRIVATE ${PROJECT_SOURCE_DIR}/lib ${AUTOGEN_DIR}/lib) target_link_libraries(pulsar-tests ${CLIENT_LIBS} pulsarStatic $<$:${GMOCKD_LIBRARY_PATH}> $<$:${GTESTD_LIBRARY_PATH}> $<$>:${GMOCK_LIBRARY_PATH}> $<$>:${GTEST_LIBRARY_PATH}>) From 1721e0005975bcc9cbd49566d6047760e6621a3b Mon Sep 17 00:00:00 2001 From: Yunze Xu Date: Mon, 14 Nov 2022 12:00:51 +0800 Subject: [PATCH 05/13] Add MessageId::batchSize() and the MessageIdBuilder (#105) Master issue: https://github.com/apache/pulsar-client-cpp/issues/87 ### Motivation To support batch index acknowledgment, we must provide a method to get the batch size of a batched message ID. ### Modifications Instead of adding another overload constructor to `MessageId`, this PR adds a `MessageIdBuilder` class to construct the `MessageId` in a more elegant way. The original constructor is counterintuitive because the partition index is the 1st argument. https://github.com/apache/pulsar-client-cpp/blob/74ef1a01f5c7a4604d251de6d040c433f9bbf56b/include/pulsar/MessageId.h#L47 Therefore, this PR marks it as deprecated and replace all invocations of it with the `MessageIdBuilder` usages. To verify the `MessageId::batchSize()`, the following tests are modified: - `BatchMessageTest.testBatchSizeInBytes`: the batch size is always 2 because of the `batchingMaxAllowedSizeInBytes` config. - `MessageChunkingTest.testEndToEnd`: the batch size field is not set (default: 0) because batching is disabled. --- include/pulsar/MessageId.h | 8 ++ include/pulsar/MessageIdBuilder.h | 115 ++++++++++++++++++++++++++++ lib/BatchAcknowledgementTracker.cc | 11 +-- lib/ClientConnection.cc | 8 +- lib/Commands.cc | 6 +- lib/Commands.h | 3 +- lib/ConsumerImpl.cc | 25 +++--- lib/Message.cc | 5 +- lib/MessageAndCallbackBatch.cc | 5 +- lib/MessageBatch.cc | 2 +- lib/MessageId.cc | 11 ++- lib/MessageIdBuilder.cc | 74 ++++++++++++++++++ lib/MessageIdImpl.h | 13 ++-- lib/MessageIdUtil.h | 5 ++ lib/NegativeAcksTracker.cc | 4 +- lib/ProducerImpl.cc | 5 +- lib/UnAckedMessageTrackerEnabled.cc | 5 +- tests/BasicEndToEndTest.cc | 4 +- tests/BatchMessageTest.cc | 6 +- tests/ConsumerTest.cc | 4 +- tests/MessageChunkingTest.cc | 2 + tests/MessageIdTest.cc | 12 +-- tests/PulsarFriend.h | 6 -- 23 files changed, 276 insertions(+), 63 deletions(-) create mode 100644 include/pulsar/MessageIdBuilder.h create mode 100644 lib/MessageIdBuilder.cc diff --git a/include/pulsar/MessageId.h b/include/pulsar/MessageId.h index 7c9626c1..28b88c85 100644 --- a/include/pulsar/MessageId.h +++ b/include/pulsar/MessageId.h @@ -37,8 +37,12 @@ class PULSAR_PUBLIC MessageId { MessageId(); /** + * @deprecated + * * Construct the MessageId * + * NOTE: This API still exists for backward compatibility, use MessageIdBuilder instead. + * * @param partition the partition number of a topic * @param ledgerId the ledger id * @param entryId the entry id @@ -88,6 +92,7 @@ class PULSAR_PUBLIC MessageId { int64_t entryId() const; int32_t batchIndex() const; int32_t partition() const; + int32_t batchSize() const; private: friend class ConsumerImpl; @@ -102,11 +107,14 @@ class PULSAR_PUBLIC MessageId { friend class PulsarWrapper; friend class PulsarFriend; friend class NegativeAcksTracker; + friend class MessageIdBuilder; friend PULSAR_PUBLIC std::ostream& operator<<(std::ostream& s, const MessageId& messageId); typedef std::shared_ptr MessageIdImplPtr; MessageIdImplPtr impl_; + + explicit MessageId(const MessageIdImplPtr& impl); }; typedef std::vector MessageIdList; diff --git a/include/pulsar/MessageIdBuilder.h b/include/pulsar/MessageIdBuilder.h new file mode 100644 index 00000000..ce2b99ba --- /dev/null +++ b/include/pulsar/MessageIdBuilder.h @@ -0,0 +1,115 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +#pragma once + +#include + +#include + +namespace pulsar { + +namespace proto { +class MessageIdData; +} + +/** + * The builder to build a MessageId. + * + * Example of building a single MessageId: + * + * ```c++ + * MessageId msgId = MessageIdBuilder() + * .ledgerId(0L) + * .entryId(0L) + * .build(); + * ``` + * + * Example of building a batched MessageId: + * + * ```c++ + * MessageId msgId = MessageIdBuilder() + * .ledgerId(0L) + * .entryId(0L) + * .batchIndex(0) + * .batchSize(2) + * .build(); + * ``` + */ +class PULSAR_PUBLIC MessageIdBuilder { + public: + explicit MessageIdBuilder(); + + /** + * Create an instance that copies the data from messageId. + */ + static MessageIdBuilder from(const MessageId& messageId); + + /** + * Create an instance from the proto::MessageIdData instance. + * + * @note It's an internal API that converts the MessageIdData defined by PulsarApi.proto + * @see https://github.com/apache/pulsar-client-cpp/blob/main/proto/PulsarApi.proto + */ + static MessageIdBuilder from(const proto::MessageIdData& messageIdData); + + /** + * Build a MessageId. + */ + MessageId build() const; + + /** + * Set the ledger ID field. + * + * Default: -1L + */ + MessageIdBuilder& ledgerId(int64_t ledgerId); + + /** + * Set the entry ID field. + * + * Default: -1L + */ + MessageIdBuilder& entryId(int64_t entryId); + + /** + * Set the partition index. + * + * Default: -1 + */ + MessageIdBuilder& partition(int32_t partition); + + /** + * Set the batch index. + * + * Default: -1 + */ + MessageIdBuilder& batchIndex(int32_t batchIndex); + + /** + * Set the batch size. + * + * Default: 0 + */ + MessageIdBuilder& batchSize(int32_t batchSize); + + private: + std::shared_ptr impl_; +}; + +} // namespace pulsar diff --git a/lib/BatchAcknowledgementTracker.cc b/lib/BatchAcknowledgementTracker.cc index 1df4984f..d1bb6a89 100644 --- a/lib/BatchAcknowledgementTracker.cc +++ b/lib/BatchAcknowledgementTracker.cc @@ -19,6 +19,7 @@ #include "BatchAcknowledgementTracker.h" #include "LogUtils.h" +#include "MessageIdUtil.h" #include "MessageImpl.h" namespace pulsar { @@ -71,8 +72,7 @@ void BatchAcknowledgementTracker::deleteAckedMessage(const MessageId& messageId, return; } - MessageId batchMessageId = - MessageId(messageId.partition(), messageId.ledgerId(), messageId.entryId(), -1 /* Batch index */); + auto batchMessageId = discardBatch(messageId); Lock lock(mutex_); if (ackType == CommandAck_AckType_Cumulative) { @@ -114,9 +114,7 @@ void BatchAcknowledgementTracker::deleteAckedMessage(const MessageId& messageId, bool BatchAcknowledgementTracker::isBatchReady(const MessageId& msgID, CommandAck_AckType ackType) { Lock lock(mutex_); - // Remove batch index - MessageId batchMessageId = - MessageId(msgID.partition(), msgID.ledgerId(), msgID.entryId(), -1 /* Batch index */); + auto batchMessageId = discardBatch(msgID); TrackerMap::iterator pos = trackerMap_.find(batchMessageId); if (pos == trackerMap_.end() || @@ -154,8 +152,7 @@ const MessageId BatchAcknowledgementTracker::getGreatestCumulativeAckReady(const Lock lock(mutex_); // Remove batch index - MessageId batchMessageId = - MessageId(messageId.partition(), messageId.ledgerId(), messageId.entryId(), -1 /* Batch index */); + auto batchMessageId = discardBatch(messageId); TrackerMap::iterator pos = trackerMap_.find(batchMessageId); // element not found diff --git a/lib/ClientConnection.cc b/lib/ClientConnection.cc index 154f0c5d..48b88b58 100644 --- a/lib/ClientConnection.cc +++ b/lib/ClientConnection.cc @@ -18,6 +18,8 @@ */ #include "ClientConnection.h" +#include + #include #include "Commands.h" @@ -43,8 +45,7 @@ static const uint32_t DefaultBufferSize = 64 * 1024; static const int KeepAliveIntervalInSeconds = 30; static MessageId toMessageId(const proto::MessageIdData& messageIdData) { - return MessageId{messageIdData.partition(), static_cast(messageIdData.ledgerid()), - static_cast(messageIdData.entryid()), messageIdData.batch_index()}; + return MessageIdBuilder::from(messageIdData).build(); } // Convert error codes from protobuf to client API Result @@ -830,8 +831,7 @@ void ClientConnection::handleIncomingCommand(BaseCommand& incomingCmd) { int producerId = sendReceipt.producer_id(); uint64_t sequenceId = sendReceipt.sequence_id(); const proto::MessageIdData& messageIdData = sendReceipt.message_id(); - MessageId messageId = MessageId(messageIdData.partition(), messageIdData.ledgerid(), - messageIdData.entryid(), messageIdData.batch_index()); + auto messageId = toMessageId(messageIdData); LOG_DEBUG(cnxString_ << "Got receipt for producer: " << producerId << " -- msg: " << sequenceId << "-- message id: " << messageId); diff --git a/lib/Commands.cc b/lib/Commands.cc index 69492c6f..f97b0eb8 100644 --- a/lib/Commands.cc +++ b/lib/Commands.cc @@ -19,6 +19,7 @@ #include "Commands.h" #include +#include #include #include @@ -807,7 +808,8 @@ uint64_t Commands::serializeSingleMessageInBatchWithPayload(const Message& msg, return msgMetadata.sequence_id(); } -Message Commands::deSerializeSingleMessageInBatch(Message& batchedMessage, int32_t batchIndex) { +Message Commands::deSerializeSingleMessageInBatch(Message& batchedMessage, int32_t batchIndex, + int32_t batchSize) { SharedBuffer& uncompressedPayload = batchedMessage.impl_->payload; // Format of batch message @@ -825,7 +827,7 @@ Message Commands::deSerializeSingleMessageInBatch(Message& batchedMessage, int32 uncompressedPayload.consume(payloadSize); const MessageId& m = batchedMessage.impl_->messageId; - MessageId singleMessageId(m.partition(), m.ledgerId(), m.entryId(), batchIndex); + auto singleMessageId = MessageIdBuilder::from(m).batchIndex(batchIndex).batchSize(batchSize).build(); Message singleMessage(singleMessageId, batchedMessage.impl_->metadata, payload, metadata, batchedMessage.impl_->getTopicName()); singleMessage.impl_->cnx_ = batchedMessage.impl_->cnx_; diff --git a/lib/Commands.h b/lib/Commands.h index 09f6f8be..6681f138 100644 --- a/lib/Commands.h +++ b/lib/Commands.h @@ -132,7 +132,8 @@ class Commands { static PULSAR_PUBLIC uint64_t serializeSingleMessageInBatchWithPayload( const Message& msg, SharedBuffer& batchPayLoad, unsigned long maxMessageSizeInBytes); - static Message deSerializeSingleMessageInBatch(Message& batchedMessage, int32_t batchIndex); + static Message deSerializeSingleMessageInBatch(Message& batchedMessage, int32_t batchIndex, + int32_t batchSize); static SharedBuffer newConsumerStats(uint64_t consumerId, uint64_t requestId); diff --git a/lib/ConsumerImpl.cc b/lib/ConsumerImpl.cc index 17d95a71..3f62a7a8 100644 --- a/lib/ConsumerImpl.cc +++ b/lib/ConsumerImpl.cc @@ -18,6 +18,8 @@ */ #include "ConsumerImpl.h" +#include + #include #include "AckGroupingTracker.h" @@ -471,8 +473,7 @@ void ConsumerImpl::messageReceived(const ClientConnectionPtr& cnx, const proto:: // Only a non-batched messages can be a chunk if (!metadata.has_num_messages_in_batch() && isChunkedMessage) { const auto& messageIdData = msg.message_id(); - MessageId messageId(messageIdData.partition(), messageIdData.ledgerid(), messageIdData.entryid(), - messageIdData.batch_index()); + auto messageId = MessageIdBuilder::from(messageIdData).build(); auto optionalPayload = processMessageChunk(payload, metadata, messageId, messageIdData, cnx); if (optionalPayload.is_present()) { payload = optionalPayload.value(); @@ -629,7 +630,7 @@ uint32_t ConsumerImpl::receiveIndividualMessagesFromBatch(const ClientConnection for (int i = 0; i < batchSize; i++) { // This is a cheap copy since message contains only one shared pointer (impl_) - Message msg = Commands::deSerializeSingleMessageInBatch(batchedMessage, i); + Message msg = Commands::deSerializeSingleMessageInBatch(batchedMessage, i, batchSize); msg.impl_->setRedeliveryCount(redeliveryCount); msg.impl_->setTopicName(batchedMessage.getTopicName()); msg.impl_->convertPayloadToKeyValue(config_.getSchema()); @@ -929,13 +930,17 @@ Optional ConsumerImpl::clearReceiveQueue() { if (incomingMessages_.peekAndClear(nextMessageInQueue)) { // There was at least one message pending in the queue const MessageId& nextMessageId = nextMessageInQueue.getMessageId(); - MessageId previousMessageId; - if (nextMessageId.batchIndex() >= 0) { - previousMessageId = MessageId(-1, nextMessageId.ledgerId(), nextMessageId.entryId(), - nextMessageId.batchIndex() - 1); - } else { - previousMessageId = MessageId(-1, nextMessageId.ledgerId(), nextMessageId.entryId() - 1, -1); - } + auto previousMessageId = (nextMessageId.batchIndex() >= 0) + ? MessageIdBuilder() + .ledgerId(nextMessageId.ledgerId()) + .entryId(nextMessageId.entryId()) + .batchIndex(nextMessageId.batchIndex() - 1) + .batchSize(nextMessageId.batchSize()) + .build() + : MessageIdBuilder() + .ledgerId(nextMessageId.ledgerId()) + .entryId(nextMessageId.entryId() - 1) + .build(); return Optional::of(previousMessageId); } else if (lastDequedMessageId_ != MessageId::earliest()) { // If the queue was empty we need to restart from the message just after the last one that has been diff --git a/lib/Message.cc b/lib/Message.cc index 46aeb473..0f28f7d9 100644 --- a/lib/Message.cc +++ b/lib/Message.cc @@ -18,6 +18,7 @@ */ #include #include +#include #include #include @@ -71,9 +72,7 @@ Message::Message(MessageImplPtr& impl) : impl_(impl) {} Message::Message(const proto::CommandMessage& msg, proto::MessageMetadata& metadata, SharedBuffer& payload, int32_t partition) : impl_(std::make_shared()) { - impl_->messageId = - MessageId(partition, msg.message_id().ledgerid(), msg.message_id().entryid(), /* batchId */ - -1); + impl_->messageId = MessageIdBuilder::from(msg.message_id()).batchIndex(-1).build(); impl_->metadata = metadata; impl_->payload = payload; } diff --git a/lib/MessageAndCallbackBatch.cc b/lib/MessageAndCallbackBatch.cc index 3f50dc02..56725389 100644 --- a/lib/MessageAndCallbackBatch.cc +++ b/lib/MessageAndCallbackBatch.cc @@ -18,6 +18,8 @@ */ #include "MessageAndCallbackBatch.h" +#include + #include "ClientConnection.h" #include "Commands.h" #include "LogUtils.h" @@ -54,8 +56,7 @@ static void completeSendCallbacks(const std::vector& callbacks, Re int32_t numOfMessages = static_cast(callbacks.size()); LOG_DEBUG("Batch complete [Result = " << result << "] [numOfMessages = " << numOfMessages << "]"); for (int32_t i = 0; i < numOfMessages; i++) { - MessageId idInBatch(id.partition(), id.ledgerId(), id.entryId(), i); - callbacks[i](result, idInBatch); + callbacks[i](result, MessageIdBuilder::from(id).batchIndex(i).batchSize(numOfMessages).build()); } } diff --git a/lib/MessageBatch.cc b/lib/MessageBatch.cc index 12144ff5..f61b56ad 100644 --- a/lib/MessageBatch.cc +++ b/lib/MessageBatch.cc @@ -47,7 +47,7 @@ MessageBatch& MessageBatch::parseFrom(const SharedBuffer& payload, uint32_t batc batch_.clear(); for (int i = 0; i < batchSize; ++i) { - batch_.push_back(Commands::deSerializeSingleMessageInBatch(batchMessage_, i)); + batch_.push_back(Commands::deSerializeSingleMessageInBatch(batchMessage_, i, batchSize)); } return *this; } diff --git a/lib/MessageId.cc b/lib/MessageId.cc index 5b133282..9a1a38c8 100644 --- a/lib/MessageId.cc +++ b/lib/MessageId.cc @@ -18,6 +18,7 @@ */ #include +#include #include #include @@ -42,14 +43,16 @@ MessageId& MessageId::operator=(const MessageId& m) { MessageId::MessageId(int32_t partition, int64_t ledgerId, int64_t entryId, int32_t batchIndex) : impl_(std::make_shared(partition, ledgerId, entryId, batchIndex)) {} +MessageId::MessageId(const MessageIdImplPtr& impl) : impl_(impl) {} + const MessageId& MessageId::earliest() { - static const MessageId _earliest(-1, -1, -1, -1); + static const auto _earliest = MessageIdBuilder().build(); return _earliest; } const MessageId& MessageId::latest() { static const int64_t long_max = std::numeric_limits::max(); - static const MessageId _latest(-1, long_max, long_max, -1); + static const auto _latest = MessageIdBuilder().ledgerId(long_max).entryId(long_max).build(); return _latest; } @@ -77,7 +80,7 @@ MessageId MessageId::deserialize(const std::string& serializedMessageId) { throw std::invalid_argument("Failed to parse serialized message id"); } - return MessageId(idData.partition(), idData.ledgerid(), idData.entryid(), idData.batch_index()); + return MessageIdBuilder::from(idData).build(); } int64_t MessageId::ledgerId() const { return impl_->ledgerId_; } @@ -88,6 +91,8 @@ int32_t MessageId::batchIndex() const { return impl_->batchIndex_; } int32_t MessageId::partition() const { return impl_->partition_; } +int32_t MessageId::batchSize() const { return impl_->batchSize_; } + PULSAR_PUBLIC std::ostream& operator<<(std::ostream& s, const pulsar::MessageId& messageId) { s << '(' << messageId.impl_->ledgerId_ << ',' << messageId.impl_->entryId_ << ',' << messageId.impl_->partition_ << ',' << messageId.impl_->batchIndex_ << ')'; diff --git a/lib/MessageIdBuilder.cc b/lib/MessageIdBuilder.cc new file mode 100644 index 00000000..8857daf5 --- /dev/null +++ b/lib/MessageIdBuilder.cc @@ -0,0 +1,74 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +#include +#include + +#include "MessageIdImpl.h" +#include "PulsarApi.pb.h" + +namespace pulsar { + +MessageIdBuilder::MessageIdBuilder() : impl_(std::make_shared()) {} + +MessageIdBuilder MessageIdBuilder::from(const MessageId& messageId) { + MessageIdBuilder builder; + *builder.impl_ = *(messageId.impl_); + return builder; +} + +MessageIdBuilder MessageIdBuilder::from(const proto::MessageIdData& messageIdData) { + return MessageIdBuilder() + .ledgerId(messageIdData.ledgerid()) + .entryId(messageIdData.entryid()) + .partition(messageIdData.partition()) + .batchIndex(messageIdData.batch_index()) + .batchSize(messageIdData.batch_size()); +} + +MessageId MessageIdBuilder::build() const { + assert(impl_->batchIndex_ < 0 || (impl_->batchSize_ > impl_->batchIndex_)); + return MessageId{impl_}; +} + +MessageIdBuilder& MessageIdBuilder::ledgerId(int64_t ledgerId) { + impl_->ledgerId_ = ledgerId; + return *this; +} + +MessageIdBuilder& MessageIdBuilder::entryId(int64_t entryId) { + impl_->entryId_ = entryId; + return *this; +} + +MessageIdBuilder& MessageIdBuilder::partition(int32_t partition) { + impl_->partition_ = partition; + return *this; +} + +MessageIdBuilder& MessageIdBuilder::batchIndex(int32_t batchIndex) { + impl_->batchIndex_ = batchIndex; + return *this; +} + +MessageIdBuilder& MessageIdBuilder::batchSize(int32_t batchSize) { + impl_->batchSize_ = batchSize; + return *this; +} + +} // namespace pulsar diff --git a/lib/MessageIdImpl.h b/lib/MessageIdImpl.h index 9db758c5..57d1c4eb 100644 --- a/lib/MessageIdImpl.h +++ b/lib/MessageIdImpl.h @@ -26,23 +26,24 @@ namespace pulsar { class MessageIdImpl { public: - MessageIdImpl() : ledgerId_(-1), entryId_(-1), partition_(-1), batchIndex_(-1), topicName_() {} + MessageIdImpl() = default; MessageIdImpl(int32_t partition, int64_t ledgerId, int64_t entryId, int32_t batchIndex) : ledgerId_(ledgerId), entryId_(entryId), partition_(partition), batchIndex_(batchIndex), topicName_() {} - const int64_t ledgerId_; - const int64_t entryId_; - const int32_t partition_; - const int32_t batchIndex_; + int64_t ledgerId_ = -1; + int64_t entryId_ = -1; + int32_t partition_ = -1; + int32_t batchIndex_ = -1; + int32_t batchSize_ = 0; const std::string& getTopicName() { return *topicName_; } void setTopicName(const std::string& topicName) { topicName_ = &topicName; } private: - const std::string* topicName_; + const std::string* topicName_ = nullptr; friend class MessageImpl; friend class MultiTopicsConsumerImpl; friend class UnAckedMessageTrackerEnabled; diff --git a/lib/MessageIdUtil.h b/lib/MessageIdUtil.h index 1f4ffd36..70af7fe0 100644 --- a/lib/MessageIdUtil.h +++ b/lib/MessageIdUtil.h @@ -17,6 +17,7 @@ * under the License. */ #include +#include namespace pulsar { @@ -35,4 +36,8 @@ inline int compareLedgerAndEntryId(const MessageId& lhs, const MessageId& rhs) { return internal::compare(lhs.entryId(), rhs.entryId()); } +inline MessageId discardBatch(const MessageId& messageId) { + return MessageIdBuilder::from(messageId).batchIndex(-1).batchSize(0).build(); +} + } // namespace pulsar diff --git a/lib/NegativeAcksTracker.cc b/lib/NegativeAcksTracker.cc index 3ccf0bea..6ff322df 100644 --- a/lib/NegativeAcksTracker.cc +++ b/lib/NegativeAcksTracker.cc @@ -26,6 +26,7 @@ #include "ConsumerImpl.h" #include "ExecutorService.h" #include "LogUtils.h" +#include "MessageIdUtil.h" DECLARE_LOG_OBJECT() namespace pulsar { @@ -90,8 +91,7 @@ void NegativeAcksTracker::add(const MessageId &m) { auto now = Clock::now(); // Erase batch id to group all nacks from same batch - MessageId batchMessageId = MessageId(m.partition(), m.ledgerId(), m.entryId(), -1); - nackedMessages_[batchMessageId] = now + nackDelay_; + nackedMessages_[discardBatch(m)] = now + nackDelay_; if (!timer_) { scheduleTimer(); diff --git a/lib/ProducerImpl.cc b/lib/ProducerImpl.cc index 23b2d96a..7fa3ff29 100644 --- a/lib/ProducerImpl.cc +++ b/lib/ProducerImpl.cc @@ -18,6 +18,8 @@ */ #include "ProducerImpl.h" +#include + #include #include "BatchMessageContainer.h" @@ -824,8 +826,7 @@ bool ProducerImpl::removeCorruptMessage(uint64_t sequenceId) { } bool ProducerImpl::ackReceived(uint64_t sequenceId, MessageId& rawMessageId) { - MessageId messageId(partition_, rawMessageId.ledgerId(), rawMessageId.entryId(), - rawMessageId.batchIndex()); + auto messageId = MessageIdBuilder::from(rawMessageId).partition(partition_).build(); Lock lock(mutex_); if (pendingMessagesQueue_.empty()) { diff --git a/lib/UnAckedMessageTrackerEnabled.cc b/lib/UnAckedMessageTrackerEnabled.cc index 0579777e..ff1b928f 100644 --- a/lib/UnAckedMessageTrackerEnabled.cc +++ b/lib/UnAckedMessageTrackerEnabled.cc @@ -24,6 +24,7 @@ #include "ConsumerImplBase.h" #include "ExecutorService.h" #include "LogUtils.h" +#include "MessageIdUtil.h" DECLARE_LOG_OBJECT(); @@ -96,7 +97,7 @@ UnAckedMessageTrackerEnabled::UnAckedMessageTrackerEnabled(long timeoutMs, long bool UnAckedMessageTrackerEnabled::add(const MessageId& msgId) { std::lock_guard acquire(lock_); - MessageId id(msgId.partition(), msgId.ledgerId(), msgId.entryId(), -1); + auto id = discardBatch(msgId); if (messageIdPartitionMap.count(id) == 0) { std::set& partition = timePartitions.back(); bool emplace = messageIdPartitionMap.emplace(id, partition).second; @@ -113,7 +114,7 @@ bool UnAckedMessageTrackerEnabled::isEmpty() { bool UnAckedMessageTrackerEnabled::remove(const MessageId& msgId) { std::lock_guard acquire(lock_); - MessageId id(msgId.partition(), msgId.ledgerId(), msgId.entryId(), -1); + auto id = discardBatch(msgId); bool removed = false; std::map&>::iterator exist = messageIdPartitionMap.find(id); diff --git a/tests/BasicEndToEndTest.cc b/tests/BasicEndToEndTest.cc index aee679e4..4b181d0f 100644 --- a/tests/BasicEndToEndTest.cc +++ b/tests/BasicEndToEndTest.cc @@ -230,10 +230,10 @@ TEST(BasicEndToEndTest, testProduceConsume) { // Send synchronously std::string content = "msg-1-content"; Message msg = MessageBuilder().setContent(content).build(); - ASSERT_EQ(MessageId(-1, -1, -1, -1), msg.getMessageId()); + ASSERT_EQ(MessageId::earliest(), msg.getMessageId()); result = producer.send(msg); ASSERT_EQ(ResultOk, result); - ASSERT_NE(MessageId(-1, -1, -1, -1), msg.getMessageId()); + ASSERT_NE(MessageId::earliest(), msg.getMessageId()); Message receivedMsg; consumer.receive(receivedMsg); diff --git a/tests/BatchMessageTest.cc b/tests/BatchMessageTest.cc index 273146ca..e46cb457 100644 --- a/tests/BatchMessageTest.cc +++ b/tests/BatchMessageTest.cc @@ -19,6 +19,7 @@ #include #include #include +#include #include #include @@ -237,7 +238,8 @@ TEST(BatchMessageTest, testBatchSizeInBytes) { std::string expectedMessageContent = prefix + std::to_string(i); LOG_DEBUG("Received Message with [ content - " << receivedMsg.getDataAsString() << "] [ messageID = " << receivedMsg.getMessageId() << "]"); - ASSERT_LT(pulsar::PulsarFriend::getBatchIndex(receivedMsg.getMessageId()), 2); + ASSERT_LT(receivedMsg.getMessageId().batchIndex(), 2); + ASSERT_EQ(receivedMsg.getMessageId().batchSize(), 2); ASSERT_EQ(receivedMsg.getProperty("msgIndex"), std::to_string(i++)); ASSERT_EQ(expectedMessageContent, receivedMsg.getDataAsString()); ASSERT_EQ(ResultOk, consumer.acknowledge(receivedMsg)); @@ -970,7 +972,7 @@ TEST(BatchMessageTest, testPraseMessageBatchEntry) { } MessageBatch messageBatch; - MessageId fakeId(0, 5000, 10, -1); + auto fakeId = MessageIdBuilder().ledgerId(5000L).entryId(10L).partition(0).build(); messageBatch.withMessageId(fakeId).parseFrom(payload, static_cast(cases.size())); const std::vector& messages = messageBatch.messages(); diff --git a/tests/ConsumerTest.cc b/tests/ConsumerTest.cc index a77636ee..a5ef32fe 100644 --- a/tests/ConsumerTest.cc +++ b/tests/ConsumerTest.cc @@ -31,6 +31,7 @@ #include "lib/ClientConnection.h" #include "lib/Future.h" #include "lib/LogUtils.h" +#include "lib/MessageIdUtil.h" #include "lib/MultiTopicsConsumerImpl.h" #include "lib/TimeUtils.h" #include "lib/UnAckedMessageTrackerDisabled.h" @@ -683,8 +684,7 @@ TEST(ConsumerTest, testBatchUnAckedMessageTracker) { Message msg; ASSERT_EQ(ResultOk, consumer.receive(msg, 1000)); MessageId msgId = msg.getMessageId(); - MessageId id(msgId.partition(), msgId.ledgerId(), msgId.entryId(), -1); - msgIdInBatchMap[id].emplace_back(msgId); + msgIdInBatchMap[discardBatch(msgId)].emplace_back(msgId); } ASSERT_EQ(batchCount, msgIdInBatchMap.size()); diff --git a/tests/MessageChunkingTest.cc b/tests/MessageChunkingTest.cc index 5818bde7..e3f01786 100644 --- a/tests/MessageChunkingTest.cc +++ b/tests/MessageChunkingTest.cc @@ -126,6 +126,8 @@ TEST_P(MessageChunkingTest, testEndToEnd) { ASSERT_EQ(ResultOk, consumer.receive(msg, 3000)); LOG_INFO("Receive " << msg.getLength() << " bytes from " << msg.getMessageId()); ASSERT_EQ(msg.getDataAsString(), largeMessage); + ASSERT_EQ(msg.getMessageId().batchIndex(), -1); + ASSERT_EQ(msg.getMessageId().batchSize(), 0); receivedMessageIds.emplace_back(msg.getMessageId()); } ASSERT_EQ(receivedMessageIds, sendMessageIds); diff --git a/tests/MessageIdTest.cc b/tests/MessageIdTest.cc index 55257d92..e653aa1c 100644 --- a/tests/MessageIdTest.cc +++ b/tests/MessageIdTest.cc @@ -17,7 +17,7 @@ * under the License. */ #include -#include +#include #include @@ -27,7 +27,7 @@ using namespace pulsar; TEST(MessageIdTest, testSerialization) { - MessageId msgId = PulsarFriend::getMessageId(-1, 1, 2, 3); + auto msgId = MessageIdBuilder().ledgerId(1L).entryId(2L).batchIndex(3L).build(); std::string serialized; msgId.serialize(serialized); @@ -38,10 +38,10 @@ TEST(MessageIdTest, testSerialization) { } TEST(MessageIdTest, testCompareLedgerAndEntryId) { - MessageId id1(-1, 2L, 1L, 0); - MessageId id2(-1, 2L, 1L, 1); - MessageId id3(-1, 2L, 2L, 0); - MessageId id4(-1, 3L, 0L, 0); + auto id1 = MessageIdBuilder().ledgerId(2L).entryId(1L).batchIndex(0).build(); + auto id2 = MessageIdBuilder::from(id1).batchIndex(1).build(); + auto id3 = MessageIdBuilder().ledgerId(2L).entryId(2L).batchIndex(0).build(); + auto id4 = MessageIdBuilder().ledgerId(3L).entryId(0L).batchIndex(0).build(); ASSERT_EQ(compareLedgerAndEntryId(id1, id2), 0); ASSERT_EQ(compareLedgerAndEntryId(id1, id2), 0); diff --git a/tests/PulsarFriend.h b/tests/PulsarFriend.h index 3272bced..878d80cb 100644 --- a/tests/PulsarFriend.h +++ b/tests/PulsarFriend.h @@ -39,12 +39,6 @@ using std::string; namespace pulsar { class PulsarFriend { public: - static MessageId getMessageId(int32_t partition, int64_t ledgerId, int64_t entryId, int32_t batchIndex) { - return MessageId(partition, ledgerId, entryId, batchIndex); - } - - static int getBatchIndex(const MessageId& mId) { return mId.batchIndex(); } - static ProducerStatsImplPtr getProducerStatsPtr(Producer producer) { ProducerImpl* producerImpl = static_cast(producer.impl_.get()); return std::static_pointer_cast(producerImpl->producerStatsBasePtr_); From ac3033d21cb8833225743ecc97b67ff4cf339489 Mon Sep 17 00:00:00 2001 From: tison Date: Tue, 15 Nov 2022 14:09:44 +0800 Subject: [PATCH 06/13] Doxygen supports darkmode toggle (#119) --- Doxyfile | 17 +++++++++++++++++ 1 file changed, 17 insertions(+) diff --git a/Doxyfile b/Doxyfile index 145948c3..6c8c36b6 100644 --- a/Doxyfile +++ b/Doxyfile @@ -1204,6 +1204,23 @@ HTML_EXTRA_STYLESHEET = HTML_EXTRA_FILES = +# The HTML_COLORSTYLE tag can be used to specify if the generated HTML output +# should be rendered with a dark or light theme. Default setting AUTO_LIGHT +# enables light output unless the user preference is dark output. Other options +# are DARK to always use dark mode, LIGHT to always use light mode, AUTO_DARK to +# default to dark mode unless the user prefers light mode, and TOGGLE to let the +# user toggle between dark and light mode via a button. +# Possible values are: LIGHT Always generate light output., DARK Always generate +# dark output., AUTO_LIGHT Automatically set the mode according to the user +# preference, use light mode if no preference is set (the default)., AUTO_DARK +# Automatically set the mode according to the user preference, use dark mode if +# no preference is set. and TOGGLE Allow to user to switch between light and +# dark mode via a button.. +# The default value is: AUTO_LIGHT. +# This tag requires that the tag GENERATE_HTML is set to YES. + +HTML_COLORSTYLE = TOGGLE + # The HTML_COLORSTYLE_HUE tag controls the color of the HTML output. Doxygen # will adjust the colors in the style sheet and background images according to # this color. Hue is specified as an angle on a colorwheel, see From 1a0ce69d2d2a5521f484b207ad575363aac6a10b Mon Sep 17 00:00:00 2001 From: Zike Yang Date: Wed, 16 Nov 2022 19:59:25 +0800 Subject: [PATCH 07/13] [feat] Support consumer seek by timestamp and reader seek for C Api (#118) Fixes #82 ### Motivation #82 ### Modifications * Support seek by timestamp for the consumer * Support seek by messageid and timestamp for the reader --- include/pulsar/c/consumer.h | 40 ++++++ include/pulsar/c/reader.h | 44 ++++++ lib/c/c_Consumer.cc | 10 ++ lib/c/c_Reader.cc | 20 +++ tests/c/c_SeekTest.cc | 265 ++++++++++++++++++++++++++++++++++++ 5 files changed, 379 insertions(+) create mode 100644 tests/c/c_SeekTest.cc diff --git a/include/pulsar/c/consumer.h b/include/pulsar/c/consumer.h index 99ff8c80..bb8ae37f 100644 --- a/include/pulsar/c/consumer.h +++ b/include/pulsar/c/consumer.h @@ -241,11 +241,51 @@ PULSAR_PUBLIC pulsar_result resume_message_listener(pulsar_consumer_t *consumer) */ PULSAR_PUBLIC void pulsar_consumer_redeliver_unacknowledged_messages(pulsar_consumer_t *consumer); +/** + * Reset the subscription associated with this consumer to a specific message id. + * + * @param consumer The consumer + * @param messageId The message id can either be a specific message or represent the first or last messages in + * the topic. + * @param callback The callback for this async operation + * @param ctx The context for the callback + */ PULSAR_PUBLIC void pulsar_consumer_seek_async(pulsar_consumer_t *consumer, pulsar_message_id_t *messageId, pulsar_result_callback callback, void *ctx); +/** + * Reset the subscription asynchronously associated with this consumer to a specific message id. + * + * @param consumer The consumer + * @param messageId The message id can either be a specific message or represent the first or last messages in + * the topic. + * @return Operation result + */ PULSAR_PUBLIC pulsar_result pulsar_consumer_seek(pulsar_consumer_t *consumer, pulsar_message_id_t *messageId); +/** + * Reset the subscription associated with this consumer to a specific message publish time. + * + * @param consumer The consumer + * @param timestamp The message publish time where to reposition the subscription. The timestamp format should + * be Unix time in milliseconds. + * @param callback The callback for this async operation + * @param ctx The context for the callback + */ +PULSAR_PUBLIC void pulsar_consumer_seek_by_timestamp_async(pulsar_consumer_t *consumer, uint64_t timestamp, + pulsar_result_callback callback, void *ctx); + +/** + * Reset the subscription asynchronously associated with this consumer to a specific message publish time. + * + * @param consumer The consumer + * @param timestamp The message publish time where to reposition the subscription. The timestamp format should + * be Unix time in milliseconds. + * @return Operation result + */ +PULSAR_PUBLIC pulsar_result pulsar_consumer_seek_by_timestamp(pulsar_consumer_t *consumer, + uint64_t timestamp); + PULSAR_PUBLIC int pulsar_consumer_is_connected(pulsar_consumer_t *consumer); PULSAR_PUBLIC pulsar_result pulsar_consumer_get_last_message_id(pulsar_consumer_t *consumer, diff --git a/include/pulsar/c/reader.h b/include/pulsar/c/reader.h index 4c546f80..12321fd2 100644 --- a/include/pulsar/c/reader.h +++ b/include/pulsar/c/reader.h @@ -59,6 +59,50 @@ PULSAR_PUBLIC pulsar_result pulsar_reader_read_next(pulsar_reader_t *reader, pul PULSAR_PUBLIC pulsar_result pulsar_reader_read_next_with_timeout(pulsar_reader_t *reader, pulsar_message_t **msg, int timeoutMs); +/** + * Reset the subscription associated with this reader to a specific message id. + * + * @param reader The reader + * @param messageId The message id can either be a specific message or represent the first or last messages in + * the topic. + * @param callback The callback for this async operation + * @param ctx The context for the callback + */ +PULSAR_PUBLIC void pulsar_reader_seek_async(pulsar_reader_t *reader, pulsar_message_id_t *messageId, + pulsar_result_callback callback, void *ctx); + +/** + * Reset the subscription asynchronously associated with this reader to a specific message id. + * + * @param reader The reader + * @param messageId The message id can either be a specific message or represent the first or last messages in + * the topic. + * @return Operation result + */ +PULSAR_PUBLIC pulsar_result pulsar_reader_seek(pulsar_reader_t *reader, pulsar_message_id_t *messageId); + +/** + * Reset the subscription associated with this reader to a specific message publish time. + * + * @param reader The reader + * @param timestamp The message publish time where to reposition the subscription. The timestamp format should + * be Unix time in milliseconds. + * @param callback The callback for this async operation + * @param ctx The context for the callback + */ +PULSAR_PUBLIC void pulsar_reader_seek_by_timestamp_async(pulsar_reader_t *reader, uint64_t timestamp, + pulsar_result_callback callback, void *ctx); + +/** + * Reset the subscription asynchronously associated with this reader to a specific message publish time. + * + * @param reader The reader + * @param timestamp The message publish time where to reposition the subscription. The timestamp format should + * be Unix time in milliseconds. + * @return Operation result + */ +PULSAR_PUBLIC pulsar_result pulsar_reader_seek_by_timestamp(pulsar_reader_t *reader, uint64_t timestamp); + PULSAR_PUBLIC pulsar_result pulsar_reader_close(pulsar_reader_t *reader); PULSAR_PUBLIC void pulsar_reader_close_async(pulsar_reader_t *reader, pulsar_result_callback callback, diff --git a/lib/c/c_Consumer.cc b/lib/c/c_Consumer.cc index 062c801f..df4c9f3f 100644 --- a/lib/c/c_Consumer.cc +++ b/lib/c/c_Consumer.cc @@ -156,6 +156,16 @@ pulsar_result pulsar_consumer_seek(pulsar_consumer_t *consumer, pulsar_message_i return (pulsar_result)consumer->consumer.seek(messageId->messageId); } +void pulsar_consumer_seek_by_timestamp_async(pulsar_consumer_t *consumer, uint64_t timestamp, + pulsar_result_callback callback, void *ctx) { + consumer->consumer.seekAsync(timestamp, + std::bind(handle_result_callback, std::placeholders::_1, callback, ctx)); +} + +pulsar_result pulsar_consumer_seek_by_timestamp(pulsar_consumer_t *consumer, uint64_t timestamp) { + return (pulsar_result)consumer->consumer.seek(timestamp); +} + int pulsar_consumer_is_connected(pulsar_consumer_t *consumer) { return consumer->consumer.isConnected(); } pulsar_result pulsar_consumer_get_last_message_id(pulsar_consumer_t *consumer, diff --git a/lib/c/c_Reader.cc b/lib/c/c_Reader.cc index 3490b540..c4bdc497 100644 --- a/lib/c/c_Reader.cc +++ b/lib/c/c_Reader.cc @@ -45,6 +45,26 @@ pulsar_result pulsar_reader_read_next_with_timeout(pulsar_reader_t *reader, puls return (pulsar_result)res; } +void pulsar_reader_seek_async(pulsar_reader_t *reader, pulsar_message_id_t *messageId, + pulsar_result_callback callback, void *ctx) { + reader->reader.seekAsync(messageId->messageId, + std::bind(handle_result_callback, std::placeholders::_1, callback, ctx)); +} + +pulsar_result pulsar_reader_seek(pulsar_reader_t *reader, pulsar_message_id_t *messageId) { + return (pulsar_result)reader->reader.seek(messageId->messageId); +} + +void pulsar_reader_seek_by_timestamp_async(pulsar_reader_t *reader, uint64_t timestamp, + pulsar_result_callback callback, void *ctx) { + reader->reader.seekAsync(timestamp, + std::bind(handle_result_callback, std::placeholders::_1, callback, ctx)); +} + +pulsar_result pulsar_reader_seek_by_timestamp(pulsar_reader_t *reader, uint64_t timestamp) { + return (pulsar_result)reader->reader.seek(timestamp); +} + pulsar_result pulsar_reader_close(pulsar_reader_t *reader) { return (pulsar_result)reader->reader.close(); } void pulsar_reader_close_async(pulsar_reader_t *reader, pulsar_result_callback callback, void *ctx) { diff --git a/tests/c/c_SeekTest.cc b/tests/c/c_SeekTest.cc new file mode 100644 index 00000000..cfa8a188 --- /dev/null +++ b/tests/c/c_SeekTest.cc @@ -0,0 +1,265 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +#include +#include +#include + +#include + +struct seek_ctx { + std::promise *promise; +}; + +static void seek_callback(pulsar_result async_result, void *ctx) { + auto *seek_ctx = (struct seek_ctx *)ctx; + seek_ctx->promise->set_value(async_result); +} + +void prepare_client(pulsar_client_t **client) { + const char *lookup_url = "pulsar://localhost:6650"; + pulsar_client_configuration_t *conf = pulsar_client_configuration_create(); + *client = pulsar_client_create(lookup_url, conf); + pulsar_client_configuration_free(conf); +} + +TEST(c_SeekTest, testConsumerSeekMessageId) { + auto topic_name_str = "test-c-seek-msgid-" + std::to_string(time(nullptr)); + const char *topic_name = topic_name_str.c_str(); + + pulsar_client_t *client; + prepare_client(&client); + + pulsar_producer_configuration_t *producer_conf = pulsar_producer_configuration_create(); + pulsar_producer_t *producer; + pulsar_result result = pulsar_client_create_producer(client, topic_name, producer_conf, &producer); + ASSERT_EQ(pulsar_result_Ok, result); + + pulsar_consumer_configuration_t *consumer_conf = pulsar_consumer_configuration_create(); + pulsar_consumer_t *consumer; + result = pulsar_client_subscribe(client, topic_name, "seek-time", consumer_conf, &consumer); + ASSERT_EQ(pulsar_result_Ok, result); + + pulsar_message_t *seek_message = nullptr; + + for (int i = 0; i < 10; i++) { + char content[10]; + sprintf(content, "msg-%d", i); + pulsar_message_t *msg = pulsar_message_create(); + pulsar_message_set_content(msg, content, strlen(content)); + pulsar_producer_send(producer, msg); + if (i == 5) { + seek_message = msg; + } else { + pulsar_message_free(msg); + } + } + + pulsar_consumer_seek(consumer, pulsar_message_get_message_id(seek_message)); + + pulsar_message_t *message; + ASSERT_EQ(pulsar_result_Ok, pulsar_consumer_receive_with_timeout(consumer, &message, 1000)); + ASSERT_STREQ((const char *)pulsar_message_get_data(message), "msg-6"); + pulsar_message_free(message); + + // Test seek asynchronously + std::promise seek_promise; + std::future seek_future = seek_promise.get_future(); + struct seek_ctx seek_ctx = {&seek_promise}; + pulsar_consumer_seek_async(consumer, pulsar_message_get_message_id(seek_message), seek_callback, + &seek_ctx); + ASSERT_EQ(pulsar_result_Ok, seek_future.get()); + ASSERT_EQ(pulsar_result_Ok, pulsar_consumer_receive_with_timeout(consumer, &message, 1000)); + ASSERT_STREQ((const char *)pulsar_message_get_data(message), "msg-6"); + + if (seek_message != NULL) { + pulsar_message_free(seek_message); + } + pulsar_consumer_free(consumer); + pulsar_consumer_configuration_free(consumer_conf); + pulsar_producer_free(producer); + pulsar_producer_configuration_free(producer_conf); + pulsar_client_free(client); +} + +TEST(c_SeekTest, testConsumerSeekTime) { + auto topic_name_str = "test-c-seek-time-" + std::to_string(time(nullptr)); + const char *topic_name = topic_name_str.c_str(); + + pulsar_client_t *client; + prepare_client(&client); + + pulsar_producer_configuration_t *producer_conf = pulsar_producer_configuration_create(); + pulsar_producer_t *producer; + pulsar_result result = pulsar_client_create_producer(client, topic_name, producer_conf, &producer); + ASSERT_EQ(pulsar_result_Ok, result); + + pulsar_consumer_configuration_t *consumer_conf = pulsar_consumer_configuration_create(); + pulsar_consumer_t *consumer; + result = pulsar_client_subscribe(client, topic_name, "seek-time", consumer_conf, &consumer); + ASSERT_EQ(pulsar_result_Ok, result); + + for (int i = 0; i < 10; i++) { + char content[10]; + sprintf(content, "msg-%d", i); + pulsar_message_t *msg = pulsar_message_create(); + pulsar_message_set_content(msg, content, strlen(content)); + pulsar_producer_send(producer, msg); + pulsar_message_free(msg); + } + + uint64_t currentTime = pulsar::TimeUtils::currentTimeMillis(); + + pulsar_consumer_seek_by_timestamp(consumer, currentTime); + + pulsar_message_t *message; + ASSERT_EQ(pulsar_result_Timeout, pulsar_consumer_receive_with_timeout(consumer, &message, 1000)); + + pulsar_consumer_seek_by_timestamp(consumer, currentTime - 100000); // Seek to 100 seconds ago + + ASSERT_EQ(pulsar_result_Ok, pulsar_consumer_receive_with_timeout(consumer, &message, 1000)); + ASSERT_STREQ((const char *)pulsar_message_get_data(message), "msg-0"); + + // Test seek asynchronously + std::promise seek_promise; + std::future seek_future = seek_promise.get_future(); + struct seek_ctx seek_ctx = {&seek_promise}; + pulsar_consumer_seek_by_timestamp_async(consumer, currentTime, seek_callback, &seek_ctx); + ASSERT_EQ(pulsar_result_Ok, seek_future.get()); + ASSERT_EQ(pulsar_result_Timeout, pulsar_consumer_receive_with_timeout(consumer, &message, 1000)); + + pulsar_consumer_free(consumer); + pulsar_consumer_configuration_free(consumer_conf); + pulsar_producer_free(producer); + pulsar_producer_configuration_free(producer_conf); + pulsar_client_free(client); +} + +TEST(c_SeekTest, testReaderSeekMessageId) { + auto topic_name_str = "test-c-reader-seek-msgid-" + std::to_string(time(nullptr)); + const char *topic_name = topic_name_str.c_str(); + + pulsar_client_t *client; + prepare_client(&client); + + pulsar_producer_configuration_t *producer_conf = pulsar_producer_configuration_create(); + pulsar_producer_t *producer; + pulsar_result result = pulsar_client_create_producer(client, topic_name, producer_conf, &producer); + ASSERT_EQ(pulsar_result_Ok, result); + + pulsar_reader_configuration_t *reader_conf = pulsar_reader_configuration_create(); + pulsar_reader_t *reader; + result = + pulsar_client_create_reader(client, topic_name, pulsar_message_id_earliest(), reader_conf, &reader); + ASSERT_EQ(pulsar_result_Ok, result); + + pulsar_message_t *seek_message = nullptr; + + for (int i = 0; i < 10; i++) { + char content[10]; + sprintf(content, "msg-%d", i); + pulsar_message_t *msg = pulsar_message_create(); + pulsar_message_set_content(msg, content, strlen(content)); + pulsar_producer_send(producer, msg); + if (i == 5) { + seek_message = msg; + } else { + pulsar_message_free(msg); + } + } + + pulsar_reader_seek(reader, pulsar_message_get_message_id(seek_message)); + + pulsar_message_t *message; + ASSERT_EQ(pulsar_result_Ok, pulsar_reader_read_next_with_timeout(reader, &message, 1000)); + ASSERT_STREQ((const char *)pulsar_message_get_data(message), "msg-6"); + pulsar_message_free(message); + + // Test seek asynchronously + std::promise seek_promise; + std::future seek_future = seek_promise.get_future(); + struct seek_ctx seek_ctx = {&seek_promise}; + pulsar_reader_seek_async(reader, pulsar_message_get_message_id(seek_message), seek_callback, &seek_ctx); + ASSERT_EQ(pulsar_result_Ok, seek_future.get()); + ASSERT_EQ(pulsar_result_Ok, pulsar_reader_read_next_with_timeout(reader, &message, 1000)); + ASSERT_STREQ((const char *)pulsar_message_get_data(message), "msg-6"); + + if (seek_message != NULL) { + pulsar_message_free(seek_message); + } + pulsar_reader_free(reader); + pulsar_reader_configuration_free(reader_conf); + pulsar_producer_free(producer); + pulsar_producer_configuration_free(producer_conf); + pulsar_client_free(client); +} + +TEST(c_SeekTest, testReaderSeekTime) { + auto topic_name_str = "test-c-reader-seek-time-" + std::to_string(time(nullptr)); + const char *topic_name = topic_name_str.c_str(); + + pulsar_client_t *client; + prepare_client(&client); + + pulsar_producer_configuration_t *producer_conf = pulsar_producer_configuration_create(); + pulsar_producer_t *producer; + pulsar_result result = pulsar_client_create_producer(client, topic_name, producer_conf, &producer); + ASSERT_EQ(pulsar_result_Ok, result); + + pulsar_reader_configuration_t *reader_conf = pulsar_reader_configuration_create(); + pulsar_reader_t *reader; + result = + pulsar_client_create_reader(client, topic_name, pulsar_message_id_earliest(), reader_conf, &reader); + ASSERT_EQ(pulsar_result_Ok, result); + + for (int i = 0; i < 10; i++) { + char content[10]; + sprintf(content, "msg-%d", i); + pulsar_message_t *msg = pulsar_message_create(); + pulsar_message_set_content(msg, content, strlen(content)); + pulsar_producer_send(producer, msg); + pulsar_message_free(msg); + } + + uint64_t currentTime = pulsar::TimeUtils::currentTimeMillis(); + + pulsar_reader_seek_by_timestamp(reader, currentTime); + + pulsar_message_t *message; + ASSERT_EQ(pulsar_result_Timeout, pulsar_reader_read_next_with_timeout(reader, &message, 1000)); + + pulsar_reader_seek_by_timestamp(reader, currentTime - 100000); // Seek to 100 seconds ago + + ASSERT_EQ(pulsar_result_Ok, pulsar_reader_read_next_with_timeout(reader, &message, 1000)); + ASSERT_STREQ((const char *)pulsar_message_get_data(message), "msg-0"); + + // Test seek asynchronously + std::promise seek_promise; + std::future seek_future = seek_promise.get_future(); + struct seek_ctx seek_ctx = {&seek_promise}; + pulsar_reader_seek_by_timestamp_async(reader, currentTime, seek_callback, &seek_ctx); + ASSERT_EQ(pulsar_result_Ok, seek_future.get()); + ASSERT_EQ(pulsar_result_Timeout, pulsar_reader_read_next_with_timeout(reader, &message, 1000)); + + pulsar_reader_free(reader); + pulsar_reader_configuration_free(reader_conf); + pulsar_producer_free(producer); + pulsar_producer_configuration_free(producer_conf); + pulsar_client_free(client); +} From de94bc52edeb429551801196dd6cbefc8bdd4bfa Mon Sep 17 00:00:00 2001 From: Baodi Shi Date: Thu, 17 Nov 2022 19:16:36 +0800 Subject: [PATCH 08/13] Fix getLastMessageId method dead recursion. (#117) --- lib/Consumer.cc | 3 ++- lib/ConsumerImpl.cc | 2 +- lib/ConsumerImpl.h | 3 +-- lib/ConsumerImplBase.h | 3 ++- lib/GetLastMessageIdResponse.h | 3 +++ lib/MultiTopicsConsumerImpl.cc | 4 ++++ lib/MultiTopicsConsumerImpl.h | 1 + tests/ConsumerTest.cc | 22 ++++++++++++++++++++++ 8 files changed, 36 insertions(+), 5 deletions(-) diff --git a/lib/Consumer.cc b/lib/Consumer.cc index 07afcea7..bda7f862 100644 --- a/lib/Consumer.cc +++ b/lib/Consumer.cc @@ -295,7 +295,8 @@ void Consumer::getLastMessageIdAsync(GetLastMessageIdCallback callback) { callback(ResultConsumerNotInitialized, MessageId()); return; } - getLastMessageIdAsync([callback](Result result, const GetLastMessageIdResponse& response) { + + impl_->getLastMessageIdAsync([callback](Result result, const GetLastMessageIdResponse& response) { callback(result, response.getLastMessageId()); }); } diff --git a/lib/ConsumerImpl.cc b/lib/ConsumerImpl.cc index 3f62a7a8..c04a59ea 100644 --- a/lib/ConsumerImpl.cc +++ b/lib/ConsumerImpl.cc @@ -1343,7 +1343,7 @@ void ConsumerImpl::getLastMessageIdAsync(BrokerGetLastMessageIdCallback callback if (state == Closed || state == Closing) { LOG_ERROR(getName() << "Client connection already closed."); if (callback) { - callback(ResultAlreadyClosed, MessageId()); + callback(ResultAlreadyClosed, GetLastMessageIdResponse()); } return; } diff --git a/lib/ConsumerImpl.h b/lib/ConsumerImpl.h index b0a24d4c..ce6c3f09 100644 --- a/lib/ConsumerImpl.h +++ b/lib/ConsumerImpl.h @@ -44,7 +44,6 @@ class BatchAcknowledgementTracker; class MessageCrypto; class GetLastMessageIdResponse; typedef std::shared_ptr MessageCryptoPtr; -typedef std::function BrokerGetLastMessageIdCallback; typedef std::shared_ptr BackoffPtr; class AckGroupingTracker; @@ -124,6 +123,7 @@ class ConsumerImpl : public ConsumerImplBase { const std::string& getName() const override; int getNumOfPrefetchedMessages() const override; void getBrokerConsumerStatsAsync(BrokerConsumerStatsCallback callback) override; + void getLastMessageIdAsync(BrokerGetLastMessageIdCallback callback) override; void seekAsync(const MessageId& msgId, ResultCallback callback) override; void seekAsync(uint64_t timestamp, ResultCallback callback) override; void negativeAcknowledge(const MessageId& msgId) override; @@ -139,7 +139,6 @@ class ConsumerImpl : public ConsumerImplBase { virtual bool isReadCompacted(); virtual void hasMessageAvailableAsync(HasMessageAvailableCallback callback); - virtual void getLastMessageIdAsync(BrokerGetLastMessageIdCallback callback); void beforeConnectionChange(ClientConnection& cnx) override; protected: diff --git a/lib/ConsumerImplBase.h b/lib/ConsumerImplBase.h index 74a8810e..5bc7e1b8 100644 --- a/lib/ConsumerImplBase.h +++ b/lib/ConsumerImplBase.h @@ -25,12 +25,12 @@ #include #include "Future.h" +#include "GetLastMessageIdResponse.h" #include "HandlerBase.h" namespace pulsar { class ConsumerImplBase; using ConsumerImplBaseWeakPtr = std::weak_ptr; - class OpBatchReceive { public: OpBatchReceive(); @@ -68,6 +68,7 @@ class ConsumerImplBase : public HandlerBase, public std::enable_shared_from_this virtual void redeliverUnacknowledgedMessages(const std::set& messageIds) = 0; virtual int getNumOfPrefetchedMessages() const = 0; virtual void getBrokerConsumerStatsAsync(BrokerConsumerStatsCallback callback) = 0; + virtual void getLastMessageIdAsync(BrokerGetLastMessageIdCallback callback) = 0; virtual void seekAsync(const MessageId& msgId, ResultCallback callback) = 0; virtual void seekAsync(uint64_t timestamp, ResultCallback callback) = 0; virtual void negativeAcknowledge(const MessageId& msgId) = 0; diff --git a/lib/GetLastMessageIdResponse.h b/lib/GetLastMessageIdResponse.h index 1ff7933e..cee754b1 100644 --- a/lib/GetLastMessageIdResponse.h +++ b/lib/GetLastMessageIdResponse.h @@ -19,6 +19,7 @@ #pragma once #include +#include #include @@ -54,4 +55,6 @@ class GetLastMessageIdResponse { bool hasMarkDeletePosition_; }; +typedef std::function BrokerGetLastMessageIdCallback; + } // namespace pulsar diff --git a/lib/MultiTopicsConsumerImpl.cc b/lib/MultiTopicsConsumerImpl.cc index d14c3cae..b8d55b49 100644 --- a/lib/MultiTopicsConsumerImpl.cc +++ b/lib/MultiTopicsConsumerImpl.cc @@ -803,6 +803,10 @@ void MultiTopicsConsumerImpl::getBrokerConsumerStatsAsync(BrokerConsumerStatsCal }); } +void MultiTopicsConsumerImpl::getLastMessageIdAsync(BrokerGetLastMessageIdCallback callback) { + callback(ResultOperationNotSupported, GetLastMessageIdResponse()); +} + void MultiTopicsConsumerImpl::handleGetConsumerStats(Result res, BrokerConsumerStats brokerConsumerStats, LatchPtr latchPtr, MultiTopicsBrokerConsumerStatsPtr statsPtr, size_t index, diff --git a/lib/MultiTopicsConsumerImpl.h b/lib/MultiTopicsConsumerImpl.h index ac25c840..da42b748 100644 --- a/lib/MultiTopicsConsumerImpl.h +++ b/lib/MultiTopicsConsumerImpl.h @@ -82,6 +82,7 @@ class MultiTopicsConsumerImpl : public ConsumerImplBase { const std::string& getName() const override; int getNumOfPrefetchedMessages() const override; void getBrokerConsumerStatsAsync(BrokerConsumerStatsCallback callback) override; + void getLastMessageIdAsync(BrokerGetLastMessageIdCallback callback) override; void seekAsync(const MessageId& msgId, ResultCallback callback) override; void seekAsync(uint64_t timestamp, ResultCallback callback) override; void negativeAcknowledge(const MessageId& msgId) override; diff --git a/tests/ConsumerTest.cc b/tests/ConsumerTest.cc index a5ef32fe..f3c8abbe 100644 --- a/tests/ConsumerTest.cc +++ b/tests/ConsumerTest.cc @@ -841,6 +841,28 @@ TEST(ConsumerTest, testPartitionsWithCloseUnblock) { thread.join(); } +TEST(ConsumerTest, testGetLastMessageId) { + Client client(lookupUrl); + const std::string topic = "testGetLastMessageId-" + std::to_string(time(nullptr)); + + Consumer consumer; + ASSERT_EQ(ResultOk, client.subscribe(topic, "test-sub", consumer)); + + MessageId msgId; + ASSERT_EQ(ResultOk, consumer.getLastMessageId(msgId)); + ASSERT_EQ(msgId, MessageId(-1, -1, -1, -1)); + + Producer producer; + ASSERT_EQ(ResultOk, client.createProducer(topic, producer)); + Message msg = MessageBuilder().setContent("message").build(); + ASSERT_EQ(ResultOk, producer.send(msg)); + + ASSERT_EQ(ResultOk, consumer.getLastMessageId(msgId)); + ASSERT_NE(msgId, MessageId(-1, -1, -1, -1)); + + client.close(); +} + TEST(ConsumerTest, testGetLastMessageIdBlockWhenConnectionDisconnected) { int operationTimeout = 5; ClientConfiguration clientConfiguration; From 9f9314a1fa4615d7fb9a01ab582ebfdd9259ca97 Mon Sep 17 00:00:00 2001 From: Yunze Xu Date: Tue, 22 Nov 2022 19:01:07 +0800 Subject: [PATCH 09/13] Fix segmentation fault during the destruction of ConsumerImpl (#121) ### Motivation When I ran the tests of Python wrapper in my local env, I observed a segmentation fault. See the key stacktrace: ``` #3 0x00007ffff6d742c5 in std::unique_lock::lock() () from /usr/local/lib/python3.8/dist-packages/_pulsar.cpython-38-x86_64-linux-gnu.so #4 0x00007ffff6d72523 in std::unique_lock::unique_lock(std::mutex&) () from /usr/local/lib/python3.8/dist-packages/_pulsar.cpython-38-x86_64-linux-gnu.so #5 0x00007ffff67de193 in pulsar::ClientImpl::newRequestId (this=0x0) at /home/xyz/github.com/apache/pulsar-client-cpp/lib/ClientImpl.cc:644 #6 0x00007ffff685d2c2 in pulsar::ConsumerImpl::~ConsumerImpl (this=0x7fff9800f9e0, __in_chrg=) at /home/xyz/github.com/apache/pulsar-client-cpp/lib/ConsumerImpl.cc:116 ``` In the destructor of `ConsumerImpl`, `client->newRequestId` might be called. However, `client` might be a null pointer because it's returned by `std::weak_ptr::lock()`. ### Modifications Add null check to avoid the segfault. --- lib/ConsumerImpl.cc | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/lib/ConsumerImpl.cc b/lib/ConsumerImpl.cc index c04a59ea..41b13ae2 100644 --- a/lib/ConsumerImpl.cc +++ b/lib/ConsumerImpl.cc @@ -127,11 +127,13 @@ ConsumerImpl::~ConsumerImpl() { ClientConnectionPtr cnx = getCnx().lock(); ClientImplPtr client = client_.lock(); - int requestId = client->newRequestId(); - if (cnx) { + 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_); + } else { + LOG_WARN(getName() << "Client is destroyed and cannot send the CloseConsumer command"); } } shutdown(); From f0268ecd29a6d0030b7d07379ec609884b4c14ff Mon Sep 17 00:00:00 2001 From: Zike Yang Date: Thu, 24 Nov 2022 20:15:10 +0800 Subject: [PATCH 10/13] [improve] Skip include debug artifact in the release and tar the windows artifacts (#124) Motivation The debug artifact for the windows system is too large. They are only used for debugging and doesn't need to include in the release. And it's better to zip tar the `windows-static` artifact to make it easier to download through the command line. The final result looks like this: https://dist.apache.org/repos/dist/dev/pulsar/pulsar-client-cpp/pulsar-client-cpp-3.1.0-candidate-1/ Modification * Skip include the windows debug artifact in the release * Tar `windows-static` artifact in the release. --- build-support/download-release-artifacts.py | 3 +++ build-support/stage-release.sh | 6 ++++++ 2 files changed, 9 insertions(+) diff --git a/build-support/download-release-artifacts.py b/build-support/download-release-artifacts.py index 774ffdc8..211faba4 100755 --- a/build-support/download-release-artifacts.py +++ b/build-support/download-release-artifacts.py @@ -45,6 +45,9 @@ data = json.loads(response.read().decode("utf-8")) for artifact in data['artifacts']: name = artifact['name'] + # Skip debug artifact + if name.endswith("-Debug"): + continue url = artifact['archive_download_url'] print('Downloading %s from %s' % (name, url)) diff --git a/build-support/stage-release.sh b/build-support/stage-release.sh index 93b5c412..2f26d383 100755 --- a/build-support/stage-release.sh +++ b/build-support/stage-release.sh @@ -39,6 +39,12 @@ cd $PULSAR_CPP_PATH build-support/generate-source-archive.sh $DEST_PATH build-support/download-release-artifacts.py $WORKFLOW_ID $DEST_PATH +pushd "$DEST_PATH" +tar cvzf x64-windows-static.tar.gz x64-windows-static +tar cvzf x86-windows-static.tar.gz x86-windows-static +rm -r x64-windows-static x86-windows-static +popd + # Sign all files cd $DEST_PATH find . -type f | xargs $PULSAR_CPP_PATH/build-support/sign-files.sh From 96d870527d2a078645c4866e9d75b3f3b97193ba Mon Sep 17 00:00:00 2001 From: TT Date: Mon, 28 Nov 2022 11:16:52 +0800 Subject: [PATCH 11/13] Fix NamedEntity::checkName regression (#127) --- lib/NamedEntity.cc | 3 ++- tests/NamespaceNameTest.cc | 7 ++++--- tests/TopicNameTest.cc | 9 +++++---- 3 files changed, 11 insertions(+), 8 deletions(-) diff --git a/lib/NamedEntity.cc b/lib/NamedEntity.cc index 484c8b1f..a920c496 100644 --- a/lib/NamedEntity.cc +++ b/lib/NamedEntity.cc @@ -22,7 +22,7 @@ /** * Allowed characters for property, namespace, cluster and topic names are - * alphanumeric (a-zA-Z_0-9) and these special chars -=:. + * alphanumeric (a-zA-Z0-9) and these special chars _-=:. * @param name * @return */ @@ -33,6 +33,7 @@ bool NamedEntity::checkName(const std::string& name) { } switch (c) { + case '_': case '-': case '=': case ':': diff --git a/tests/NamespaceNameTest.cc b/tests/NamespaceNameTest.cc index 132506ea..f8fcfc72 100644 --- a/tests/NamespaceNameTest.cc +++ b/tests/NamespaceNameTest.cc @@ -44,9 +44,10 @@ TEST(NamespaceNameTest, testNamespaceNameV2) { } TEST(NamespaceNameTest, testNamespaceNameLegalCharacters) { - std::shared_ptr nn1 = NamespaceName::get("cluster-1:=.", "namespace-1:=."); - ASSERT_EQ("cluster-1:=.", nn1->getProperty()); + std::shared_ptr nn1 = NamespaceName::get("cluster-1:=._", "namespace-1:=._"); + ASSERT_TRUE(nn1); + ASSERT_EQ("cluster-1:=._", nn1->getProperty()); ASSERT_TRUE(nn1->getCluster().empty()); - ASSERT_EQ("namespace-1:=.", nn1->getLocalName()); + ASSERT_EQ("namespace-1:=._", nn1->getLocalName()); ASSERT_TRUE(nn1->isV2()); } diff --git a/tests/TopicNameTest.cc b/tests/TopicNameTest.cc index 44b3dc2d..41838f5d 100644 --- a/tests/TopicNameTest.cc +++ b/tests/TopicNameTest.cc @@ -143,12 +143,13 @@ TEST(TopicNameTest, testIllegalCharacters) { } TEST(TopicNameTest, testLegalNonAlphaCharacters) { - std::shared_ptr topicName = TopicName::get("persistent://cluster-1:=./namespace-1:=./topic"); + std::shared_ptr topicName = + TopicName::get("persistent://cluster-1:=._/namespace-1:=._/topic_"); ASSERT_TRUE(topicName); - ASSERT_EQ("cluster-1:=.", topicName->getProperty()); - ASSERT_EQ("namespace-1:=.", topicName->getNamespacePortion()); + ASSERT_EQ("cluster-1:=._", topicName->getProperty()); + ASSERT_EQ("namespace-1:=._", topicName->getNamespacePortion()); ASSERT_EQ("persistent", topicName->getDomain()); - ASSERT_EQ("topic", topicName->getLocalName()); + ASSERT_EQ("topic_", topicName->getLocalName()); } TEST(TopicNameTest, testIllegalUrl) { From 85b1b53f1e9ed571d813276bde4f9ad7b31b5659 Mon Sep 17 00:00:00 2001 From: erobot Date: Mon, 28 Nov 2022 15:37:31 +0800 Subject: [PATCH 12/13] [fix] Fix PartitionedProducerImpl::closeAsync to close sub-producers properly (#125) ### Motivation PartitionedProducerImpl do not close sub-producers properly when any sub-producer creation fails. Continuing to retry creating producer will eventually reach the maximum producer limit. It seems a regression caused by #54. When sub-producer creation fails, state_ is set to Failed. PartitionedProducerImpl::closeAsync only do cleanup when state_==Ready and sub-producers do not close when state_==Failed. https://github.com/apache/pulsar-client-cpp/blob/f0268ecd29a6d0030b7d07379ec609884b4c14ff/lib/PartitionedProducerImpl.cc#L273-L276 ### Modifications Close sub-producers when state != Closed. --- lib/PartitionedProducerImpl.cc | 7 ++--- tests/ProducerTest.cc | 51 ++++++++++++++++++++++++++++++++++ 2 files changed, 53 insertions(+), 5 deletions(-) diff --git a/lib/PartitionedProducerImpl.cc b/lib/PartitionedProducerImpl.cc index 26d5796a..0b5d4527 100644 --- a/lib/PartitionedProducerImpl.cc +++ b/lib/PartitionedProducerImpl.cc @@ -266,14 +266,11 @@ void PartitionedProducerImpl::closeAsync(CloseCallback originalCallback) { originalCallback(result); } }; - if (state_ == Closed) { + + if (state_ == Closed || state_.exchange(Closing) == Closing) { closeCallback(ResultAlreadyClosed); return; } - State expectedState = Ready; - if (!state_.compare_exchange_strong(expectedState, Closing)) { - return; - } cancelTimers(); diff --git a/tests/ProducerTest.cc b/tests/ProducerTest.cc index a0b1e7e2..74d3cf20 100644 --- a/tests/ProducerTest.cc +++ b/tests/ProducerTest.cc @@ -373,4 +373,55 @@ TEST_P(ProducerTest, testFlushNoBatch) { client.close(); } +TEST(ProducerTest, testCloseSubProducerWhenFail) { + Client client(serviceUrl); + + std::string ns = "test-close-sub-producer-when-fail"; + std::string localName = std::string("testCloseSubProducerWhenFail") + std::to_string(time(nullptr)); + std::string topicName = "persistent://public/" + ns + '/' + localName; + const int maxProducersPerTopic = 10; + const int partitionNum = 5; + + // call admin api to create namespace with max prodcuer limit + std::string url = adminUrl + "admin/v2/namespaces/public/" + ns; + int res = + makePutRequest(url, "{\"max_producers_per_topic\": " + std::to_string(maxProducersPerTopic) + "}"); + ASSERT_TRUE(res == 204 || res == 409) << "res:" << res; + + // call admin api to create partitioned topic + res = makePutRequest(adminUrl + "admin/v2/persistent/public/" + ns + "/" + localName + "/partitions", + std::to_string(partitionNum)); + ASSERT_TRUE(res == 204 || res == 409) << "res: " << res; + + ProducerConfiguration producerConfiguration; + producerConfiguration.setBatchingEnabled(false); + + // create producers for partition-0 up to max producer limit + std::vector producers; + for (int i = 0; i < maxProducersPerTopic; ++i) { + Producer producer; + ASSERT_EQ(ResultOk, + client.createProducer(topicName + "-partition-0", producerConfiguration, producer)); + producers.push_back(producer); + } + + // create partitioned producer, should fail because partition-0 already reach max producer limit + for (int i = 0; i < maxProducersPerTopic; ++i) { + Producer producer; + ASSERT_EQ(ResultProducerBusy, client.createProducer(topicName, producer)); + } + + std::this_thread::sleep_for(std::chrono::seconds(1)); + + // create producer for partition-1, should succeed + Producer producer; + ASSERT_EQ(ResultOk, client.createProducer(topicName + "-partition-1", producerConfiguration, producer)); + producers.push_back(producer); + + for (auto& producer : producers) { + producer.close(); + } + client.close(); +} + INSTANTIATE_TEST_CASE_P(Pulsar, ProducerTest, ::testing::Values(true, false)); From 2ac32f2db5f855bc712be8f40660ca67c7d720d3 Mon Sep 17 00:00:00 2001 From: Baodi Shi Date: Tue, 6 Dec 2022 14:27:40 +0800 Subject: [PATCH 13/13] dlq one 1. --- include/pulsar/ConsumerConfiguration.h | 15 + include/pulsar/DeadLetterPolicy.h | 72 ++++ include/pulsar/DeadLetterPolicyBuilder.h | 82 +++++ include/pulsar/ProducerConfiguration.h | 12 + lib/BinaryProtoLookupService.cc | 5 + lib/BinaryProtoLookupService.h | 3 + lib/Commands.cc | 6 +- lib/Commands.h | 3 +- lib/ConsumerConfiguration.cc | 6 + lib/ConsumerConfigurationImpl.h | 1 + lib/ConsumerImpl.cc | 155 ++++++++- lib/ConsumerImpl.h | 20 ++ lib/DeadLetterPolicyBuilder.cc | 54 +++ lib/DeadLetterPolicyImpl.cc | 38 +++ lib/DeadLetterPolicyImpl.h | 31 ++ lib/HTTPLookupService.cc | 4 + lib/HTTPLookupService.h | 2 + lib/LookupService.h | 8 + lib/MessageId.cc | 29 +- lib/MultiTopicsConsumerImpl.cc | 21 +- lib/MultiTopicsConsumerImpl.h | 1 + lib/NegativeAcksTracker.cc | 2 +- lib/ProducerConfiguration.cc | 12 + lib/ProducerConfigurationImpl.h | 1 + lib/ProducerImpl.cc | 3 +- lib/RetryableLookupService.h | 4 + lib/SynchronizedHashMap.h | 2 +- tests/ConsumerConfigurationTest.cc | 14 + tests/ConsumerTest.cc | 65 ++++ tests/DeadLetterPolicyTest.cc | 43 +++ tests/DeadLetterQueueTest.cc | 405 +++++++++++++++++++++++ tests/ProducerTest.cc | 1 + 32 files changed, 1107 insertions(+), 13 deletions(-) create mode 100644 include/pulsar/DeadLetterPolicy.h create mode 100644 include/pulsar/DeadLetterPolicyBuilder.h create mode 100644 lib/DeadLetterPolicyBuilder.cc create mode 100644 lib/DeadLetterPolicyImpl.cc create mode 100644 lib/DeadLetterPolicyImpl.h create mode 100644 tests/DeadLetterPolicyTest.cc create mode 100644 tests/DeadLetterQueueTest.cc diff --git a/include/pulsar/ConsumerConfiguration.h b/include/pulsar/ConsumerConfiguration.h index 520901c2..071e3378 100644 --- a/include/pulsar/ConsumerConfiguration.h +++ b/include/pulsar/ConsumerConfiguration.h @@ -34,6 +34,7 @@ #include #include "BatchReceivePolicy.h" +#include "DeadLetterPolicy.h" namespace pulsar { @@ -398,6 +399,20 @@ class PULSAR_PUBLIC ConsumerConfiguration { */ const BatchReceivePolicy& getBatchReceivePolicy() const; + /** + * Set dead letter policy. + * + * @param deadLetterPolicy thd default is empty + */ + void setDeadLetterPolicy(const DeadLetterPolicy& deadLetterPolicy); + + /** + * Get dead letter policy. + * + * @return dead letter policy + */ + const DeadLetterPolicy& getDeadLetterPolicy() const; + /** * Set whether the subscription status should be replicated. * The default value is `false`. diff --git a/include/pulsar/DeadLetterPolicy.h b/include/pulsar/DeadLetterPolicy.h new file mode 100644 index 00000000..95bf9250 --- /dev/null +++ b/include/pulsar/DeadLetterPolicy.h @@ -0,0 +1,72 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +#ifndef DEAD_LETTER_POLICY_HPP_ +#define DEAD_LETTER_POLICY_HPP_ + +#include + +#include + +namespace pulsar { + +struct DeadLetterPolicyImpl; + +/** + * Configuration for the "dead letter queue" feature in consumer. + * + * see @DeadLetterPolicyBuilder + */ +class PULSAR_PUBLIC DeadLetterPolicy { + public: + + DeadLetterPolicy(); + + /** + * Get dead letter topic + * + * @return + */ + std::string getDeadLetterTopic() const; + + /** + * Get max redeliver count + * + * @return + */ + int getMaxRedeliverCount() const; + + /** + * Get initial subscription name + * + * @return + */ + std::string getInitialSubscriptionName() const; + + private: + friend class DeadLetterPolicyBuilder; + + typedef std::shared_ptr DeadLetterPolicyImplPtr; + DeadLetterPolicyImplPtr impl_; + + explicit DeadLetterPolicy(const DeadLetterPolicyImplPtr& impl); + +}; +} // namespace pulsar + +#endif /* DEAD_LETTER_POLICY_HPP_ */ diff --git a/include/pulsar/DeadLetterPolicyBuilder.h b/include/pulsar/DeadLetterPolicyBuilder.h new file mode 100644 index 00000000..705335e9 --- /dev/null +++ b/include/pulsar/DeadLetterPolicyBuilder.h @@ -0,0 +1,82 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +#ifndef DEAD_LETTER_POLICY_BUILD_HPP_ +#define DEAD_LETTER_POLICY_BUILD_HPP_ + +#include +#include + +#include + +namespace pulsar { + +struct DeadLetterPolicyImpl; + +/** + * The builder to build a DeadLetterPolicyBuilder + * + * Example of building DeadLetterPolicy: + * + * ```c++ + * DeadLetterPolicy dlqPolicy = DeadLetterPolicyBuilder() + * .deadLetterTopic("dlq-topic") + * .maxRedeliverCount(10) + * .initialSubscriptionName("init-sub-name") + * .build(); + * ``` + */ +class PULSAR_PUBLIC DeadLetterPolicyBuilder { + public: + + DeadLetterPolicyBuilder(); + + /** + * Set dead letter topic + * + * @return + */ + DeadLetterPolicyBuilder& deadLetterTopic(const std::string& deadLetterTopic); + + /** + * Set max redeliver count + * + * @return + */ + DeadLetterPolicyBuilder& maxRedeliverCount(int maxRedeliverCount); + + /** + * Set initial subscription name + * + * @return + */ + DeadLetterPolicyBuilder& initialSubscriptionName(const std::string& initialSubscriptionName); + + /** + * Build DeadLetterPolicy. + * + * @return + */ + DeadLetterPolicy build(); + + private: + std::shared_ptr impl_; +}; +} // namespace pulsar + +#endif /* DEAD_LETTER_POLICY_BUILD_HPP_ */ diff --git a/include/pulsar/ProducerConfiguration.h b/include/pulsar/ProducerConfiguration.h index 873e1383..a4fae26c 100644 --- a/include/pulsar/ProducerConfiguration.h +++ b/include/pulsar/ProducerConfiguration.h @@ -532,6 +532,18 @@ class PULSAR_PUBLIC ProducerConfiguration { */ ProducerAccessMode getAccessMode() const; + /** + * Use this configuration to automatically create an initial subscription when creating a topic. + * + * If this field is not set, the initial subscription is not created. + */ + ProducerConfiguration& setInitialSubscriptionName(const std::string& initialSubscriptionName); + + /** + * Get initial subscription name. + */ + const std::string& getInitialSubscriptionName() const; + friend class PulsarWrapper; private: diff --git a/lib/BinaryProtoLookupService.cc b/lib/BinaryProtoLookupService.cc index b863d529..4c242377 100644 --- a/lib/BinaryProtoLookupService.cc +++ b/lib/BinaryProtoLookupService.cc @@ -155,6 +155,11 @@ Future BinaryProtoLookupService::getTopicsOfNamespac return promise->getFuture(); } + +Future BinaryProtoLookupService::getSchema(const TopicNamePtr& topicName) { + return Promise().getFuture(); +} + void BinaryProtoLookupService::sendGetTopicsOfNamespaceRequest(const std::string& nsName, Result result, const ClientConnectionWeakPtr& clientCnx, NamespaceTopicsPromisePtr promise) { diff --git a/lib/BinaryProtoLookupService.h b/lib/BinaryProtoLookupService.h index 9adb6483..129f6dd0 100644 --- a/lib/BinaryProtoLookupService.h +++ b/lib/BinaryProtoLookupService.h @@ -20,6 +20,7 @@ #define _PULSAR_BINARY_LOOKUP_SERVICE_HEADER_ #include +#include #include @@ -45,6 +46,8 @@ class PULSAR_PUBLIC BinaryProtoLookupService : public LookupService { Future getTopicsOfNamespaceAsync(const NamespaceNamePtr& nsName) override; + Future getSchema(const TopicNamePtr& topicName) override; + private: std::mutex mutex_; uint64_t requestIdGenerator_ = 0; diff --git a/lib/Commands.cc b/lib/Commands.cc index f97b0eb8..4ae11e2e 100644 --- a/lib/Commands.cc +++ b/lib/Commands.cc @@ -383,7 +383,8 @@ SharedBuffer Commands::newProducer(const std::string& topic, uint64_t producerId const std::map& metadata, const SchemaInfo& schemaInfo, uint64_t epoch, bool userProvidedProducerName, bool encrypted, - ProducerAccessMode accessMode, Optional topicEpoch) { + ProducerAccessMode accessMode, Optional topicEpoch, + std::string initialSubscriptionName) { BaseCommand cmd; cmd.set_type(BaseCommand::PRODUCER); CommandProducer* producer = cmd.mutable_producer(); @@ -397,6 +398,9 @@ SharedBuffer Commands::newProducer(const std::string& topic, uint64_t producerId if (topicEpoch.is_present()) { producer->set_topic_epoch(topicEpoch.value()); } + if (!initialSubscriptionName.empty()) { + producer->set_initial_subscription_name(initialSubscriptionName); + } for (std::map::const_iterator it = metadata.begin(); it != metadata.end(); it++) { diff --git a/lib/Commands.h b/lib/Commands.h index 6681f138..b5edcc66 100644 --- a/lib/Commands.h +++ b/lib/Commands.h @@ -107,7 +107,8 @@ class Commands { const std::map& metadata, const SchemaInfo& schemaInfo, uint64_t epoch, bool userProvidedProducerName, bool encrypted, - ProducerAccessMode accessMode, Optional topicEpoch); + ProducerAccessMode accessMode, Optional topicEpoch, + std::string initialSubscriptionName); static SharedBuffer newAck(uint64_t consumerId, int64_t ledgerId, int64_t entryId, CommandAck_AckType ackType, CommandAck_ValidationError validationError); diff --git a/lib/ConsumerConfiguration.cc b/lib/ConsumerConfiguration.cc index f37e042d..fba142bb 100644 --- a/lib/ConsumerConfiguration.cc +++ b/lib/ConsumerConfiguration.cc @@ -287,4 +287,10 @@ const BatchReceivePolicy& ConsumerConfiguration::getBatchReceivePolicy() const { return impl_->batchReceivePolicy; } +void ConsumerConfiguration::setDeadLetterPolicy(const DeadLetterPolicy& deadLetterPolicy) { + impl_->deadLetterPolicy = deadLetterPolicy; +} + +const DeadLetterPolicy& ConsumerConfiguration::getDeadLetterPolicy() const { return impl_->deadLetterPolicy; } + } // namespace pulsar diff --git a/lib/ConsumerConfigurationImpl.h b/lib/ConsumerConfigurationImpl.h index 259b9354..f795ed6a 100644 --- a/lib/ConsumerConfigurationImpl.h +++ b/lib/ConsumerConfigurationImpl.h @@ -46,6 +46,7 @@ struct ConsumerConfigurationImpl { bool readCompacted{false}; InitialPosition subscriptionInitialPosition{InitialPosition::InitialPositionLatest}; BatchReceivePolicy batchReceivePolicy{}; + DeadLetterPolicy deadLetterPolicy; int patternAutoDiscoveryPeriod{60}; bool replicateSubscriptionStateEnabled{false}; std::map properties; diff --git a/lib/ConsumerImpl.cc b/lib/ConsumerImpl.cc index 41b13ae2..da2137d9 100644 --- a/lib/ConsumerImpl.cc +++ b/lib/ConsumerImpl.cc @@ -18,6 +18,7 @@ */ #include "ConsumerImpl.h" +#include #include #include @@ -113,6 +114,21 @@ ConsumerImpl::ConsumerImpl(const ClientImplPtr client, const std::string& topic, msgCrypto_ = std::make_shared(consumerStr_, false); } + // Config dlq + auto deadLetterPolicy = conf.getDeadLetterPolicy(); + if (deadLetterPolicy.getMaxRedeliverCount() > 0) { + auto deadLetterPolicyBuilder = + DeadLetterPolicyBuilder() + .maxRedeliverCount(deadLetterPolicy.getMaxRedeliverCount()) + .initialSubscriptionName(deadLetterPolicy.getInitialSubscriptionName()); + if (deadLetterPolicy.getDeadLetterTopic().empty()) { + deadLetterPolicyBuilder.deadLetterTopic(topic + "-" + subscriptionName + DLQ_GROUP_TOPIC_SUFFIX); + } else { + deadLetterPolicyBuilder.deadLetterTopic(deadLetterPolicy.getDeadLetterTopic()); + } + deadLetterPolicy_ = deadLetterPolicyBuilder.build(); + } + checkExpiredChunkedTimer_ = executor_->createDeadlineTimer(); } @@ -460,6 +476,7 @@ void ConsumerImpl::messageReceived(const ClientConnectionPtr& cnx, const proto:: return; } + auto redeliveryCount = msg.redelivery_count(); const bool isMessageUndecryptable = metadata.encryption_keys_size() > 0 && !config_.getCryptoKeyReader().get() && config_.getCryptoFailureAction() == ConsumerCryptoFailureAction::CONSUME; @@ -520,6 +537,11 @@ void ConsumerImpl::messageReceived(const ClientConnectionPtr& cnx, const proto:: << startMessageId.value()); return; } + LOG_INFO("Receive msg redeliveryCount: " << redeliveryCount << " msg: " << m.getMessageId()); + if (redeliveryCount >= deadLetterPolicy_.getMaxRedeliverCount()) { + possibleSendToDeadLetterTopicMessages_.emplace(m.getMessageId(), std::vector{m}); + increaseAvailablePermits(cnx); + } executeNotifyCallback(m); } @@ -566,6 +588,8 @@ void ConsumerImpl::failPendingReceiveCallback() { } void ConsumerImpl::executeNotifyCallback(Message& msg) { + LOG_INFO("exece msg" + msg.getDataAsString() + " " << msg.getRedeliveryCount()); + Lock lock(pendingReceiveMutex_); // if asyncReceive is waiting then notify callback without adding to incomingMessages queue bool asyncReceivedWaiting = !pendingReceives_.empty(); @@ -630,6 +654,7 @@ uint32_t ConsumerImpl::receiveIndividualMessagesFromBatch(const ClientConnection int skippedMessages = 0; + std::vector possibleToDeadLetter; for (int i = 0; i < batchSize; i++) { // This is a cheap copy since message contains only one shared pointer (impl_) Message msg = Commands::deSerializeSingleMessageInBatch(batchedMessage, i, batchSize); @@ -652,9 +677,20 @@ uint32_t ConsumerImpl::receiveIndividualMessagesFromBatch(const ClientConnection } } + LOG_INFO("batch receive: " << msg.getMessageId() << " originMsgId" << batchedMessage.getMessageId() + << " redeliverCount: " << redeliveryCount); + if (redeliveryCount >= deadLetterPolicy_.getMaxRedeliverCount()) { + LOG_INFO("add to batch dlq queue: " << msg.getMessageId()); + possibleToDeadLetter.emplace_back(msg); + } + executeNotifyCallback(msg); } + if (!possibleToDeadLetter.empty()) { + possibleSendToDeadLetterTopicMessages_.emplace(batchedMessage.getMessageId(), possibleToDeadLetter); + } + if (skippedMessages > 0) { increaseAvailablePermits(cnx, skippedMessages); } @@ -1184,11 +1220,32 @@ void ConsumerImpl::redeliverUnacknowledgedMessages(const std::set& me if (messageIds.empty()) { return; } + if (config_.getConsumerType() != ConsumerShared && config_.getConsumerType() != ConsumerKeyShared) { redeliverUnacknowledgedMessages(); return; } - redeliverMessages(messageIds); + + ClientConnectionPtr cnx = getCnx().lock(); + if (cnx) { + if (cnx->getServerProtocolVersion() >= proto::v2) { + auto needRedeliverMsgs = std::make_shared>(); + auto needCallBack = std::make_shared>(messageIds.size()); + for (const auto& msgId : messageIds) { + processPossibleToDLQ(msgId, + [this, needRedeliverMsgs, &msgId, needCallBack](bool processSuccess) { + if (!processSuccess) { + needRedeliverMsgs->emplace(msgId); + } + if (--(*needCallBack) == 0 && !needRedeliverMsgs->empty()) { + redeliverMessages(*needRedeliverMsgs); + } + }); + } + } + } else { + LOG_WARN("Connection not ready for Consumer - " << getConsumerId()); + } } void ConsumerImpl::redeliverMessages(const std::set& messageIds) { @@ -1196,10 +1253,11 @@ void ConsumerImpl::redeliverMessages(const std::set& messageIds) { if (cnx) { if (cnx->getServerProtocolVersion() >= proto::v2) { cnx->sendCommand(Commands::newRedeliverUnacknowledgedMessages(consumerId_, messageIds)); - LOG_DEBUG("Sending RedeliverUnacknowledgedMessages command for Consumer - " << getConsumerId()); + // todo debug log + LOG_INFO("Sending RedeliverUnacknowledgedMessages command for Consumer - " << getConsumerId()); } } else { - LOG_DEBUG("Connection not ready for Consumer - " << getConsumerId()); + LOG_WARN("Connection not ready for Consumer - " << getConsumerId()); } } @@ -1505,4 +1563,95 @@ void ConsumerImpl::cancelTimers() noexcept { checkExpiredChunkedTimer_->cancel(ec); } +void ConsumerImpl::processPossibleToDLQ(const MessageId& messageId, ProcessDLQCallBack cb) { + auto messages = possibleSendToDeadLetterTopicMessages_.find(messageId); + if (messages.is_empty()) { + cb(false); + return; + } + + // Initialize deadLetterProducer_ + if (!deadLetterProducer_) { + Lock createLock(createProducerLock_); + if (!deadLetterProducer_) { + deadLetterProducer_ = std::make_shared>(); + ProducerConfiguration producerConfiguration; + producerConfiguration.setBlockIfQueueFull(false); + if (!deadLetterPolicy_.getInitialSubscriptionName().empty()) { + producerConfiguration.setInitialSubscriptionName( + deadLetterPolicy_.getInitialSubscriptionName()); + } + ClientImplPtr client = client_.lock(); + if (client) { + client->createProducerAsync( + deadLetterPolicy_.getDeadLetterTopic(), producerConfiguration, + [this](Result res, Producer producer) { + if (res == ResultOk) { + deadLetterProducer_->setValue(producer); + } else { + LOG_ERROR("Dead letter producer create exception with topic: " + << deadLetterPolicy_.getDeadLetterTopic() << " ex: " << res); + deadLetterProducer_.reset(); + } + }); + } else { + LOG_WARN(getName() << "Client is destroyed and cannot create dead letter producer."); + } + } + createLock.unlock(); + } + + for (const auto& message : messages.value()) { + auto self = get_shared_this_ptr(); + deadLetterProducer_->getFuture().addListener([self, message, cb](Result res, Producer producer) { + auto originMessageId = message.getMessageId(); + std::stringstream originMessageIdStr; + originMessageIdStr << originMessageId; + MessageBuilder msgBuilder; + msgBuilder.setAllocatedContent(const_cast(message.getData()), message.getLength()) + .setProperties(message.getProperties()) + .setProperty(PROPERTY_ORIGIN_MESSAGE_ID, originMessageIdStr.str()) + .setProperty(SYSTEM_PROPERTY_REAL_TOPIC, message.getTopicName()); + if (message.hasPartitionKey()) { + msgBuilder.setPartitionKey(message.getPartitionKey()); + } + if (message.hasOrderingKey()) { + msgBuilder.setOrderingKey(message.getOrderingKey()); + } + producer.sendAsync(msgBuilder.build(), [self, originMessageId, cb](Result res, + const MessageId& messageId) { + if (res == ResultOk) { + if (self->state_ != Ready) { + LOG_WARN( + "Send to the DLQ successfully, but consumer is not ready. ignore acknowledge : " + << self->state_); + cb(false); + return; + } + self->acknowledgeAsync(originMessageId, [self, originMessageId, cb](Result result) { + if (result != ResultOk) { + LOG_WARN("{" << self->topic_ << "} {" << self->subscription_ << "} {" + << self->consumerName_ << "} Failed to acknowledge the message {" + << originMessageId + << "} of the original topic but send to the DLQ successfully : " + << result); + cb(false); + } else { + LOG_DEBUG("Send msg:" << originMessageId + << "to DLQ success and acknowledge success."); + cb(true); + } + }); + } else { + LOG_WARN("{" << self->topic_ << "} {" << self->subscription_ << "} {" + << self->consumerName_ << "} Failed to send DLQ message to {" + << self->deadLetterPolicy_.getDeadLetterTopic() << "} for message id " + << "{" << originMessageId << "} : " << res); + cb(false); + } + }); + }); + } +} + } /* namespace pulsar */ diff --git a/lib/ConsumerImpl.h b/lib/ConsumerImpl.h index ce6c3f09..1286d9db 100644 --- a/lib/ConsumerImpl.h +++ b/lib/ConsumerImpl.h @@ -35,6 +35,7 @@ #include "TestUtil.h" #include "TimeUtils.h" #include "UnboundedBlockingQueue.h" +#include "lib/SynchronizedHashMap.h" namespace pulsar { class UnAckedMessageTrackerInterface; @@ -45,6 +46,7 @@ class MessageCrypto; class GetLastMessageIdResponse; typedef std::shared_ptr MessageCryptoPtr; typedef std::shared_ptr BackoffPtr; +typedef std::function ProcessDLQCallBack; class AckGroupingTracker; using AckGroupingTrackerPtr = std::shared_ptr; @@ -64,6 +66,10 @@ enum ConsumerTopicType Partitioned }; +const static std::string SYSTEM_PROPERTY_REAL_TOPIC = "REAL_TOPIC"; +const static std::string PROPERTY_ORIGIN_MESSAGE_ID = "ORIGIN_MESSAGE_ID"; +const static std::string DLQ_GROUP_TOPIC_SUFFIX = "-DLQ"; + class ConsumerImpl : public ConsumerImplBase { public: ConsumerImpl(const ClientImplPtr client, const std::string& topic, const std::string& subscriptionName, @@ -196,9 +202,16 @@ class ConsumerImpl : public ConsumerImplBase { Optional clearReceiveQueue(); void seekAsyncInternal(long requestId, SharedBuffer seek, const MessageId& seekId, long timestamp, ResultCallback callback); + /** + * send msg to DLQ, This method is an asynchronous method that does not throw any exceptions. + * + * @param message + */ + void processPossibleToDLQ(const MessageId& messageId, ProcessDLQCallBack cb); std::mutex mutexForReceiveWithZeroQueueSize; const ConsumerConfiguration config_; + DeadLetterPolicy deadLetterPolicy_; const std::string subscription_; std::string originalSubscriptionName_; const bool isPersistent_; @@ -230,6 +243,10 @@ class ConsumerImpl : public ConsumerImplBase { MessageCryptoPtr msgCrypto_; const bool readCompacted_; + SynchronizedHashMap> possibleSendToDeadLetterTopicMessages_; + std::shared_ptr> deadLetterProducer_; + std::mutex createProducerLock_; + // Make the access to `lastDequedMessageId_` and `lastMessageIdInBroker_` thread safe mutable std::mutex mutexForMessageId_; MessageId lastDequedMessageId_{MessageId::earliest()}; @@ -335,6 +352,9 @@ class ConsumerImpl : public ConsumerImplBase { FRIEND_TEST(ConsumerTest, testPartitionedConsumerUnAckedMessageRedelivery); FRIEND_TEST(ConsumerTest, testMultiTopicsConsumerUnAckedMessageRedelivery); FRIEND_TEST(ConsumerTest, testBatchUnAckedMessageTracker); + FRIEND_TEST(DeadLetterQueueTest, testAutoSetDLQTopicName); + FRIEND_TEST(DeadLetterQueueTest, testSendDLQTriggerByAckTimeOutAndNeAck); + FRIEND_TEST(DeadLetterQueueTest, testWithoutConsumerReceiveImmediately); }; } /* namespace pulsar */ diff --git a/lib/DeadLetterPolicyBuilder.cc b/lib/DeadLetterPolicyBuilder.cc new file mode 100644 index 00000000..a98eb118 --- /dev/null +++ b/lib/DeadLetterPolicyBuilder.cc @@ -0,0 +1,54 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +#include "DeadLetterPolicyImpl.h" + +#include +#include + +using namespace pulsar; + +namespace pulsar { + +DeadLetterPolicyBuilder::DeadLetterPolicyBuilder() : impl_(std::make_shared()) {} + +DeadLetterPolicyBuilder& DeadLetterPolicyBuilder::deadLetterTopic(const std::string& deadLetterTopic) { + impl_->deadLetterTopic = deadLetterTopic; + return *this; +} + +DeadLetterPolicyBuilder& DeadLetterPolicyBuilder::maxRedeliverCount(int maxRedeliverCount) { + impl_->maxRedeliverCount = maxRedeliverCount; + return *this; +} + +DeadLetterPolicyBuilder& DeadLetterPolicyBuilder::initialSubscriptionName( + const std::string& initialSubscriptionName) { + impl_->initialSubscriptionName = initialSubscriptionName; + return *this; +} + +DeadLetterPolicy DeadLetterPolicyBuilder::build() { + if (impl_->maxRedeliverCount <= 0) { + throw std::invalid_argument( "maxRedeliverCount must be > 0."); + } + return DeadLetterPolicy(impl_); +} + +} // namespace pulsar diff --git a/lib/DeadLetterPolicyImpl.cc b/lib/DeadLetterPolicyImpl.cc new file mode 100644 index 00000000..cd172671 --- /dev/null +++ b/lib/DeadLetterPolicyImpl.cc @@ -0,0 +1,38 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +#include "DeadLetterPolicyImpl.h" + +#include + +using namespace pulsar; + +namespace pulsar { + +DeadLetterPolicy::DeadLetterPolicy() : impl_(std::make_shared()) {} + +std::string DeadLetterPolicy::getDeadLetterTopic() const { return impl_->deadLetterTopic; } + +int DeadLetterPolicy::getMaxRedeliverCount() const { return impl_->maxRedeliverCount; } + +std::string DeadLetterPolicy::getInitialSubscriptionName() const { return impl_->initialSubscriptionName; } + +DeadLetterPolicy::DeadLetterPolicy(const DeadLetterPolicy::DeadLetterPolicyImplPtr& impl) : impl_(impl) {} + +} // namespace pulsar diff --git a/lib/DeadLetterPolicyImpl.h b/lib/DeadLetterPolicyImpl.h new file mode 100644 index 00000000..288f80fb --- /dev/null +++ b/lib/DeadLetterPolicyImpl.h @@ -0,0 +1,31 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +#pragma once + +#include "string" + +namespace pulsar { + +struct DeadLetterPolicyImpl { + std::string deadLetterTopic; + int maxRedeliverCount{INT_MAX}; + std::string initialSubscriptionName; +}; + +} // namespace pulsar diff --git a/lib/HTTPLookupService.cc b/lib/HTTPLookupService.cc index 8167b641..b27aec65 100644 --- a/lib/HTTPLookupService.cc +++ b/lib/HTTPLookupService.cc @@ -143,6 +143,10 @@ Future HTTPLookupService::getTopicsOfNamespaceAsync( return promise.getFuture(); } +Future HTTPLookupService::getSchema(const TopicNamePtr& topicName) { + return Promise().getFuture(); +} + static size_t curlWriteCallback(void *contents, size_t size, size_t nmemb, void *responseDataPtr) { ((std::string *)responseDataPtr)->append((char *)contents, size * nmemb); return size * nmemb; diff --git a/lib/HTTPLookupService.h b/lib/HTTPLookupService.h index 929d7ab1..02f0d180 100644 --- a/lib/HTTPLookupService.h +++ b/lib/HTTPLookupService.h @@ -72,6 +72,8 @@ class HTTPLookupService : public LookupService, public std::enable_shared_from_t Future getPartitionMetadataAsync(const TopicNamePtr&) override; + Future getSchema(const TopicNamePtr& topicName) override; + Future getTopicsOfNamespaceAsync(const NamespaceNamePtr& nsName) override; }; } // namespace pulsar diff --git a/lib/LookupService.h b/lib/LookupService.h index 6af290ce..98f4a7f7 100644 --- a/lib/LookupService.h +++ b/lib/LookupService.h @@ -72,6 +72,14 @@ class LookupService { */ virtual Future getTopicsOfNamespaceAsync(const NamespaceNamePtr& nsName) = 0; + /** + * returns current SchemaInfo {@link SchemaInfo} for a given topic. + * + * @param topicName topic-name + * @return SchemaInfo + */ + virtual Future getSchema(const TopicNamePtr& topicName) = 0; + virtual ~LookupService() {} }; diff --git a/lib/MessageId.cc b/lib/MessageId.cc index 9a1a38c8..29f5def7 100644 --- a/lib/MessageId.cc +++ b/lib/MessageId.cc @@ -19,7 +19,7 @@ #include #include - +#include #include #include #include @@ -28,6 +28,33 @@ #include "MessageIdImpl.h" #include "PulsarApi.pb.h" +namespace std { + +template <> +struct hash +{ + std::size_t operator()(const pulsar::MessageId& msgId) const + { + using boost::hash_value; + using boost::hash_combine; + + // Start with a hash value of 0 . + std::size_t seed = 0; + + // Modify 'seed' by XORing and bit-shifting in + // one member of 'Key' after the other: + hash_combine(seed,hash_value(msgId.ledgerId())); + hash_combine(seed,hash_value(msgId.entryId())); + hash_combine(seed,hash_value(msgId.batchIndex())); + hash_combine(seed,hash_value(msgId.partition())); + + // Return the result. + return seed; + } +}; + +} // namespace std + namespace pulsar { MessageId::MessageId() { diff --git a/lib/MultiTopicsConsumerImpl.cc b/lib/MultiTopicsConsumerImpl.cc index b8d55b49..9156b57c 100644 --- a/lib/MultiTopicsConsumerImpl.cc +++ b/lib/MultiTopicsConsumerImpl.cc @@ -770,10 +770,23 @@ void MultiTopicsConsumerImpl::redeliverUnacknowledgedMessages(const std::setredeliverUnacknowledgedMessages(messageIds); - }); + + LOG_INFO("Sending RedeliverUnacknowledgedMessages command for partitioned consumer."); + std::unordered_map> topicToMessageId; + for (const MessageId& messageId : messageIds) { + auto topicName = messageId.getTopicName(); + LOG_INFO("multi consumer: " << topicName) + topicToMessageId[topicName].emplace(messageId); + } + + for (const auto& kv : topicToMessageId) { + auto optConsumer = consumers_.find(kv.first); + if (optConsumer.is_present()) { + optConsumer.value()->redeliverUnacknowledgedMessages(kv.second); + } else { + LOG_ERROR("Message of topic: " << kv.first << " not in consumers"); + } + } } int MultiTopicsConsumerImpl::getNumOfPrefetchedMessages() const { return incomingMessages_.size(); } diff --git a/lib/MultiTopicsConsumerImpl.h b/lib/MultiTopicsConsumerImpl.h index da42b748..a3082c8c 100644 --- a/lib/MultiTopicsConsumerImpl.h +++ b/lib/MultiTopicsConsumerImpl.h @@ -167,6 +167,7 @@ class MultiTopicsConsumerImpl : public ConsumerImplBase { FRIEND_TEST(ConsumerTest, testMultiTopicsConsumerUnAckedMessageRedelivery); FRIEND_TEST(ConsumerTest, testPartitionedConsumerUnAckedMessageRedelivery); + FRIEND_TEST(DeadLetterQueueTest, testSendDLQTriggerByAckTimeOutAndNeAck); }; typedef std::shared_ptr MultiTopicsConsumerImplPtr; diff --git a/lib/NegativeAcksTracker.cc b/lib/NegativeAcksTracker.cc index 6ff322df..9dcca20f 100644 --- a/lib/NegativeAcksTracker.cc +++ b/lib/NegativeAcksTracker.cc @@ -80,7 +80,7 @@ void NegativeAcksTracker::handleTimer(const boost::system::error_code &ec) { } if (!messagesToRedeliver.empty()) { - consumer_.redeliverMessages(messagesToRedeliver); + consumer_.redeliverUnacknowledgedMessages(messagesToRedeliver); } scheduleTimer(); } diff --git a/lib/ProducerConfiguration.cc b/lib/ProducerConfiguration.cc index 9b3fdfbd..75e0f232 100644 --- a/lib/ProducerConfiguration.cc +++ b/lib/ProducerConfiguration.cc @@ -16,6 +16,8 @@ * specific language governing permissions and limitations * under the License. */ +#include "pulsar/ProducerConfiguration.h" + #include #include "ProducerConfigurationImpl.h" @@ -266,4 +268,14 @@ ProducerConfiguration::ProducerAccessMode ProducerConfiguration::getAccessMode() return impl_->accessMode; } +ProducerConfiguration& ProducerConfiguration::setInitialSubscriptionName( + const std::string& initialSubscriptionName) { + impl_->initialSubscriptionName = initialSubscriptionName; + return *this; +} + +const std::string& ProducerConfiguration::getInitialSubscriptionName() const { + return impl_->initialSubscriptionName; +} + } // namespace pulsar diff --git a/lib/ProducerConfigurationImpl.h b/lib/ProducerConfigurationImpl.h index 6c2b19da..ddf1816f 100644 --- a/lib/ProducerConfigurationImpl.h +++ b/lib/ProducerConfigurationImpl.h @@ -51,6 +51,7 @@ struct ProducerConfigurationImpl { std::map properties; bool chunkingEnabled{false}; ProducerConfiguration::ProducerAccessMode accessMode{ProducerConfiguration::Shared}; + std::string initialSubscriptionName; }; } // namespace pulsar diff --git a/lib/ProducerImpl.cc b/lib/ProducerImpl.cc index 7fa3ff29..50efd949 100644 --- a/lib/ProducerImpl.cc +++ b/lib/ProducerImpl.cc @@ -151,7 +151,8 @@ void ProducerImpl::connectionOpened(const ClientConnectionPtr& cnx) { SharedBuffer cmd = Commands::newProducer( topic_, producerId_, producerName_, requestId, conf_.getProperties(), conf_.getSchema(), epoch_, userProvidedProducerName_, conf_.isEncryptionEnabled(), - static_cast(conf_.getAccessMode()), topicEpoch); + static_cast(conf_.getAccessMode()), topicEpoch, + conf_.getInitialSubscriptionName()); cnx->sendRequestWithId(cmd, requestId) .addListener(std::bind(&ProducerImpl::handleCreateProducer, shared_from_this(), cnx, std::placeholders::_1, std::placeholders::_2)); diff --git a/lib/RetryableLookupService.h b/lib/RetryableLookupService.h index 7d704ec5..2614b257 100644 --- a/lib/RetryableLookupService.h +++ b/lib/RetryableLookupService.h @@ -66,6 +66,10 @@ class RetryableLookupService : public LookupService, [this, nsName] { return lookupService_->getTopicsOfNamespaceAsync(nsName); }); } + Future getSchema(const TopicNamePtr& topicName) override { + return Promise().getFuture(); + } + template Future executeAsync(const std::string& key, std::function()> f) { Promise promise; diff --git a/lib/SynchronizedHashMap.h b/lib/SynchronizedHashMap.h index b8a8c91e..dbb2eb0d 100644 --- a/lib/SynchronizedHashMap.h +++ b/lib/SynchronizedHashMap.h @@ -37,7 +37,7 @@ class SynchronizedHashMap { public: using OptValue = Optional; using PairVector = std::vector>; - using MapType = std::unordered_map; + using MapType = std::map; using Iterator = typename MapType::iterator; SynchronizedHashMap() = default; diff --git a/tests/ConsumerConfigurationTest.cc b/tests/ConsumerConfigurationTest.cc index fde87364..b9b7f091 100644 --- a/tests/ConsumerConfigurationTest.cc +++ b/tests/ConsumerConfigurationTest.cc @@ -26,6 +26,7 @@ DECLARE_LOG_OBJECT() #include "../lib/Future.h" #include "../lib/Utils.h" +#include "pulsar/DeadLetterPolicyBuilder.h" using namespace pulsar; @@ -317,3 +318,16 @@ TEST(ConsumerConfigurationTest, testResetAckTimeOut) { config.setUnAckedMessagesTimeoutMs(0); ASSERT_EQ(0, config.getUnAckedMessagesTimeoutMs()); } + + +TEST(ConsumerConfigurationTest, testDeadLetterPolicy) { + ConsumerConfiguration config; + auto dlqPolicy = config.getDeadLetterPolicy(); + ASSERT_TRUE(dlqPolicy.getDeadLetterTopic().empty()); + ASSERT_EQ(dlqPolicy.getMaxRedeliverCount(), -1); + ASSERT_TRUE(dlqPolicy.getInitialSubscriptionName().empty()); + + config.setDeadLetterPolicy(DeadLetterPolicyBuilder().maxRedeliverCount(10).build()); + auto dlqPolicy2 = config.getDeadLetterPolicy(); + ASSERT_EQ(dlqPolicy2.getMaxRedeliverCount(), 10); +} diff --git a/tests/ConsumerTest.cc b/tests/ConsumerTest.cc index f3c8abbe..5272ff62 100644 --- a/tests/ConsumerTest.cc +++ b/tests/ConsumerTest.cc @@ -38,6 +38,7 @@ #include "lib/UnAckedMessageTrackerEnabled.h" #include "lib/Utils.h" #include "lib/stats/ProducerStatsImpl.h" +#include "pulsar/DeadLetterPolicyBuilder.h" static const std::string lookupUrl = "pulsar://localhost:6650"; static const std::string adminUrl = "http://localhost:8080/"; @@ -958,6 +959,70 @@ TEST_P(ConsumerSeekTest, testSeekForMessageId) { producer.close(); } +// batch(true, falsee) multiConsumer(true, false) +// trigger: Nack , ackTimueOut, 这两种混搭? +// isInitSubscribeption +// auto set dlqTopicName? +// 堆积多个消息时, 并发处理. 有一部分到达送达dlq, 有一部分没有到达. +// 断言properties? +// 其他呢??? +TEST(ConsumerTest, testSendDLQTriggerByNegativeAcknowledge) { + Client client(lookupUrl); + const std::string topic = "testSendDLQTriggerByNegativeAcknowledge-" + std::to_string(time(nullptr)); + const std::string subName = "dlq-sub"; + const std::string dlqTopic = topic + subName + "DLQ"; + + + bool multiConsumer = false; + if (multiConsumer) { + // call admin api to make it partitioned + std::string url = adminUrl + "admin/v2/persistent/public/default/" + topic + "/partitions"; + int res = makePutRequest(url, "5"); + LOG_INFO("res = " << res); + ASSERT_FALSE(res != 204 && res != 409); + } + + auto dlqPolicy = DeadLetterPolicyBuilder() + .maxRedeliverCount(3) + .initialSubscriptionName("init-sub") + .deadLetterTopic(dlqTopic) + .build(); + ConsumerConfiguration consumerConfig; + consumerConfig.setDeadLetterPolicy(dlqPolicy); + consumerConfig.setNegativeAckRedeliveryDelayMs(100); + consumerConfig.setConsumerType(ConsumerType::ConsumerShared); + Consumer consumer; + ASSERT_EQ(ResultOk, client.subscribe(topic, subName, consumerConfig, consumer)); + + ProducerConfiguration producerConfiguration; + producerConfiguration.setBatchingEnabled(false); + Producer producer; + ASSERT_EQ(ResultOk, client.createProducer(topic, producerConfiguration, producer)); + + const int num = 50; + Message msg; + for (int i = 0; i < num; ++i) { + msg = MessageBuilder().setContent(std::to_string(i)).build(); + ASSERT_EQ(ResultOk, producer.send(msg)); + } + + // Each message nack 3 times, a total of 15 times + for (int i = 0; i < dlqPolicy.getMaxRedeliverCount() * num; ++i) { + ASSERT_EQ(ResultOk, consumer.receive(msg)); + consumer.negativeAcknowledge(msg); + } + + Consumer deadLetterQueueConsumer; + ASSERT_EQ(ResultOk, client.subscribe(topic, dlqTopic, deadLetterQueueConsumer)); + for (int i = 0; i < num; i++) { + ASSERT_EQ(ResultOk, consumer.receive(msg)); + ASSERT_EQ(msg.getDataAsString(), std::to_string(i)); + ASSERT_EQ(msg.getRedeliveryCount(), dlqPolicy.getMaxRedeliverCount()); + } + + client.close(); +} + INSTANTIATE_TEST_CASE_P(Pulsar, ConsumerSeekTest, ::testing::Values(true, false)); } // namespace pulsar diff --git a/tests/DeadLetterPolicyTest.cc b/tests/DeadLetterPolicyTest.cc new file mode 100644 index 00000000..f58e84da --- /dev/null +++ b/tests/DeadLetterPolicyTest.cc @@ -0,0 +1,43 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +#include +#include + +using namespace pulsar; + +TEST(DeadLetterPolicy, testDeadLetterPolicy) { + + // test default value. + DeadLetterPolicy deadLetterPolicy; + ASSERT_EQ(deadLetterPolicy.getMaxRedeliverCount(), INT_MAX); + ASSERT_TRUE(deadLetterPolicy.getDeadLetterTopic().empty()); + ASSERT_TRUE(deadLetterPolicy.getInitialSubscriptionName().empty()); + + // test don't allowed max redeliver count less than 0. + ASSERT_THROW(DeadLetterPolicyBuilder().maxRedeliverCount(-1).build(), std::invalid_argument); + + // test create DeadLetterPolicy by builder. + deadLetterPolicy = DeadLetterPolicyBuilder().maxRedeliverCount(10) + .deadLetterTopic("topic-subscription-DLQ") + .initialSubscriptionName("init-DLQ-subscription") + .build(); + ASSERT_EQ(deadLetterPolicy.getMaxRedeliverCount(), 10); + ASSERT_EQ(deadLetterPolicy.getDeadLetterTopic(), "topic-subscription-DLQ"); + ASSERT_EQ(deadLetterPolicy.getInitialSubscriptionName(), "init-DLQ-subscription"); +} diff --git a/tests/DeadLetterQueueTest.cc b/tests/DeadLetterQueueTest.cc new file mode 100644 index 00000000..0aba4fef --- /dev/null +++ b/tests/DeadLetterQueueTest.cc @@ -0,0 +1,405 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +#include +#include + +#include +#include +#include +#include +#include +#include + +#include "HttpHelper.h" +#include "PulsarFriend.h" +#include "lib/ClientConnection.h" +#include "lib/Future.h" +#include "lib/LogUtils.h" +#include "lib/MessageIdUtil.h" +#include "lib/MultiTopicsConsumerImpl.h" +#include "lib/TimeUtils.h" +#include "lib/UnAckedMessageTrackerDisabled.h" +#include "lib/UnAckedMessageTrackerEnabled.h" +#include "lib/Utils.h" +#include "lib/stats/ProducerStatsImpl.h" +#include "pulsar/DeadLetterPolicyBuilder.h" + +static const std::string lookupUrl = "pulsar://localhost:6650"; +static const std::string adminUrl = "http://localhost:8080/"; + +DECLARE_LOG_OBJECT() + +namespace pulsar { + +// Because isSchemaValidationEnforced config defaults to false. +// Therefore, in scenarios where AUTO_PUBLISH schemas are not supported now, the unit test can pass. +// When implementing AUTO_PUBLISH schema, we should set isSchemaValidationEnforced to true to revalidate. +TEST(DeadLetterQueueTest, testAutoSchema) { + Client client(lookupUrl); + const std::string topic = "testAutoSchema-" + std::to_string(time(nullptr)); + const std::string subName = "dlq-sub"; + + static const std::string jsonSchema = + R"({"type":"record","name":"cpx","fields":[{"name":"re","type":"double"},{"name":"im","type":"double"}]})"; + SchemaInfo schemaInfo(AUTO_PUBLISH, "test-json", jsonSchema); + + auto dlqPolicy = DeadLetterPolicyBuilder() + .maxRedeliverCount(3) + .deadLetterTopic(topic + subName + "-DLQ") + .initialSubscriptionName("init-sub") + .build(); + ConsumerConfiguration consumerConfig; + consumerConfig.setDeadLetterPolicy(dlqPolicy); + consumerConfig.setNegativeAckRedeliveryDelayMs(100); + consumerConfig.setConsumerType(ConsumerType::ConsumerShared); + consumerConfig.setSchema(schemaInfo); + Consumer consumer; + ASSERT_EQ(ResultOk, client.subscribe(topic, subName, consumerConfig, consumer)); + + // Initialize the DLQ subscription first and make sure that DLQ topic is created and a schema exists. + ConsumerConfiguration dlqConsumerConfig; + dlqConsumerConfig.setConsumerType(ConsumerType::ConsumerShared); + dlqConsumerConfig.setSchema(schemaInfo); + Consumer deadLetterConsumer; + ASSERT_EQ(ResultOk, client.subscribe(dlqPolicy.getDeadLetterTopic(), subName, dlqConsumerConfig, + deadLetterConsumer)); + + Producer producer; + ProducerConfiguration producerConfig; + producerConfig.setSchema(schemaInfo); + ASSERT_EQ(ResultOk, client.createProducer(topic, producerConfig, producer)); + std::string data = "{\"re\":2.1,\"im\":1.23}"; + const int num = 1; + for (int i = 0; i < num; ++i) { + ASSERT_EQ(ResultOk, producer.send(MessageBuilder().setContent(data).build())); + } + + // nack all msg. + Message msg; + for (int i = 0; i < dlqPolicy.getMaxRedeliverCount() * num + num; ++i) { + ASSERT_EQ(ResultOk, consumer.receive(msg)); + consumer.negativeAcknowledge(msg); + } + + // assert dlq msg. + for (int i = 0; i < num; i++) { + ASSERT_EQ(ResultOk, deadLetterConsumer.receive(msg, 5000)); + ASSERT_TRUE(!msg.getDataAsString().empty()); + ASSERT_TRUE(msg.getProperty(SYSTEM_PROPERTY_REAL_TOPIC).find(topic)); + ASSERT_TRUE(!msg.getProperty(PROPERTY_ORIGIN_MESSAGE_ID).empty()); + } + ASSERT_EQ(ResultTimeout, deadLetterConsumer.receive(msg, 200)); + + client.close(); +} + +// If the user never receives this message, the message should not be delivered to the DLQ. +TEST(DeadLetterQueueTest, testWithoutConsumerReceiveImmediately) { + Client client(lookupUrl); + const std::string topic = "testWithoutConsumerReceiveImmediately-" + std::to_string(time(nullptr)); + const std::string subName = "dlq-sub"; + auto dlqPolicy = + DeadLetterPolicyBuilder().maxRedeliverCount(3).initialSubscriptionName("init-sub").build(); + ConsumerConfiguration consumerConfig; + consumerConfig.setDeadLetterPolicy(dlqPolicy); + consumerConfig.setNegativeAckRedeliveryDelayMs(100); + consumerConfig.setConsumerType(ConsumerType::ConsumerShared); + Consumer consumer; + ASSERT_EQ(ResultOk, client.subscribe(topic, subName, consumerConfig, consumer)); + + // set ack timeout is 10 ms. + auto &consumerImpl = PulsarFriend::getConsumerImpl(consumer); + consumerImpl.unAckedMessageTrackerPtr_.reset( + new UnAckedMessageTrackerEnabled(10, PulsarFriend::getClientImplPtr(client), consumerImpl)); + + Producer producer; + ASSERT_EQ(ResultOk, client.createProducer(topic, producer)); + producer.send(MessageBuilder().setContent("msg").build()); + + // Wait a while, message should not be send to DLQ + sleep(2); + + Message msg; + ASSERT_EQ(ResultOk, consumer.receive(msg)); + client.close(); +} + +TEST(DeadLetterQueueTest, testAutoSetDLQTopicName) { + Client client(lookupUrl); + const std::string topic = "testAutoSetDLQName-" + std::to_string(time(nullptr)); + const std::string subName = "dlq-sub"; + const std::string dlqTopic = "persistent://public/default/" + topic + "-" + subName + "-DLQ"; + auto dlqPolicy = + DeadLetterPolicyBuilder().maxRedeliverCount(3).initialSubscriptionName("init-sub").build(); + ConsumerConfiguration consumerConfig; + consumerConfig.setDeadLetterPolicy(dlqPolicy); + consumerConfig.setNegativeAckRedeliveryDelayMs(100); + consumerConfig.setConsumerType(ConsumerType::ConsumerShared); + Consumer consumer; + ASSERT_EQ(ResultOk, client.subscribe(topic, subName, consumerConfig, consumer)); + + auto &consumerImpl = PulsarFriend::getConsumerImpl(consumer); + ASSERT_EQ(consumerImpl.deadLetterPolicy_.getDeadLetterTopic(), dlqTopic); + + client.close(); +} + +class DeadLetterQueueTest : public ::testing::TestWithParam> { + public: + void SetUp() override { + isProducerBatch_ = std::get<0>(GetParam()); + isMultiConsumer_ = std::get<1>(GetParam()); + consumerType_ = std::get<2>(GetParam()); + producerConf_ = ProducerConfiguration().setBatchingEnabled(isProducerBatch_); + } + + void TearDown() override { client_.close(); } + + void initTopic(std::string topicName) { + if (isMultiConsumer_) { + // call admin api to make it partitioned + std::string url = adminUrl + "admin/v2/persistent/public/default/" + topicName + "/partitions"; + int res = makePutRequest(url, "5"); + LOG_INFO("res = " << res); + ASSERT_FALSE(res != 204 && res != 409); + } + } + + protected: + Client client_{lookupUrl}; + ProducerConfiguration producerConf_; + bool isMultiConsumer_; + bool isProducerBatch_; + ConsumerType consumerType_; +}; + +TEST_P(DeadLetterQueueTest, testSendDLQTriggerByAckTimeOutAndNeAck) { + Client client(lookupUrl); + const std::string topic = "testSendDLQTriggerByAckTimeOut-" + std::to_string(time(nullptr)) + + std::to_string(isMultiConsumer_) + std::to_string(isProducerBatch_) + + std::to_string(consumerType_); + const std::string subName = "dlq-sub"; + const std::string dlqTopic = topic + "-" + subName + "-DLQ"; + initTopic(topic); + + auto dlqPolicy = DeadLetterPolicyBuilder().maxRedeliverCount(3).deadLetterTopic(dlqTopic).build(); + ConsumerConfiguration consumerConfig; + consumerConfig.setDeadLetterPolicy(dlqPolicy); + consumerConfig.setNegativeAckRedeliveryDelayMs(100); + consumerConfig.setConsumerType(consumerType_); + Consumer consumer; + ASSERT_EQ(ResultOk, client.subscribe(topic, subName, consumerConfig, consumer)); + + // Reset the unack timeout + long unackTimeOut = 200; + if (isMultiConsumer_) { + auto multiConsumer = PulsarFriend::getMultiTopicsConsumerImplPtr(consumer); + multiConsumer->unAckedMessageTrackerPtr_.reset(new UnAckedMessageTrackerEnabled( + unackTimeOut, PulsarFriend::getClientImplPtr(client), *multiConsumer)); + multiConsumer->consumers_.forEachValue([&client, unackTimeOut](ConsumerImplPtr consumer) { + consumer->unAckedMessageTrackerPtr_.reset(new UnAckedMessageTrackerEnabled( + unackTimeOut, PulsarFriend::getClientImplPtr(client), *consumer)); + }); + } else { + auto &consumerImpl = PulsarFriend::getConsumerImpl(consumer); + consumerImpl.unAckedMessageTrackerPtr_.reset(new UnAckedMessageTrackerEnabled( + unackTimeOut, PulsarFriend::getClientImplPtr(client), consumerImpl)); + } + + Producer producer; + ASSERT_EQ(ResultOk, client.createProducer(topic, producerConf_, producer)); + const int num = 100; + Message msg; + for (int i = 0; i < num; ++i) { + msg = MessageBuilder() + .setContent(std::to_string(i)) + .setPartitionKey("p-key") + .setOrderingKey("o-key") + .setProperty("pk-1", "pv-1") + .build(); + producer.sendAsync(msg, [](Result res, const MessageId &msgId) { ASSERT_EQ(res, ResultOk); }); + } + + // receive messages and don't ack. + for (int i = 0; i < dlqPolicy.getMaxRedeliverCount() * num + num; ++i) { + ASSERT_EQ(ResultOk, consumer.receive(msg)); + // Randomly specify some messages manually negativeAcknowledge. + if (rand() % 2 == 0) { + consumer.negativeAcknowledge(msg); + } + } + + // assert dlq msg. + Consumer deadLetterQueueConsumer; + ConsumerConfiguration dlqConsumerConfig; + dlqConsumerConfig.setSubscriptionInitialPosition(InitialPositionEarliest); + ASSERT_EQ(ResultOk, client.subscribe(dlqTopic, "dlq-sub", dlqConsumerConfig, deadLetterQueueConsumer)); + for (int i = 0; i < num; i++) { + ASSERT_EQ(ResultOk, deadLetterQueueConsumer.receive(msg)); + ASSERT_TRUE(!msg.getDataAsString().empty()); + ASSERT_EQ(msg.getPartitionKey(), "p-key"); + ASSERT_EQ(msg.getOrderingKey(), "o-key"); + ASSERT_EQ(msg.getProperty("pk-1"), "pv-1"); + ASSERT_TRUE(msg.getProperty(SYSTEM_PROPERTY_REAL_TOPIC).find(topic)); + ASSERT_TRUE(!msg.getProperty(PROPERTY_ORIGIN_MESSAGE_ID).empty()); + } + + ASSERT_EQ(ResultTimeout, deadLetterQueueConsumer.receive(msg, 200)); +} + +TEST_P(DeadLetterQueueTest, testSendDLQTriggerByNegativeAcknowledge) { + Client client(lookupUrl); + const std::string topic = "testSendDLQTriggerByNegativeAcknowledge-" + std::to_string(time(nullptr)) + + std::to_string(isMultiConsumer_) + std::to_string(isProducerBatch_) + + std::to_string(consumerType_); + const std::string subName = "dlq-sub"; + const std::string dlqTopic = topic + subName + "DLQ"; + initTopic(topic); + + auto dlqPolicy = DeadLetterPolicyBuilder().maxRedeliverCount(3).deadLetterTopic(dlqTopic).build(); + ConsumerConfiguration consumerConfig; + consumerConfig.setDeadLetterPolicy(dlqPolicy); + consumerConfig.setNegativeAckRedeliveryDelayMs(100); + consumerConfig.setConsumerType(consumerType_); + Consumer consumer; + ASSERT_EQ(ResultOk, client.subscribe(topic, subName, consumerConfig, consumer)); + + Producer producer; + ASSERT_EQ(ResultOk, client.createProducer(topic, producerConf_, producer)); + + const int num = 10; + Message msg; + for (int i = 0; i < num; ++i) { + msg = MessageBuilder() + .setContent(std::to_string(i)) + .setPartitionKey("p-key") + .setOrderingKey("o-key") + .setProperty("pk-1", "pv-1") + .build(); + producer.sendAsync(msg, [](Result res, const MessageId &msgId) { ASSERT_EQ(res, ResultOk); }); + } + + // nack all msg. + for (int i = 0; i < dlqPolicy.getMaxRedeliverCount() * num + num; ++i) { + ASSERT_EQ(ResultOk, consumer.receive(msg)); + consumer.negativeAcknowledge(msg); + } + + // assert dlq msg. + Consumer deadLetterQueueConsumer; + ConsumerConfiguration dlqConsumerConfig; + dlqConsumerConfig.setSubscriptionInitialPosition(InitialPositionEarliest); + ASSERT_EQ(ResultOk, client.subscribe(dlqTopic, "dlq-sub", dlqConsumerConfig, deadLetterQueueConsumer)); + for (int i = 0; i < num; i++) { + ASSERT_EQ(ResultOk, deadLetterQueueConsumer.receive(msg)); + ASSERT_TRUE(!msg.getDataAsString().empty()); + ASSERT_EQ(msg.getPartitionKey(), "p-key"); + ASSERT_EQ(msg.getOrderingKey(), "o-key"); + ASSERT_EQ(msg.getProperty("pk-1"), "pv-1"); + ASSERT_TRUE(msg.getProperty(SYSTEM_PROPERTY_REAL_TOPIC).find(topic)); + ASSERT_TRUE(!msg.getProperty(PROPERTY_ORIGIN_MESSAGE_ID).empty()); + } + ASSERT_EQ(ResultTimeout, deadLetterQueueConsumer.receive(msg, 200)); +} + +TEST_P(DeadLetterQueueTest, testInitSubscription) { + Client client(lookupUrl); + const std::string topic = "testInitSubscription-" + std::to_string(time(nullptr)) + + std::to_string(isMultiConsumer_) + std::to_string(isProducerBatch_) + + std::to_string(consumerType_); + const std::string subName = "dlq-sub"; + const std::string dlqTopic = topic + subName + "DLQ"; + const std::string dlqInitSub = "dlq-init-sub"; + initTopic(topic); + + auto dlqPolicy = DeadLetterPolicyBuilder() + .maxRedeliverCount(3) + .initialSubscriptionName(dlqInitSub) + .deadLetterTopic(dlqTopic) + .build(); + ConsumerConfiguration consumerConfig; + consumerConfig.setDeadLetterPolicy(dlqPolicy); + consumerConfig.setNegativeAckRedeliveryDelayMs(100); + consumerConfig.setConsumerType(consumerType_); + Consumer consumer; + ASSERT_EQ(ResultOk, client.subscribe(topic, subName, consumerConfig, consumer)); + + Consumer deadLetterQueueConsumer; + ConsumerConfiguration dlqConsumerConfig; + dlqConsumerConfig.setSubscriptionInitialPosition(InitialPositionEarliest); + ASSERT_EQ(ResultOk, client.subscribe(dlqTopic, "dlq-sub", dlqConsumerConfig, deadLetterQueueConsumer)); + + Producer producer; + ASSERT_EQ(ResultOk, client.createProducer(topic, producerConf_, producer)); + + const int num = 10; + Message msg; + for (int i = 0; i < num; ++i) { + msg = MessageBuilder().setContent(std::to_string(i)).build(); + ASSERT_EQ(ResultOk, producer.send(msg)); + } + + // nack all msg. + for (int i = 0; i < dlqPolicy.getMaxRedeliverCount() * num + num; ++i) { + ASSERT_EQ(ResultOk, consumer.receive(msg)); + consumer.negativeAcknowledge(msg); + } + + // Use this subscription to ensure that messages are sent to the DLQ. + for (int i = 0; i < num; i++) { + ASSERT_EQ(ResultOk, deadLetterQueueConsumer.receive(msg)); + ASSERT_TRUE(!msg.getDataAsString().empty()); + ASSERT_TRUE(msg.getProperty(SYSTEM_PROPERTY_REAL_TOPIC).find(topic)); + ASSERT_TRUE(!msg.getProperty(PROPERTY_ORIGIN_MESSAGE_ID).empty()); + } + + // If there is no initial subscription, then the subscription will not receive the DLQ messages sent + // before the subscription. + Consumer initDLQConsumer; + ConsumerConfiguration initDLQConsumerConfig; + dlqConsumerConfig.setSubscriptionInitialPosition(InitialPositionLatest); + ASSERT_EQ(ResultOk, client.subscribe(dlqTopic, dlqInitSub, initDLQConsumerConfig, initDLQConsumer)); + for (int i = 0; i < num; i++) { + ASSERT_EQ(ResultOk, initDLQConsumer.receive(msg, 1000)); + ASSERT_TRUE(!msg.getDataAsString().empty()); + ASSERT_TRUE(msg.getProperty(SYSTEM_PROPERTY_REAL_TOPIC).find(topic)); + ASSERT_TRUE(!msg.getProperty(PROPERTY_ORIGIN_MESSAGE_ID).empty()); + } + ASSERT_EQ(ResultTimeout, initDLQConsumer.receive(msg, 200)); +} + +bool isBatchs[2] = {true, false}; +bool isMultiTopics[2] = {true, false}; +ConsumerType subTypes[2] = {ConsumerType::ConsumerShared, ConsumerType::ConsumerKeyShared}; + +std::vector> getValues() { + std::vector> values; + for (const auto isBatch : isBatchs) { + for (const auto isMultiTopic : isMultiTopics) { + for (const auto subType : subTypes) { + values.emplace_back(std::make_tuple(isBatch, isMultiTopic, subType)); + } + } + } + return values; +} + +INSTANTIATE_TEST_CASE_P(Pulsar, DeadLetterQueueTest, ::testing::ValuesIn(getValues())); + +} // namespace pulsar diff --git a/tests/ProducerTest.cc b/tests/ProducerTest.cc index 74d3cf20..ee0c3bec 100644 --- a/tests/ProducerTest.cc +++ b/tests/ProducerTest.cc @@ -112,6 +112,7 @@ TEST(ProducerTest, testIsConnected) { const std::string partitionedTopic = "testProducerIsConnectedPartitioned-" + std::to_string(time(nullptr)); + Producer producer; ASSERT_FALSE(producer.isConnected()); // ProducerImpl