Remove support for Python 3.9 (end of life) (#1116)

* Remove Pytnon 3.9 support (end of life)

* Safe ruff fixes

* Unsafe ruff fixes

* Manual fixes

Author: ods

.github/workflows/publish.yml

@@ -15,7 +15,7 @@ jobs:
       - uses: actions/checkout@v4
       - uses: actions/setup-python@v5
         with:
-          python-version: "3.9"
+          python-version: "3.10"
       - name: Prepare C files to include
         run: |
           python -m pip install --upgrade pip build
@@ -52,7 +52,7 @@ jobs:
       - uses: actions/checkout@v4
       - uses: actions/setup-python@v5
         with:
-          python-version: "3.9"
+          python-version: "3.10"
       - name: Set up QEMU
         if: ${{ matrix.arch == 'aarch64' }}
         uses: docker/setup-qemu-action@v3
@@ -82,10 +82,8 @@ jobs:
 
     strategy:
       matrix:
-        python: ["3.9", "3.10", "3.11", "3.12", "3.13"]
+        python: ["3.10", "3.11", "3.12", "3.13"]
         include:
-          - python: "3.9"
-            aiokafka_whl: dist/aiokafka-*-cp39-cp39-win_amd64.whl
           - python: "3.10"
             aiokafka_whl: dist/aiokafka-*-cp310-cp310-win_amd64.whl
           - python: "3.11"
@@ -127,10 +125,8 @@ jobs:
 
     strategy:
       matrix:
-        python: ["3.9", "3.10", "3.11", "3.12", "3.13"]
+        python: ["3.10", "3.11", "3.12", "3.13"]
         include:
-          - python: "3.9"
-            aiokafka_whl: dist/aiokafka-*-cp39-cp39-macosx_*_x86_64.whl
           - python: "3.10"
             aiokafka_whl: dist/aiokafka-*-cp310-cp310-macosx_*_x86_64.whl
           - python: "3.11"
@@ -170,10 +166,8 @@ jobs:
 
     strategy:
       matrix:
-        python: ["3.9", "3.10", "3.11", "3.12", "3.13"]
+        python: ["3.10", "3.11", "3.12", "3.13"]
         include:
-          - python: "3.9"
-            aiokafka_whl: dist/aiokafka-*-cp39-cp39-macosx_*_arm64.whl
           - python: "3.10"
             aiokafka_whl: dist/aiokafka-*-cp310-cp310-macosx_*_arm64.whl
           - python: "3.11"
@@ -213,10 +207,8 @@ jobs:
 
     strategy:
       matrix:
-        python: ["3.9", "3.10", "3.11", "3.12", "3.13"]
+        python: ["3.10", "3.11", "3.12", "3.13"]
         include:
-          - python: "3.9"
-            aiokafka_whl: dist/aiokafka-*-cp39-cp39-manylinux*_x86_64.whl
           - python: "3.10"
             aiokafka_whl: dist/aiokafka-*-cp310-cp310-manylinux*_x86_64.whl
           - python: "3.11"

.github/workflows/tests.yml

@@ -22,7 +22,7 @@ jobs:
     - name: Set up Python
       uses: actions/setup-python@v5
       with:
-        python-version: 3.9
+        python-version: "3.10"
 
     - name: Install system dependencies
       run: |
@@ -39,10 +39,10 @@ jobs:
       uses: actions/cache@v4
       with:
         path: ${{ steps.pip-cache.outputs.dir }}
-        key: ${{ runner.os }}-py-3.9-${{ hashFiles('requirements-ci.txt') }}-${{ hashFiles('setup.py') }}
+        key: ${{ runner.os }}-py-3.10-${{ hashFiles('requirements-ci.txt') }}-${{ hashFiles('setup.py') }}
         # If miss on key takes any other cache with different hashes, will download correct ones on next step anyway
         restore-keys: |
-          ${{ runner.os }}-py-3.9-
+          ${{ runner.os }}-py-3.10-
 
     - name: Install python dependencies
       run: |
@@ -73,7 +73,7 @@ jobs:
 
     strategy:
       matrix:
-        python: ["3.9", "3.10", "3.11", "3.12", "3.13"]
+        python: ["3.10", "3.11", "3.12", "3.13"]
 
     steps:
     - uses: actions/checkout@v4
@@ -141,7 +141,7 @@ jobs:
 
     strategy:
       matrix:
-        python: ["3.9", "3.10", "3.11", "3.12", "3.13"]
+        python: ["3.10", "3.11", "3.12", "3.13"]
 
     steps:
     - uses: actions/checkout@v4
@@ -215,9 +215,6 @@ jobs:
             scala: "2.13"
 
           # Older python versions against latest broker
-          - python: "3.9"
-            kafka: "2.8.1"
-            scala: "2.13"
           - python: "3.10"
             kafka: "2.8.1"
             scala: "2.13"

aiokafka/admin/client.py

@@ -3,7 +3,7 @@
 from collections import defaultdict
 from collections.abc import Sequence
 from ssl import SSLContext
-from typing import Any, Optional, Union
+from typing import Any
 
 import async_timeout
 
@@ -89,21 +89,21 @@ def __init__(
         self,
         *,
         loop=None,
-        bootstrap_servers: Union[str, list[str]] = "localhost",
+        bootstrap_servers: str | list[str] = "localhost",
         client_id: str = "aiokafka-" + __version__,
         request_timeout_ms: int = 40000,
         connections_max_idle_ms: int = 540000,
         retry_backoff_ms: int = 100,
         metadata_max_age_ms: int = 300000,
         security_protocol: str = "PLAINTEXT",
-        ssl_context: Optional[SSLContext] = None,
+        ssl_context: SSLContext | None = None,
         api_version: str = "auto",
         sasl_mechanism: str = "PLAIN",
-        sasl_plain_username: Optional[str] = None,
-        sasl_plain_password: Optional[str] = None,
+        sasl_plain_username: str | None = None,
+        sasl_plain_password: str | None = None,
         sasl_kerberos_service_name: str = "kafka",
-        sasl_kerberos_domain_name: Optional[str] = None,
-        sasl_oauth_token_provider: Optional[str] = None,
+        sasl_kerberos_domain_name: str | None = None,
+        sasl_oauth_token_provider: str | None = None,
     ):
         self._closed = False
         self._started = False
@@ -141,7 +141,7 @@ async def close(self):
     async def _send_request(
         self,
         request: Request,
-        node_id: Optional[int] = None,
+        node_id: int | None = None,
     ) -> Response:
         if node_id is None:
             node_id = self._client.get_random_node()
@@ -227,7 +227,7 @@ def _convert_new_topic_request(new_topic):
     async def create_topics(
         self,
         new_topics: list[NewTopic],
-        timeout_ms: Optional[int] = None,
+        timeout_ms: int | None = None,
         validate_only: bool = False,
     ) -> Response:
         """Create new topics in the cluster.
@@ -269,7 +269,7 @@ async def create_topics(
     async def delete_topics(
         self,
         topics: list[str],
-        timeout_ms: Optional[int] = None,
+        timeout_ms: int | None = None,
     ) -> Response:
         """Delete topics from the cluster.
 
@@ -285,7 +285,7 @@ async def delete_topics(
 
     async def _get_cluster_metadata(
         self,
-        topics: Optional[list[str]] = None,
+        topics: list[str] | None = None,
     ) -> Response:
         """
         Retrieve cluster metadata
@@ -303,7 +303,7 @@ async def list_topics(self) -> list[str]:
 
     async def describe_topics(
         self,
-        topics: Optional[list[str]] = None,
+        topics: list[str] | None = None,
     ) -> list[Any]:
         metadata = await self._get_cluster_metadata(topics=topics)
         obj = metadata.to_object()
@@ -426,7 +426,7 @@ def _convert_topic_partitions(topic_partitions: dict[str, NewPartitions]):
     async def create_partitions(
         self,
         topic_partitions: dict[str, NewPartitions],
-        timeout_ms: Optional[int] = None,
+        timeout_ms: int | None = None,
         validate_only: bool = False,
     ) -> Response:
         """Create additional partitions for an existing topic.
@@ -457,7 +457,7 @@ async def create_partitions(
     async def describe_consumer_groups(
         self,
         group_ids: list[str],
-        group_coordinator_id: Optional[int] = None,
+        group_coordinator_id: int | None = None,
         include_authorized_operations: bool = False,
     ) -> list[Response]:
         """Describe a set of consumer groups.
@@ -509,7 +509,7 @@ async def describe_consumer_groups(
 
     async def list_consumer_groups(
         self,
-        broker_ids: Optional[list[int]] = None,
+        broker_ids: list[int] | None = None,
     ) -> list[tuple[Any, ...]]:
         """List all consumer groups known to the cluster.
 
@@ -578,8 +578,8 @@ async def find_coordinator(self, group_id: str, coordinator_type: int = 0) -> in
     async def list_consumer_group_offsets(
         self,
         group_id: str,
-        group_coordinator_id: Optional[int] = None,
-        partitions: Optional[list[TopicPartition]] = None,
+        group_coordinator_id: int | None = None,
+        partitions: list[TopicPartition] | None = None,
     ) -> dict[TopicPartition, OffsetAndMetadata]:
         """Fetch Consumer Offsets for a single consumer group.
 
@@ -638,7 +638,7 @@ async def list_consumer_group_offsets(
     async def delete_records(
         self,
         records_to_delete: dict[TopicPartition, RecordsToDelete],
-        timeout_ms: Optional[int] = None,
+        timeout_ms: int | None = None,
     ) -> dict[TopicPartition, int]:
         """Delete records from partitions.
 

aiokafka/cluster.py

@@ -4,7 +4,6 @@
 import threading
 import time
 from concurrent.futures import Future
-from typing import Optional
 
 from aiokafka import errors as Errors
 from aiokafka.conn import collect_hosts
@@ -104,7 +103,7 @@ def broker_metadata(self, broker_id):
             or self._coordinator_brokers.get(broker_id)
         )
 
-    def partitions_for_topic(self, topic: str) -> Optional[set[int]]:
+    def partitions_for_topic(self, topic: str) -> set[int] | None:
         """Return set of all partitions for topic (whether available or not)
 
         Arguments:

aiokafka/codec.py

@@ -1,7 +1,6 @@
 import gzip
 import io
 import struct
-from typing import Optional
 
 from typing_extensions import Buffer
 
@@ -30,7 +29,7 @@ def has_lz4() -> bool:
     return cramjam is not None
 
 
-def gzip_encode(payload: Buffer, compresslevel: Optional[int] = None) -> bytes:
+def gzip_encode(payload: Buffer, compresslevel: int | None = None) -> bytes:
     if not compresslevel:
         compresslevel = 9
 
@@ -94,7 +93,7 @@ def snappy_encode(
         return cramjam.snappy.compress_raw(payload)
 
     out = io.BytesIO()
-    for fmt, dat in zip(_XERIAL_V1_FORMAT, _XERIAL_V1_HEADER):
+    for fmt, dat in zip(_XERIAL_V1_FORMAT, _XERIAL_V1_HEADER, strict=False):
         out.write(struct.pack("!" + fmt, dat))
 
     payload = memoryview(payload)
@@ -186,7 +185,7 @@ def lz4_decode(payload: Buffer) -> bytes:
     return bytes(cramjam.lz4.decompress(payload))
 
 
-def zstd_encode(payload: Buffer, level: Optional[int] = None) -> bytes:
+def zstd_encode(payload: Buffer, level: int | None = None) -> bytes:
     if not has_zstd():
         raise NotImplementedError("Zstd codec is not available")
 

aiokafka/conn.py

@@ -411,7 +411,7 @@ def _on_read_task_error(cls, self_ref, read_task):
         try:
             read_task.result()
         except Exception as exc:
-            if not isinstance(exc, (OSError, EOFError, ConnectionError)):
+            if not isinstance(exc, OSError | EOFError | ConnectionError):
                 log.exception("Unexpected exception in AIOKafkaConnection")
 
             self = self_ref()
@@ -783,7 +783,7 @@ def create_salted_password(self, salt, iterations):
 
     @staticmethod
     def _xor_bytes(left, right):
-        return bytes(lb ^ rb for lb, rb in zip(left, right))
+        return bytes(lb ^ rb for lb, rb in zip(left, right, strict=False))
 
 
 class OAuthAuthenticator(BaseSaslAuthenticator):

aiokafka/consumer/consumer.py

@@ -452,7 +452,7 @@ async def _wait_topics(self):
                 await self._client._wait_on_metadata(topic)
 
     def _validate_topics(self, topics):
-        if not isinstance(topics, (tuple, set, list)):
+        if not isinstance(topics, tuple | set | list):
             raise TypeError("Topics should be list of strings")
         return set(topics)
 

aiokafka/consumer/group_coordinator.py

@@ -442,7 +442,7 @@ async def _perform_assignment(self, response: Response):
                 member_id, group_instance_id, metadata_bytes = member
             elif isinstance(
                 response,
-                (JoinGroupResponse[0], JoinGroupResponse[1], JoinGroupResponse[2]),
+                JoinGroupResponse[0] | JoinGroupResponse[1] | JoinGroupResponse[2],
             ):
                 member_id, metadata_bytes = member
             else:
@@ -986,11 +986,9 @@ def _is_commit_retriable(self, error):
         # as retriable.
         return error.retriable or isinstance(
             error,
-            (
-                Errors.UnknownMemberIdError,
-                Errors.IllegalGenerationError,
-                Errors.RebalanceInProgressError,
-            ),
+            Errors.UnknownMemberIdError
+            | Errors.IllegalGenerationError
+            | Errors.RebalanceInProgressError,
         )
 
     async def _maybe_do_last_autocommit(self, assignment):

aiokafka/consumer/subscription_state.py

@@ -391,7 +391,7 @@ class Assignment:
     """
 
     def __init__(self, topic_partitions: Iterable[TopicPartition]):
-        assert isinstance(topic_partitions, (list, set, tuple))
+        assert isinstance(topic_partitions, list | set | tuple)
 
         self._topic_partitions = frozenset(topic_partitions)
 

aiokafka/coordinator/assignors/sticky/sorted_set.py

@@ -1,9 +1,7 @@
-from collections.abc import Collection, Iterable, Iterator
+from collections.abc import Callable, Collection, Iterable, Iterator
 from typing import (
     Any,
-    Callable,
     Generic,
-    Optional,
     TypeVar,
     final,
 )
@@ -15,16 +13,16 @@
 class SortedSet(Generic[T], Collection[T]):
     def __init__(
         self,
-        iterable: Optional[Iterable[T]] = None,
-        key: Optional[Callable[[T], Any]] = None,
+        iterable: Iterable[T] | None = None,
+        key: Callable[[T], Any] | None = None,
     ) -> None:
         self._key: Callable[[T], Any] = key if key is not None else lambda x: x
         self._set: set[T] = set(iterable) if iterable is not None else set()
 
-        self._cached_last: Optional[T] = None
-        self._cached_first: Optional[T] = None
+        self._cached_last: T | None = None
+        self._cached_first: T | None = None
 
-    def first(self) -> Optional[T]:
+    def first(self) -> T | None:
         if self._cached_first is not None:
             return self._cached_first
 
@@ -35,7 +33,7 @@ def first(self) -> Optional[T]:
         self._cached_first = first
         return first
 
-    def last(self) -> Optional[T]:
+    def last(self) -> T | None:
         if self._cached_last is not None:
             return self._cached_last