{
"info": {
"author": "Robert Thille",
"author_email": "rthille@ciena.com",
"bugtrack_url": null,
"classifiers": [
"Development Status :: 5 - Production/Stable",
"Framework :: Twisted",
"Intended Audience :: Developers",
"License :: OSI Approved :: Apache Software License",
"Operating System :: MacOS :: MacOS X",
"Operating System :: POSIX",
"Programming Language :: Python :: 3",
"Programming Language :: Python :: 3.5",
"Programming Language :: Python :: 3.6",
"Programming Language :: Python :: 3.7",
"Programming Language :: Python :: 3.8",
"Programming Language :: Python :: Implementation :: CPython",
"Programming Language :: Python :: Implementation :: PyPy",
"Topic :: Communications",
"Topic :: System :: Distributed Computing"
],
"description": "\nAfkak is a [Twisted](https://twistedmatrix.com/)-native [Apache Kafka](https://kafka.apache.org/) client library.\nIt provides support for:\n\n* Producing messages, with automatic batching and optional compression.\n* Consuming messages, with group coordination and automatic commit.\n\nLearn more in the **[documentation](https://afkak.readthedocs.io/en/latest/)**, download [from PyPI](https://pypi.org/projects/afkak), or review the [contribution guidelines](./CONTRIBUTING.md).\nPlease report any issues [on GitHub](https://github.com/ciena/afkak/issues).\n\n# Status\n\nAfkak supports these Pythons:\n\n- CPython 3.5, 3.6, 3.7, 3.8, and 3.9\n- PyPy3\n\nWe aim to support Kafka 1.1.x and later.\nIntegration tests are run against these Kafka broker versions:\n\n- 0.9.0.1\n- 1.1.1\n\nTesting against 2.0.0 is planned (see [#45](https://github.com/ciena/afkak/issues/45)).\n\nNewer broker releases will generally function, but not all Afkak features will work on older brokers.\nIn particular, the coordinated consumer won\u2019t work before Kafka 0.9.0.1.\nWe don\u2019t recommend deploying such old releases of Kafka anyway, as they have serious bugs.\n\n# Usage\n\n### High level\n\nNote: This code is not meant to be runnable. See [producer\\_example](./producer_example) and [consumer\\_example](./consumer_example) for runnable example code.\n\n```python\nfrom afkak.client import KafkaClient\nfrom afkak.consumer import Consumer\nfrom afkak.producer import Producer\nfrom afkak.common import (OFFSET_EARLIEST, PRODUCER_ACK_ALL_REPLICAS,\n PRODUCER_ACK_LOCAL_WRITE)\n\nkClient = KafkaClient(\"localhost:9092\")\n\n# To send messages\nproducer = Producer(kClient)\nd1 = producer.send_messages(\"my-topic\", msgs=[b\"some message\"])\nd2 = producer.send_messages(\"my-topic\", msgs=[b\"takes a list\", b\"of messages\"])\n# To get confirmations/errors on the sends, add callbacks to the returned deferreds\nd1.addCallbacks(handleResponses, handleErrors)\n\n# To wait for acknowledgements\n# PRODUCER_ACK_LOCAL_WRITE : server will wait till the data is written to\n# a local log before sending response\n# [ the default ]\n# PRODUCER_ACK_ALL_REPLICAS : server will block until the message is committed\n# by all in sync replicas before sending a response\nproducer = Producer(kClient,\n req_acks=Producer.PRODUCER_ACK_LOCAL_WRITE,\n ack_timeout=2000)\n\nresponseD = producer.send_messages(\"my-topic\", msgs=[b\"message\"])\n\n# Using twisted's @inlineCallbacks:\nresponses = yield responseD\nif response:\n print(response[0].error)\n print(response[0].offset)\n\n# To send messages in batch: You can use a producer with any of the\n# partitioners for doing this. The following producer will collect\n# messages in batch and send them to Kafka after 20 messages are\n# collected or every 60 seconds (whichever comes first). You can\n# also batch by number of bytes.\n# Notes:\n# * If the producer dies before the messages are sent, the caller would\n# * not have had the callbacks called on the send_messages() returned\n# * deferreds, and so can retry.\n# * Calling producer.stop() before the messages are sent will\n# errback() the deferred(s) returned from the send_messages call(s)\nproducer = Producer(kClient, batch_send=True,\n batch_send_every_n=20,\n batch_send_every_t=60)\nresponseD1 = producer.send_messages(\"my-topic\", msgs=[b\"message\"])\nresponseD2 = producer.send_messages(\"my-topic\", msgs=[b\"message 2\"])\n\n# To consume messages\n# define a function which takes a list of messages to process and\n# possibly returns a deferred which fires when the processing is\n# complete.\ndef processor_func(consumer, messages):\n # Store_Messages_In_Database may return a deferred\n result = store_messages_in_database(messages)\n # record last processed message\n consumer.commit()\n return result\n\nthe_partition = 3 # Consume only from partition 3.\nconsumer = Consumer(kClient, \"my-topic\", the_partition, processor_func)\nd = consumer.start(OFFSET_EARLIEST) # Start reading at earliest message\n# The deferred returned by consumer.start() will fire when an error\n# occurs that can't handled by the consumer, or when consumer.stop()\n# is called\nyield d\n\nconsumer.stop()\nkClient.close()\n```\n\n#### Keyed messages\n```python\nfrom afkak.client import KafkaClient\nfrom afkak.producer import Producer\nfrom afkak.partitioner import HashedPartitioner, RoundRobinPartitioner\n\nkafka = KafkaClient(\"localhost:9092\")\n\n# Use the HashedPartitioner so that the producer will use the optional key\n# argument on send_messages()\nproducer = Producer(kafka, partitioner_class=HashedPartitioner)\nproducer.send_messages(\"my-topic\", \"key1\", [b\"some message\"])\nproducer.send_messages(\"my-topic\", \"key2\", [b\"this method\"])\n\n\n```\n\n### Low level\n\n```python\nfrom afkak.client import KafkaClient\nkafka = KafkaClient(\"localhost:9092\")\nreq = ProduceRequest(topic=\"my-topic\", partition=1,\n messages=[KafkaProtocol.encode_message(b\"some message\")])\nresps = afkak.send_produce_request(payloads=[req], fail_on_error=True)\nkafka.close()\n\nresps[0].topic # b\"my-topic\"\nresps[0].partition # 1\nresps[0].error # 0 (hopefully)\nresps[0].offset # offset of the first message sent in this request\n```\n\n# Install\n\nAfkak releases are [available on PyPI][afkak-pypi].\n\nBecause the Afkak dependencies [Twisted][twisted] and [python-snappy][python-snappy] have binary extension modules you will need to install the Python development headers for the interpreter you wish to use:\n\n[afkak-pypi]: https://pypi.python.org/pypi/afkak\n[twisted]: https://pypi.python.org/pypi/Twisted\n[python-snappy]: https://pypi.python.org/pypi/python-snappy\n\n