feat: support batch (getmany) in aiokafka instrumentation

CHANGELOG.md

@@ -29,6 +29,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   ([#3385](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/3385))
 - `opentelemetry-instrumentation` Make auto instrumentation use the same dependency resolver as manual instrumentation does
   ([#3202](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/3202))
+- `opentelemetry-instrumentation-aiokafka` Add instrumentation of `consumer.getmany` (batch)
+  ([#3257](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/3257))
 
 ### Fixed
 

instrumentation/opentelemetry-instrumentation-aiokafka/pyproject.toml

@@ -29,6 +29,7 @@ dependencies = [
   "opentelemetry-api ~= 1.27",
   "opentelemetry-instrumentation == 0.54b0.dev",
   "opentelemetry-semantic-conventions == 0.54b0.dev",
+  "typing_extensions ~= 4.1",
 ]
 
 [project.optional-dependencies]

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

@@ -69,15 +69,20 @@ async def async_consume_hook(span, record, args, kwargs):
 ___
 """
 
+from __future__ import annotations
+
 from asyncio import iscoroutinefunction
-from typing import Collection
+from typing import TYPE_CHECKING, Collection
 
 import aiokafka
-from wrapt import wrap_function_wrapper
+from wrapt import (
+    wrap_function_wrapper,  # type: ignore[reportUnknownVariableType]
+)
 
 from opentelemetry import trace
 from opentelemetry.instrumentation.aiokafka.package import _instruments
 from opentelemetry.instrumentation.aiokafka.utils import (
+    _wrap_getmany,
     _wrap_getone,
     _wrap_send,
 )
@@ -86,6 +91,21 @@ async def async_consume_hook(span, record, args, kwargs):
 from opentelemetry.instrumentation.utils import unwrap
 from opentelemetry.semconv.schemas import Schemas
 
+if TYPE_CHECKING:
+    from typing import TypedDict
+
+    from typing_extensions import Unpack
+
+    from .utils import ConsumeHookT, ProduceHookT
+
+    class InstrumentKwargs(TypedDict, total=False):
+        tracer_provider: trace.TracerProvider
+        async_produce_hook: ProduceHookT
+        async_consume_hook: ConsumeHookT
+
+    class UninstrumentKwargs(TypedDict, total=False):
+        pass
+
 
 class AIOKafkaInstrumentor(BaseInstrumentor):
     """An instrumentor for kafka module
@@ -95,7 +115,7 @@ class AIOKafkaInstrumentor(BaseInstrumentor):
     def instrumentation_dependencies(self) -> Collection[str]:
         return _instruments
 
-    def _instrument(self, **kwargs):
+    def _instrument(self, **kwargs: Unpack[InstrumentKwargs]):
         """Instruments the kafka module
 
         Args:
@@ -131,7 +151,13 @@ def _instrument(self, **kwargs):
             "getone",
             _wrap_getone(tracer, async_consume_hook),
         )
+        wrap_function_wrapper(
+            aiokafka.AIOKafkaConsumer,
+            "getmany",
+            _wrap_getmany(tracer, async_consume_hook),
+        )
 
-    def _uninstrument(self, **kwargs):
+    def _uninstrument(self, **kwargs: Unpack[UninstrumentKwargs]):
         unwrap(aiokafka.AIOKafkaProducer, "send")
         unwrap(aiokafka.AIOKafkaConsumer, "getone")
+        unwrap(aiokafka.AIOKafkaConsumer, "getmany")