From 96bddec1f856abe5086913b10c4c30d1373a176d Mon Sep 17 00:00:00 2001 From: Adam Kotwasinski Date: Sun, 11 Dec 2022 00:12:34 -0800 Subject: [PATCH 1/4] kafka: upstream config Signed-off-by: Adam Kotwasinski --- .../filters/network/source/mesh/upstream_config.cc | 11 ++++++++++- .../filters/network/source/mesh/upstream_config.h | 7 ++++++- .../test/mesh/command_handlers/metadata_unit_test.cc | 2 +- .../network/test/mesh/upstream_config_unit_test.cc | 10 ++++++++-- .../test/mesh/upstream_kafka_facade_unit_test.cc | 12 ++++++++---- 5 files changed, 33 insertions(+), 9 deletions(-) diff --git a/contrib/kafka/filters/network/source/mesh/upstream_config.cc b/contrib/kafka/filters/network/source/mesh/upstream_config.cc index 8e6917df034c8..48b1fefe31b95 100644 --- a/contrib/kafka/filters/network/source/mesh/upstream_config.cc +++ b/contrib/kafka/filters/network/source/mesh/upstream_config.cc @@ -16,6 +16,8 @@ using KafkaClusterDefinition = envoy::extensions::filters::network::kafka_mesh::v3alpha::KafkaClusterDefinition; using ForwardingRule = envoy::extensions::filters::network::kafka_mesh::v3alpha::ForwardingRule; +const std::string DEFAULT_CONSUMER_GROUP_ID = "envoy"; + UpstreamKafkaConfigurationImpl::UpstreamKafkaConfigurationImpl(const KafkaMeshProtoConfig& config) : advertised_address_{config.advertised_host(), config.advertised_port()} { @@ -43,8 +45,15 @@ UpstreamKafkaConfigurationImpl::UpstreamKafkaConfigurationImpl(const KafkaMeshPr upstream_cluster_definition.producer_config().begin(), upstream_cluster_definition.producer_config().end()}; producer_configs["bootstrap.servers"] = upstream_cluster_definition.bootstrap_servers(); + + // TODO (adam.kotwasinski) This needs to read configs just like producer does. + std::map consumer_configs = {}; + consumer_configs["bootstrap.servers"] = upstream_cluster_definition.bootstrap_servers(); + // TODO (adam.kotwasinski) When configs are read, use this only if absent. + consumer_configs["group.id"] = DEFAULT_CONSUMER_GROUP_ID; + ClusterConfig cluster_config = {cluster_name, upstream_cluster_definition.partition_count(), - producer_configs}; + producer_configs, consumer_configs}; cluster_name_to_cluster_config[cluster_name] = cluster_config; } diff --git a/contrib/kafka/filters/network/source/mesh/upstream_config.h b/contrib/kafka/filters/network/source/mesh/upstream_config.h index ad49f2f1304d6..a8b14084ce9d7 100644 --- a/contrib/kafka/filters/network/source/mesh/upstream_config.h +++ b/contrib/kafka/filters/network/source/mesh/upstream_config.h @@ -39,9 +39,14 @@ struct ClusterConfig { // producer property. std::map upstream_producer_properties_; + // This map always contains entries with keys 'bootstrap.servers' and 'group.id', as these are the + // only mandatory consumer properties. + std::map upstream_consumer_properties_; + bool operator==(const ClusterConfig& rhs) const { return name_ == rhs.name_ && partition_count_ == rhs.partition_count_ && - upstream_producer_properties_ == rhs.upstream_producer_properties_; + upstream_producer_properties_ == rhs.upstream_producer_properties_ && + upstream_consumer_properties_ == rhs.upstream_consumer_properties_; } }; diff --git a/contrib/kafka/filters/network/test/mesh/command_handlers/metadata_unit_test.cc b/contrib/kafka/filters/network/test/mesh/command_handlers/metadata_unit_test.cc index ba144e3ce74c2..132946cd84019 100644 --- a/contrib/kafka/filters/network/test/mesh/command_handlers/metadata_unit_test.cc +++ b/contrib/kafka/filters/network/test/mesh/command_handlers/metadata_unit_test.cc @@ -33,7 +33,7 @@ TEST(MetadataTest, shouldBeAlwaysReadyForAnswer) { const std::pair advertised_address = {"host", 1234}; EXPECT_CALL(configuration, getAdvertisedAddress()).WillOnce(Return(advertised_address)); // First topic is going to have configuration present (42 partitions for each topic). - const ClusterConfig topic1config = {"", 42, {}}; + const ClusterConfig topic1config = {"", 42, {}, {}}; EXPECT_CALL(configuration, computeClusterConfigForTopic("topic1")) .WillOnce(Return(absl::make_optional(topic1config))); // Second topic is not going to have configuration present. diff --git a/contrib/kafka/filters/network/test/mesh/upstream_config_unit_test.cc b/contrib/kafka/filters/network/test/mesh/upstream_config_unit_test.cc index 23bfb039b9a03..3ecb0b5fd6792 100644 --- a/contrib/kafka/filters/network/test/mesh/upstream_config_unit_test.cc +++ b/contrib/kafka/filters/network/test/mesh/upstream_config_unit_test.cc @@ -108,8 +108,14 @@ advertised_port: 42 TestUtility::loadFromYamlAndValidate(yaml, proto_config); const UpstreamKafkaConfiguration& testee = UpstreamKafkaConfigurationImpl{proto_config}; - const ClusterConfig cluster1 = {"cluster1", 1, {{"bootstrap.servers", "s1"}}}; - const ClusterConfig cluster2 = {"cluster2", 2, {{"bootstrap.servers", "s2"}}}; + const ClusterConfig cluster1 = {"cluster1", + 1, + {{"bootstrap.servers", "s1"}}, + {{"bootstrap.servers", "s1"}, {"group.id", "envoy"}}}; + const ClusterConfig cluster2 = {"cluster2", + 2, + {{"bootstrap.servers", "s2"}}, + {{"bootstrap.servers", "s2"}, {"group.id", "envoy"}}}; // when, then (advertised address is returned properly) const auto address = testee.getAdvertisedAddress(); diff --git a/contrib/kafka/filters/network/test/mesh/upstream_kafka_facade_unit_test.cc b/contrib/kafka/filters/network/test/mesh/upstream_kafka_facade_unit_test.cc index d4a0497c21476..baa6c55baf538 100644 --- a/contrib/kafka/filters/network/test/mesh/upstream_kafka_facade_unit_test.cc +++ b/contrib/kafka/filters/network/test/mesh/upstream_kafka_facade_unit_test.cc @@ -36,7 +36,8 @@ TEST(UpstreamKafkaFacadeTest, shouldCreateProducerOnlyOnceForTheSameCluster) { const std::string topic2 = "topic2"; MockUpstreamKafkaConfiguration configuration; - const ClusterConfig cluster_config = {"cluster", 1, {{"bootstrap.servers", "localhost:9092"}}}; + const ClusterConfig cluster_config = { + "cluster", 1, {{"bootstrap.servers", "localhost:9092"}}, {}}; EXPECT_CALL(configuration, computeClusterConfigForTopic(topic1)).WillOnce(Return(cluster_config)); EXPECT_CALL(configuration, computeClusterConfigForTopic(topic2)).WillOnce(Return(cluster_config)); ThreadLocal::MockInstance slot_allocator; @@ -61,10 +62,12 @@ TEST(UpstreamKafkaFacadeTest, shouldCreateDifferentProducersForDifferentClusters MockUpstreamKafkaConfiguration configuration; // Notice it's the cluster name that matters, not the producer config. - const ClusterConfig cluster_config1 = {"cluster1", 1, {{"bootstrap.servers", "localhost:9092"}}}; + const ClusterConfig cluster_config1 = { + "cluster1", 1, {{"bootstrap.servers", "localhost:9092"}}, {}}; EXPECT_CALL(configuration, computeClusterConfigForTopic(topic1)) .WillOnce(Return(cluster_config1)); - const ClusterConfig cluster_config2 = {"cluster2", 1, {{"bootstrap.servers", "localhost:9092"}}}; + const ClusterConfig cluster_config2 = { + "cluster2", 1, {{"bootstrap.servers", "localhost:9092"}}, {}}; EXPECT_CALL(configuration, computeClusterConfigForTopic(topic2)) .WillOnce(Return(cluster_config2)); ThreadLocal::MockInstance slot_allocator; @@ -87,7 +90,8 @@ TEST(UpstreamKafkaFacadeTest, shouldThrowIfThereIsNoConfigurationForGivenTopic) const std::string topic = "topic1"; MockUpstreamKafkaConfiguration configuration; - const ClusterConfig cluster_config = {"cluster", 1, {{"bootstrap.servers", "localhost:9092"}}}; + const ClusterConfig cluster_config = { + "cluster", 1, {{"bootstrap.servers", "localhost:9092"}}, {}}; EXPECT_CALL(configuration, computeClusterConfigForTopic(topic)).WillOnce(Return(absl::nullopt)); ThreadLocal::MockInstance slot_allocator; EXPECT_CALL(slot_allocator, allocateSlot()) From 58ae1ce1122ee1db1e805ebe11b0a8f798ef60ae Mon Sep 17 00:00:00 2001 From: Adam Kotwasinski Date: Wed, 16 Feb 2022 12:15:16 -0800 Subject: [PATCH 2/4] kafka: shared consumer manager Signed-off-by: Adam Kotwasinski --- .../filters/network/source/kafka_types.h | 6 + .../kafka/filters/network/source/mesh/BUILD | 32 +++ .../source/mesh/shared_consumer_manager.h | 33 +++ .../mesh/shared_consumer_manager_impl.cc | 243 ++++++++++++++++++ .../mesh/shared_consumer_manager_impl.h | 116 +++++++++ .../source/mesh/upstream_kafka_consumer.h | 28 ++ contrib/kafka/filters/network/test/mesh/BUILD | 9 + .../shared_consumer_manager_impl_unit_test.cc | 25 ++ 8 files changed, 492 insertions(+) create mode 100644 contrib/kafka/filters/network/source/mesh/shared_consumer_manager.h create mode 100644 contrib/kafka/filters/network/source/mesh/shared_consumer_manager_impl.cc create mode 100644 contrib/kafka/filters/network/source/mesh/shared_consumer_manager_impl.h create mode 100644 contrib/kafka/filters/network/test/mesh/shared_consumer_manager_impl_unit_test.cc diff --git a/contrib/kafka/filters/network/source/kafka_types.h b/contrib/kafka/filters/network/source/kafka_types.h index d01c304984e4c..36ce3b430b83b 100644 --- a/contrib/kafka/filters/network/source/kafka_types.h +++ b/contrib/kafka/filters/network/source/kafka_types.h @@ -2,6 +2,7 @@ #include #include +#include #include #include "absl/types/optional.h" @@ -45,6 +46,11 @@ struct Uuid { bool operator==(const Uuid& rhs) const { return msb_ == rhs.msb_ && lsb_ == rhs.lsb_; }; }; +/** + * Kafka topic-partition pair. + */ +using KafkaPartition = std::pair; + } // namespace Kafka } // namespace NetworkFilters } // namespace Extensions diff --git a/contrib/kafka/filters/network/source/mesh/BUILD b/contrib/kafka/filters/network/source/mesh/BUILD index 71546a58bef01..3df2b411496e1 100644 --- a/contrib/kafka/filters/network/source/mesh/BUILD +++ b/contrib/kafka/filters/network/source/mesh/BUILD @@ -110,6 +110,38 @@ envoy_cc_library( ], ) +envoy_cc_library( + name = "shared_consumer_manager_lib", + srcs = [ + ], + hdrs = [ + "shared_consumer_manager.h", + ], + tags = ["skip_on_windows"], + deps = [ + ":upstream_kafka_consumer_lib", + "//source/common/common:minimal_logger_lib", + ], +) + +envoy_cc_library( + name = "shared_consumer_manager_impl_lib", + srcs = [ + "shared_consumer_manager_impl.cc", + ], + hdrs = [ + "shared_consumer_manager_impl.h", + ], + tags = ["skip_on_windows"], + deps = [ + ":shared_consumer_manager_lib", + ":upstream_config_lib", + ":upstream_kafka_consumer_impl_lib", + "//contrib/kafka/filters/network/source:kafka_types_lib", + "//source/common/common:minimal_logger_lib", + ], +) + envoy_cc_library( name = "inbound_record_lib", srcs = [ diff --git a/contrib/kafka/filters/network/source/mesh/shared_consumer_manager.h b/contrib/kafka/filters/network/source/mesh/shared_consumer_manager.h new file mode 100644 index 0000000000000..1873406adc201 --- /dev/null +++ b/contrib/kafka/filters/network/source/mesh/shared_consumer_manager.h @@ -0,0 +1,33 @@ +#pragma once + +#include +#include + +#include "contrib/kafka/filters/network/source/mesh/upstream_kafka_consumer.h" + +namespace Envoy { +namespace Extensions { +namespace NetworkFilters { +namespace Kafka { +namespace Mesh { + +/** + * Manages (raw) Kafka consumers pointing to upstream Kafka clusters. + * It is expected to have only one instance of this object in the runtime. + */ +class SharedConsumerManager { +public: + virtual ~SharedConsumerManager() = default; + + virtual void getRecordsOrRegisterCallback(const RecordCbSharedPtr& callback) PURE; + + virtual void removeCallback(const RecordCbSharedPtr& callback) PURE; +}; + +using SharedConsumerManagerSharedPtr = std::shared_ptr; + +} // namespace Mesh +} // namespace Kafka +} // namespace NetworkFilters +} // namespace Extensions +} // namespace Envoy diff --git a/contrib/kafka/filters/network/source/mesh/shared_consumer_manager_impl.cc b/contrib/kafka/filters/network/source/mesh/shared_consumer_manager_impl.cc new file mode 100644 index 0000000000000..751aa5c7ce4b2 --- /dev/null +++ b/contrib/kafka/filters/network/source/mesh/shared_consumer_manager_impl.cc @@ -0,0 +1,243 @@ +#include "contrib/kafka/filters/network/source/mesh/shared_consumer_manager_impl.h" + +#include + +#include "source/common/common/fmt.h" + +#include "contrib/kafka/filters/network/source/mesh/upstream_kafka_consumer_impl.h" + +namespace Envoy { +namespace Extensions { +namespace NetworkFilters { +namespace Kafka { +namespace Mesh { + +SharedConsumerManagerImpl::SharedConsumerManagerImpl( + const UpstreamKafkaConfiguration& configuration, Thread::ThreadFactory& thread_factory) + : configuration_{configuration}, thread_factory_{thread_factory} {} + +SharedConsumerManagerImpl::~SharedConsumerManagerImpl() { doShutdown(); } + +void SharedConsumerManagerImpl::getRecordsOrRegisterCallback(const RecordCbSharedPtr& callback) { + // For every fetch topic, figure out the upstream cluster, + // create consumer if needed ... + const TopicToPartitionsMap interest = callback->interest(); + for (const auto& fetch : interest) { + const std::string& topic = fetch.first; + getOrCreateConsumer(topic); // XXX this method should not be named 'get...' + } + + // ... and start processing. + store_->getRecordsOrRegisterCallback(callback); +} + +KafkaConsumer& SharedConsumerManagerImpl::getOrCreateConsumer(const std::string& topic) { + absl::MutexLock lock(&consumers_mutex_); + const auto it = topic_to_consumer_.find(topic); + // Return consumer already present or create new one and register it. + return (topic_to_consumer_.end() == it) ? registerNewConsumer(topic) : *(it->second); +} + +KafkaConsumer& SharedConsumerManagerImpl::registerNewConsumer(const std::string& topic) { + ENVOY_LOG(info, "Creating consumer for topic [{}]", topic); + // Compute which upstream cluster corresponds to the topic. + const absl::optional cluster_config = + configuration_.computeClusterConfigForTopic(topic); + if (!cluster_config) { + throw EnvoyException( + fmt::format("Could not compute upstream cluster configuration for topic [{}]", topic)); + } + // Create the consumer and register it. + KafkaConsumerPtr new_consumer = std::make_unique( + *store_, thread_factory_, topic, cluster_config->partition_count_, + cluster_config->upstream_consumer_properties_); + ENVOY_LOG(info, "Registering new Kafka consumer for topic [{}], consuming from cluster [{}]", + topic, cluster_config->name_); + auto result = topic_to_consumer_.emplace(topic, std::move(new_consumer)); + return *(result.first->second); +} + +void SharedConsumerManagerImpl::removeCallback(const RecordCbSharedPtr& callback) { + // Real work - let's remove the callback. + store_->removeCallback(callback); +} + +void SharedConsumerManagerImpl::doShutdown() { + ENVOY_LOG(info, "Shutting down consumer manager"); + absl::MutexLock consumers_lock(&consumers_mutex_); + ENVOY_LOG(info, "There are {} consumers to close", topic_to_consumer_.size()); + topic_to_consumer_.clear(); + ENVOY_LOG(info, "Consumers have been shut down"); + store_ = nullptr; + ENVOY_LOG(info, "Data stored has been erased"); +} + +bool Store::waitUntilInterest(const std::string& topic, const int32_t timeout_ms) const { + auto store_has_interest = [this, &topic]() { return hasInterest(topic); }; + + // Effectively this means "has an interest appeared within timeout". + // If not, we let the user know so they could do something else instead of being infinitely + // blocked. + bool can_poll = callbacks_mutex_.LockWhenWithTimeout(absl::Condition(&store_has_interest), + absl::Milliseconds(timeout_ms)); + callbacks_mutex_.Unlock(); + return can_poll; +} + +bool Store::hasInterest(const std::string& topic) const { + for (const auto& e : partition_to_callbacks_) { // XXX to ma beznadziejny warning + if (topic == e.first.first && !e.second.empty()) { + return true; + } + } + return false; +} + +void Store::getRecordsOrRegisterCallback(const RecordCbSharedPtr& callback) { + + TopicToPartitionsMap requested = callback->interest(); + ENVOY_LOG(info, "Registering callback {}", callback->debugId()); + + bool fulfilled_at_startup = false; + + { + absl::MutexLock lock(&messages_mutex_); + for (const auto& topic_and_partitions : requested) { + const std::string topic = topic_and_partitions.first; + for (const int32_t partition : topic_and_partitions.second) { + const KafkaPartition kp = {topic, partition}; + + auto& stored_messages = messages_waiting_for_interest_[kp]; + if (0 != stored_messages.size()) { + ENVOY_LOG(info, "Early notification for callback {}, as there are {} messages ready", + callback->debugId(), stored_messages.size()); + } + + for (auto it = stored_messages.begin(); it != stored_messages.end();) { + Reply callback_status = callback->receive(*it); + bool callback_finished; + switch (callback_status) { + case Reply::ACCEPTED_AND_FINISHED: { + callback_finished = true; + it = stored_messages.erase(it); + break; + } + case Reply::ACCEPTED_AND_WANT_MORE: { + callback_finished = false; + it = stored_messages.erase(it); + break; + } + case Reply::REJECTED: { + callback_finished = true; + break; + } + } /* switch */ + if (callback_finished) { + fulfilled_at_startup = true; + break; // Callback does not want any messages anymore. + } + } /* for-messages */ + } /* for-partitions */ + } /* for-topic_and_partitions */ + } /* lock */ + + if (!fulfilled_at_startup) { + // Usual path: the request was not fulfilled at receive time (there were no buffered messages). + // So we just register the callback. + absl::MutexLock lock(&callbacks_mutex_); + + for (const auto& topic_and_partitions : requested) { + const std::string topic = topic_and_partitions.first; + for (const int32_t partition : topic_and_partitions.second) { + const KafkaPartition kp = {topic, partition}; + auto& partition_callbacks = partition_to_callbacks_[kp]; + partition_callbacks.push_back(callback); + } + } + } else { + ENVOY_LOG(info, "No registration for callback {} due to successful early processing", + callback->debugId()); + } +} + +void Store::receive(InboundRecordSharedPtr message) { // XXX this api is inefficient, as we acquire + // lock PER MESSAGE (instead of batch) + + const KafkaPartition kafka_partition = {message->topic_, message->partition_}; + + // Whether this message has been consumed by any of the callbacks. + // Because then we can safely throw it away instead of storing. + bool consumed_by_callback = false; + + { + absl::MutexLock lock(&callbacks_mutex_); + auto& callbacks = partition_to_callbacks_[kafka_partition]; + + std::vector satisfied_callbacks = {}; + + // Typical case: there is some interest in messages for given partition. Notify the callback and + // remove it. + for (const auto& callback : callbacks) { + Reply callback_status = callback->receive(message); + switch (callback_status) { + case Reply::ACCEPTED_AND_FINISHED: { + consumed_by_callback = true; + // A callback is finally satisfied, it will never want more messages. + satisfied_callbacks.push_back(callback); + break; + } + case Reply::ACCEPTED_AND_WANT_MORE: { + consumed_by_callback = true; + break; + } + case Reply::REJECTED: { + break; + } + } /* switch */ + + /* Some callback has taken the message - this is good, no more iterating. */ + if (consumed_by_callback) { + break; + } + } + + for (const auto& callback : satisfied_callbacks) { + // Lock was acquired at the beggining of the block. + removeCallbackWithoutLocking(callback, partition_to_callbacks_); + } + } + + // Noone is interested in our message, so we are going to store it in a local cache. + if (!consumed_by_callback) { + absl::MutexLock lock(&messages_mutex_); + auto& stored_messages = messages_waiting_for_interest_[kafka_partition]; + stored_messages.push_back(message); // XXX there should be buffer limits here + ENVOY_LOG(info, "Stored message [{}]", message->toString()); + } +} + +void Store::removeCallback(const RecordCbSharedPtr& callback) { + absl::MutexLock lock(&callbacks_mutex_); + removeCallbackWithoutLocking(callback, + partition_to_callbacks_); // Lock was acquired a line above. +} + +void Store::removeCallbackWithoutLocking( + const RecordCbSharedPtr& callback, + std::map>& partition_to_callbacks) { + + ENVOY_LOG(info, "Removing callback {}", callback->debugId()); + for (auto& e : partition_to_callbacks) { + auto& partition_callbacks = e.second; + partition_callbacks.erase( + std::remove(partition_callbacks.begin(), partition_callbacks.end(), callback), + partition_callbacks.end()); + } + ENVOY_LOG(info, "Removed callback {}", callback->debugId()); +} + +} // namespace Mesh +} // namespace Kafka +} // namespace NetworkFilters +} // namespace Extensions +} // namespace Envoy diff --git a/contrib/kafka/filters/network/source/mesh/shared_consumer_manager_impl.h b/contrib/kafka/filters/network/source/mesh/shared_consumer_manager_impl.h new file mode 100644 index 0000000000000..2f5abdaac5188 --- /dev/null +++ b/contrib/kafka/filters/network/source/mesh/shared_consumer_manager_impl.h @@ -0,0 +1,116 @@ +#pragma once + +#include +#include +#include + +#include "envoy/thread/thread.h" + +#include "source/common/common/logger.h" + +#include "contrib/kafka/filters/network/source/kafka_types.h" +#include "contrib/kafka/filters/network/source/mesh/shared_consumer_manager.h" +#include "contrib/kafka/filters/network/source/mesh/upstream_config.h" +#include "contrib/kafka/filters/network/source/mesh/upstream_kafka_consumer.h" + +namespace Envoy { +namespace Extensions { +namespace NetworkFilters { +namespace Kafka { +namespace Mesh { + +/** + * Meaningful state for upstream-pointing consumer. + * Keeps messages received so far that nobody was interested in. + * Keeps callbacks that are interested in messages. + * + * Mutex locking order: + * 1. store's consumer list mutex (consumers_mutex_) + * 2. message's data mutex (data_mutex_) + */ +class Store : public InboundRecordProcessor, private Logger::Loggable { +public: + // InboundRecordProcessor + bool waitUntilInterest(const std::string& topic, const int32_t timeout_ms) const override; + + // InboundRecordProcessor + void receive(InboundRecordSharedPtr message) override; + + // XXX + void getRecordsOrRegisterCallback(const RecordCbSharedPtr& callback); + + // XXX + void removeCallback(const RecordCbSharedPtr& callback); + +private: + // XXX + bool hasInterest(const std::string& topic) const; + + // HAX! + void removeCallbackWithoutLocking( + const RecordCbSharedPtr& callback, + std::map>& partition_to_callbacks); + + /** + * Invariant: for every i: KafkaPartition, the following holds: + * !(partition_to_callbacks_[i].size() >= 0 && messages_waiting_for_interest_[i].size() >= 0) + */ + + mutable absl::Mutex callbacks_mutex_; + std::map> + partition_to_callbacks_ ABSL_GUARDED_BY(callbacks_mutex_); + + mutable absl::Mutex messages_mutex_; + std::map> + messages_waiting_for_interest_ ABSL_GUARDED_BY(messages_mutex_); +}; + +using StorePtr = std::unique_ptr; + +// ============================================================================================================= + +/** + * Implements SCM interface by maintaining a collection of Kafka consumers on per-topic basis. + * Maintains a message cache for messages that had no interest but might be requested later. + */ +class SharedConsumerManagerImpl : public SharedConsumerManager, + private Logger::Loggable { +public: + SharedConsumerManagerImpl(const UpstreamKafkaConfiguration& configuration, + Thread::ThreadFactory& thread_factory); + + ~SharedConsumerManagerImpl() override; + + /** + * Registers a callback that is interested in messages for particular partitions. + */ + void getRecordsOrRegisterCallback(const RecordCbSharedPtr& callback) override; + + void removeCallback(const RecordCbSharedPtr& callback) override; + +private: + KafkaConsumer& getOrCreateConsumer(const std::string& topic); + // Mutates 'topic_to_consumer_'. + KafkaConsumer& registerNewConsumer(const std::string& topic); + + // Disables this instance (so it can no longer be used by requests). + // After this method finishes, no requests (RecordCbSharedPtr) are ever held by this object (what + // means they are held only by originating filter). + void doShutdown(); + + StorePtr store_{std::make_unique()}; + + const UpstreamKafkaConfiguration& configuration_; + Thread::ThreadFactory& thread_factory_; + + mutable absl::Mutex consumers_mutex_; + std::map topic_to_consumer_ ABSL_GUARDED_BY(consumers_mutex_); +}; + +// ============================================================================================================= + +} // namespace Mesh +} // namespace Kafka +} // namespace NetworkFilters +} // namespace Extensions +} // namespace Envoy diff --git a/contrib/kafka/filters/network/source/mesh/upstream_kafka_consumer.h b/contrib/kafka/filters/network/source/mesh/upstream_kafka_consumer.h index 3b7646de449ce..498684fdc692e 100644 --- a/contrib/kafka/filters/network/source/mesh/upstream_kafka_consumer.h +++ b/contrib/kafka/filters/network/source/mesh/upstream_kafka_consumer.h @@ -14,6 +14,34 @@ namespace NetworkFilters { namespace Kafka { namespace Mesh { +// Topic name to topic partitions. +using TopicToPartitionsMap = std::map>; + +enum class Reply { + REJECTED, + ACCEPTED_AND_WANT_MORE, + ACCEPTED_AND_FINISHED, +}; + +// XXX rename +// Callback for objects that want to be notified that new Kafka record has been received. +class RecordCb { +public: + virtual ~RecordCb() = default; + + // Notify the callback that with a message. + // @return whether the callback could accept the message + virtual Reply receive(InboundRecordSharedPtr message) PURE; + + virtual TopicToPartitionsMap interest() const PURE; + + virtual std::string debugId() const PURE; +}; + +using RecordCbSharedPtr = std::shared_ptr; + +// ================================================================================================ + /** * An entity that is interested in inbound records delivered by Kafka consumer. */ diff --git a/contrib/kafka/filters/network/test/mesh/BUILD b/contrib/kafka/filters/network/test/mesh/BUILD index 47c556411c0c2..ebb23eba960bd 100644 --- a/contrib/kafka/filters/network/test/mesh/BUILD +++ b/contrib/kafka/filters/network/test/mesh/BUILD @@ -74,6 +74,15 @@ envoy_cc_test( ], ) +envoy_cc_test( + name = "shared_consumer_manager_impl_unit_test", + srcs = ["shared_consumer_manager_impl_unit_test.cc"], + tags = ["skip_on_windows"], + deps = [ + "//contrib/kafka/filters/network/source/mesh:shared_consumer_manager_impl_lib", + ], +) + envoy_cc_test( name = "upstream_kafka_consumer_impl_unit_test", srcs = ["upstream_kafka_consumer_impl_unit_test.cc"], diff --git a/contrib/kafka/filters/network/test/mesh/shared_consumer_manager_impl_unit_test.cc b/contrib/kafka/filters/network/test/mesh/shared_consumer_manager_impl_unit_test.cc new file mode 100644 index 0000000000000..5df31d81e7acb --- /dev/null +++ b/contrib/kafka/filters/network/test/mesh/shared_consumer_manager_impl_unit_test.cc @@ -0,0 +1,25 @@ +#include "contrib/kafka/filters/network/source/mesh/shared_consumer_manager_impl.h" +#include "gmock/gmock.h" +#include "gtest/gtest.h" + +namespace Envoy { +namespace Extensions { +namespace NetworkFilters { +namespace Kafka { +namespace Mesh { + +class SCMTest : public testing::Test { +protected: +}; + +TEST_F(SCMTest, ShouldTest) { + // given + // when + // then +} + +} // namespace Mesh +} // namespace Kafka +} // namespace NetworkFilters +} // namespace Extensions +} // namespace Envoy From 207bd44ba300851df61a5ecb99e46f0417591475 Mon Sep 17 00:00:00 2001 From: Adam Kotwasinski Date: Sun, 11 Dec 2022 16:13:22 -0800 Subject: [PATCH 3/4] kafka: simplify this PR by removing the "request&receive" part of SCM Signed-off-by: Adam Kotwasinski --- .../source/mesh/shared_consumer_manager.h | 5 +- .../mesh/shared_consumer_manager_impl.cc | 217 ++---------------- .../mesh/shared_consumer_manager_impl.h | 73 +----- .../source/mesh/upstream_kafka_consumer.h | 28 --- 4 files changed, 33 insertions(+), 290 deletions(-) diff --git a/contrib/kafka/filters/network/source/mesh/shared_consumer_manager.h b/contrib/kafka/filters/network/source/mesh/shared_consumer_manager.h index 1873406adc201..c3ef2fc646b81 100644 --- a/contrib/kafka/filters/network/source/mesh/shared_consumer_manager.h +++ b/contrib/kafka/filters/network/source/mesh/shared_consumer_manager.h @@ -19,9 +19,8 @@ class SharedConsumerManager { public: virtual ~SharedConsumerManager() = default; - virtual void getRecordsOrRegisterCallback(const RecordCbSharedPtr& callback) PURE; - - virtual void removeCallback(const RecordCbSharedPtr& callback) PURE; + // Start the consumer (if there is none) to make sure that records can be received from the topic. + virtual void registerConsumerIfAbsent(const std::string& topic) PURE; }; using SharedConsumerManagerSharedPtr = std::shared_ptr; diff --git a/contrib/kafka/filters/network/source/mesh/shared_consumer_manager_impl.cc b/contrib/kafka/filters/network/source/mesh/shared_consumer_manager_impl.cc index 751aa5c7ce4b2..b06bfe4b7f5f8 100644 --- a/contrib/kafka/filters/network/source/mesh/shared_consumer_manager_impl.cc +++ b/contrib/kafka/filters/network/source/mesh/shared_consumer_manager_impl.cc @@ -14,32 +14,21 @@ namespace Mesh { SharedConsumerManagerImpl::SharedConsumerManagerImpl( const UpstreamKafkaConfiguration& configuration, Thread::ThreadFactory& thread_factory) - : configuration_{configuration}, thread_factory_{thread_factory} {} + : record_processor_{std::make_unique()}, configuration_{configuration}, + thread_factory_{thread_factory} {} -SharedConsumerManagerImpl::~SharedConsumerManagerImpl() { doShutdown(); } - -void SharedConsumerManagerImpl::getRecordsOrRegisterCallback(const RecordCbSharedPtr& callback) { - // For every fetch topic, figure out the upstream cluster, - // create consumer if needed ... - const TopicToPartitionsMap interest = callback->interest(); - for (const auto& fetch : interest) { - const std::string& topic = fetch.first; - getOrCreateConsumer(topic); // XXX this method should not be named 'get...' - } - - // ... and start processing. - store_->getRecordsOrRegisterCallback(callback); -} - -KafkaConsumer& SharedConsumerManagerImpl::getOrCreateConsumer(const std::string& topic) { +void SharedConsumerManagerImpl::registerConsumerIfAbsent(const std::string& topic) { absl::MutexLock lock(&consumers_mutex_); const auto it = topic_to_consumer_.find(topic); // Return consumer already present or create new one and register it. - return (topic_to_consumer_.end() == it) ? registerNewConsumer(topic) : *(it->second); + if (topic_to_consumer_.end() == it) { + registerNewConsumer(topic); + } } -KafkaConsumer& SharedConsumerManagerImpl::registerNewConsumer(const std::string& topic) { - ENVOY_LOG(info, "Creating consumer for topic [{}]", topic); +void SharedConsumerManagerImpl::registerNewConsumer(const std::string& topic) { + ENVOY_LOG(debug, "Creating consumer for topic [{}]", topic); + // Compute which upstream cluster corresponds to the topic. const absl::optional cluster_config = configuration_.computeClusterConfigForTopic(topic); @@ -47,193 +36,25 @@ KafkaConsumer& SharedConsumerManagerImpl::registerNewConsumer(const std::string& throw EnvoyException( fmt::format("Could not compute upstream cluster configuration for topic [{}]", topic)); } + // Create the consumer and register it. KafkaConsumerPtr new_consumer = std::make_unique( - *store_, thread_factory_, topic, cluster_config->partition_count_, + *record_processor_, thread_factory_, topic, cluster_config->partition_count_, cluster_config->upstream_consumer_properties_); - ENVOY_LOG(info, "Registering new Kafka consumer for topic [{}], consuming from cluster [{}]", + ENVOY_LOG(debug, "Registering new Kafka consumer for topic [{}], consuming from cluster [{}]", topic, cluster_config->name_); - auto result = topic_to_consumer_.emplace(topic, std::move(new_consumer)); - return *(result.first->second); + topic_to_consumer_.emplace(topic, std::move(new_consumer)); } -void SharedConsumerManagerImpl::removeCallback(const RecordCbSharedPtr& callback) { - // Real work - let's remove the callback. - store_->removeCallback(callback); -} - -void SharedConsumerManagerImpl::doShutdown() { - ENVOY_LOG(info, "Shutting down consumer manager"); - absl::MutexLock consumers_lock(&consumers_mutex_); - ENVOY_LOG(info, "There are {} consumers to close", topic_to_consumer_.size()); - topic_to_consumer_.clear(); - ENVOY_LOG(info, "Consumers have been shut down"); - store_ = nullptr; - ENVOY_LOG(info, "Data stored has been erased"); -} - -bool Store::waitUntilInterest(const std::string& topic, const int32_t timeout_ms) const { - auto store_has_interest = [this, &topic]() { return hasInterest(topic); }; - - // Effectively this means "has an interest appeared within timeout". - // If not, we let the user know so they could do something else instead of being infinitely - // blocked. - bool can_poll = callbacks_mutex_.LockWhenWithTimeout(absl::Condition(&store_has_interest), - absl::Milliseconds(timeout_ms)); - callbacks_mutex_.Unlock(); - return can_poll; -} - -bool Store::hasInterest(const std::string& topic) const { - for (const auto& e : partition_to_callbacks_) { // XXX to ma beznadziejny warning - if (topic == e.first.first && !e.second.empty()) { - return true; - } - } +// InboundRecordProcessor +bool SharedProcessor::waitUntilInterest(const std::string&, const int32_t) const { + // TODO (adam.kotwasinski) To implement in future commits. return false; } -void Store::getRecordsOrRegisterCallback(const RecordCbSharedPtr& callback) { - - TopicToPartitionsMap requested = callback->interest(); - ENVOY_LOG(info, "Registering callback {}", callback->debugId()); - - bool fulfilled_at_startup = false; - - { - absl::MutexLock lock(&messages_mutex_); - for (const auto& topic_and_partitions : requested) { - const std::string topic = topic_and_partitions.first; - for (const int32_t partition : topic_and_partitions.second) { - const KafkaPartition kp = {topic, partition}; - - auto& stored_messages = messages_waiting_for_interest_[kp]; - if (0 != stored_messages.size()) { - ENVOY_LOG(info, "Early notification for callback {}, as there are {} messages ready", - callback->debugId(), stored_messages.size()); - } - - for (auto it = stored_messages.begin(); it != stored_messages.end();) { - Reply callback_status = callback->receive(*it); - bool callback_finished; - switch (callback_status) { - case Reply::ACCEPTED_AND_FINISHED: { - callback_finished = true; - it = stored_messages.erase(it); - break; - } - case Reply::ACCEPTED_AND_WANT_MORE: { - callback_finished = false; - it = stored_messages.erase(it); - break; - } - case Reply::REJECTED: { - callback_finished = true; - break; - } - } /* switch */ - if (callback_finished) { - fulfilled_at_startup = true; - break; // Callback does not want any messages anymore. - } - } /* for-messages */ - } /* for-partitions */ - } /* for-topic_and_partitions */ - } /* lock */ - - if (!fulfilled_at_startup) { - // Usual path: the request was not fulfilled at receive time (there were no buffered messages). - // So we just register the callback. - absl::MutexLock lock(&callbacks_mutex_); - - for (const auto& topic_and_partitions : requested) { - const std::string topic = topic_and_partitions.first; - for (const int32_t partition : topic_and_partitions.second) { - const KafkaPartition kp = {topic, partition}; - auto& partition_callbacks = partition_to_callbacks_[kp]; - partition_callbacks.push_back(callback); - } - } - } else { - ENVOY_LOG(info, "No registration for callback {} due to successful early processing", - callback->debugId()); - } -} - -void Store::receive(InboundRecordSharedPtr message) { // XXX this api is inefficient, as we acquire - // lock PER MESSAGE (instead of batch) - - const KafkaPartition kafka_partition = {message->topic_, message->partition_}; - - // Whether this message has been consumed by any of the callbacks. - // Because then we can safely throw it away instead of storing. - bool consumed_by_callback = false; - - { - absl::MutexLock lock(&callbacks_mutex_); - auto& callbacks = partition_to_callbacks_[kafka_partition]; - - std::vector satisfied_callbacks = {}; - - // Typical case: there is some interest in messages for given partition. Notify the callback and - // remove it. - for (const auto& callback : callbacks) { - Reply callback_status = callback->receive(message); - switch (callback_status) { - case Reply::ACCEPTED_AND_FINISHED: { - consumed_by_callback = true; - // A callback is finally satisfied, it will never want more messages. - satisfied_callbacks.push_back(callback); - break; - } - case Reply::ACCEPTED_AND_WANT_MORE: { - consumed_by_callback = true; - break; - } - case Reply::REJECTED: { - break; - } - } /* switch */ - - /* Some callback has taken the message - this is good, no more iterating. */ - if (consumed_by_callback) { - break; - } - } - - for (const auto& callback : satisfied_callbacks) { - // Lock was acquired at the beggining of the block. - removeCallbackWithoutLocking(callback, partition_to_callbacks_); - } - } - - // Noone is interested in our message, so we are going to store it in a local cache. - if (!consumed_by_callback) { - absl::MutexLock lock(&messages_mutex_); - auto& stored_messages = messages_waiting_for_interest_[kafka_partition]; - stored_messages.push_back(message); // XXX there should be buffer limits here - ENVOY_LOG(info, "Stored message [{}]", message->toString()); - } -} - -void Store::removeCallback(const RecordCbSharedPtr& callback) { - absl::MutexLock lock(&callbacks_mutex_); - removeCallbackWithoutLocking(callback, - partition_to_callbacks_); // Lock was acquired a line above. -} - -void Store::removeCallbackWithoutLocking( - const RecordCbSharedPtr& callback, - std::map>& partition_to_callbacks) { - - ENVOY_LOG(info, "Removing callback {}", callback->debugId()); - for (auto& e : partition_to_callbacks) { - auto& partition_callbacks = e.second; - partition_callbacks.erase( - std::remove(partition_callbacks.begin(), partition_callbacks.end(), callback), - partition_callbacks.end()); - } - ENVOY_LOG(info, "Removed callback {}", callback->debugId()); +// InboundRecordProcessor +void SharedProcessor::receive(InboundRecordSharedPtr) { + // TODO (adam.kotwasinski) To implement in future commits. } } // namespace Mesh diff --git a/contrib/kafka/filters/network/source/mesh/shared_consumer_manager_impl.h b/contrib/kafka/filters/network/source/mesh/shared_consumer_manager_impl.h index 2f5abdaac5188..48541eedf32bf 100644 --- a/contrib/kafka/filters/network/source/mesh/shared_consumer_manager_impl.h +++ b/contrib/kafka/filters/network/source/mesh/shared_consumer_manager_impl.h @@ -1,8 +1,6 @@ #pragma once #include -#include -#include #include "envoy/thread/thread.h" @@ -20,58 +18,24 @@ namespace Kafka { namespace Mesh { /** - * Meaningful state for upstream-pointing consumer. - * Keeps messages received so far that nobody was interested in. - * Keeps callbacks that are interested in messages. - * - * Mutex locking order: - * 1. store's consumer list mutex (consumers_mutex_) - * 2. message's data mutex (data_mutex_) + * Placeholder interface for now. + * In future: + * Processor implementation that stores received records (that had no interest), and callbacks + * waiting for records (that had no matching records delivered yet). */ -class Store : public InboundRecordProcessor, private Logger::Loggable { +class SharedProcessor : public InboundRecordProcessor { public: // InboundRecordProcessor bool waitUntilInterest(const std::string& topic, const int32_t timeout_ms) const override; // InboundRecordProcessor void receive(InboundRecordSharedPtr message) override; - - // XXX - void getRecordsOrRegisterCallback(const RecordCbSharedPtr& callback); - - // XXX - void removeCallback(const RecordCbSharedPtr& callback); - -private: - // XXX - bool hasInterest(const std::string& topic) const; - - // HAX! - void removeCallbackWithoutLocking( - const RecordCbSharedPtr& callback, - std::map>& partition_to_callbacks); - - /** - * Invariant: for every i: KafkaPartition, the following holds: - * !(partition_to_callbacks_[i].size() >= 0 && messages_waiting_for_interest_[i].size() >= 0) - */ - - mutable absl::Mutex callbacks_mutex_; - std::map> - partition_to_callbacks_ ABSL_GUARDED_BY(callbacks_mutex_); - - mutable absl::Mutex messages_mutex_; - std::map> - messages_waiting_for_interest_ ABSL_GUARDED_BY(messages_mutex_); }; -using StorePtr = std::unique_ptr; - -// ============================================================================================================= +using SharedProcessorPtr = std::unique_ptr; /** - * Implements SCM interface by maintaining a collection of Kafka consumers on per-topic basis. - * Maintains a message cache for messages that had no interest but might be requested later. + * Implements SCM interface by maintaining a collection of Kafka consumers (one per topic). */ class SharedConsumerManagerImpl : public SharedConsumerManager, private Logger::Loggable { @@ -79,26 +43,15 @@ class SharedConsumerManagerImpl : public SharedConsumerManager, SharedConsumerManagerImpl(const UpstreamKafkaConfiguration& configuration, Thread::ThreadFactory& thread_factory); - ~SharedConsumerManagerImpl() override; - - /** - * Registers a callback that is interested in messages for particular partitions. - */ - void getRecordsOrRegisterCallback(const RecordCbSharedPtr& callback) override; - - void removeCallback(const RecordCbSharedPtr& callback) override; + // SharedConsumerManager + void registerConsumerIfAbsent(const std::string& topic) override; private: - KafkaConsumer& getOrCreateConsumer(const std::string& topic); // Mutates 'topic_to_consumer_'. - KafkaConsumer& registerNewConsumer(const std::string& topic); - - // Disables this instance (so it can no longer be used by requests). - // After this method finishes, no requests (RecordCbSharedPtr) are ever held by this object (what - // means they are held only by originating filter). - void doShutdown(); + void registerNewConsumer(const std::string& topic) + ABSL_EXCLUSIVE_LOCKS_REQUIRED(consumers_mutex_); - StorePtr store_{std::make_unique()}; + SharedProcessorPtr record_processor_; const UpstreamKafkaConfiguration& configuration_; Thread::ThreadFactory& thread_factory_; @@ -107,8 +60,6 @@ class SharedConsumerManagerImpl : public SharedConsumerManager, std::map topic_to_consumer_ ABSL_GUARDED_BY(consumers_mutex_); }; -// ============================================================================================================= - } // namespace Mesh } // namespace Kafka } // namespace NetworkFilters diff --git a/contrib/kafka/filters/network/source/mesh/upstream_kafka_consumer.h b/contrib/kafka/filters/network/source/mesh/upstream_kafka_consumer.h index 498684fdc692e..3b7646de449ce 100644 --- a/contrib/kafka/filters/network/source/mesh/upstream_kafka_consumer.h +++ b/contrib/kafka/filters/network/source/mesh/upstream_kafka_consumer.h @@ -14,34 +14,6 @@ namespace NetworkFilters { namespace Kafka { namespace Mesh { -// Topic name to topic partitions. -using TopicToPartitionsMap = std::map>; - -enum class Reply { - REJECTED, - ACCEPTED_AND_WANT_MORE, - ACCEPTED_AND_FINISHED, -}; - -// XXX rename -// Callback for objects that want to be notified that new Kafka record has been received. -class RecordCb { -public: - virtual ~RecordCb() = default; - - // Notify the callback that with a message. - // @return whether the callback could accept the message - virtual Reply receive(InboundRecordSharedPtr message) PURE; - - virtual TopicToPartitionsMap interest() const PURE; - - virtual std::string debugId() const PURE; -}; - -using RecordCbSharedPtr = std::shared_ptr; - -// ================================================================================================ - /** * An entity that is interested in inbound records delivered by Kafka consumer. */ From 9cd16ede417c3be846214bf4f5a0b782df312f8c Mon Sep 17 00:00:00 2001 From: Adam Kotwasinski Date: Mon, 12 Dec 2022 11:43:39 -0800 Subject: [PATCH 4/4] kafka: add some tests Signed-off-by: Adam Kotwasinski --- .../kafka/filters/network/source/mesh/BUILD | 1 + .../source/mesh/shared_consumer_manager.h | 8 +- .../mesh/shared_consumer_manager_impl.cc | 58 +++++++++++--- .../mesh/shared_consumer_manager_impl.h | 33 +++++++- .../shared_consumer_manager_impl_unit_test.cc | 79 ++++++++++++++++++- 5 files changed, 159 insertions(+), 20 deletions(-) diff --git a/contrib/kafka/filters/network/source/mesh/BUILD b/contrib/kafka/filters/network/source/mesh/BUILD index 3df2b411496e1..cfc0b024924f2 100644 --- a/contrib/kafka/filters/network/source/mesh/BUILD +++ b/contrib/kafka/filters/network/source/mesh/BUILD @@ -134,6 +134,7 @@ envoy_cc_library( ], tags = ["skip_on_windows"], deps = [ + ":librdkafka_utils_lib", ":shared_consumer_manager_lib", ":upstream_config_lib", ":upstream_kafka_consumer_impl_lib", diff --git a/contrib/kafka/filters/network/source/mesh/shared_consumer_manager.h b/contrib/kafka/filters/network/source/mesh/shared_consumer_manager.h index c3ef2fc646b81..98eac5cad8699 100644 --- a/contrib/kafka/filters/network/source/mesh/shared_consumer_manager.h +++ b/contrib/kafka/filters/network/source/mesh/shared_consumer_manager.h @@ -1,7 +1,7 @@ #pragma once -#include -#include +#include +#include #include "contrib/kafka/filters/network/source/mesh/upstream_kafka_consumer.h" @@ -13,7 +13,7 @@ namespace Mesh { /** * Manages (raw) Kafka consumers pointing to upstream Kafka clusters. - * It is expected to have only one instance of this object in the runtime. + * It is expected to have only one instance of this object per mesh-filter type. */ class SharedConsumerManager { public: @@ -23,7 +23,7 @@ class SharedConsumerManager { virtual void registerConsumerIfAbsent(const std::string& topic) PURE; }; -using SharedConsumerManagerSharedPtr = std::shared_ptr; +using SharedConsumerManagerPtr = std::unique_ptr; } // namespace Mesh } // namespace Kafka diff --git a/contrib/kafka/filters/network/source/mesh/shared_consumer_manager_impl.cc b/contrib/kafka/filters/network/source/mesh/shared_consumer_manager_impl.cc index b06bfe4b7f5f8..1550c16cb0569 100644 --- a/contrib/kafka/filters/network/source/mesh/shared_consumer_manager_impl.cc +++ b/contrib/kafka/filters/network/source/mesh/shared_consumer_manager_impl.cc @@ -1,7 +1,5 @@ #include "contrib/kafka/filters/network/source/mesh/shared_consumer_manager_impl.h" -#include - #include "source/common/common/fmt.h" #include "contrib/kafka/filters/network/source/mesh/upstream_kafka_consumer_impl.h" @@ -12,10 +10,45 @@ namespace NetworkFilters { namespace Kafka { namespace Mesh { +// KafkaConsumerFactoryImpl + +class KafkaConsumerFactoryImpl : public KafkaConsumerFactory { +public: + // KafkaConsumerFactory + KafkaConsumerPtr createConsumer(InboundRecordProcessor& record_processor, + Thread::ThreadFactory& thread_factory, const std::string& topic, + const int32_t partition_count, + const RawKafkaConfig& configuration) const override; + + // Default singleton accessor. + static const KafkaConsumerFactory& getDefaultInstance(); +}; + +KafkaConsumerPtr +KafkaConsumerFactoryImpl::createConsumer(InboundRecordProcessor& record_processor, + Thread::ThreadFactory& thread_factory, + const std::string& topic, const int32_t partition_count, + const RawKafkaConfig& configuration) const { + return std::make_unique(record_processor, thread_factory, topic, + partition_count, configuration); +} + +const KafkaConsumerFactory& KafkaConsumerFactoryImpl::getDefaultInstance() { + CONSTRUCT_ON_FIRST_USE(KafkaConsumerFactoryImpl); +} + +// SharedConsumerManagerImpl + SharedConsumerManagerImpl::SharedConsumerManagerImpl( const UpstreamKafkaConfiguration& configuration, Thread::ThreadFactory& thread_factory) - : record_processor_{std::make_unique()}, configuration_{configuration}, - thread_factory_{thread_factory} {} + : SharedConsumerManagerImpl{configuration, thread_factory, + KafkaConsumerFactoryImpl::getDefaultInstance()} {} + +SharedConsumerManagerImpl::SharedConsumerManagerImpl( + const UpstreamKafkaConfiguration& configuration, Thread::ThreadFactory& thread_factory, + const KafkaConsumerFactory& consumer_factory) + : distributor_{std::make_unique()}, configuration_{configuration}, + thread_factory_{thread_factory}, consumer_factory_{consumer_factory} {} void SharedConsumerManagerImpl::registerConsumerIfAbsent(const std::string& topic) { absl::MutexLock lock(&consumers_mutex_); @@ -38,22 +71,27 @@ void SharedConsumerManagerImpl::registerNewConsumer(const std::string& topic) { } // Create the consumer and register it. - KafkaConsumerPtr new_consumer = std::make_unique( - *record_processor_, thread_factory_, topic, cluster_config->partition_count_, + KafkaConsumerPtr new_consumer = consumer_factory_.createConsumer( + *distributor_, thread_factory_, topic, cluster_config->partition_count_, cluster_config->upstream_consumer_properties_); ENVOY_LOG(debug, "Registering new Kafka consumer for topic [{}], consuming from cluster [{}]", topic, cluster_config->name_); topic_to_consumer_.emplace(topic, std::move(new_consumer)); } -// InboundRecordProcessor -bool SharedProcessor::waitUntilInterest(const std::string&, const int32_t) const { +size_t SharedConsumerManagerImpl::getConsumerCountForTest() const { + absl::MutexLock lock(&consumers_mutex_); + return topic_to_consumer_.size(); +} + +// RecordDistributor + +bool RecordDistributor::waitUntilInterest(const std::string&, const int32_t) const { // TODO (adam.kotwasinski) To implement in future commits. return false; } -// InboundRecordProcessor -void SharedProcessor::receive(InboundRecordSharedPtr) { +void RecordDistributor::receive(InboundRecordSharedPtr) { // TODO (adam.kotwasinski) To implement in future commits. } diff --git a/contrib/kafka/filters/network/source/mesh/shared_consumer_manager_impl.h b/contrib/kafka/filters/network/source/mesh/shared_consumer_manager_impl.h index 48541eedf32bf..849d4076442cf 100644 --- a/contrib/kafka/filters/network/source/mesh/shared_consumer_manager_impl.h +++ b/contrib/kafka/filters/network/source/mesh/shared_consumer_manager_impl.h @@ -7,6 +7,7 @@ #include "source/common/common/logger.h" #include "contrib/kafka/filters/network/source/kafka_types.h" +#include "contrib/kafka/filters/network/source/mesh/librdkafka_utils.h" #include "contrib/kafka/filters/network/source/mesh/shared_consumer_manager.h" #include "contrib/kafka/filters/network/source/mesh/upstream_config.h" #include "contrib/kafka/filters/network/source/mesh/upstream_kafka_consumer.h" @@ -22,8 +23,9 @@ namespace Mesh { * In future: * Processor implementation that stores received records (that had no interest), and callbacks * waiting for records (that had no matching records delivered yet). + * Basically core of Fetch-handling business logic. */ -class SharedProcessor : public InboundRecordProcessor { +class RecordDistributor : public InboundRecordProcessor { public: // InboundRecordProcessor bool waitUntilInterest(const std::string& topic, const int32_t timeout_ms) const override; @@ -32,29 +34,52 @@ class SharedProcessor : public InboundRecordProcessor { void receive(InboundRecordSharedPtr message) override; }; -using SharedProcessorPtr = std::unique_ptr; +using RecordDistributorPtr = std::unique_ptr; /** - * Implements SCM interface by maintaining a collection of Kafka consumers (one per topic). + * Injectable for tests. + */ +class KafkaConsumerFactory { +public: + virtual ~KafkaConsumerFactory() = default; + + // Create a Kafka consumer. + virtual KafkaConsumerPtr createConsumer(InboundRecordProcessor& record_processor, + Thread::ThreadFactory& thread_factory, + const std::string& topic, const int32_t partition_count, + const RawKafkaConfig& configuration) const PURE; +}; + +/** + * Maintains a collection of Kafka consumers (one per topic). */ class SharedConsumerManagerImpl : public SharedConsumerManager, private Logger::Loggable { public: + // Main constructor. SharedConsumerManagerImpl(const UpstreamKafkaConfiguration& configuration, Thread::ThreadFactory& thread_factory); + // Visible for testing. + SharedConsumerManagerImpl(const UpstreamKafkaConfiguration& configuration, + Thread::ThreadFactory& thread_factory, + const KafkaConsumerFactory& consumer_factory); + // SharedConsumerManager void registerConsumerIfAbsent(const std::string& topic) override; + size_t getConsumerCountForTest() const; + private: // Mutates 'topic_to_consumer_'. void registerNewConsumer(const std::string& topic) ABSL_EXCLUSIVE_LOCKS_REQUIRED(consumers_mutex_); - SharedProcessorPtr record_processor_; + RecordDistributorPtr distributor_; const UpstreamKafkaConfiguration& configuration_; Thread::ThreadFactory& thread_factory_; + const KafkaConsumerFactory& consumer_factory_; mutable absl::Mutex consumers_mutex_; std::map topic_to_consumer_ ABSL_GUARDED_BY(consumers_mutex_); diff --git a/contrib/kafka/filters/network/test/mesh/shared_consumer_manager_impl_unit_test.cc b/contrib/kafka/filters/network/test/mesh/shared_consumer_manager_impl_unit_test.cc index 5df31d81e7acb..2ec7f57b5f342 100644 --- a/contrib/kafka/filters/network/test/mesh/shared_consumer_manager_impl_unit_test.cc +++ b/contrib/kafka/filters/network/test/mesh/shared_consumer_manager_impl_unit_test.cc @@ -1,3 +1,6 @@ +#include "envoy/common/exception.h" +#include "envoy/thread/thread.h" + #include "contrib/kafka/filters/network/source/mesh/shared_consumer_manager_impl.h" #include "gmock/gmock.h" #include "gtest/gtest.h" @@ -7,17 +10,89 @@ namespace Extensions { namespace NetworkFilters { namespace Kafka { namespace Mesh { +namespace { + +using testing::_; +using testing::Return; + +class MockThreadFactory : public Thread::ThreadFactory { +public: + MOCK_METHOD(Thread::ThreadPtr, createThread, (std::function, Thread::OptionsOptConstRef)); + MOCK_METHOD(Thread::ThreadId, currentThreadId, ()); +}; + +class MockUpstreamKafkaConfiguration : public UpstreamKafkaConfiguration { +public: + MOCK_METHOD(absl::optional, computeClusterConfigForTopic, (const std::string&), + (const)); + MOCK_METHOD((std::pair), getAdvertisedAddress, (), (const)); +}; -class SCMTest : public testing::Test { +class MockKafkaConsumerFactory : public KafkaConsumerFactory { +public: + MOCK_METHOD(KafkaConsumerPtr, createConsumer, + (InboundRecordProcessor&, Thread::ThreadFactory&, const std::string&, const int32_t, + const RawKafkaConfig&), + (const)); +}; + +class SharedConsumerManagerTest : public testing::Test { protected: + MockThreadFactory thread_factory_; + MockUpstreamKafkaConfiguration configuration_; + MockKafkaConsumerFactory consumer_factory_; + + std::unique_ptr makeTestee() { + return std::make_unique(configuration_, thread_factory_, + consumer_factory_); + } }; -TEST_F(SCMTest, ShouldTest) { +TEST_F(SharedConsumerManagerTest, ShouldRegisterTopicOnlyOnce) { // given + const std::string topic1 = "topic1"; + const std::string topic2 = "topic2"; + + const ClusterConfig cluster_config = {"cluster", 1, {}, {}}; + EXPECT_CALL(configuration_, computeClusterConfigForTopic(topic1)) + .WillOnce(Return(cluster_config)); + EXPECT_CALL(configuration_, computeClusterConfigForTopic(topic2)) + .WillOnce(Return(cluster_config)); + + EXPECT_CALL(consumer_factory_, createConsumer(_, _, _, _, _)).Times(2).WillRepeatedly([]() { + return nullptr; + }); + + auto testee = makeTestee(); + // when + for (int i = 0; i < 3; ++i) { + testee->registerConsumerIfAbsent(topic1); + } + for (int i = 0; i < 3; ++i) { + testee->registerConsumerIfAbsent(topic2); + } + // then + ASSERT_EQ(testee->getConsumerCountForTest(), 2); +} + +TEST_F(SharedConsumerManagerTest, ShouldHandleMissingConfig) { + // given + const std::string topic = "topic"; + + EXPECT_CALL(configuration_, computeClusterConfigForTopic(topic)).WillOnce(Return(absl::nullopt)); + + EXPECT_CALL(consumer_factory_, createConsumer(_, _, _, _, _)).Times(0); + + auto testee = makeTestee(); + + // when, then - construction throws and nothing gets registered. + EXPECT_THROW(testee->registerConsumerIfAbsent(topic), EnvoyException); + ASSERT_EQ(testee->getConsumerCountForTest(), 0); } +} // namespace } // namespace Mesh } // namespace Kafka } // namespace NetworkFilters