kafka: upgrade to kafka 3.3.1 (#23354)

Signed-off-by: Adam Kotwasinski <[email protected]>

Author: adamkotwasinski

bazel/repository_locations.bzl

@@ -1205,13 +1205,13 @@ REPOSITORY_LOCATIONS_SPEC = dict(
         project_name = "Kafka (source)",
         project_desc = "Open-source distributed event streaming platform",
         project_url = "https://kafka.apache.org",
-        version = "3.2.3",
-        sha256 = "a5b45221e215696769f6ccb741c1e91fa1e18194d7ce02fade797f09efaae03a",
+        version = "3.3.1",
+        sha256 = "aab244e6ad1d63a830af1776e1810d355e36900be1e8e4e66b7555af8639e2d2",
         strip_prefix = "kafka-{version}/clients/src/main/resources/common/message",
         urls = ["https://github.com/apache/kafka/archive/{version}.zip"],
         use_category = ["dataplane_ext"],
         extensions = ["envoy.filters.network.kafka_broker", "envoy.filters.network.kafka_mesh"],
-        release_date = "2022-09-13",
+        release_date = "2022-09-29",
         cpe = "cpe:2.3:a:apache:kafka:*",
         license = "Apache-2.0",
         license_url = "https://github.com/apache/kafka/blob/{version}/LICENSE",
@@ -1235,11 +1235,11 @@ REPOSITORY_LOCATIONS_SPEC = dict(
         project_name = "Kafka (server binary)",
         project_desc = "Open-source distributed event streaming platform",
         project_url = "https://kafka.apache.org",
-        version = "3.2.3",
-        sha256 = "b6f91bc013fcdccd73977d49e20eaebb8fcb121a89a0803d11a9b8f1fc93db80",
+        version = "3.3.1",
+        sha256 = "18ad8a365fb111de249d3bb8bf3c96cd1af060ec8fb3e3d1fc4a7ae10d9042de",
         strip_prefix = "kafka_2.13-{version}",
         urls = ["https://archive.apache.org/dist/kafka/{version}/kafka_2.13-{version}.tgz"],
-        release_date = "2022-09-09",
+        release_date = "2022-10-02",
         use_category = ["test_only"],
     ),
     kafka_python_client = dict(

contrib/kafka/filters/network/source/protocol/generator.py

@@ -238,8 +238,8 @@ def parse_field(self, field_spec, highest_possible_version):
 
     def parse_type(self, type_name, field_spec, highest_possible_version):
         """
-    Parse a given type element - returns an array type, primitive (e.g. uint32_t) or complex one.
-    """
+        Parse a given type element - returns an array type, primitive (e.g. uint32_t) or complex one.
+        """
         if (type_name.startswith('[]')):
             # In spec files, array types are defined as `[]underlying_type` instead of having its own
             # element with type inside.
@@ -474,8 +474,8 @@ def is_printable(self):
 
 class Primitive(TypeSpecification):
     """
-  Represents a Kafka primitive value.
-  """
+    Represents a Kafka primitive value.
+    """
 
     USABLE_PRIMITIVE_TYPE_NAMES = [
         'bool', 'int8', 'int16', 'int32', 'int64', 'uint16', 'float64', 'string', 'bytes',
@@ -562,7 +562,7 @@ class Primitive(TypeSpecification):
     def __init__(self, name, custom_default_value):
         self.original_name = name
         self.name = Primitive.compute(name, Primitive.KAFKA_TYPE_TO_ENVOY_TYPE)
-        self.custom_default_value = custom_default_value
+        self.custom_default_value = Primitive.sanitize_value(self.name, custom_default_value)
 
     @staticmethod
     def compute(name, map):
@@ -571,6 +571,18 @@ def compute(name, map):
         else:
             raise ValueError(name)
 
+    @staticmethod
+    def sanitize_value(type, arg):
+        """
+        Unfortunately we cannot print Python True/False straight into C++ code, so we lowercase.
+        """
+        if arg is None:
+            return None
+        if 'bool' == type:
+            return str(arg).lower()
+        else:
+            return arg
+
     def compute_declaration_chain(self):
         # Primitives need no declarations.
         return []

docs/root/configuration/listeners/network_filters/kafka_broker_filter.rst

@@ -5,7 +5,7 @@ Kafka Broker filter
 
 The Apache Kafka broker filter decodes the client protocol for
 `Apache Kafka <https://kafka.apache.org/>`_, both the requests and responses in the payload.
-The message versions in `Kafka 3.2.3 <http://kafka.apache.org/32/protocol.html#protocol_api_keys>`_
+The message versions in `Kafka 3.3.1 <http://kafka.apache.org/33/protocol.html#protocol_api_keys>`_
 are supported.
 The filter attempts not to influence the communication between client and brokers, so the messages
 that could not be decoded (due to Kafka client or broker running a newer version than supported by

docs/root/configuration/listeners/network_filters/kafka_mesh_filter.rst

@@ -6,7 +6,7 @@ Kafka Mesh filter
 The Apache Kafka mesh filter provides a facade for `Apache Kafka <https://kafka.apache.org/>`_
 producers. Produce requests sent to this filter insance can be forwarded to one of multiple
 clusters, depending on configured forwarding rules. Corresponding message versions from
-Kafka 3.2.3 are supported.
+Kafka 3.3.1 are supported.
 
 * This filter should be configured with the type URL ``type.googleapis.com/envoy.extensions.filters.network.kafka_mesh.v3alpha.KafkaMesh``.
 * :ref:`v3 API reference <envoy_v3_api_msg_extensions.filters.network.kafka_mesh.v3alpha.KafkaMesh>`