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/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 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 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/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/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/include/pulsar/ProducerConfiguration.h b/include/pulsar/ProducerConfiguration.h index 39ecbe09..a4fae26c 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(); @@ -527,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/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/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/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/ClientConnection.cc b/lib/ClientConnection.cc index b3df8310..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); @@ -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/Commands.cc b/lib/Commands.cc index 69492c6f..4ae11e2e 100644 --- a/lib/Commands.cc +++ b/lib/Commands.cc @@ -19,6 +19,7 @@ #include "Commands.h" #include +#include #include #include @@ -382,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(); @@ -396,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++) { @@ -807,7 +812,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 +831,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..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); @@ -132,7 +133,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/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/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 09667294..da2137d9 100644 --- a/lib/ConsumerImpl.cc +++ b/lib/ConsumerImpl.cc @@ -18,6 +18,9 @@ */ #include "ConsumerImpl.h" +#include +#include + #include #include "AckGroupingTracker.h" @@ -111,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(); } @@ -125,11 +143,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(); @@ -322,6 +342,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 +380,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 +411,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; @@ -459,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; @@ -474,8 +492,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(); @@ -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,9 +654,10 @@ 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); + Message msg = Commands::deSerializeSingleMessageInBatch(batchedMessage, i, batchSize); msg.impl_->setRedeliveryCount(redeliveryCount); msg.impl_->setTopicName(batchedMessage.getTopicName()); msg.impl_->convertPayloadToKeyValue(config_.getSchema()); @@ -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); } @@ -932,13 +968,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 @@ -1180,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) { @@ -1192,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()); } } @@ -1341,7 +1403,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; } @@ -1501,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 9ba65773..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; @@ -44,8 +45,8 @@ class BatchAcknowledgementTracker; class MessageCrypto; class GetLastMessageIdResponse; typedef std::shared_ptr MessageCryptoPtr; -typedef std::function BrokerGetLastMessageIdCallback; typedef std::shared_ptr BackoffPtr; +typedef std::function ProcessDLQCallBack; class AckGroupingTracker; using AckGroupingTrackerPtr = std::shared_ptr; @@ -65,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, @@ -124,6 +129,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 +145,6 @@ class ConsumerImpl : public ConsumerImplBase { virtual bool isReadCompacted(); virtual void hasMessageAvailableAsync(HasMessageAvailableCallback callback); - virtual void getLastMessageIdAsync(BrokerGetLastMessageIdCallback callback); void beforeConnectionChange(ClientConnection& cnx) override; protected: @@ -197,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_; @@ -231,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()}; @@ -307,6 +323,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 @@ -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/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/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/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/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/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/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/Message.cc b/lib/Message.cc index 84f203f7..0f28f7d9 100644 --- a/lib/Message.cc +++ b/lib/Message.cc @@ -18,6 +18,7 @@ */ #include #include +#include #include #include @@ -54,7 +55,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_() {} @@ -63,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..29f5def7 100644 --- a/lib/MessageId.cc +++ b/lib/MessageId.cc @@ -18,7 +18,8 @@ */ #include - +#include +#include #include #include #include @@ -27,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() { @@ -42,14 +70,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 +107,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 +118,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/MultiTopicsConsumerImpl.cc b/lib/MultiTopicsConsumerImpl.cc index d14c3cae..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(); } @@ -803,6 +816,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..a3082c8c 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; @@ -166,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/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/lib/NegativeAcksTracker.cc b/lib/NegativeAcksTracker.cc index 3ccf0bea..9dcca20f 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 { @@ -79,7 +80,7 @@ void NegativeAcksTracker::handleTimer(const boost::system::error_code &ec) { } if (!messagesToRedeliver.empty()) { - consumer_.redeliverMessages(messagesToRedeliver); + consumer_.redeliverUnacknowledgedMessages(messagesToRedeliver); } scheduleTimer(); } @@ -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/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/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 f3e61204..50efd949 100644 --- a/lib/ProducerImpl.cc +++ b/lib/ProducerImpl.cc @@ -18,6 +18,8 @@ */ #include "ProducerImpl.h" +#include + #include #include "BatchMessageContainer.h" @@ -149,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)); @@ -236,7 +239,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 +389,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: @@ -813,8 +827,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/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/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/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/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/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}>) 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 a77636ee..5272ff62 100644 --- a/tests/ConsumerTest.cc +++ b/tests/ConsumerTest.cc @@ -31,12 +31,14 @@ #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/"; @@ -683,8 +685,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()); @@ -841,6 +842,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; @@ -936,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/MessageChunkingTest.cc b/tests/MessageChunkingTest.cc index 8675886f..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); @@ -180,6 +182,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, 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/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/ProducerTest.cc b/tests/ProducerTest.cc index 77a79e1a..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 @@ -275,7 +276,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 +298,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); @@ -336,4 +374,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)); 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_); 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) { 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); +}