Drop Python 3.5 support (should fix CI) (#667)

.appveyor.yml

@@ -3,10 +3,6 @@ environment:
   KAFKA_VERSION: "0.10.2.1"
   SCALA_VERSION: "2.11"
   matrix:
-    - PYTHON: "C:\\Python35"
-      SNAPPY_WHEEL: "tools\\python_snappy-0.5.4-cp35-cp35m-win32.whl"
-    - PYTHON: "C:\\Python35-x64"
-      SNAPPY_WHEEL: "tools\\python_snappy-0.5.4-cp35-cp35m-win_amd64.whl"
     - PYTHON: "C:\\Python36"
       SNAPPY_WHEEL: "tools\\python_snappy-0.5.4-cp36-cp36m-win32.whl"
     - PYTHON: "C:\\Python36-x64"

.github/workflows/publish.yml

@@ -46,7 +46,6 @@ jobs:
           # On windows and mac we should have z library preinstalled
           CIBW_BEFORE_BUILD: pip install -r requirements-cython.txt
           CIBW_BUILD_VERBOSITY: 2
-          CIBW_SKIP: cp27-* cp35-* pp27-*
         run: |
           python -m pip install --upgrade pip setuptools
           pip install cibuildwheel

.github/workflows/tests.yml

@@ -4,11 +4,7 @@
 name: Tests
 
 on:
-  push:
-    branches: [ master ]
-    tags:
-      - "v0.[0-9]+.[0-9]+"
-      - "v0.[0-9]+.[0-9]+.dev*"
+  push: {}
   pull_request:
     branches: [ master ]
 
@@ -219,9 +215,6 @@ jobs:
             scala: "2.12"
 
           # Older python versions against latest broker
-          - python: 3.5
-            kafka: "2.4.0"
-            scala: "2.12"
           - python: 3.6
             kafka: "2.4.0"
             scala: "2.12"

.travis.yml_bak

@@ -35,21 +35,6 @@ matrix:
       - make check-readme
 
     # All Package build combinations for newest Kafka
-    - stage: *stage_functional
-      sudo: true
-      python: 3.5
-      services:
-      - docker
-      env: KAFKA_VERSION=2.4.0 SCALA_VERSION=2.12 PYTHONASYNCIODEBUG=1
-      script:
-      - make ci-test-all
-    - sudo: true
-      python: 3.5
-      services:
-      - docker
-      env: KAFKA_VERSION=2.4.0 SCALA_VERSION=2.12 PYTHONASYNCIODEBUG=1 AIOKAFKA_NO_EXTENSIONS=1
-      script:
-      - make ci-test-all
     - sudo: true
       python: 3.6
       services:

CHANGES.rst

@@ -7,6 +7,9 @@ Fix memory leak in kafka consumer when consumer is in idle state not consuming a
 =======
 618.feature
 added `OAUTHBEARER` as a new `sasl_mechanism`.
+=======
+667.bugfix
+Drop support for Python 3.5
 
 0.6.0 (2020-05-15)
 ==================

aiokafka/__init__.py

@@ -8,7 +8,7 @@
 from .structs import (
     TopicPartition, ConsumerRecord, OffsetAndTimestamp, OffsetAndMetadata
 )
-from .util import PY_35, ensure_future
+from .util import ensure_future
 
 
 __all__ = [
@@ -24,4 +24,4 @@
     "OffsetAndMetadata"
 ]
 
-(PY_35, ensure_future, AIOKafkaClient)
+(ensure_future, AIOKafkaClient)

aiokafka/abc.py

@@ -87,11 +87,7 @@ def on_partitions_assigned(self, assigned):
         pass
 
 
-# This statement is compatible with both Python 2.7 & 3+
-ABC = abc.ABCMeta('ABC', (object,), {'__slots__': ()})
-
-
-class AbstractTokenProvider(ABC):
+class AbstractTokenProvider(abc.ABC):
     """
     A Token Provider must be used for the SASL OAuthBearer protocol.
     The implementation should ensure token reuse so that multiple

aiokafka/conn.py

@@ -20,7 +20,7 @@
     GroupCoordinatorResponse_v0 as GroupCoordinatorResponse)
 
 import aiokafka.errors as Errors
-from aiokafka.util import ensure_future, create_future, PY_36
+from aiokafka.util import ensure_future, create_future
 
 from aiokafka.abc import AbstractTokenProvider
 
@@ -183,13 +183,9 @@ def __init__(self, host, port, *, loop, client_id='aiokafka',
     # that
     def __del__(self, _warnings=warnings):
         if self.connected():
-            if PY_36:
-                kwargs = {'source': self}
-            else:
-                kwargs = {}
             _warnings.warn("Unclosed AIOKafkaConnection {!r}".format(self),
                            ResourceWarning,
-                           **kwargs)
+                           source=self)
             if self._loop.is_closed():
                 return
 

aiokafka/consumer/consumer.py

@@ -16,7 +16,7 @@
 )
 from aiokafka.structs import TopicPartition
 from aiokafka.util import (
-    PY_36, commit_structure_validate, get_running_loop
+    commit_structure_validate, get_running_loop
 )
 from aiokafka import __version__
 
@@ -318,13 +318,9 @@ def __init__(self, *topics, loop=None,
 
     def __del__(self, _warnings=warnings):
         if self._closed is False:
-            if PY_36:
-                kwargs = {'source': self}
-            else:
-                kwargs = {}
             _warnings.warn("Unclosed AIOKafkaConsumer {!r}".format(self),
                            ResourceWarning,
-                           **kwargs)
+                           source=self)
             context = {'consumer': self,
                        'message': 'Unclosed AIOKafkaConsumer'}
             if self._source_traceback is not None:

aiokafka/producer/producer.py

@@ -13,7 +13,7 @@
 from aiokafka.record.legacy_records import LegacyRecordBatchBuilder
 from aiokafka.structs import TopicPartition
 from aiokafka.util import (
-    INTEGER_MAX_VALUE, PY_36, commit_structure_validate, get_running_loop
+    INTEGER_MAX_VALUE, commit_structure_validate, get_running_loop
 )
 
 from .message_accumulator import MessageAccumulator
@@ -274,13 +274,9 @@ def __init__(self, *, loop=None, bootstrap_servers='localhost',
     # We don't attempt to close the Consumer, as __del__ is synchronous
     def __del__(self, _warnings=warnings):
         if self._closed is False:
-            if PY_36:
-                kwargs = {'source': self}
-            else:
-                kwargs = {}
             _warnings.warn("Unclosed AIOKafkaProducer {!r}".format(self),
                            ResourceWarning,
-                           **kwargs)
+                           source=self)
             context = {'producer': self,
                        'message': 'Unclosed AIOKafkaProducer'}
             if self._source_traceback is not None:

aiokafka/util.py

@@ -1,14 +1,13 @@
 import asyncio
 import os
-import sys
 from asyncio import AbstractEventLoop
 from distutils.version import StrictVersion
 from typing import Dict, Tuple, TypeVar, Union
 
 from .structs import OffsetAndMetadata, TopicPartition
 
 
-__all__ = ["ensure_future", "create_future", "PY_35"]
+__all__ = ["ensure_future", "create_future"]
 
 
 try:
@@ -70,9 +69,6 @@ def get_running_loop() -> asyncio.AbstractEventLoop:
     return loop
 
 
-PY_35 = sys.version_info >= (3, 5)
-PY_352 = sys.version_info >= (3, 5, 2)
-PY_36 = sys.version_info >= (3, 6)
 NO_EXTENSIONS = bool(os.environ.get("AIOKAFKA_NO_EXTENSIONS"))
 
 INTEGER_MAX_VALUE = 2 ** 31 - 1

docs/examples.rst

@@ -9,7 +9,7 @@ Examples
    Manual commit <examples/manual_commit>
    Group consumer <examples/group_consumer>
    Custom partitioner <examples/custom_partitioner>
-   Python 3.5 async syntax <examples/python35_examples>
+   Python async for syntax <examples/python35_examples>
    SSL usage <examples/ssl_consume_produce>
    Local state consumer <examples/local_state_consumer>
    Batch producer <examples/batch_produce>

docs/examples/python35_examples.rst

@@ -1,8 +1,8 @@
 
-Python 3.5 async usage
+Python async for usage
 ======================
 
-``aiokafka`` supports Python3.5 ``async def`` syntax and adds some sugar using
+``aiokafka`` supports ``async for`` syntax and adds some sugar using
 this syntax.
 
 

requirements-ci.txt

@@ -12,6 +12,6 @@ xxhash==1.4.3
 python-snappy==0.5.4
 docutils==0.16
 Pygments==2.6.1
-gssapi==1.6.2 # pyup: <= 1.6.2  # For some reason 1.6.5 does not install with py35
+gssapi==1.6.5
 dataclasses==0.5; python_version<"3.7"
 async_generator==1.10; python_version<"3.7"

setup.py

@@ -102,12 +102,8 @@ def build_extension(self, ext):
 
 PY_VER = sys.version_info
 
-if PY_VER >= (3, 5):
-    pass
-elif PY_VER >= (3, 4):
-    install_requires.append("typing")
-else:
-    raise RuntimeError("aiokafka doesn't support Python earlier than 3.4")
+if PY_VER < (3, 6):
+    raise RuntimeError("aiokafka doesn't support Python earlier than 3.6")
 
 
 def read(f):
@@ -135,7 +131,6 @@ def read_version():
     "License :: OSI Approved :: Apache Software License",
     "Intended Audience :: Developers",
     "Programming Language :: Python :: 3",
-    "Programming Language :: Python :: 3.5",
     "Programming Language :: Python :: 3.6",
     "Programming Language :: Python :: 3.7",
     "Operating System :: OS Independent",

tests/_testutil.py

@@ -26,7 +26,19 @@
 
 __all__ = ['KafkaIntegrationTestCase', 'random_string']
 
-run_until_complete = pytest.mark.asyncio
+
+def run_until_complete(fun):
+    if not asyncio.iscoroutinefunction(fun):
+        fun = asyncio.coroutine(fun)
+
+    @wraps(fun)
+    def wrapper(test, *args, **kw):
+        loop = test.loop
+        timeout = getattr(test, "TEST_TIMEOUT", 120)
+        ret = loop.run_until_complete(
+            asyncio.wait_for(fun(test, *args, **kw), timeout, loop=loop))
+        return ret
+    return wrapper
 
 
 def kafka_versions(*versions):
@@ -378,7 +390,7 @@ def assert_message_count(self, messages, num_messages):
         self.assertEqual(len(messages), num_messages)
 
         # Make sure there are no duplicates
-        self.assertEqual(len(set(messages)), num_messages)
+        self.assertTrue(all(messages.count(m) == 1 for m in messages))
 
     def create_ssl_context(self):
         context = create_ssl_context(

tests/conftest.py

@@ -275,9 +275,3 @@ def setup_test_class(request, loop, kafka_server, ssl_folder, acl_manager,
     docker_image = request.config.getoption('--docker-image')
     kafka_version = docker_image.split(":")[-1].split("_")[-1]
     request.cls.kafka_version = kafka_version
-
-
-def pytest_ignore_collect(path, config):
-    if 'pep492' in str(path):
-        if sys.version_info < (3, 5, 0):
-            return True

tests/test_helpers.py

@@ -1,5 +1,4 @@
 import ssl
-import sys
 import unittest
 
 from aiokafka.helpers import create_ssl_context
@@ -21,8 +20,6 @@ def _check_ssl_dir(self):
         self.assertTrue(keyfile.exists(), str(keyfile))
         return cafile, certfile, keyfile
 
-    @pytest.mark.skipif(sys.version_info < (3, 4),
-                        reason="requires python3.4")
     def test_create_ssl_context(self):
         cafile, certfile, keyfile = self._check_ssl_dir()