feat: Update kafka python to support kafka-python 3 and remove support for kafka-python-ng

Author: rjduffner

.changelog/4785.added

@@ -0,0 +1 @@
+Update kafka python to support kafka-python 3 and remove support for kafka-python-ng

instrumentation/opentelemetry-instrumentation-kafka-python/pyproject.toml

@@ -31,10 +31,8 @@ dependencies = [
 ]
 
 [project.optional-dependencies]
-instruments = []
-instruments-any = [
-  "kafka-python >= 2.0, < 3.0",
-  "kafka-python-ng >= 2.0, < 3.0"
+instruments = [
+  "kafka-python >= 2.0, < 4.0",
 ]
 
 [project.entry-points.opentelemetry_instrumentor]

instrumentation/opentelemetry-instrumentation-kafka-python/src/opentelemetry/instrumentation/kafka/__init__.py

@@ -82,19 +82,14 @@ def process_msg(message):
 ---
 """
 
-from importlib.metadata import PackageNotFoundError, distribution
 from typing import Collection
 
 import kafka
 from wrapt import wrap_function_wrapper
 
 from opentelemetry import trace
 from opentelemetry.instrumentation.instrumentor import BaseInstrumentor
-from opentelemetry.instrumentation.kafka.package import (
-    _instruments_any,
-    _instruments_kafka_python,
-    _instruments_kafka_python_ng,
-)
+from opentelemetry.instrumentation.kafka.package import _instruments
 from opentelemetry.instrumentation.kafka.utils import _wrap_next, _wrap_send
 from opentelemetry.instrumentation.kafka.version import __version__
 from opentelemetry.instrumentation.utils import unwrap
@@ -106,24 +101,7 @@ class KafkaInstrumentor(BaseInstrumentor):
     """
 
     def instrumentation_dependencies(self) -> Collection[str]:
-        # Determine which package of kafka-python is installed
-        # Right now there are two packages, kafka-python and kafka-python-ng
-        # The latter is a fork of the former because the former is connected
-        # to a pypi namespace that the current maintainers cannot access
-        # https://github.com/dpkp/kafka-python/issues/2431
-        try:
-            distribution("kafka-python-ng")
-            return (_instruments_kafka_python_ng,)
-        except PackageNotFoundError:
-            pass
-
-        try:
-            distribution("kafka-python")
-            return (_instruments_kafka_python,)
-        except PackageNotFoundError:
-            pass
-
-        return _instruments_any
+        return _instruments
 
     def _instrument(self, **kwargs):
         """Instruments the kafka module

instrumentation/opentelemetry-instrumentation-kafka-python/src/opentelemetry/instrumentation/kafka/package.py

@@ -2,8 +2,6 @@
 # SPDX-License-Identifier: Apache-2.0
 
 
-_instruments_kafka_python = "kafka-python >= 2.0, < 3.0"
-_instruments_kafka_python_ng = "kafka-python-ng >= 2.0, < 3.0"
+_instruments_kafka_python = "kafka-python >= 2.0, < 4.0"
 
-_instruments = ()
-_instruments_any = (_instruments_kafka_python, _instruments_kafka_python_ng)
+_instruments = (_instruments_kafka_python,)

…ation-kafka-python/test-requirements.txt …ion-kafka-python/test-requirements-0.txtinstrumentation/opentelemetry-instrumentation-kafka-python/test-requirements.txt renamed to instrumentation/opentelemetry-instrumentation-kafka-python/test-requirements-0.txt instrumentation/opentelemetry-instrumentation-kafka-python/test-requirements.txt renamed to instrumentation/opentelemetry-instrumentation-kafka-python/test-requirements-0.txt

@@ -1,3 +1,4 @@
+kafka-python>=2.0.0,<3.0.0
 asgiref==3.8.1
 Deprecated==1.2.14
 iniconfig==2.0.0

…on-kafka-python/test-requirements-ng.txt …ion-kafka-python/test-requirements-1.txtinstrumentation/opentelemetry-instrumentation-kafka-python/test-requirements-ng.txt renamed to instrumentation/opentelemetry-instrumentation-kafka-python/test-requirements-1.txt instrumentation/opentelemetry-instrumentation-kafka-python/test-requirements-ng.txt renamed to instrumentation/opentelemetry-instrumentation-kafka-python/test-requirements-1.txt

@@ -1,13 +1,14 @@
+kafka-python>=3.0.0,<4.0.0
 asgiref==3.8.1
 Deprecated==1.2.14
 iniconfig==2.0.0
-kafka-python-ng==2.2.2
+kafka-python==2.0.2
 packaging==24.0
 pluggy==1.6.0
 py-cpuinfo==9.0.0
 pytest==7.4.4
 tomli==2.0.1
-typing_extensions==4.9.0
+typing_extensions==4.12.2
 wrapt==1.16.0
 zipp==3.19.2
 -e opentelemetry-instrumentation

instrumentation/opentelemetry-instrumentation-kafka-python/tests/test_instrumentation.py

@@ -10,7 +10,6 @@
 from opentelemetry.instrumentation.kafka import KafkaInstrumentor
 from opentelemetry.instrumentation.kafka.package import (
     _instruments_kafka_python,
-    _instruments_kafka_python_ng,
 )
 
 
@@ -54,50 +53,6 @@ def _distribution(name):
         )
         self.assertEqual(package_to_instrument, (_instruments_kafka_python,))
 
-    @patch("opentelemetry.instrumentation.kafka.distribution")
-    def test_instrumentation_dependencies_kafka_python_ng_installed(
-        self, mock_distribution
-    ) -> None:
-        instrumentation = KafkaInstrumentor()
-
-        def _distribution(name):
-            if name == "kafka-python-ng":
-                return None
-            raise PackageNotFoundError
-
-        mock_distribution.side_effect = _distribution
-        package_to_instrument = instrumentation.instrumentation_dependencies()
-
-        self.assertEqual(mock_distribution.call_count, 1)
-        self.assertEqual(
-            mock_distribution.mock_calls, [call("kafka-python-ng")]
-        )
-        self.assertEqual(
-            package_to_instrument, (_instruments_kafka_python_ng,)
-        )
-
-    @patch("opentelemetry.instrumentation.kafka.distribution")
-    def test_instrumentation_dependencies_both_installed(
-        self, mock_distribution
-    ) -> None:
-        instrumentation = KafkaInstrumentor()
-
-        def _distribution(name):
-            # The function returns None here for all names
-            # to simulate both packages being installed
-            return None
-
-        mock_distribution.side_effect = _distribution
-        package_to_instrument = instrumentation.instrumentation_dependencies()
-
-        self.assertEqual(mock_distribution.call_count, 1)
-        self.assertEqual(
-            mock_distribution.mock_calls, [call("kafka-python-ng")]
-        )
-        self.assertEqual(
-            package_to_instrument, (_instruments_kafka_python_ng,)
-        )
-
     @patch("opentelemetry.instrumentation.kafka.distribution")
     def test_instrumentation_dependencies_none_installed(
         self, mock_distribution
@@ -122,7 +77,4 @@ def _distribution(name):
                 call("kafka-python"),
             ],
         )
-        self.assertEqual(
-            package_to_instrument,
-            (_instruments_kafka_python, _instruments_kafka_python_ng),
-        )
+        self.assertEqual(package_to_instrument, (_instruments_kafka_python,))

tox.ini

@@ -415,10 +415,8 @@ envlist =
     lint-instrumentation-aiokafka
 
     ; opentelemetry-instrumentation-kafka-python
-    py3{10,11}-test-instrumentation-kafka-python
-    py3{10,11,12,13,14}-test-instrumentation-kafka-pythonng
+    py3{10,11,12,13,14}-test-instrumentation-kafka-python{0,1}
     pypy3-test-instrumentation-kafka-python
-    pypy3-test-instrumentation-kafka-pythonng
     lint-instrumentation-kafka-python
 
     ; opentelemetry-instrumentation-confluent-kafka
@@ -539,10 +537,8 @@ deps =
   aiokafka: -r {toxinidir}/instrumentation/opentelemetry-instrumentation-aiokafka/test-requirements.txt
 
   kafka-python: {[testenv]test_deps}
-  kafka-python: -r {toxinidir}/instrumentation/opentelemetry-instrumentation-kafka-python/test-requirements.txt
-
-  kafka-pythonng: {[testenv]test_deps}
-  kafka-pythonng: -r {toxinidir}/instrumentation/opentelemetry-instrumentation-kafka-python/test-requirements-ng.txt
+  kafka-python-0: -r {toxinidir}/instrumentation/opentelemetry-instrumentation-kafka-python/test-requirements-0.txt
+  kafka-python-1: -r {toxinidir}/instrumentation/opentelemetry-instrumentation-kafka-python/test-requirements-1.txt
 
   confluent-kafka: {[testenv]test_deps}
   confluent-kafka: -r {toxinidir}/instrumentation/opentelemetry-instrumentation-confluent-kafka/test-requirements.txt
@@ -889,9 +885,6 @@ commands =
   test-instrumentation-kafka-python: pytest {toxinidir}/instrumentation/opentelemetry-instrumentation-kafka-python/tests {posargs}
   lint-instrumentation-kafka-python: sh -c "cd instrumentation && pylint --rcfile ../.pylintrc opentelemetry-instrumentation-kafka-python"
 
-  ; Test only for kafka-pythonng instrumentation as the only difference between kafka-python and kafka-pythonng is the version of kafka-python
-  test-instrumentation-kafka-pythonng: pytest {toxinidir}/instrumentation/opentelemetry-instrumentation-kafka-python/tests {posargs}
-
   test-instrumentation-confluent-kafka: pytest {toxinidir}/instrumentation/opentelemetry-instrumentation-confluent-kafka/tests {posargs}
   lint-instrumentation-confluent-kafka: sh -c "cd instrumentation && pylint --rcfile ../.pylintrc opentelemetry-instrumentation-confluent-kafka"