diff --git a/.github/dependabot.yml b/.github/dependabot.yml new file mode 100644 index 000000000..2c7d17083 --- /dev/null +++ b/.github/dependabot.yml @@ -0,0 +1,7 @@ +version: 2 +updates: + # Maintain dependencies for GitHub Actions + - package-ecosystem: "github-actions" + directory: "/" + schedule: + interval: "daily" diff --git a/.github/workflows/codeql-analysis.yml b/.github/workflows/codeql-analysis.yml new file mode 100644 index 000000000..4f6360b71 --- /dev/null +++ b/.github/workflows/codeql-analysis.yml @@ -0,0 +1,67 @@ +--- +# For most projects, this workflow file will not need changing; you simply need +# to commit it to your repository. +# +# You may wish to alter this file to override the set of languages analyzed, +# or to provide custom queries or build logic. +# +# ******** NOTE ******** +# We have attempted to detect the languages in your repository. Please check +# the `language` matrix defined below to confirm you have the correct set of +# supported CodeQL languages. +# +name: CodeQL +on: + push: + branches: [master] + pull_request: + # The branches below must be a subset of the branches above + branches: [master] + schedule: + - cron: 19 10 * * 6 +jobs: + analyze: + name: Analyze + runs-on: ubuntu-latest + permissions: + actions: read + contents: read + security-events: write + strategy: + fail-fast: false + matrix: + language: [python] + # CodeQL supports [ 'cpp', 'csharp', 'go', 'java', 'javascript', 'python' ] + # Learn more: + # https://docs.github.com/en/free-pro-team@latest/github/finding-security-vulnerabilities-and-errors-in-your-code/configuring-code-scanning#changing-the-languages-that-are-analyzed + steps: + - name: Checkout repository + uses: actions/checkout@v4 + + # Initializes the CodeQL tools for scanning. + - name: Initialize CodeQL + uses: github/codeql-action/init@v3 + with: + languages: ${{ matrix.language }} + # If you wish to specify custom queries, you can do so here or in a config file. + # By default, queries listed here will override any specified in a config file. + # Prefix the list here with "+" to use these queries and those in the config file. + # queries: ./path/to/local/query, your-org/your-repo/queries@main + + # Autobuild attempts to build any compiled languages (C/C++, C#, or Java). + # If this step fails, then you should remove it and run the build manually (see below) + - name: Autobuild + uses: github/codeql-action/autobuild@v3 + + # ℹī¸ Command-line programs to run using the OS shell. + # 📚 https://git.io/JvXDl + + # ✏ī¸ If the Autobuild fails above, remove it and uncomment the following three lines + # and modify them (or add more) to build your code if your project + # uses a compiled language + + #- run: | + # make bootstrap + # make release + - name: Perform CodeQL Analysis + uses: github/codeql-action/analyze@v3 diff --git a/.github/workflows/python-package.yml b/.github/workflows/python-package.yml new file mode 100644 index 000000000..59ad718cf --- /dev/null +++ b/.github/workflows/python-package.yml @@ -0,0 +1,198 @@ +name: CI/CD + +on: + push: + branches: ["master"] + pull_request: + branches: ["master"] + release: + types: [created] + branches: + - 'master' + workflow_dispatch: + +env: + FORCE_COLOR: "1" # Make tools pretty. + PIP_DISABLE_PIP_VERSION_CHECK: "1" + PIP_NO_PYTHON_VERSION_WARNING: "1" + PYTHON_LATEST: "3.12" + KAFKA_LATEST: "2.6.0" + + # For re-actors/checkout-python-sdist + sdist-artifact: python-package-distributions + +jobs: + + build-sdist: + name: đŸ“Ļ Build the source distribution + runs-on: ubuntu-latest + steps: + - name: Checkout project + uses: actions/checkout@v4 + with: + fetch-depth: 0 + - name: Set up Python + uses: actions/setup-python@v5 + with: + python-version: ${{ env.PYTHON_LATEST }} + cache: pip + - run: python -m pip install build + name: Install core libraries for build and install + - name: Build artifacts + run: python -m build + - name: Upload built artifacts for testing + uses: actions/upload-artifact@v3 + with: + name: ${{ env.sdist-artifact }} + # NOTE: Exact expected file names are specified here + # NOTE: as a safety measure — if anything weird ends + # NOTE: up being in this dir or not all dists will be + # NOTE: produced, this will fail the workflow. + path: dist/${{ env.sdist-name }} + retention-days: 15 + + test-python: + name: Tests on ${{ matrix.python-version }} + needs: + - build-sdist + runs-on: ubuntu-latest + continue-on-error: ${{ matrix.experimental }} + strategy: + fail-fast: false + matrix: + python-version: + - "3.8" + - "3.9" + - "3.10" + - "3.11" + - "3.12" + experimental: [ false ] + include: + - python-version: "pypy3.9" + experimental: true + - python-version: "~3.13.0-0" + experimental: true + steps: + - name: Checkout the source code + uses: actions/checkout@v4 + with: + fetch-depth: 0 + - name: Setup java + uses: actions/setup-java@v4 + with: + distribution: temurin + java-version: 11 + - name: Set up Python + uses: actions/setup-python@v5 + with: + python-version: ${{ matrix.python-version }} + cache: pip + cache-dependency-path: | + requirements-dev.txt + - name: Check Java installation + run: source travis_java_install.sh + - name: Pull Kafka releases + run: ./build_integration.sh + env: + PLATFORM: ${{ matrix.platform }} + KAFKA_VERSION: ${{ env.KAFKA_LATEST }} + # TODO: Cache releases to expedite testing + - name: Install dependencies + run: | + sudo apt install -y libsnappy-dev libzstd-dev + python -m pip install --upgrade pip + python -m pip install tox tox-gh-actions + pip install . + pip install -r requirements-dev.txt + - name: Test with tox + run: tox + env: + PLATFORM: ${{ matrix.platform }} + KAFKA_VERSION: ${{ env.KAFKA_LATEST }} + + test-kafka: + name: Tests for Kafka ${{ matrix.kafka-version }} + needs: + - build-sdist + runs-on: ubuntu-latest + strategy: + fail-fast: false + matrix: + kafka-version: + - "0.8.2.2" + - "0.9.0.1" + - "0.10.2.2" + - "0.11.0.2" + - "0.11.0.3" + - "1.1.1" + - "2.4.0" + - "2.5.0" + - "2.6.0" + steps: + - name: Checkout the source code + uses: actions/checkout@v4 + with: + fetch-depth: 0 + - name: Setup java + uses: actions/setup-java@v4 + with: + distribution: temurin + java-version: 8 + - name: Set up Python + uses: actions/setup-python@v5 + with: + python-version: ${{ env.PYTHON_LATEST }} + cache: pip + cache-dependency-path: | + requirements-dev.txt + - name: Pull Kafka releases + run: ./build_integration.sh + env: + # This is fast enough as long as you pull only one release at a time, + # no need to worry about caching + PLATFORM: ${{ matrix.platform }} + KAFKA_VERSION: ${{ matrix.kafka-version }} + - name: Install dependencies + run: | + sudo apt install -y libsnappy-dev libzstd-dev + python -m pip install --upgrade pip + python -m pip install tox tox-gh-actions + pip install . + pip install -r requirements-dev.txt + - name: Test with tox + run: tox + env: + PLATFORM: ${{ matrix.platform }} + KAFKA_VERSION: ${{ matrix.kafka-version }} + + check: # This job does nothing and is only used for the branch protection + name: ✅ Ensure the required checks passing + if: always() + needs: + - build-sdist + - test-python + - test-kafka + runs-on: ubuntu-latest + steps: + - name: Decide whether the needed jobs succeeded or failed + uses: re-actors/alls-green@release/v1 + with: + jobs: ${{ toJSON(needs) }} + publish: + name: đŸ“Ļ Publish to PyPI + runs-on: ubuntu-latest + needs: [build-sdist] + permissions: + id-token: write + environment: pypi + if: github.event_name == 'release' && github.event.action == 'created' + steps: + - name: Download the sdist artifact + uses: actions/download-artifact@v3 + with: + name: artifact + path: dist + - name: Publish package to PyPI + uses: pypa/gh-action-pypi-publish@release/v1 + with: + password: ${{ secrets.PYPI_API_TOKEN }} diff --git a/Makefile b/Makefile index b4dcbffc9..fc8fa5b21 100644 --- a/Makefile +++ b/Makefile @@ -20,14 +20,14 @@ test37: build-integration test27: build-integration KAFKA_VERSION=$(KAFKA_VERSION) SCALA_VERSION=$(SCALA_VERSION) tox -e py27 -- $(FLAGS) -# Test using py.test directly if you want to use local python. Useful for other +# Test using pytest directly if you want to use local python. Useful for other # platforms that require manual installation for C libraries, ie. Windows. test-local: build-integration - KAFKA_VERSION=$(KAFKA_VERSION) SCALA_VERSION=$(SCALA_VERSION) py.test \ + KAFKA_VERSION=$(KAFKA_VERSION) SCALA_VERSION=$(SCALA_VERSION) pytest \ --pylint --pylint-rcfile=pylint.rc --pylint-error-types=EF $(FLAGS) kafka test cov-local: build-integration - KAFKA_VERSION=$(KAFKA_VERSION) SCALA_VERSION=$(SCALA_VERSION) py.test \ + KAFKA_VERSION=$(KAFKA_VERSION) SCALA_VERSION=$(SCALA_VERSION) pytest \ --pylint --pylint-rcfile=pylint.rc --pylint-error-types=EF --cov=kafka \ --cov-config=.covrc --cov-report html $(FLAGS) kafka test @echo "open file://`pwd`/htmlcov/index.html" diff --git a/README.rst b/README.rst index 5f834442c..78a92a884 100644 --- a/README.rst +++ b/README.rst @@ -7,10 +7,16 @@ Kafka Python client :target: https://pypi.python.org/pypi/kafka-python .. image:: https://coveralls.io/repos/dpkp/kafka-python/badge.svg?branch=master&service=github :target: https://coveralls.io/github/dpkp/kafka-python?branch=master -.. image:: https://travis-ci.org/dpkp/kafka-python.svg?branch=master - :target: https://travis-ci.org/dpkp/kafka-python .. image:: https://img.shields.io/badge/license-Apache%202-blue.svg :target: https://github.com/dpkp/kafka-python/blob/master/LICENSE +.. image:: https://img.shields.io/pypi/dw/kafka-python.svg + :target: https://pypistats.org/packages/kafka-python +.. image:: https://img.shields.io/pypi/v/kafka-python.svg + :target: https://pypi.org/project/kafka-python +.. image:: https://img.shields.io/pypi/implementation/kafka-python + :target: https://github.com/dpkp/kafka-python/blob/master/setup.py + + Python client for the Apache Kafka distributed stream processing system. kafka-python is designed to function much like the official java client, with a diff --git a/docs/usage.rst b/docs/usage.rst index 1cf1aa414..047bbad77 100644 --- a/docs/usage.rst +++ b/docs/usage.rst @@ -8,6 +8,8 @@ KafkaConsumer .. code:: python from kafka import KafkaConsumer + import json + import msgpack # To consume latest messages and auto-commit offsets consumer = KafkaConsumer('my-topic', @@ -57,6 +59,8 @@ KafkaProducer from kafka import KafkaProducer from kafka.errors import KafkaError + import msgpack + import json producer = KafkaProducer(bootstrap_servers=['broker1:1234']) @@ -108,3 +112,52 @@ KafkaProducer # configure multiple retries producer = KafkaProducer(retries=5) + + +ClusterMetadata +============= +.. code:: python + + from kafka.cluster import ClusterMetadata + + clusterMetadata = ClusterMetadata(bootstrap_servers=['broker1:1234']) + + # get all brokers metadata + print(clusterMetadata.brokers()) + + # get specific broker metadata + print(clusterMetadata.broker_metadata('bootstrap-0')) + + # get all partitions of a topic + print(clusterMetadata.partitions_for_topic("topic")) + + # list topics + print(clusterMetadata.topics()) + + +KafkaAdminClient +============= +.. code:: python + from kafka import KafkaAdminClient + from kafka.admin import NewTopic + + admin = KafkaAdminClient(bootstrap_servers=['broker1:1234']) + + # create a new topic + topics_list = [] + topics_list.append(NewTopic(name="testtopic", num_partitions=1, replication_factor=1)) + admin.create_topics(topics_list,timeout_ms=None, validate_only=False) + + # delete a topic + admin.delete_topics(['testtopic']) + + # list consumer groups + print(admin.list_consumer_groups()) + + # get consumer group details + print(admin.describe_consumer_groups('cft-plt-qa.connect')) + + # get consumer group offset + print(admin.list_consumer_group_offsets('cft-plt-qa.connect')) + + diff --git a/kafka/admin/client.py b/kafka/admin/client.py index fd4d66110..62d487543 100644 --- a/kafka/admin/client.py +++ b/kafka/admin/client.py @@ -147,6 +147,7 @@ class KafkaAdminClient(object): sasl_oauth_token_provider (AbstractTokenProvider): OAuthBearer token provider instance. (See kafka.oauth.abstract). Default: None kafka_client (callable): Custom class / callable for creating KafkaClient instances + socks5_proxy (str): Socks5 proxy url. Default: None """ DEFAULT_CONFIG = { @@ -182,6 +183,7 @@ class KafkaAdminClient(object): 'sasl_kerberos_service_name': 'kafka', 'sasl_kerberos_domain_name': None, 'sasl_oauth_token_provider': None, + 'socks5_proxy': None, # metrics configs 'metric_reporters': [], @@ -272,7 +274,7 @@ def _refresh_controller_id(self): version = self._matching_api_version(MetadataRequest) if 1 <= version <= 6: request = MetadataRequest[version]() - future = self._send_request_to_node(self._client.least_loaded_node(), request) + future = self._send_request_to_least_loaded_node(request) self._wait_for_futures([future]) @@ -310,7 +312,7 @@ def _find_coordinator_id_send_request(self, group_id): raise NotImplementedError( "Support for GroupCoordinatorRequest_v{} has not yet been added to KafkaAdminClient." .format(version)) - return self._send_request_to_node(self._client.least_loaded_node(), request) + return self._send_request_to_least_loaded_node(request) def _find_coordinator_id_process_response(self, response): """Process a FindCoordinatorResponse. @@ -355,13 +357,41 @@ def _find_coordinator_ids(self, group_ids): } return groups_coordinators + def _send_request_to_least_loaded_node(self, request): + """Send a Kafka protocol message to the least loaded broker. + + Returns a future that may be polled for status and results. + + :param request: The message to send. + :return: A future object that may be polled for status and results. + :exception: The exception if the message could not be sent. + """ + node_id = self._client.least_loaded_node() + while not self._client.ready(node_id): + # poll until the connection to broker is ready, otherwise send() + # will fail with NodeNotReadyError + self._client.poll() + + # node_id is not part of the cluster anymore, choose a new broker + # to connect to + if self._client.cluster.broker_metadata(node_id) is None: + node_id = self._client.least_loaded_node() + + return self._client.send(node_id, request) + def _send_request_to_node(self, node_id, request): """Send a Kafka protocol message to a specific broker. + .. note:: + + This function will enter in an infinite loop if `node_id` is + removed from the cluster. + Returns a future that may be polled for status and results. :param node_id: The broker id to which to send the message. :param request: The message to send. + :param wakeup: Optional flag to disable thread-wakeup. :return: A future object that may be polled for status and results. :exception: The exception if the message could not be sent. """ @@ -369,7 +399,7 @@ def _send_request_to_node(self, node_id, request): # poll until the connection to broker is ready, otherwise send() # will fail with NodeNotReadyError self._client.poll() - return self._client.send(node_id, request) + return self._client.send(node_id, request, wakeup) def _send_request_to_controller(self, request): """Send a Kafka protocol message to the cluster controller. @@ -382,10 +412,23 @@ def _send_request_to_controller(self, request): tries = 2 # in case our cached self._controller_id is outdated while tries: tries -= 1 - future = self._send_request_to_node(self._controller_id, request) + future = self._client.send(self._controller_id, request) self._wait_for_futures([future]) + if future.exception is not None: + log.error( + "Sending request to controller_id %s failed with %s", + self._controller_id, + future.exception, + ) + is_outdated_controler = ( + self._client.cluster.broker_metadata(self._controller_id) is None + ) + if is_outdated_controler: + self._refresh_controller_id() + continue + response = future.value # In Java, the error field name is inconsistent: # - CreateTopicsResponse / CreatePartitionsResponse uses topic_errors @@ -506,10 +549,7 @@ def _get_cluster_metadata(self, topics=None, auto_topic_creation=False): allow_auto_topic_creation=auto_topic_creation ) - future = self._send_request_to_node( - self._client.least_loaded_node(), - request - ) + future = self._send_request_to_least_loaded_node(request) self._wait_for_futures([future]) return future.value @@ -601,7 +641,7 @@ def describe_acls(self, acl_filter): .format(version) ) - future = self._send_request_to_node(self._client.least_loaded_node(), request) + future = self._send_request_to_least_loaded_node(request) self._wait_for_futures([future]) response = future.value @@ -692,7 +732,7 @@ def create_acls(self, acls): .format(version) ) - future = self._send_request_to_node(self._client.least_loaded_node(), request) + future = self._send_request_to_least_loaded_node(request) self._wait_for_futures([future]) response = future.value @@ -786,7 +826,7 @@ def delete_acls(self, acl_filters): .format(version) ) - future = self._send_request_to_node(self._client.least_loaded_node(), request) + future = self._send_request_to_least_loaded_node(request) self._wait_for_futures([future]) response = future.value @@ -846,8 +886,7 @@ def describe_configs(self, config_resources, include_synonyms=False): )) if len(topic_resources) > 0: - futures.append(self._send_request_to_node( - self._client.least_loaded_node(), + futures.append(self._send_request_to_least_loaded_node( DescribeConfigsRequest[version](resources=topic_resources) )) @@ -867,8 +906,7 @@ def describe_configs(self, config_resources, include_synonyms=False): )) if len(topic_resources) > 0: - futures.append(self._send_request_to_node( - self._client.least_loaded_node(), + futures.append(self._send_request_to_least_loaded_node( DescribeConfigsRequest[version](resources=topic_resources, include_synonyms=include_synonyms) )) else: @@ -915,7 +953,7 @@ def alter_configs(self, config_resources): # // a single request that may be sent to any broker. # # So this is currently broken as it always sends to the least_loaded_node() - future = self._send_request_to_node(self._client.least_loaded_node(), request) + future = self._send_request_to_least_loaded_node(request) self._wait_for_futures([future]) response = future.value diff --git a/kafka/client_async.py b/kafka/client_async.py index 58f22d4ec..9a720112d 100644 --- a/kafka/client_async.py +++ b/kafka/client_async.py @@ -154,6 +154,7 @@ class KafkaClient(object): sasl mechanism handshake. Default: one of bootstrap servers sasl_oauth_token_provider (AbstractTokenProvider): OAuthBearer token provider instance. (See kafka.oauth.abstract). Default: None + socks5_proxy (str): Socks5 proxy URL. Default: None """ DEFAULT_CONFIG = { @@ -192,7 +193,8 @@ class KafkaClient(object): 'sasl_plain_password': None, 'sasl_kerberos_service_name': 'kafka', 'sasl_kerberos_domain_name': None, - 'sasl_oauth_token_provider': None + 'sasl_oauth_token_provider': None, + 'socks5_proxy': None, } def __init__(self, **configs): @@ -368,18 +370,26 @@ def _maybe_connect(self, node_id): conn = self._conns.get(node_id) if conn is None: - broker = self.cluster.broker_metadata(node_id) - assert broker, 'Broker id %s not in current metadata' % (node_id,) - - log.debug("Initiating connection to node %s at %s:%s", - node_id, broker.host, broker.port) - host, port, afi = get_ip_port_afi(broker.host) - cb = WeakMethod(self._conn_state_change) - conn = BrokerConnection(host, broker.port, afi, - state_change_callback=cb, - node_id=node_id, - **self.config) - self._conns[node_id] = conn + broker_metadata = self.cluster.broker_metadata(node_id) + + # The broker may have been removed from the cluster after the + # call to `maybe_connect`. At this point there is no way to + # recover, so just ignore the connection + if broker_metadata is None: + log.debug("Node %s is not available anymore, discarding connection", node_id) + if node_id in self._connecting: + self._connecting.remove(node_id) + return False + else: + log.debug("Initiating connection to node %s at %s:%s", + node_id, broker_metadata.host, broker_metadata.port) + host, port, afi = get_ip_port_afi(broker_metadata.host) + cb = WeakMethod(self._conn_state_change) + conn = BrokerConnection(host, broker_metadata.port, afi, + state_change_callback=cb, + node_id=node_id, + **self.config) + self._conns[node_id] = conn # Check if existing connection should be recreated because host/port changed elif self._should_recycle_connection(conn): @@ -637,6 +647,9 @@ def _poll(self, timeout): self._sensors.select_time.record((end_select - start_select) * 1000000000) for key, events in ready: + if key.fileobj.fileno() < 0: + time.sleep(0.1) + if key.fileobj is self._wake_r: self._clear_wake_fd() continue @@ -899,7 +912,13 @@ def check_version(self, node_id=None, timeout=2, strict=False): self._lock.release() raise Errors.NoBrokersAvailable() self._maybe_connect(try_node) - conn = self._conns[try_node] + try: + conn = self._conns[try_node] + except KeyError: + if node_id is not None: + self._lock.release() + raise Errors.NodeNotReadyError() + continue # We will intentionally cause socket failures # These should not trigger metadata refresh diff --git a/kafka/codec.py b/kafka/codec.py index 917400e74..c740a181c 100644 --- a/kafka/codec.py +++ b/kafka/codec.py @@ -187,7 +187,7 @@ def _detect_xerial_stream(payload): The version is the version of this format as written by xerial, in the wild this is currently 1 as such we only support v1. - Compat is there to claim the miniumum supported version that + Compat is there to claim the minimum supported version that can read a xerial block stream, presently in the wild this is 1. """ diff --git a/kafka/conn.py b/kafka/conn.py index cac354875..df903f264 100644 --- a/kafka/conn.py +++ b/kafka/conn.py @@ -33,6 +33,7 @@ from kafka.protocol.parser import KafkaProtocol from kafka.protocol.types import Int32, Int8 from kafka.scram import ScramClient +from kafka.socks5_wrapper import Socks5Wrapper from kafka.version import __version__ @@ -78,7 +79,7 @@ class SSLWantWriteError(Exception): try: import gssapi from gssapi.raw.misc import GSSError -except ImportError: +except (ImportError, OSError): #no gssapi available, will disable gssapi mechanism gssapi = None GSSError = None @@ -191,6 +192,7 @@ class BrokerConnection(object): sasl mechanism handshake. Default: one of bootstrap servers sasl_oauth_token_provider (AbstractTokenProvider): OAuthBearer token provider instance. (See kafka.oauth.abstract). Default: None + socks5_proxy (str): Socks5 proxy url. Default: None """ DEFAULT_CONFIG = { @@ -224,7 +226,8 @@ class BrokerConnection(object): 'sasl_plain_password': None, 'sasl_kerberos_service_name': 'kafka', 'sasl_kerberos_domain_name': None, - 'sasl_oauth_token_provider': None + 'sasl_oauth_token_provider': None, + 'socks5_proxy': None, } SECURITY_PROTOCOLS = ('PLAINTEXT', 'SSL', 'SASL_PLAINTEXT', 'SASL_SSL') SASL_MECHANISMS = ('PLAIN', 'GSSAPI', 'OAUTHBEARER', "SCRAM-SHA-256", "SCRAM-SHA-512") @@ -236,6 +239,7 @@ def __init__(self, host, port, afi, **configs): self._sock_afi = afi self._sock_addr = None self._api_versions = None + self._socks5_proxy = None self.config = copy.copy(self.DEFAULT_CONFIG) for key in self.config: @@ -368,7 +372,11 @@ def connect(self): log.debug('%s: creating new socket', self) assert self._sock is None self._sock_afi, self._sock_addr = next_lookup - self._sock = socket.socket(self._sock_afi, socket.SOCK_STREAM) + if self.config["socks5_proxy"] is not None: + self._socks5_proxy = Socks5Wrapper(self.config["socks5_proxy"], self.afi) + self._sock = self._socks5_proxy.socket(self._sock_afi, socket.SOCK_STREAM) + else: + self._sock = socket.socket(self._sock_afi, socket.SOCK_STREAM) for option in self.config['socket_options']: log.debug('%s: setting socket option %s', self, option) @@ -385,7 +393,10 @@ def connect(self): # to check connection status ret = None try: - ret = self._sock.connect_ex(self._sock_addr) + if self._socks5_proxy: + ret = self._socks5_proxy.connect_ex(self._sock_addr) + else: + ret = self._sock.connect_ex(self._sock_addr) except socket.error as err: ret = err.errno diff --git a/kafka/consumer/group.py b/kafka/consumer/group.py index a1d1dfa37..969969932 100644 --- a/kafka/consumer/group.py +++ b/kafka/consumer/group.py @@ -245,6 +245,7 @@ class KafkaConsumer(six.Iterator): sasl_oauth_token_provider (AbstractTokenProvider): OAuthBearer token provider instance. (See kafka.oauth.abstract). Default: None kafka_client (callable): Custom class / callable for creating KafkaClient instances + socks5_proxy (str): Socks5 proxy URL. Default: None Note: Configuration parameters are described in more detail at @@ -308,6 +309,7 @@ class KafkaConsumer(six.Iterator): 'sasl_oauth_token_provider': None, 'legacy_iterator': False, # enable to revert to < 1.4.7 iterator 'kafka_client': KafkaClient, + 'socks5_proxy': None, } DEFAULT_SESSION_TIMEOUT_MS_0_9 = 30000 diff --git a/kafka/coordinator/base.py b/kafka/coordinator/base.py index 5e41309df..e71984108 100644 --- a/kafka/coordinator/base.py +++ b/kafka/coordinator/base.py @@ -952,7 +952,7 @@ def _run_once(self): # disable here to prevent propagating an exception to this # heartbeat thread # must get client._lock, or maybe deadlock at heartbeat - # failure callbak in consumer poll + # failure callback in consumer poll self.coordinator._client.poll(timeout_ms=0) with self.coordinator._lock: diff --git a/kafka/producer/kafka.py b/kafka/producer/kafka.py index dd1cc508c..431642776 100644 --- a/kafka/producer/kafka.py +++ b/kafka/producer/kafka.py @@ -281,6 +281,7 @@ class KafkaProducer(object): sasl_oauth_token_provider (AbstractTokenProvider): OAuthBearer token provider instance. (See kafka.oauth.abstract). Default: None kafka_client (callable): Custom class / callable for creating KafkaClient instances + socks5_proxy (str): Socks5 proxy URL. Default: None Note: Configuration parameters are described in more detail at @@ -335,6 +336,7 @@ class KafkaProducer(object): 'sasl_kerberos_domain_name': None, 'sasl_oauth_token_provider': None, 'kafka_client': KafkaClient, + 'socks5_proxy': None, } _COMPRESSORS = { diff --git a/kafka/record/_crc32c.py b/kafka/record/_crc32c.py index ecff48f5e..9b51ad8a9 100644 --- a/kafka/record/_crc32c.py +++ b/kafka/record/_crc32c.py @@ -105,7 +105,7 @@ def crc_update(crc, data): Returns: 32-bit updated CRC-32C as long. """ - if type(data) != array.array or data.itemsize != 1: + if not isinstance(data, array.array) or data.itemsize != 1: buf = array.array("B", data) else: buf = data diff --git a/kafka/record/abc.py b/kafka/record/abc.py index d5c172aaa..8509e23e5 100644 --- a/kafka/record/abc.py +++ b/kafka/record/abc.py @@ -85,7 +85,7 @@ def build(self): class ABCRecordBatch(object): - """ For v2 incapsulates a RecordBatch, for v0/v1 a single (maybe + """ For v2 encapsulates a RecordBatch, for v0/v1 a single (maybe compressed) message. """ __metaclass__ = abc.ABCMeta diff --git a/kafka/record/legacy_records.py b/kafka/record/legacy_records.py index e2ee5490c..2f8523fcb 100644 --- a/kafka/record/legacy_records.py +++ b/kafka/record/legacy_records.py @@ -263,7 +263,7 @@ def __iter__(self): # When magic value is greater than 0, the timestamp # of a compressed message depends on the - # typestamp type of the wrapper message: + # timestamp type of the wrapper message: if timestamp_type == self.LOG_APPEND_TIME: timestamp = self._timestamp diff --git a/kafka/socks5_wrapper.py b/kafka/socks5_wrapper.py new file mode 100644 index 000000000..18bea7c8d --- /dev/null +++ b/kafka/socks5_wrapper.py @@ -0,0 +1,248 @@ +try: + from urllib.parse import urlparse +except ImportError: + from urlparse import urlparse + +import errno +import logging +import random +import socket +import struct + +log = logging.getLogger(__name__) + + +class ProxyConnectionStates: + DISCONNECTED = '' + CONNECTING = '' + NEGOTIATE_PROPOSE = '' + NEGOTIATING = '' + AUTHENTICATING = '' + REQUEST_SUBMIT = '' + REQUESTING = '' + READ_ADDRESS = '' + COMPLETE = '' + + +class Socks5Wrapper: + """Socks5 proxy wrapper + + Manages connection through socks5 proxy with support for username/password + authentication. + """ + + def __init__(self, proxy_url, afi): + self._buffer_in = b'' + self._buffer_out = b'' + self._proxy_url = urlparse(proxy_url) + self._sock = None + self._state = ProxyConnectionStates.DISCONNECTED + self._target_afi = socket.AF_UNSPEC + + proxy_addrs = self.dns_lookup(self._proxy_url.hostname, self._proxy_url.port, afi) + # TODO raise error on lookup failure + self._proxy_addr = random.choice(proxy_addrs) + + @classmethod + def is_inet_4_or_6(cls, gai): + """Given a getaddrinfo struct, return True iff ipv4 or ipv6""" + return gai[0] in (socket.AF_INET, socket.AF_INET6) + + @classmethod + def dns_lookup(cls, host, port, afi=socket.AF_UNSPEC): + """Returns a list of getaddrinfo structs, optionally filtered to an afi (ipv4 / ipv6)""" + # XXX: all DNS functions in Python are blocking. If we really + # want to be non-blocking here, we need to use a 3rd-party + # library like python-adns, or move resolution onto its + # own thread. This will be subject to the default libc + # name resolution timeout (5s on most Linux boxes) + try: + return list(filter(cls.is_inet_4_or_6, + socket.getaddrinfo(host, port, afi, + socket.SOCK_STREAM))) + except socket.gaierror as ex: + log.warning("DNS lookup failed for proxy %s:%d, %r", host, port, ex) + return [] + + def socket(self, family, sock_type): + """Open and record a socket. + + Returns the actual underlying socket + object to ensure e.g. selects and ssl wrapping works as expected. + """ + self._target_afi = family # Store the address family of the target + afi, _, _, _, _ = self._proxy_addr + self._sock = socket.socket(afi, sock_type) + return self._sock + + def _flush_buf(self): + """Send out all data that is stored in the outgoing buffer. + + It is expected that the caller handles error handling, including non-blocking + as well as connection failure exceptions. + """ + while self._buffer_out: + sent_bytes = self._sock.send(self._buffer_out) + self._buffer_out = self._buffer_out[sent_bytes:] + + def _peek_buf(self, datalen): + """Ensure local inbound buffer has enough data, and return that data without + consuming the local buffer + + It's expected that the caller handles e.g. blocking exceptions""" + while True: + bytes_remaining = datalen - len(self._buffer_in) + if bytes_remaining <= 0: + break + data = self._sock.recv(bytes_remaining) + if not data: + break + self._buffer_in = self._buffer_in + data + + return self._buffer_in[:datalen] + + def _read_buf(self, datalen): + """Read and consume bytes from socket connection + + It's expected that the caller handles e.g. blocking exceptions""" + buf = self._peek_buf(datalen) + if buf: + self._buffer_in = self._buffer_in[len(buf):] + return buf + + def connect_ex(self, addr): + """Runs a state machine through connection to authentication to + proxy connection request. + + The somewhat strange setup is to facilitate non-intrusive use from + BrokerConnection state machine. + + This function is called with a socket in non-blocking mode. Both + send and receive calls can return in EWOULDBLOCK/EAGAIN which we + specifically avoid handling here. These are handled in main + BrokerConnection connection loop, which then would retry calls + to this function.""" + + if self._state == ProxyConnectionStates.DISCONNECTED: + self._state = ProxyConnectionStates.CONNECTING + + if self._state == ProxyConnectionStates.CONNECTING: + _, _, _, _, sockaddr = self._proxy_addr + ret = self._sock.connect_ex(sockaddr) + if not ret or ret == errno.EISCONN: + self._state = ProxyConnectionStates.NEGOTIATE_PROPOSE + else: + return ret + + if self._state == ProxyConnectionStates.NEGOTIATE_PROPOSE: + if self._proxy_url.username and self._proxy_url.password: + # Propose username/password + self._buffer_out = b"\x05\x01\x02" + else: + # Propose no auth + self._buffer_out = b"\x05\x01\x00" + self._state = ProxyConnectionStates.NEGOTIATING + + if self._state == ProxyConnectionStates.NEGOTIATING: + self._flush_buf() + buf = self._read_buf(2) + if buf[0:1] != b"\x05": + log.error("Unrecognized SOCKS version") + self._state = ProxyConnectionStates.DISCONNECTED + self._sock.close() + return errno.ECONNREFUSED + + if buf[1:2] == b"\x00": + # No authentication required + self._state = ProxyConnectionStates.REQUEST_SUBMIT + elif buf[1:2] == b"\x02": + # Username/password authentication selected + userlen = len(self._proxy_url.username) + passlen = len(self._proxy_url.password) + self._buffer_out = struct.pack( + "!bb{}sb{}s".format(userlen, passlen), + 1, # version + userlen, + self._proxy_url.username.encode(), + passlen, + self._proxy_url.password.encode(), + ) + self._state = ProxyConnectionStates.AUTHENTICATING + else: + log.error("Unrecognized SOCKS authentication method") + self._state = ProxyConnectionStates.DISCONNECTED + self._sock.close() + return errno.ECONNREFUSED + + if self._state == ProxyConnectionStates.AUTHENTICATING: + self._flush_buf() + buf = self._read_buf(2) + if buf == b"\x01\x00": + # Authentication succesful + self._state = ProxyConnectionStates.REQUEST_SUBMIT + else: + log.error("Socks5 proxy authentication failure") + self._state = ProxyConnectionStates.DISCONNECTED + self._sock.close() + return errno.ECONNREFUSED + + if self._state == ProxyConnectionStates.REQUEST_SUBMIT: + if self._target_afi == socket.AF_INET: + addr_type = 1 + addr_len = 4 + elif self._target_afi == socket.AF_INET6: + addr_type = 4 + addr_len = 16 + else: + log.error("Unknown address family, %r", self._target_afi) + self._state = ProxyConnectionStates.DISCONNECTED + self._sock.close() + return errno.ECONNREFUSED + + self._buffer_out = struct.pack( + "!bbbb{}sh".format(addr_len), + 5, # version + 1, # command: connect + 0, # reserved + addr_type, # 1 for ipv4, 4 for ipv6 address + socket.inet_pton(self._target_afi, addr[0]), # either 4 or 16 bytes of actual address + addr[1], # port + ) + self._state = ProxyConnectionStates.REQUESTING + + if self._state == ProxyConnectionStates.REQUESTING: + self._flush_buf() + buf = self._read_buf(2) + if buf[0:2] == b"\x05\x00": + self._state = ProxyConnectionStates.READ_ADDRESS + else: + log.error("Proxy request failed: %r", buf[1:2]) + self._state = ProxyConnectionStates.DISCONNECTED + self._sock.close() + return errno.ECONNREFUSED + + if self._state == ProxyConnectionStates.READ_ADDRESS: + # we don't really care about the remote endpoint address, but need to clear the stream + buf = self._peek_buf(2) + if buf[0:2] == b"\x00\x01": + _ = self._read_buf(2 + 4 + 2) # ipv4 address + port + elif buf[0:2] == b"\x00\x05": + _ = self._read_buf(2 + 16 + 2) # ipv6 address + port + else: + log.error("Unrecognized remote address type %r", buf[1:2]) + self._state = ProxyConnectionStates.DISCONNECTED + self._sock.close() + return errno.ECONNREFUSED + self._state = ProxyConnectionStates.COMPLETE + + if self._state == ProxyConnectionStates.COMPLETE: + return 0 + + # not reached; + # Send and recv will raise socket error on EWOULDBLOCK/EAGAIN that is assumed to be handled by + # the caller. The caller re-enters this state machine from retry logic with timer or via select & family + log.error("Internal error, state %r not handled correctly", self._state) + self._state = ProxyConnectionStates.DISCONNECTED + if self._sock: + self._sock.close() + return errno.ECONNREFUSED diff --git a/kafka/vendor/selectors34.py b/kafka/vendor/selectors34.py index ebf5d515e..787490340 100644 --- a/kafka/vendor/selectors34.py +++ b/kafka/vendor/selectors34.py @@ -15,7 +15,11 @@ from __future__ import absolute_import from abc import ABCMeta, abstractmethod -from collections import namedtuple, Mapping +from collections import namedtuple +try: + from collections.abc import Mapping +except ImportError: + from collections import Mapping from errno import EINTR import math import select diff --git a/kafka/vendor/six.py b/kafka/vendor/six.py index 3621a0ab4..319821353 100644 --- a/kafka/vendor/six.py +++ b/kafka/vendor/six.py @@ -1,6 +1,6 @@ # pylint: skip-file -# Copyright (c) 2010-2017 Benjamin Peterson +# Copyright (c) 2010-2020 Benjamin Peterson # # Permission is hereby granted, free of charge, to any person obtaining a copy # of this software and associated documentation files (the "Software"), to deal @@ -31,7 +31,7 @@ import types __author__ = "Benjamin Peterson " -__version__ = "1.11.0" +__version__ = "1.16.0" # Useful for very coarse version differentiation. @@ -77,6 +77,11 @@ def __len__(self): # https://github.com/dpkp/kafka-python/pull/979#discussion_r100403389 # del X +if PY34: + from importlib.util import spec_from_loader +else: + spec_from_loader = None + def _add_doc(func, doc): """Add documentation to a function.""" @@ -192,6 +197,11 @@ def find_module(self, fullname, path=None): return self return None + def find_spec(self, fullname, path, target=None): + if fullname in self.known_modules: + return spec_from_loader(fullname, self) + return None + def __get_module(self, fullname): try: return self.known_modules[fullname] @@ -229,6 +239,12 @@ def get_code(self, fullname): return None get_source = get_code # same as get_code + def create_module(self, spec): + return self.load_module(spec.name) + + def exec_module(self, module): + pass + _importer = _SixMetaPathImporter(__name__) @@ -253,7 +269,7 @@ class _MovedItems(_LazyModule): MovedAttribute("reduce", "__builtin__", "functools"), MovedAttribute("shlex_quote", "pipes", "shlex", "quote"), MovedAttribute("StringIO", "StringIO", "io"), - MovedAttribute("UserDict", "UserDict", "collections"), + MovedAttribute("UserDict", "UserDict", "collections", "IterableUserDict", "UserDict"), MovedAttribute("UserList", "UserList", "collections"), MovedAttribute("UserString", "UserString", "collections"), MovedAttribute("xrange", "__builtin__", "builtins", "xrange", "range"), @@ -261,9 +277,11 @@ class _MovedItems(_LazyModule): MovedAttribute("zip_longest", "itertools", "itertools", "izip_longest", "zip_longest"), MovedModule("builtins", "__builtin__"), MovedModule("configparser", "ConfigParser"), + MovedModule("collections_abc", "collections", "collections.abc" if sys.version_info >= (3, 3) else "collections"), MovedModule("copyreg", "copy_reg"), MovedModule("dbm_gnu", "gdbm", "dbm.gnu"), - MovedModule("_dummy_thread", "dummy_thread", "_dummy_thread"), + MovedModule("dbm_ndbm", "dbm", "dbm.ndbm"), + MovedModule("_dummy_thread", "dummy_thread", "_dummy_thread" if sys.version_info < (3, 9) else "_thread"), MovedModule("http_cookiejar", "cookielib", "http.cookiejar"), MovedModule("http_cookies", "Cookie", "http.cookies"), MovedModule("html_entities", "htmlentitydefs", "html.entities"), @@ -643,13 +661,16 @@ def u(s): import io StringIO = io.StringIO BytesIO = io.BytesIO + del io _assertCountEqual = "assertCountEqual" if sys.version_info[1] <= 1: _assertRaisesRegex = "assertRaisesRegexp" _assertRegex = "assertRegexpMatches" + _assertNotRegex = "assertNotRegexpMatches" else: _assertRaisesRegex = "assertRaisesRegex" _assertRegex = "assertRegex" + _assertNotRegex = "assertNotRegex" else: def b(s): return s @@ -671,6 +692,7 @@ def indexbytes(buf, i): _assertCountEqual = "assertItemsEqual" _assertRaisesRegex = "assertRaisesRegexp" _assertRegex = "assertRegexpMatches" + _assertNotRegex = "assertNotRegexpMatches" _add_doc(b, """Byte literal""") _add_doc(u, """Text literal""") @@ -687,6 +709,10 @@ def assertRegex(self, *args, **kwargs): return getattr(self, _assertRegex)(*args, **kwargs) +def assertNotRegex(self, *args, **kwargs): + return getattr(self, _assertNotRegex)(*args, **kwargs) + + if PY3: exec_ = getattr(moves.builtins, "exec") @@ -722,16 +748,7 @@ def exec_(_code_, _globs_=None, _locs_=None): """) -if sys.version_info[:2] == (3, 2): - exec_("""def raise_from(value, from_value): - try: - if from_value is None: - raise value - raise value from from_value - finally: - value = None -""") -elif sys.version_info[:2] > (3, 2): +if sys.version_info[:2] > (3,): exec_("""def raise_from(value, from_value): try: raise value from from_value @@ -811,13 +828,33 @@ def print_(*args, **kwargs): _add_doc(reraise, """Reraise an exception.""") if sys.version_info[0:2] < (3, 4): + # This does exactly the same what the :func:`py3:functools.update_wrapper` + # function does on Python versions after 3.2. It sets the ``__wrapped__`` + # attribute on ``wrapper`` object and it doesn't raise an error if any of + # the attributes mentioned in ``assigned`` and ``updated`` are missing on + # ``wrapped`` object. + def _update_wrapper(wrapper, wrapped, + assigned=functools.WRAPPER_ASSIGNMENTS, + updated=functools.WRAPPER_UPDATES): + for attr in assigned: + try: + value = getattr(wrapped, attr) + except AttributeError: + continue + else: + setattr(wrapper, attr, value) + for attr in updated: + getattr(wrapper, attr).update(getattr(wrapped, attr, {})) + wrapper.__wrapped__ = wrapped + return wrapper + _update_wrapper.__doc__ = functools.update_wrapper.__doc__ + def wraps(wrapped, assigned=functools.WRAPPER_ASSIGNMENTS, updated=functools.WRAPPER_UPDATES): - def wrapper(f): - f = functools.wraps(wrapped, assigned, updated)(f) - f.__wrapped__ = wrapped - return f - return wrapper + return functools.partial(_update_wrapper, wrapped=wrapped, + assigned=assigned, updated=updated) + wraps.__doc__ = functools.wraps.__doc__ + else: wraps = functools.wraps @@ -830,7 +867,15 @@ def with_metaclass(meta, *bases): class metaclass(type): def __new__(cls, name, this_bases, d): - return meta(name, bases, d) + if sys.version_info[:2] >= (3, 7): + # This version introduced PEP 560 that requires a bit + # of extra care (we mimic what is done by __build_class__). + resolved_bases = types.resolve_bases(bases) + if resolved_bases is not bases: + d['__orig_bases__'] = bases + else: + resolved_bases = bases + return meta(name, resolved_bases, d) @classmethod def __prepare__(cls, name, this_bases): @@ -850,13 +895,75 @@ def wrapper(cls): orig_vars.pop(slots_var) orig_vars.pop('__dict__', None) orig_vars.pop('__weakref__', None) + if hasattr(cls, '__qualname__'): + orig_vars['__qualname__'] = cls.__qualname__ return metaclass(cls.__name__, cls.__bases__, orig_vars) return wrapper +def ensure_binary(s, encoding='utf-8', errors='strict'): + """Coerce **s** to six.binary_type. + + For Python 2: + - `unicode` -> encoded to `str` + - `str` -> `str` + + For Python 3: + - `str` -> encoded to `bytes` + - `bytes` -> `bytes` + """ + if isinstance(s, binary_type): + return s + if isinstance(s, text_type): + return s.encode(encoding, errors) + raise TypeError("not expecting type '%s'" % type(s)) + + +def ensure_str(s, encoding='utf-8', errors='strict'): + """Coerce *s* to `str`. + + For Python 2: + - `unicode` -> encoded to `str` + - `str` -> `str` + + For Python 3: + - `str` -> `str` + - `bytes` -> decoded to `str` + """ + # Optimization: Fast return for the common case. + if type(s) is str: + return s + if PY2 and isinstance(s, text_type): + return s.encode(encoding, errors) + elif PY3 and isinstance(s, binary_type): + return s.decode(encoding, errors) + elif not isinstance(s, (text_type, binary_type)): + raise TypeError("not expecting type '%s'" % type(s)) + return s + + +def ensure_text(s, encoding='utf-8', errors='strict'): + """Coerce *s* to six.text_type. + + For Python 2: + - `unicode` -> `unicode` + - `str` -> `unicode` + + For Python 3: + - `str` -> `str` + - `bytes` -> decoded to `str` + """ + if isinstance(s, binary_type): + return s.decode(encoding, errors) + elif isinstance(s, text_type): + return s + else: + raise TypeError("not expecting type '%s'" % type(s)) + + def python_2_unicode_compatible(klass): """ - A decorator that defines __unicode__ and __str__ methods under Python 2. + A class decorator that defines __unicode__ and __str__ methods under Python 2. Under Python 3 it does nothing. To support Python 2 and 3 with a single code base, define a __str__ method diff --git a/requirements-dev.txt b/requirements-dev.txt index 00ad68c22..1fa933da2 100644 --- a/requirements-dev.txt +++ b/requirements-dev.txt @@ -1,17 +1,17 @@ -coveralls==2.1.2 -crc32c==2.1 -docker-py==1.10.6 -flake8==3.8.3 -lz4==3.1.0 -mock==4.0.2 -py==1.9.0 -pylint==2.6.0 -pytest==6.0.2 -pytest-cov==2.10.1 -pytest-mock==3.3.1 -pytest-pylint==0.17.0 -python-snappy==0.5.4 -Sphinx==3.2.1 -sphinx-rtd-theme==0.5.0 -tox==3.20.0 -xxhash==2.0.0 +coveralls +crc32c +docker-py +flake8 +lz4 +mock +py +pylint +pytest +pytest-cov +pytest-mock +pytest-pylint +python-snappy +Sphinx +sphinx-rtd-theme +tox +xxhash diff --git a/setup.py b/setup.py index fe8a594f3..77043da04 100644 --- a/setup.py +++ b/setup.py @@ -40,7 +40,7 @@ def run(cls): "crc32c": ["crc32c"], "lz4": ["lz4"], "snappy": ["python-snappy"], - "zstd": ["python-zstandard"], + "zstd": ["zstandard"], }, cmdclass={"test": Tox}, packages=find_packages(exclude=['test']), @@ -50,7 +50,10 @@ def run(cls): license="Apache License 2.0", description="Pure Python client for Apache Kafka", long_description=README, - keywords="apache kafka", + keywords=[ + "apache kafka", + "kafka", + ], classifiers=[ "Development Status :: 5 - Production/Stable", "Intended Audience :: Developers", @@ -64,6 +67,11 @@ def run(cls): "Programming Language :: Python :: 3.6", "Programming Language :: Python :: 3.7", "Programming Language :: Python :: 3.8", + "Programming Language :: Python :: 3.9", + "Programming Language :: Python :: 3.10", + "Programming Language :: Python :: 3.11", + "Programming Language :: Python :: 3.12", + "Programming Language :: Python :: Implementation :: CPython", "Programming Language :: Python :: Implementation :: PyPy", "Topic :: Software Development :: Libraries :: Python Modules", ] diff --git a/test/fixtures.py b/test/fixtures.py index 26fb5e89d..d9c072b86 100644 --- a/test/fixtures.py +++ b/test/fixtures.py @@ -25,7 +25,7 @@ def get_open_port(): sock = socket.socket() - sock.bind(("", 0)) + sock.bind(("127.0.0.1", 0)) port = sock.getsockname()[1] sock.close() return port diff --git a/test/test_assignors.py b/test/test_assignors.py index 67e91e131..858ef426d 100644 --- a/test/test_assignors.py +++ b/test/test_assignors.py @@ -655,7 +655,7 @@ def test_conflicting_previous_assignments(mocker): 'execution_number,n_topics,n_consumers', [(i, randint(10, 20), randint(20, 40)) for i in range(100)] ) def test_reassignment_with_random_subscriptions_and_changes(mocker, execution_number, n_topics, n_consumers): - all_topics = set(['t{}'.format(i) for i in range(1, n_topics + 1)]) + all_topics = sorted(['t{}'.format(i) for i in range(1, n_topics + 1)]) partitions = dict([(t, set(range(1, i + 1))) for i, t in enumerate(all_topics)]) cluster = create_cluster(mocker, topics=all_topics, topic_partitions_lambda=lambda t: partitions[t]) diff --git a/test/test_client_async.py b/test/test_client_async.py index 74da66a36..b0592086a 100644 --- a/test/test_client_async.py +++ b/test/test_client_async.py @@ -71,13 +71,8 @@ def test_can_connect(cli, conn): def test_maybe_connect(cli, conn): - try: - # Node not in metadata, raises AssertionError - cli._maybe_connect(2) - except AssertionError: - pass - else: - assert False, 'Exception not raised' + # Node not in metadata should be ignored + cli._maybe_connect(2) # New node_id creates a conn object assert 0 not in cli._conns @@ -220,12 +215,12 @@ def test_send(cli, conn): request = ProduceRequest[0](0, 0, []) assert request.expect_response() is False ret = cli.send(0, request) - assert conn.send.called_with(request) + conn.send.assert_called_with(request, blocking=False) assert isinstance(ret, Future) request = MetadataRequest[0]([]) cli.send(0, request) - assert conn.send.called_with(request) + conn.send.assert_called_with(request, blocking=False) def test_poll(mocker): diff --git a/tox.ini b/tox.ini index 10e9911dc..d9b1e36d4 100644 --- a/tox.ini +++ b/tox.ini @@ -1,17 +1,26 @@ [tox] -envlist = py{26,27,34,35,36,37,38,py}, docs +envlist = py{38,39,310,311,312,py}, docs [pytest] testpaths = kafka test addopts = --durations=10 log_format = %(created)f %(filename)-23s %(threadName)s %(message)s +[gh-actions] +python = + 3.8: py38 + 3.9: py39 + 3.10: py310 + 3.11: py311 + 3.12: py312 + pypy-3.9: pypy + [testenv] deps = pytest pytest-cov - py{27,34,35,36,37,38,py}: pylint - py{27,34,35,36,37,38,py}: pytest-pylint + pylint + pytest-pylint pytest-mock mock python-snappy @@ -20,19 +29,16 @@ deps = xxhash crc32c commands = - py.test {posargs:--pylint --pylint-rcfile=pylint.rc --pylint-error-types=EF --cov=kafka --cov-config=.covrc} + pytest {posargs:--pylint --pylint-rcfile=pylint.rc --pylint-error-types=EF --cov=kafka --cov-config=.covrc} setenv = CRC32C_SW_MODE = auto PROJECT_ROOT = {toxinidir} passenv = KAFKA_VERSION -[testenv:py26] -# pylint doesn't support python2.6 -commands = py.test {posargs:--cov=kafka --cov-config=.covrc} [testenv:pypy] # pylint is super slow on pypy... -commands = py.test {posargs:--cov=kafka --cov-config=.covrc} +commands = pytest {posargs:--cov=kafka --cov-config=.covrc} [testenv:docs] deps = diff --git a/travis_java_install.sh b/travis_java_install.sh old mode 100644 new mode 100755