Merge remote-tracking branch 'envoy/main' into main

Signed-off-by: Adam Kotwasinski <adam.kotwasinski@gmail.com>

source/extensions/filters/network/kafka/mesh/command_handlers/BUILD

@@ -11,55 +11,56 @@ licenses(["notice"])  # Apache 2
 envoy_extension_package()
 
 envoy_cc_library(
-    name = "api_versions_lib",
+    name = "metadata_lib",
     srcs = [
-        "api_versions.cc",
+        "metadata.cc",
     ],
     hdrs = [
-        "api_versions.h",
+        "metadata.h",
     ],
     tags = ["skip_on_windows"],
     deps = [
         "//source/common/common:minimal_logger_lib",
         "//source/extensions/filters/network/kafka:kafka_request_parser_lib",
         "//source/extensions/filters/network/kafka:kafka_response_parser_lib",
-        "//source/extensions/filters/network/kafka:tagged_fields_lib",
         "//source/extensions/filters/network/kafka/mesh:abstract_command_lib",
+        "//source/extensions/filters/network/kafka/mesh:upstream_config_lib",
     ],
 )
 
 envoy_cc_library(
-    name = "metadata_lib",
+    name = "produce_lib",
     srcs = [
-        "metadata.cc",
+        "produce.cc",
     ],
     hdrs = [
-        "metadata.h",
+        "produce.h",
     ],
     tags = ["skip_on_windows"],
     deps = [
         "//source/common/common:minimal_logger_lib",
         "//source/extensions/filters/network/kafka:kafka_request_parser_lib",
         "//source/extensions/filters/network/kafka:kafka_response_parser_lib",
         "//source/extensions/filters/network/kafka/mesh:abstract_command_lib",
-        "//source/extensions/filters/network/kafka/mesh:upstream_config_lib",
+        "//source/extensions/filters/network/kafka/mesh:upstream_kafka_facade_lib",
     ],
 )
 
 envoy_cc_library(
-    name = "produce_lib",
+    name = "api_versions_lib",
     srcs = [
-        "produce.cc",
+        "api_versions.cc",
     ],
     hdrs = [
-        "produce.h",
+        "api_versions.h",
     ],
     tags = ["skip_on_windows"],
     deps = [
         "//source/common/common:minimal_logger_lib",
         "//source/extensions/filters/network/kafka:kafka_request_parser_lib",
         "//source/extensions/filters/network/kafka:kafka_response_parser_lib",
+        "//source/extensions/filters/network/kafka:tagged_fields_lib",
         "//source/extensions/filters/network/kafka/mesh:abstract_command_lib",
-        "//source/extensions/filters/network/kafka/mesh:upstream_kafka_facade_lib",
     ],
 )
+

source/extensions/filters/network/kafka/mesh/command_handlers/api_versions.cc

@@ -22,19 +22,20 @@ constexpr int16_t MAX_PRODUCE_SUPPORTED = PRODUCE_REQUEST_MAX_VERSION; /* Genera
 constexpr int16_t MIN_METADATA_SUPPORTED = 1;
 constexpr int16_t MAX_METADATA_SUPPORTED = METADATA_REQUEST_MAX_VERSION; /* Generated value. */
 
-ApiVersionsRequestHolder::ApiVersionsRequestHolder(
-    AbstractRequestListener& filter, const std::shared_ptr<Request<ApiVersionsRequest>> request)
-    : BaseInFlightRequest{filter}, request_{request} {}
+ApiVersionsRequestHolder::ApiVersionsRequestHolder(AbstractRequestListener& filter,
+                                                   const RequestHeader request_header)
+    : BaseInFlightRequest{filter}, request_header_{request_header} {}
 
 // Api Versions requests are immediately ready for answer (as they do not need to reach upstream).
 void ApiVersionsRequestHolder::startProcessing() { notifyFilter(); }
 
