fix: confluent-kafka: fix incorrect number of arguments (#3922)

* confluent-kafka: fix incorrect number of arguments

* update confluent-kafka dep

* changelog

---------

Co-authored-by: Riccardo Magliocchetti <riccardo.magliocchetti@gmail.com>

Author: papigers

CHANGELOG.md

@@ -94,6 +94,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   ([#4081](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/4081))
 - `opentelemetry-instrumentation-system-metrics`: Use proper numeric `cpython.gc.generation` attribute in CPython metrics, out of spec `generation` attribute is deprecated and will be removed in the future
   ([#4092](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/4092))
+- `opentelemetry-instrumentation-confluent-kafka`: Fix incorrect number of argument to `_inner_wrap_close`
+  ([#3922](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/3922))
 
 ### Breaking changes
 

instrumentation/opentelemetry-instrumentation-confluent-kafka/src/opentelemetry/instrumentation/confluent_kafka/__init__.py

@@ -134,8 +134,8 @@ def produce(self, topic, value=None, *args, **kwargs):  # pylint: disable=keywor
 
 
 class AutoInstrumentedConsumer(Consumer):
-    def __init__(self, config):
-        super().__init__(config)
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
         self._current_consume_span = None
 
     # This method is deliberately implemented in order to allow wrapt to wrap this function
@@ -185,9 +185,9 @@ def __init__(self, consumer: Consumer, tracer: Tracer):
         self._current_consume_span = None
         self._current_context_token = None
 
-    def close(self):
+    def close(self, *args, **kwargs):
         return ConfluentKafkaInstrumentor.wrap_close(
-            self._consumer.close, self
+            self._consumer.close, self, args, kwargs
         )
 
     def committed(self, partitions, timeout=-1):
@@ -308,8 +308,10 @@ def _inner_wrap_consume(func, instance, args, kwargs):
                 func, instance, self._tracer, args, kwargs
             )
 
-        def _inner_wrap_close(func, instance):
-            return ConfluentKafkaInstrumentor.wrap_close(func, instance)
+        def _inner_wrap_close(func, instance, args, kwargs):
+            return ConfluentKafkaInstrumentor.wrap_close(
+                func, instance, args, kwargs
+            )
 
         wrapt.wrap_function_wrapper(
             AutoInstrumentedProducer,
@@ -421,7 +423,7 @@ def wrap_consume(func, instance, tracer, args, kwargs):
         return records
 
     @staticmethod
-    def wrap_close(func, instance):
+    def wrap_close(func, instance, args, kwargs):
         if instance._current_consume_span:
             _end_current_consume_span(instance)
-        func()
+        func(*args, **kwargs)