envoyproxy · mattklein123 · Dec 13, 2022 · Dec 11, 2022 · Feb 16, 2022 · Dec 12, 2022
@@ -2,6 +2,7 @@
 
 #include <memory>
 #include <string>
+#include <utility>
 #include <vector>
 
 #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<std::string, int32_t>;
+
 } // namespace Kafka
 } // namespace NetworkFilters
 } // namespace Extensions

@@ -110,6 +110,39 @@ 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 = [
+        ":librdkafka_utils_lib",
+        ":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 = [

@@ -0,0 +1,32 @@
+#pragma once
+
+#include <memory>
+#include <string>
+
+#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 per mesh-filter type.
+ */
+class SharedConsumerManager {
+public:
+  virtual ~SharedConsumerManager() = default;
+
+  // 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 SharedConsumerManagerPtr = std::unique_ptr<SharedConsumerManager>;
+
+} // namespace Mesh
+} // namespace Kafka
+} // namespace NetworkFilters
+} // namespace Extensions
+} // namespace Envoy
@@ -0,0 +1,102 @@
+#include "contrib/kafka/filters/network/source/mesh/shared_consumer_manager_impl.h"
+
+#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 {
+
+// 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<RichKafkaConsumer>(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)
+    : SharedConsumerManagerImpl{configuration, thread_factory,
+                                KafkaConsumerFactoryImpl::getDefaultInstance()} {}
+
+SharedConsumerManagerImpl::SharedConsumerManagerImpl(
+    const UpstreamKafkaConfiguration& configuration, Thread::ThreadFactory& thread_factory,
+    const KafkaConsumerFactory& consumer_factory)
+    : distributor_{std::make_unique<RecordDistributor>()}, configuration_{configuration},
+      thread_factory_{thread_factory}, consumer_factory_{consumer_factory} {}
+
+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.
+  if (topic_to_consumer_.end() == it) {
+    registerNewConsumer(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<ClusterConfig> 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 = 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));
+}
+
+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;
+}
+
+void RecordDistributor::receive(InboundRecordSharedPtr) {
+  // TODO (adam.kotwasinski) To implement in future commits.
+}
+
+} // namespace Mesh
+} // namespace Kafka
+} // namespace NetworkFilters
+} // namespace Extensions
+} // namespace Envoy
@@ -0,0 +1,92 @@
+#pragma once
+
+#include <map>
+
+#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/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"
+
+namespace Envoy {
+namespace Extensions {
+namespace NetworkFilters {
+namespace Kafka {
+namespace Mesh {
+
+/**
+ * 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).
+ * Basically core of Fetch-handling business logic.
+ */
+class RecordDistributor : public InboundRecordProcessor {
+public:
+  // InboundRecordProcessor
+  bool waitUntilInterest(const std::string& topic, const int32_t timeout_ms) const override;
+
+  // InboundRecordProcessor
+  void receive(InboundRecordSharedPtr message) override;
+};
+
+using RecordDistributorPtr = std::unique_ptr<RecordDistributor>;
+
+/**
+ * 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<Logger::Id::kafka> {
+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_);
+
+  RecordDistributorPtr distributor_;
+
+  const UpstreamKafkaConfiguration& configuration_;
+  Thread::ThreadFactory& thread_factory_;
+  const KafkaConsumerFactory& consumer_factory_;
+
+  mutable absl::Mutex consumers_mutex_;
+  std::map<std::string, KafkaConsumerPtr> topic_to_consumer_ ABSL_GUARDED_BY(consumers_mutex_);
+};
+
+} // namespace Mesh
+} // namespace Kafka
+} // namespace NetworkFilters
+} // namespace Extensions
+} // namespace Envoy
@@ -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<std::string, std::string> 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;
   }
 

@@ -39,9 +39,14 @@ struct ClusterConfig {
   // producer property.
   std::map<std::string, std::string> 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<std::string, std::string> 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_;
   }
 };
 

@@ -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"],

@@ -33,7 +33,7 @@ TEST(MetadataTest, shouldBeAlwaysReadyForAnswer) {
   const std::pair<std::string, int32_t> 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.