-// Because these requests can be trivially handled, they are okay to be sent downstream at any time.
+// Because these requests can be trivially handled, the responses are okay to be sent downstream at
+// any time.
 bool ApiVersionsRequestHolder::finished() const { return true; }
 
 AbstractResponseSharedPtr ApiVersionsRequestHolder::computeAnswer() const {
-  const auto& header = request_->request_header_;
-  const ResponseMetadata metadata = {header.api_key_, header.api_version_, header.correlation_id_};
+  const ResponseMetadata metadata = {request_header_.api_key_, request_header_.api_version_,
+                                     request_header_.correlation_id_};
 
   const int16_t error_code = 0;
   const ApiVersionsResponseKey produce_entry = {PRODUCE_REQUEST_API_KEY, MIN_PRODUCE_SUPPORTED,

source/extensions/filters/network/kafka/mesh/command_handlers/api_versions.h

@@ -9,10 +9,14 @@ namespace NetworkFilters {
 namespace Kafka {
 namespace Mesh {
 
+/**
+ * Api version requests are the first requests sent by Kafka clients to brokers.
+ * We send our customized response to fail clients that might be trying to accomplish something more
+ * than this filter supports.
+ */
 class ApiVersionsRequestHolder : public BaseInFlightRequest {
 public:
-  ApiVersionsRequestHolder(AbstractRequestListener& filter,
-                           const std::shared_ptr<Request<ApiVersionsRequest>> request);
+  ApiVersionsRequestHolder(AbstractRequestListener& filter, const RequestHeader request_header);
 
   void startProcessing() override;
 
@@ -21,8 +25,8 @@ class ApiVersionsRequestHolder : public BaseInFlightRequest {
   AbstractResponseSharedPtr computeAnswer() const override;
 
 private:
-  // Original request.
-  const std::shared_ptr<Request<ApiVersionsRequest>> request_;
+  // Original request header.
+  const RequestHeader request_header_;
 };
 
 } // namespace Mesh

source/extensions/filters/network/kafka/mesh/request_processor.cc

@@ -26,27 +26,22 @@ static void throwOnUnsupportedRequest(const std::string& reason, const RequestHe
 
 void RequestProcessor::onMessage(AbstractRequestSharedPtr arg) {
   switch (arg->request_header_.api_key_) {
-  case /* Produce */ 0:
+  case PRODUCE_REQUEST_API_KEY:
     process(std::dynamic_pointer_cast<Request<ProduceRequest>>(arg));
     break;
-  case /* Metadata */ 3:
+  case METADATA_REQUEST_API_KEY:
     process(std::dynamic_pointer_cast<Request<MetadataRequest>>(arg));
     break;
-  case /* ApiVersions */ 18:
+  case API_VERSIONS_REQUEST_API_KEY:
     process(std::dynamic_pointer_cast<Request<ApiVersionsRequest>>(arg));
     break;
   default:
-    // We got something else than typical Produce request.
+    // Client sent a request we cannot handle right now.
     throwOnUnsupportedRequest("unsupported (bad client API invoked?)", arg->request_header_);
     break;
   } // switch
 }
 
-// We got something that the parser could not handle.
-void RequestProcessor::onFailedParse(RequestParseFailureSharedPtr arg) {
-  throwOnUnsupportedRequest("unknown", arg->request_header_);
-}
-
 void RequestProcessor::process(const std::shared_ptr<Request<ProduceRequest>> request) const {
   auto res = std::make_shared<ProduceRequestHolder>(origin_, upstream_kafka_facade_, request);
   origin_.onRequest(res);
@@ -58,10 +53,15 @@ void RequestProcessor::process(const std::shared_ptr<Request<MetadataRequest>> r
 }
 
 void RequestProcessor::process(const std::shared_ptr<Request<ApiVersionsRequest>> request) const {
-  auto res = std::make_shared<ApiVersionsRequestHolder>(origin_, request);
+  auto res = std::make_shared<ApiVersionsRequestHolder>(origin_, request->request_header_);
   origin_.onRequest(res);
 }
 
+// We got something that the parser could not handle.
+void RequestProcessor::onFailedParse(RequestParseFailureSharedPtr arg) {
+  throwOnUnsupportedRequest("unknown", arg->request_header_);
+}
+
 } // namespace Mesh
 } // namespace Kafka
 } // namespace NetworkFilters

test/extensions/filters/network/kafka/mesh/command_handlers/BUILD

@@ -12,36 +12,37 @@ licenses(["notice"])  # Apache 2
 envoy_package()
 
 envoy_extension_cc_test(
-    name = "api_versions_unit_test",
-    srcs = ["api_versions_unit_test.cc"],
+    name = "metadata_unit_test",
+    srcs = ["metadata_unit_test.cc"],
     extension_names = ["envoy.filters.network.kafka_broker"],
     tags = ["skip_on_windows"],
     deps = [
-        "//source/extensions/filters/network/kafka/mesh/command_handlers:api_versions_lib",
+        "//source/extensions/filters/network/kafka/mesh/command_handlers:metadata_lib",
         "//test/mocks/network:network_mocks",
         "//test/mocks/stats:stats_mocks",
     ],
 )
 
 envoy_extension_cc_test(
-    name = "metadata_unit_test",
-    srcs = ["metadata_unit_test.cc"],
+    name = "produce_unit_test",
+    srcs = ["produce_unit_test.cc"],
     extension_names = ["envoy.filters.network.kafka_broker"],
     tags = ["skip_on_windows"],
     deps = [
-        "//source/extensions/filters/network/kafka/mesh/command_handlers:metadata_lib",
+        "//source/extensions/filters/network/kafka/mesh/command_handlers:produce_lib",
         "//test/mocks/network:network_mocks",
         "//test/mocks/stats:stats_mocks",
     ],
 )
 
 envoy_extension_cc_test(
-    name = "produce_unit_test",
-    srcs = ["produce_unit_test.cc"],
+    name = "api_versions_unit_test",
+    srcs = ["api_versions_unit_test.cc"],
+    # This name needs to be changed after we have the mesh filter ready.
     extension_names = ["envoy.filters.network.kafka_broker"],
     tags = ["skip_on_windows"],
     deps = [
-        "//source/extensions/filters/network/kafka/mesh/command_handlers:produce_lib",
+        "//source/extensions/filters/network/kafka/mesh/command_handlers:api_versions_lib",
         "//test/mocks/network:network_mocks",
         "//test/mocks/stats:stats_mocks",
     ],

test/extensions/filters/network/kafka/mesh/command_handlers/api_versions_unit_test.cc

@@ -20,10 +20,8 @@ TEST(ApiVersionsTest, shouldBeAlwaysReadyForAnswer) {
   // given
   MockAbstractRequestListener filter;
   EXPECT_CALL(filter, onRequestReadyForAnswer());
-  const RequestHeader header = {0, 0, 0, absl::nullopt};
-  const ApiVersionsRequest data = {};
-  const auto message = std::make_shared<Request<ApiVersionsRequest>>(header, data);
-  ApiVersionsRequestHolder testee = {filter, message};
+  const RequestHeader header = {API_VERSIONS_REQUEST_API_KEY, 0, 0, absl::nullopt};
+  ApiVersionsRequestHolder testee = {filter, header};
 
   // when, then - invoking should immediately notify the filter.
   testee.startProcessing();

test/extensions/filters/network/kafka/mesh/request_processor_unit_test.cc

@@ -47,7 +47,7 @@ class RequestProcessorTest : public testing::Test {
 
 TEST_F(RequestProcessorTest, ShouldProcessProduceRequest) {
   // given
-  const RequestHeader header = {0, 0, 0, absl::nullopt};
+  const RequestHeader header = {PRODUCE_REQUEST_API_KEY, 0, 0, absl::nullopt};
   const ProduceRequest data = {0, 0, {}};
   const auto message = std::make_shared<Request<ProduceRequest>>(header, data);
 
@@ -63,7 +63,7 @@ TEST_F(RequestProcessorTest, ShouldProcessProduceRequest) {
 
 TEST_F(RequestProcessorTest, ShouldProcessMetadataRequest) {
   // given
-  const RequestHeader header = {3, 0, 0, absl::nullopt};
+  const RequestHeader header = {METADATA_REQUEST_API_KEY, 0, 0, absl::nullopt};
   const MetadataRequest data = {absl::nullopt};
   const auto message = std::make_shared<Request<MetadataRequest>>(header, data);
 
@@ -79,7 +79,7 @@ TEST_F(RequestProcessorTest, ShouldProcessMetadataRequest) {
 
 TEST_F(RequestProcessorTest, ShouldProcessApiVersionsRequest) {
   // given
-  const RequestHeader header = {18, 0, 0, absl::nullopt};
+  const RequestHeader header = {API_VERSIONS_REQUEST_API_KEY, 0, 0, absl::nullopt};
   const ApiVersionsRequest data = {};
   const auto message = std::make_shared<Request<ApiVersionsRequest>>(header, data);
 
@@ -95,7 +95,7 @@ TEST_F(RequestProcessorTest, ShouldProcessApiVersionsRequest) {
 
 TEST_F(RequestProcessorTest, ShouldHandleUnsupportedRequest) {
   // given
-  const RequestHeader header = {2, 0, 0, absl::nullopt};
+  const RequestHeader header = {LIST_OFFSET_REQUEST_API_KEY, 0, 0, absl::nullopt};
   const ListOffsetRequest data = {0, {}};
   const auto message = std::make_shared<Request<ListOffsetRequest>>(header, data);