Merge branch 'develop' into dependabot/github_actions/docker/setup-buildx-action-3.11.1

Author: leandrodamascena

CHANGELOG.md

@@ -4,36 +4,65 @@
 <a name="unreleased"></a>
 # Unreleased
 
+## Maintenance
+
+* **deps:** bump redis from 5.3.0 to 6.2.0 ([#6827](https://github.com/aws-powertools/powertools-lambda-python/issues/6827))
+
+
+<a name="v3.15.1"></a>
+## [v3.15.1] - 2025-06-20
+## Features
+
+* **kafka:** add logic to handle protobuf deserialization ([#6841](https://github.com/aws-powertools/powertools-lambda-python/issues/6841))
+
+## Maintenance
+
+* version bump
+* **ci:** new pre-release 3.15.1a0 ([#6839](https://github.com/aws-powertools/powertools-lambda-python/issues/6839))
+
+
+<a name="v3.15.0"></a>
+## [v3.15.0] - 2025-06-19
 ## Bug Fixes
 
 * **bedrock_agent:** fix querystring field resolution ([#6777](https://github.com/aws-powertools/powertools-lambda-python/issues/6777))
 
 ## Documentation
 
+* **kafka:** add kafka documentation ([#6834](https://github.com/aws-powertools/powertools-lambda-python/issues/6834))
 * **public_reference:** add Instil as a public reference ([#6763](https://github.com/aws-powertools/powertools-lambda-python/issues/6763))
 
+## Features
+
+* **kafka:** add support for Confluence Producers ([#6833](https://github.com/aws-powertools/powertools-lambda-python/issues/6833))
+* **kafka:** New Kafka utility ([#6821](https://github.com/aws-powertools/powertools-lambda-python/issues/6821))
+
 ## Maintenance
 
+* version bump
+* **ci:** new pre-release 3.14.1a6 ([#6830](https://github.com/aws-powertools/powertools-lambda-python/issues/6830))
+* **ci:** new pre-release 3.14.1a5 ([#6820](https://github.com/aws-powertools/powertools-lambda-python/issues/6820))
+* **ci:** new pre-release 3.14.1a0 ([#6773](https://github.com/aws-powertools/powertools-lambda-python/issues/6773))
 * **ci:** new pre-release 3.14.1a4 ([#6812](https://github.com/aws-powertools/powertools-lambda-python/issues/6812))
 * **ci:** new pre-release 3.14.1a3 ([#6797](https://github.com/aws-powertools/powertools-lambda-python/issues/6797))
 * **ci:** new pre-release 3.14.1a1 ([#6778](https://github.com/aws-powertools/powertools-lambda-python/issues/6778))
 * **ci:** new pre-release 3.14.1a2 ([#6788](https://github.com/aws-powertools/powertools-lambda-python/issues/6788))
-* **ci:** new pre-release 3.14.1a0 ([#6773](https://github.com/aws-powertools/powertools-lambda-python/issues/6773))
-* **deps:** bump mkdocstrings-python from 1.16.11 to 1.16.12 ([#6765](https://github.com/aws-powertools/powertools-lambda-python/issues/6765))
 * **deps:** bump mkdocstrings-python from 1.16.11 to 1.16.12 in /docs ([#6768](https://github.com/aws-powertools/powertools-lambda-python/issues/6768))
+* **deps:** bump mkdocstrings-python from 1.16.11 to 1.16.12 ([#6765](https://github.com/aws-powertools/powertools-lambda-python/issues/6765))
 * **deps:** bump protobuf from 6.31.0 to 6.31.1 ([#6815](https://github.com/aws-powertools/powertools-lambda-python/issues/6815))
+* **deps-dev:** bump boto3-stubs from 1.38.29 to 1.38.30 ([#6772](https://github.com/aws-powertools/powertools-lambda-python/issues/6772))
 * **deps-dev:** bump aws-cdk from 2.1017.1 to 2.1018.0 ([#6775](https://github.com/aws-powertools/powertools-lambda-python/issues/6775))
+* **deps-dev:** bump boto3-stubs from 1.38.33 to 1.38.35 ([#6796](https://github.com/aws-powertools/powertools-lambda-python/issues/6796))
+* **deps-dev:** bump aws-cdk from 2.1018.0 to 2.1018.1 ([#6803](https://github.com/aws-powertools/powertools-lambda-python/issues/6803))
+* **deps-dev:** bump boto3-stubs from 1.38.30 to 1.38.31 ([#6776](https://github.com/aws-powertools/powertools-lambda-python/issues/6776))
+* **deps-dev:** bump requests from 2.32.3 to 2.32.4 ([#6789](https://github.com/aws-powertools/powertools-lambda-python/issues/6789))
+* **deps-dev:** bump boto3-stubs from 1.38.28 to 1.38.29 ([#6764](https://github.com/aws-powertools/powertools-lambda-python/issues/6764))
 * **deps-dev:** bump ruff from 0.11.12 to 0.11.13 ([#6780](https://github.com/aws-powertools/powertools-lambda-python/issues/6780))
-* **deps-dev:** bump requests from 2.32.3 to 2.32.4 ([#6787](https://github.com/aws-powertools/powertools-lambda-python/issues/6787))
-* **deps-dev:** bump boto3-stubs from 1.38.29 to 1.38.30 ([#6772](https://github.com/aws-powertools/powertools-lambda-python/issues/6772))
 * **deps-dev:** bump boto3-stubs from 1.38.31 to 1.38.33 ([#6786](https://github.com/aws-powertools/powertools-lambda-python/issues/6786))
-* **deps-dev:** bump boto3-stubs from 1.38.28 to 1.38.29 ([#6764](https://github.com/aws-powertools/powertools-lambda-python/issues/6764))
-* **deps-dev:** bump requests from 2.32.3 to 2.32.4 ([#6789](https://github.com/aws-powertools/powertools-lambda-python/issues/6789))
-* **deps-dev:** bump boto3-stubs from 1.38.33 to 1.38.35 ([#6796](https://github.com/aws-powertools/powertools-lambda-python/issues/6796))
 * **deps-dev:** bump aws-cdk-aws-lambda-python-alpha from 2.200.0a0 to 2.200.1a0 ([#6766](https://github.com/aws-powertools/powertools-lambda-python/issues/6766))
 * **deps-dev:** bump aws-cdk-lib from 2.200.0 to 2.200.1 ([#6767](https://github.com/aws-powertools/powertools-lambda-python/issues/6767))
-* **deps-dev:** bump aws-cdk from 2.1018.0 to 2.1018.1 ([#6803](https://github.com/aws-powertools/powertools-lambda-python/issues/6803))
-* **deps-dev:** bump boto3-stubs from 1.38.30 to 1.38.31 ([#6776](https://github.com/aws-powertools/powertools-lambda-python/issues/6776))
+* **deps-dev:** bump pytest-cov from 6.1.1 to 6.2.1 ([#6800](https://github.com/aws-powertools/powertools-lambda-python/issues/6800))
+* **deps-dev:** bump requests from 2.32.3 to 2.32.4 ([#6787](https://github.com/aws-powertools/powertools-lambda-python/issues/6787))
 
 
 <a name="v3.14.0"></a>
@@ -6648,7 +6677,9 @@
 * Merge pull request [#5](https://github.com/aws-powertools/powertools-lambda-python/issues/5) from jfuss/feat/python38
 
 
-[Unreleased]: https://github.com/aws-powertools/powertools-lambda-python/compare/v3.14.0...HEAD
+[Unreleased]: https://github.com/aws-powertools/powertools-lambda-python/compare/v3.15.1...HEAD
+[v3.15.1]: https://github.com/aws-powertools/powertools-lambda-python/compare/v3.15.0...v3.15.1
+[v3.15.0]: https://github.com/aws-powertools/powertools-lambda-python/compare/v3.14.0...v3.15.0
 [v3.14.0]: https://github.com/aws-powertools/powertools-lambda-python/compare/v3.13.0...v3.14.0
 [v3.13.0]: https://github.com/aws-powertools/powertools-lambda-python/compare/v3.12.0...v3.13.0
 [v3.12.0]: https://github.com/aws-powertools/powertools-lambda-python/compare/v3.11.0...v3.12.0

aws_lambda_powertools/shared/version.py

@@ -1,3 +1,3 @@
 """Exposes version constant to avoid circular dependencies."""
 
-VERSION = "3.14.1a5"
+VERSION = "3.15.1"

aws_lambda_powertools/utilities/kafka/consumer_records.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import logging
 from functools import cached_property
 from typing import TYPE_CHECKING, Any
 
@@ -13,6 +14,8 @@
 
     from aws_lambda_powertools.utilities.kafka.schema_config import SchemaConfig
 
+logger = logging.getLogger(__name__)
+
 
 class ConsumerRecordRecords(KafkaEventRecordBase):
     """
@@ -31,18 +34,24 @@ def key(self) -> Any:
         if not key:
             return None
 
+        logger.debug("Deserializing key field")
+
         # Determine schema type and schema string
         schema_type = None
-        schema_str = None
+        schema_value = None
         output_serializer = None
 
         if self.schema_config and self.schema_config.key_schema_type:
             schema_type = self.schema_config.key_schema_type
-            schema_str = self.schema_config.key_schema
+            schema_value = self.schema_config.key_schema
             output_serializer = self.schema_config.key_output_serializer
 
         # Always use get_deserializer if None it will default to DEFAULT
-        deserializer = get_deserializer(schema_type, schema_str)
+        deserializer = get_deserializer(
+            schema_type=schema_type,
+            schema_value=schema_value,
+            field_metadata=self.key_schema_metadata,
+        )
         deserialized_value = deserializer.deserialize(key)
 
         # Apply output serializer if specified
@@ -57,16 +66,22 @@ def value(self) -> Any:
 
         # Determine schema type and schema string
         schema_type = None
-        schema_str = None
+        schema_value = None
         output_serializer = None
 
+        logger.debug("Deserializing value field")
+
         if self.schema_config and self.schema_config.value_schema_type:
             schema_type = self.schema_config.value_schema_type
-            schema_str = self.schema_config.value_schema
+            schema_value = self.schema_config.value_schema
             output_serializer = self.schema_config.value_output_serializer
 
         # Always use get_deserializer if None it will default to DEFAULT
-        deserializer = get_deserializer(schema_type, schema_str)
+        deserializer = get_deserializer(
+            schema_type=schema_type,
+            schema_value=schema_value,
+            field_metadata=self.value_schema_metadata,
+        )
         deserialized_value = deserializer.deserialize(value)
 
         # Apply output serializer if specified

aws_lambda_powertools/utilities/kafka/deserializer/avro.py

@@ -1,6 +1,8 @@
 from __future__ import annotations
 
 import io
+import logging
+from typing import Any
 
 from avro.io import BinaryDecoder, DatumReader
 from avro.schema import parse as parse_schema
@@ -9,8 +11,11 @@
 from aws_lambda_powertools.utilities.kafka.exceptions import (
     KafkaConsumerAvroSchemaParserError,
     KafkaConsumerDeserializationError,
+    KafkaConsumerDeserializationFormatMismatch,
 )
 
+logger = logging.getLogger(__name__)
+
 
 class AvroDeserializer(DeserializerBase):
     """
@@ -20,10 +25,11 @@ class AvroDeserializer(DeserializerBase):
     a provided Avro schema definition.
     """
 
-    def __init__(self, schema_str: str):
+    def __init__(self, schema_str: str, field_metadata: dict[str, Any] | None = None):
         try:
             self.parsed_schema = parse_schema(schema_str)
             self.reader = DatumReader(self.parsed_schema)
+            self.field_metatada = field_metadata
         except Exception as e:
             raise KafkaConsumerAvroSchemaParserError(
                 f"Invalid Avro schema. Please ensure the provided avro schema is valid: {type(e).__name__}: {str(e)}",
@@ -60,6 +66,13 @@ def deserialize(self, data: bytes | str) -> object:
         ... except KafkaConsumerDeserializationError as e:
         ...     print(f"Failed to deserialize: {e}")
         """
+        data_format = self.field_metatada.get("dataFormat") if self.field_metatada else None
+
+        if data_format and data_format != "AVRO":
+            raise KafkaConsumerDeserializationFormatMismatch(f"Expected data is AVRO but you sent {data_format}")
+
+        logger.debug("Deserializing data with AVRO format")
+
         try:
             value = self._decode_input(data)
             bytes_reader = io.BytesIO(value)

aws_lambda_powertools/utilities/kafka/deserializer/default.py

@@ -1,9 +1,12 @@
 from __future__ import annotations
 
 import base64
+import logging
 
 from aws_lambda_powertools.utilities.kafka.deserializer.base import DeserializerBase
 
+logger = logging.getLogger(__name__)
+
 
 class DefaultDeserializer(DeserializerBase):
     """
@@ -43,4 +46,5 @@ def deserialize(self, data: bytes | str) -> str:
         >>> result = deserializer.deserialize(bytes_data)
         >>> print(result == bytes_data)  # Output: True
         """
+        logger.debug("Deserializing data with primitives types")
         return base64.b64decode(data).decode("utf-8")

aws_lambda_powertools/utilities/kafka/deserializer/deserializer.py

@@ -13,21 +13,27 @@
 _deserializer_cache: dict[str, DeserializerBase] = {}
 
 
-def _get_cache_key(schema_type: str | object, schema_value: Any) -> str:
+def _get_cache_key(schema_type: str | object, schema_value: Any, field_metadata: dict[str, Any]) -> str:
+    schema_metadata = None
+
+    if field_metadata:
+        schema_metadata = field_metadata.get("schemaId")
+
     if schema_value is None:
-        return str(schema_type)
+        schema_hash = f"{str(schema_type)}_{schema_metadata}"
 
     if isinstance(schema_value, str):
+        hashable_value = f"{schema_value}_{schema_metadata}"
         # For string schemas like Avro, hash the content
-        schema_hash = hashlib.md5(schema_value.encode("utf-8"), usedforsecurity=False).hexdigest()
+        schema_hash = hashlib.md5(hashable_value.encode("utf-8"), usedforsecurity=False).hexdigest()
     else:
         # For objects like Protobuf, use the object id
-        schema_hash = str(id(schema_value))
+        schema_hash = f"{str(id(schema_value))}_{schema_metadata}"
 
     return f"{schema_type}_{schema_hash}"
 
 
-def get_deserializer(schema_type: str | object, schema_value: Any) -> DeserializerBase:
+def get_deserializer(schema_type: str | object, schema_value: Any, field_metadata: Any) -> DeserializerBase:
     """
     Factory function to get the appropriate deserializer based on schema type.
 
@@ -75,7 +81,7 @@ def get_deserializer(schema_type: str | object, schema_value: Any) -> Deserializ
     """
 
     # Generate a cache key based on schema type and value
-    cache_key = _get_cache_key(schema_type, schema_value)
+    cache_key = _get_cache_key(schema_type, schema_value, field_metadata)
 
     # Check if we already have this deserializer in cache
     if cache_key in _deserializer_cache:
@@ -87,14 +93,14 @@ def get_deserializer(schema_type: str | object, schema_value: Any) -> Deserializ
         # Import here to avoid dependency if not used
         from aws_lambda_powertools.utilities.kafka.deserializer.avro import AvroDeserializer
 
-        deserializer = AvroDeserializer(schema_value)
+        deserializer = AvroDeserializer(schema_str=schema_value, field_metadata=field_metadata)
     elif schema_type == "PROTOBUF":
         # Import here to avoid dependency if not used
         from aws_lambda_powertools.utilities.kafka.deserializer.protobuf import ProtobufDeserializer
 
-        deserializer = ProtobufDeserializer(schema_value)
+        deserializer = ProtobufDeserializer(message_class=schema_value, field_metadata=field_metadata)
     elif schema_type == "JSON":
-        deserializer = JsonDeserializer()
+        deserializer = JsonDeserializer(field_metadata=field_metadata)
 
     else:
         # Default to no-op deserializer

aws_lambda_powertools/utilities/kafka/deserializer/json.py

@@ -2,9 +2,16 @@
 
 import base64
 import json
+import logging
+from typing import Any
 
 from aws_lambda_powertools.utilities.kafka.deserializer.base import DeserializerBase
-from aws_lambda_powertools.utilities.kafka.exceptions import KafkaConsumerDeserializationError
+from aws_lambda_powertools.utilities.kafka.exceptions import (
+    KafkaConsumerDeserializationError,
+    KafkaConsumerDeserializationFormatMismatch,
+)
+
+logger = logging.getLogger(__name__)
 
 
 class JsonDeserializer(DeserializerBase):
@@ -15,6 +22,9 @@ class JsonDeserializer(DeserializerBase):
     into Python dictionaries.
     """
 
+    def __init__(self, field_metadata: dict[str, Any] | None = None):
+        self.field_metatada = field_metadata
+
     def deserialize(self, data: bytes | str) -> dict:
         """
         Deserialize JSON data to a Python dictionary.
@@ -45,6 +55,14 @@ def deserialize(self, data: bytes | str) -> dict:
         ... except KafkaConsumerDeserializationError as e:
         ...     print(f"Failed to deserialize: {e}")
         """
+
+        data_format = self.field_metatada.get("dataFormat") if self.field_metatada else None
+
+        if data_format and data_format != "JSON":
+            raise KafkaConsumerDeserializationFormatMismatch(f"Expected data is JSON but you sent {data_format}")
+
+        logger.debug("Deserializing data with JSON format")
+
         try:
             return json.loads(base64.b64decode(data).decode("utf-8"))
         except Exception as e: