Merge branch 'main' into fix/aiokafka-tests-fixes

Author: xrmx

CHANGELOG.md

@@ -42,6 +42,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   ([#4259](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/4259))
 - `opentelemetry-instrumentation-aiokafka`: fix `Unclosed AIOKafkaProducer` warning and `RuntimeWarning: coroutine was never awaited` in tests
   ([#4384](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/4384)) 
+- `opentelemetry-instrumentation-aiokafka`: Fix compatibility with aiokafka 0.13 by calling
+  `_key_serializer`/`_value_serializer` directly instead of the internal `_serialize` method
+  whose signature changed in 0.13 from `(topic, key, value)` to `(key, value, headers)`
+  ([#4379](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/4379))
 
 ### Breaking changes
 

instrumentation/opentelemetry-instrumentation-aiokafka/src/opentelemetry/instrumentation/aiokafka/utils.py

@@ -162,9 +162,17 @@ async def _extract_send_partition(
         key = _extract_send_key(args, kwargs)
         value = _extract_send_value(args, kwargs)
         partition = _extract_argument("partition", 3, None, args, kwargs)
-        key_bytes, value_bytes = cast(
-            "tuple[bytes | None, bytes | None]",
-            instance._serialize(topic, key, value),  # type: ignore[reportUnknownMemberType]
+        key_bytes = cast(
+            "bytes | None",
+            instance._key_serializer(key)  # type: ignore[reportUnknownMemberType]
+            if instance._key_serializer  # type: ignore[reportUnknownMemberType]
+            else key,
+        )
+        value_bytes = cast(
+            "bytes | None",
+            instance._value_serializer(value)  # type: ignore[reportUnknownMemberType]
+            if instance._value_serializer  # type: ignore[reportUnknownMemberType]
+            else value,
         )
         valid_types = (bytes, bytearray, memoryview, type(None))
         if (

instrumentation/opentelemetry-instrumentation-aiokafka/test-requirements.txt

@@ -1,4 +1,5 @@
-aiokafka==0.11.0
+aiokafka==0.13.0; python_version >= "3.10"
+aiokafka==0.12.0; python_version < "3.10"
 pytest==7.4.4
 -e opentelemetry-instrumentation
 -e instrumentation/opentelemetry-instrumentation-aiokafka

instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_instrumentation.py

@@ -14,9 +14,10 @@
 
 from __future__ import annotations
 
+import sys
 import uuid
 from typing import Any, Sequence, cast
-from unittest import IsolatedAsyncioTestCase, TestCase, mock
+from unittest import IsolatedAsyncioTestCase, TestCase, mock, skipIf
 
 import aiokafka
 from aiokafka import (
@@ -63,6 +64,10 @@ def test_instrument_api(self) -> None:
         )
 
 
+@skipIf(
+    sys.version_info < (3, 10),
+    "aiokafka >= 0.13 requires Python 3.10+",
+)
 class TestAIOKafkaInstrumentation(TestBase, IsolatedAsyncioTestCase):
     @staticmethod
     def consumer_record_factory(
@@ -126,7 +131,7 @@ async def consumer_factory(**consumer_kwargs: Any) -> AIOKafkaConsumer:
 
     @staticmethod
     async def producer_factory() -> AIOKafkaProducer:
-        producer = AIOKafkaProducer(api_version="1.0")
+        producer = AIOKafkaProducer()
 
         producer.client._wait_on_metadata = mock.AsyncMock()
         producer.client.bootstrap = mock.AsyncMock()

instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_utils.py

@@ -371,7 +371,8 @@ async def test_create_consumer_span(
 
     async def test_kafka_properties_extractor(self):
         aiokafka_instance_mock = mock.Mock()
-        aiokafka_instance_mock._serialize.return_value = None, None
+        aiokafka_instance_mock._key_serializer = None
+        aiokafka_instance_mock._value_serializer = None
         aiokafka_instance_mock._partition.return_value = "partition"
         aiokafka_instance_mock.client._wait_on_metadata = mock.AsyncMock()
         assert (

util/opentelemetry-util-genai/CHANGELOG.md

@@ -18,6 +18,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   ([#4320](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/4320))
 - Add workflow invocation type to genAI utils
   ([https://github.com/open-telemetry/opentelemetry-python-contrib/pull/4310](#4310))
+- Check if upload works at startup in initializer of the `UploadCompletionHook`, instead
+of repeatedly failing on every upload ([https://github.com/open-telemetry/opentelemetry-python-contrib/pull/4390](#4390)).
 
 ## Version 0.3b0 (2026-02-20)
 

util/opentelemetry-util-genai/src/opentelemetry/util/genai/_upload/completion_hook.py

@@ -156,6 +156,30 @@ def __init__(
                 f"Invalid {upload_format=}. Must be one of {_FORMATS}"
             )
         self._format = upload_format
+        self._content_type = (
+            "application/json"
+            if self._format == "json"
+            else "application/jsonl"
+        )
+        test_path = posixpath.join(
+            self._base_path,
+            f".one_off_test_to_see_if_upload_works.{self._format}",
+        )
+        try:
+            with self._fs.open(
+                test_path, "w", content_type=self._content_type
+            ) as file:
+                file.write("\n")
+        except Exception as exception:  # pylint: disable=broad-exception-caught
+            raise ValueError(
+                f"Failed to write file to the following path, upload is not working: {test_path}.\n Got error: {exception}"
+            )
+        # Try to delete the file.. But we don't explicitly ask people to grant the GCS delete IAM permission in our
+        # docs, so if delete fails just leave the file..
+        try:
+            self._fs.rm_file(test_path)  # pyright: ignore[reportUnknownMemberType]
+        except Exception:  # pylint: disable=broad-exception-caught
+            pass
 
         # Use a ThreadPoolExecutor for its queueing and thread management. The semaphore
         # limits the number of queued tasks. If the queue is full, data will be dropped.
@@ -271,13 +295,7 @@ def _do_upload(
             for message_idx, line in enumerate(message_lines):
                 line[_MESSAGE_INDEX_KEY] = message_idx
 
-        content_type = (
-            "application/json"
-            if self._format == "json"
-            else "application/jsonl"
-        )
-
-        with self._fs.open(path, "w", content_type=content_type) as file:
+        with self._fs.open(path, "w", content_type=self._content_type) as file:
             for message in message_lines:
                 gen_ai_json_dump(message, file)
                 file.write("\n")

util/opentelemetry-util-genai/tests/test_upload.py

@@ -102,6 +102,8 @@ def setUp(self):
         self.hook = UploadCompletionHook(
             base_path=BASE_PATH, max_queue_size=MAXSIZE, lru_cache_max_size=5
         )
+        # 1 upload is done when creating the UploadHook to ensure upload works. Reset mock.
+        self.mock_fs.reset_mock()
 
     def tearDown(self) -> None:
         self.hook.shutdown()
@@ -145,6 +147,18 @@ def test_upload_then_shutdown(self):
             "should have uploaded 4 files",
         )
 
+    def test_failed_upload_causes_initializer_to_throw(self):
+        self.mock_fs.open.side_effect = ValueError("Failed for some reason!")
+        with self.assertRaisesRegex(
+            ValueError,
+            "Failed to write file to the following path, upload is not working:",
+        ):
+            UploadCompletionHook(
+                base_path=BASE_PATH,
+                max_queue_size=MAXSIZE,
+                lru_cache_max_size=5,
+            )
+
     def test_lru_cache_works(self):
         record = LogRecord()
         self.hook.on_completion(