Kombu: Kafka support

[rdiomar]

Any plans to add Kafka to the mix? mahendra implemented it here. It would be amazing if that gets pulled in.

[ask]

The first order would be a pull request, but I'm now hesitant to add additional transports unless the transport supports async I/O, as that is needed for the celery worker's prefork pool to be stable.

[ask]

Please open pull request :)

[mahendra]

cool. i will work on this. was a bit held up in the last few months :-/

However, I am not so sure of using kafka as a celery broker. Will send out details later.

@ask can you elaborate on the async I/O part?

[rrmckinley]

@mahendra can you share some of your thoughts on kafka as a celery broker? I want to pursue this, but it sounds like you have concerns?

[ninowalker]

@ask - I'd also appreciate if you could elaborate on the async I/O concern. The kafka consumer(s) have quite different behaviors (and capabilities) relative to the transports already supported.

/cc @mahendra

[anthonyrisinger]

@mahendra any chance you can expand? even if briefly?

we are investigating the viability of using kafka as a message broker, and then making celery behave as sort of stream processor, to unify our uses of kafka and celery.

since your kafka transport has not received any love since 2013, and since you seem to have some reservations, i would very much like to know what they are before committing more time!

[rafaelhbarros]

@mahendra any updates?

[ninowalker]

@rafaelhbarros @mahendra @xtfxme - got clarification on the async I/O question:

"Just means we need to register the socket(s) in the event loop, i.e. at least consuming cannot be blocking" - Ask, https://twitter.com/asksol/status/609424548396707840

[rafaelhbarros]

Is there any interest in the community to add support for Kafka?

[ask]

Yeah, I think there's quite some interest. I would really like that to happen!

[mahendra]

@rafaelhbarros i was originally planning to use kafka with celery (batch) for collecting metrics coming from clients. Not sure if kafka is a good fit for non-batch operations with celery. To get this working with kombu/celery there are few patches that I have to merge into python-kafka. Give me a week or two, let me check on the status of the python-kafka patches. I will get back.

Apologies for going incommunicado on this :-(

[rafaelhbarros]

@mahendra @ask I appreciate the responses, I'll keep watching this issue with hopes!

[thedrow]

@mahendra That's awesome!

[cerdman]

+1

[davidmashburn]

+1

[shazeline]

+1

[lavr]

+1

[cerdman]

anyone have thoughts on where i could start (in terms of building a broker interface/adapter for kafka) on at least developing a proof of concept for this? i am good with kafka / python - have not used celery much at all. i have an urgent need to use celery now in a project however, and the data for part of it is already in kafka - figure its as good a time as any to see if i can help.

[thedrow]

@cerdman The first step is to pick a Python client for Kafka that can work for the versions that we support.
https://github.com/mumrah/kafka-python seems like a possible choice but it doesn't work well with gevent as of yet. See https://github.com/mumrah/kafka-python/issues/37 https://github.com/mumrah/kafka-python/issues/271 https://github.com/mumrah/kafka-python/issues/257
It also seems that the client is not tested with eventlet as well.
I think that the first step would be to add a connection pool to that project and ensure it supports both gevent and eventlet.

Another option would be https://github.com/Parsely/pykafka which is not tested with 3.3 as of yet and is not tested with either gevent and eventlet.

After we chose a client we need to implement the transport interface: https://github.com/celery/kombu/blob/master/kombu/transport/base.py

[hwsdien]

+1

[johnsonc]

+1 @mahendra This would be really useful!!

[BenjaminDavison]

Hey All,

I had a stab at the code here:

https://github.com/BenjaminDavison/kombu/tree/3.0/kombu/transport

Worth noting, I used pykafka instead of kafka-python, as pykafka has just added gevent/greenlet support.

Code is messy at the moment etc (yay for weekend hack projects :dancer: ) was just wondering how to go about testing this code, I have run through the example in the README.md and it produces and consumes fine, going to try and get celery running soon.

(Also note, I just lifted and shifted what @mahendra had done, and replaced all the kafka-python code with pykafka)

Thanks,

Ben

[thedrow]

pykafka also supports librdkafka on CPython which is awesome.
I'll review it as soon as I can. If I forget, ping me.

[BenjaminDavison]

I've managed to get it running in celery, but for some reason when it tries to consume from the pid inbox the consumer seems to exit (if I explicitly stop it consuming from the pid inbox queue it works fine)

I've just pushed a change with some fixes.

[BenjaminDavison]

So I was able to fix the celery pidbox issue (the consumer blocks, so you need to set it to not block)

I was just wondering about the "transport supports async I/O" issue @ask pointed out, pykafka works with gevent, but I don't know if that's enough. What specifically should I be looking for?

I also have a bit of a tangential question around Celery and keeping track of the queue and how that might work with Kafka.

For example:

Person A puts 5 tasks the queue (task1, task2, task3, task4, task5) and let's say for example task 4 fails - how does celery communicate back to (in this example we'll use rabbitmq) that there's a failure? And how does it communicate a success of a task back down to Kombu? Because at the moment I'm just auto committing the offset back to Kafka, so if any of those tasks fail no other workers can get that task again.

Thanks,

Ben

[thedrow]

With RabbitMQ we're using nack to indicate failure and if a dead letter exchange is defined at the server it will be rerouted there. No idea about Kafka though.

[erickt]

@BenjaminDavison: heh I just was mocking out a similar kafka implementation. My approach to solve that whole outstanding tasks was to copy the qpid transport and only allow one outstanding message. That would let me ack and nack without doing fancy things like using a dead letter queue.

I'm not sure how well this would interact with celery though. In order to blend well with Kafka's transactional model, I suspect we'd really need to have a one-to-one Kafka consumer to Celery consumer.

[dpkp]

Update frim kafka-python: we've rewritten the core client to use nonblocking sockets and would love to support kombu integration. Open ticket is https://github.com/dpkp/kafka-python/issues/528

Does kombu have any general documentation on relationship between messages channels transports and connections? In particular I'm looking for ack/reject semantics and expectations for coordination between multiple hosts/processes. The Kafka consumer model generally uses groups of consumers that are assigned different partitions of a topic and each "ack" messages in order on each assigned partition.

[cerdman]

@dpkp - I am still a greenhorn as far as kombu - but am wondering if this might be a good start since it is the module which sets up AMQ emulation for non-AMQ transports and I assume how we would build integration.

https://github.com/celery/kombu/blob/master/kombu/transport/virtual/__init__.py

[erickt]

fyi, here's the branch of my latest kombu + kafka, based off of @BenjaminDavison's branch. I've submitted a PR to merge my branch with his, but it hasn't landed yet. @BenjaminDavison: any chance you could take a look at it?

[dpkp]

cool -- if you merge a pykafka transport, you might think about naming it kombu.transport.pykafka so it doesn't end up conflicting with a transport based on the kafka-python library.

[BenjaminDavison]

@erickt I'm going to be on holiday for the next month, if you want you can create the PR and submit if needed (please also attribute to @mahendra )

[ptallada]

+1

[alanjds]

Sirs, what behavior was implemented to handle failures? I had read the code but I am still confused.

On the reject+requeue, how is that implemented?

[alanjds]

@erickt & @BenjaminDavison: I am trying your fork. Handled some API change on pykafka (https://github.com/Parsely/pykafka/commit/02235816b9f4a43a9d2ee656b6ae6ce2fb3981db) on https://github.com/alanjds/kombu/tree/kafka-erickt and fixed some funtests but still the Kafka tests are running weird. Test basic_get hangs forever on my machine, for example.

Will be back to this on monday, but if you have some insights, please let me know.
Thanks.

[alanjds]

After tried with Kafka 0.9 I found that the tests loops forever even on test_basic_get of funtests. Occurs due a known "issue" (KAFKA-2985) of Kafka 0.9, marked as "Not a Problem" on its JIRA. The comments suggest that there is a client implementation error, maybe on pykafka maybe on kombu.

The commit https://github.com/celery/kombu/commit/42d74181b871700cb5253622ec55f306f51e6114 on branch kombu/kafka-erickt@alanjds raises the error instead of swallowing it. The error does not occur on Kafka 0.8.2.1

[alanjds]

On Kafka 0.8.2.1, the tests fails because of _purge(). It does send the "next message" pointer to the end of the stream, but each time it answers with the whole stream size as the purged count. This made all TransportCase.purge() calls to loop forever. The commit https://github.com/celery/kombu/commit/3b57466b59ee01eb01abacf2227e93759eabb350 on branch kafka-erickt-nopurge@alanjds "fixes" by ignoring this behaviour.

On this branch, the only test not passing is the specific one testing the count of purged messages. I don't know yet if is the kafka transport Channel._purge() or the funtests TransportCase.setUp() that leads to this error and should be fixed.

After talking to people around me used to Kafka (I am not used to it), they got the impression that a purge call makes no semantic sense on the Kafka world. However, I still know not how to deal with it.

Have anyone suggestions please?

[ask]

@alanjds I don't know much about Kafka, and I don't remember much from @mahendra's patch.

Purge is only used for administration, if for some reason you want to clear out the queues.
It would be acceptable for a transport to not implement it, if it does not make sense there.

[ask]

It's also acceptable to not implement basic_get.

This command is used for synchronously fetching messages, and it's almost never used in practice.
The Celery amqp and rpc result backends take use of it though, to implement the in-hindsight-too-much-magic AsyncResult.state/AsyncResult.result, etc properties that has the side effect of polling the backend until the result is ready. Using the properties like this should probably be considered a bad practice, and instead result.get() should always be used first to wait for the result (in Celery 4.0 you will be able to do result.then(callback).

[ask]

Btw, does Kafka have an async client (e.g. for tornado or asyncio)? Writing new synchronous transports is better than nothing, but it's much better if this is asynchronous so that we can use the modern async worker path (used by amqp, redis, qpid, and sqs in master)

[alanjds]

Hum... the pykafka.producer.Producer is async by default via threads and capable of using gevent, by their docs. One of it is created here.

I had a look on redis transport, but was not able to perceive if the actual kafka transport can be considered async in the way you mean.

[alanjds]

...just noticed the reliable_purge test option. Now the pykafka transport have green tests (!!!) on branch kafka-erickt-nopurge@alanjds

[dpkp]

kafka-python is also using an async IO loop by design and as I understand it works w/ gevent. I've also been working on exposing the internal bytes pump so that any arbitrary event loop could be used (asyncio, twisted, etc).

[ask]

@alanjds No threads are allowed in the worker, as you cannot mix threads and fork! The compat worker
path uses threads, which is why it's prone to deadlocks.

It would be a mistake to think that using threads in the background is something the user will benefit from, they have no place in a Python library! It's OK if the API requires the user to start them manually, but as a side effect of instantiating something like a Connection(), or conn.send() or similar: kittens will die painfully!

@dpkp That's awesome! That should mean we can connect kafka-python to our event loop. I've also successfully connected our loop to Tornado, and vice versa.

[alanjds]

I had read somewhere that pykafka use threads to pack multiple messages together to raise the final throughput, but I am not the best to talk about any of this. In fact, the best I could find right now is this old thread: https://news.ycombinator.com/item?id=10013875

[alanjds]

Details about pykafka async implementation: https://github.com/Parsely/pykafka/pull/177

[ashwin31]

:+1:

[madhur230491]

Hi ,

Am looking for some help here . Am using celery workers to send some data to kafka brokers. Am starting celery with evenlet and pykafka client initialization is not working .

Below are the option with which I start celery -
CELERYD_OPTS="-Q prppush,productemails -P eventlet"

I read that recenlty supprt for greenlets and gevent was added but not working for me .
This is how am initialising my client -
self._kafka_connection = KafkaClient(hosts="kafkabroker0.moengage.com:10252,kafkabroker1.moengage.com:10252,kafkabroker2.moengage.com:10252" ,use_greenlets=True)

My python version - Python 2.7.6
My pykafka version - latest
kafka brokers - 0.8.2.1

My implementation is below -

class PyKafkaProducerConsumer(object):

Put the bellow in all classes which needs singleton!

metaclass = SingletonMetaClass

_kafka_connection = None

Moved this to uwsgi init section, check delight/init.py

def init(self,topic="test"):

print "*********** entered pykafka init"
self._kafka_connection = KafkaClient(hosts="kafkabroker0.moengage.com:10252,kafkabroker1.moengage.com:10252,kafkabroker2.moengage.com:10252",use_greenlets=True)
print "*********** exit pykafka init"
#self._topic_connection = self._kafka_connection.topics[topic]

def sendToKafka(self, topic="test" ,msg={}):
print "** sending thru pykafka"
topic_connection = self._kafka_connection.topics["test"]
json_msg=json.dumps(msg)
# Asynchronous by default
producer=topic_connection.get_sync_producer(min_queued_messages=1,delivery_reports=True)
print producer
producer.produce(json_msg)
I only see "**** entered pykafka init" when I try to initialize the client , and it doesnt return the client object form there . Any help would be very much appreciated as am stuck on this from a long time .

Also if celery is started without eventlet , this works fine .

[alanjds]

Hi, @madhur230491.

Does it works for you with the prefork (default) parallelism?

[madhur230491]

@alanjds : Hi , Yeah it works if I use normal celery concurrency without eventlet .

[yufeiminds]

+1 👍

[alanjds]

@siddharth96: I had read that you managed to bake a confluent-kafka backend. Would you mind to shere it here please?

[siddharth96]

@alanjds Hi, Below is the code. It's based on the work that you'd done on top of @mahendra 's work.

"""
kombu.transport.kafka
=====================

Kafka transport.

:copyright: (c) 2010 - 2013 by Mahendra M.
:license: BSD, see LICENSE for more details.

**Synopsis**

Connects to kafka (0.8.x+) as <server>:<port>/<vhost>
The <vhost> becomes the group for all the clients. So we can use
it like a vhost

It is recommended that the queue be created in advance, by specifying the
number of partitions. The partition configuration determines how many
consumers can fetch data in parallel

**Limitations**
* The client API needs to modified to fetch data only from a single
  partition. This can be used for effective load balancing also.


"""

from __future__ import absolute_import

import calendar
import datetime
import confluent_kafka
import traceback
from anyjson import loads, dumps

from kombu.five import Empty
from kombu.transport import virtual
from kombu.utils.compat import OrderedDict
from kombu.utils.url import parse_url

from confluent_kafka import Producer, Consumer
from confluent_kafka.cimpl import KafkaError


KAFKA_CONNECTION_ERRORS = ()
KAFKA_CHANNEL_ERRORS = ()

DEFAULT_PORT = 9092
DEFAULT_KAFKA_GROUP = 'kombu-consumer-group'


class QoS(object):
    def __init__(self, channel):
        self.prefetch_count = 1
        self._channel = channel
        self._not_yet_acked = OrderedDict()

    def can_consume(self):
        """Returns True if the :class:`Channel` can consume more messages, else
        False.

        :returns: True, if this QoS object can accept a message.
        :rtype: bool
        """
        return not self.prefetch_count or len(self._not_yet_acked) < self \
            .prefetch_count

    def can_consume_max_estimate(self):
        if self.prefetch_count:
            return self.prefetch_count - len(self._not_yet_acked)
        else:
            return 1

    def append(self, message, delivery_tag):
        self._not_yet_acked[delivery_tag] = message

    def get(self, delivery_tag):
        return self._not_yet_acked[delivery_tag]

    def ack(self, delivery_tag):
        # Uncomment, if you're not using Kafka's autocommit
        # message = self._not_yet_acked.pop(delivery_tag)
        # topic = self._get_queue(message)
        # # exchange = message.properties['delivery_info']['exchange']
        # consumer = self._channel._get_consumer(topic)
        # consumer.commit()
        pass

    def reject(self, delivery_tag, requeue=False):
        """Reject a message by delivery tag.

        If requeue is True, then the last consumed message is reverted so it'll
        be refetched on the next attempt. If False, that message is consumed and
        ignored.
        In order to revert a message, it's first acknowledged and then re-queued
        """
        if requeue:
            message = self._not_yet_acked.get(delivery_tag)
            self.on_requeue(message, delivery_tag)
        self.ack(delivery_tag)

    def on_requeue(self, message, delivery_tag):
        """
        This approach is risky and reset_offsets is also not recommended
        Feel free to come up with your own re-queue approach.
        :type message: kombu.transport.virtual.Message
        :type delivery_tag: str
        :return:
        """
        raise NotImplementedError()

    def restore_unacked_once(self):
        pass

    def _get_queue(self, message):
        q = None
        # Read queue name from channel
        if self._channel._active_queues and len(self._channel._active_queues) == 1:
            q = self._channel._active_queues[0]
        # If channel is bound to multiple queues then try reading the queue
        # from the message using a 'jugaad'
        elif 'args' in message.payload:
            try:
                # Try reading queue from passed in thread-locals
                # Refer "shared_context" variable in bbcelery/queue.py -> _queueable_impl()
                # i.e., message.payload['args'][1] = "shared_context" in the above method
                q = message.payload['args'][1]['queue']
            except (IndexError, KeyError):
                q = None
        if not q:
            # Since routing-key is same as queue right now
            q = message.properties['delivery_info']['routing_key']
        return q


class Channel(virtual.Channel):
    QoS = QoS

    def __init__(self, *args, **kwargs):
        super(Channel, self).__init__(*args, **kwargs)

        self._kafka_consumers = {}
        self._kafka_producers = {}

    def exchange_unbind(self, destination, source='', routing_key='', nowait=False,
                        arguments=None):
        """
        Not applicable for Kafka
        :param destination:
        :param source:
        :param routing_key:
        :param nowait:
        :param arguments:
        :return:
        """
        pass

    def queue_unbind(self, queue, exchange=None, routing_key='', arguments=None, **kwargs):
        """
        Not applicable for Kafka
        :param queue:
        :param exchange:
        :param routing_key:
        :param arguments:
        :param kwargs:
        :return:
        """
        pass

    def exchange_bind(self, destination, source='', routing_key='', nowait=False, arguments=None):
        """
        Not applicable for Kafka
        :param destination:
        :param source:
        :param routing_key:
        :param nowait:
        :param arguments:
        :return:
        """
        pass

    def flow(self, active=True):
        """
        Not applicable for Kafka
        :param active:
        :return:
        """
        pass

    def fetch_offsets(self, client, topic, offset):
        # This hasn't been migrated to confluent-kafka yet
        """Fetch raw offset data from a topic.
        note: stolen from the pykafka cli
        :param client: KafkaClient connected to the cluster.
        :type client:  :class:`pykafka.KafkaClient`
        :param topic:  Name of the topic.
        :type topic:  :class:`pykafka.topic.Topic`
        :param offset: Offset to reset to. Can be earliest, latest or a
            datetime.  Using a datetime will reset the offset to the latest
            message published *before* the datetime.
        :type offset: :class:`pykafka.common.OffsetType` or
            :class:`datetime.datetime`
        :returns: {
                partition_id: :class:`pykafka.protocol.OffsetPartitionResponse`
            }
        """
        # if offset.lower() == 'earliest':
        #     return topic.earliest_available_offsets()
        # elif offset.lower() == 'latest':
        #     return topic.latest_available_offsets()
        # else:
        #     offset = datetime.datetime.strptime(offset, "%Y-%m-%dT%H:%M:%S")
        #     offset = int(calendar.timegm(offset.utctimetuple()) * 1000)
        #     return topic.fetch_offset_limits(offset)
        pass

    def sanitize_queue_name(self, queue):
        """Need to sanitize the queue name, celery sometimes pushes in @
        signs"""
        return str(queue).replace('@', '')

    def _get_producer(self, queue):
        """Create/get a producer instance for the given topic/queue
        :rtype: Producer
        """
        queue = self.sanitize_queue_name(queue)
        producer = self._kafka_producers.get(queue, None)
        if producer is None:
            config = self.get_producer_config(self._get_hosts())
            producer = Producer(config)
            self._kafka_producers[queue] = producer
        return producer

    @classmethod
    def get_producer_config(cls, hosts):
        return {'bootstrap.servers': hosts,
                'api.version.request': True,
                'group.id': DEFAULT_KAFKA_GROUP
                }

    @classmethod
    def get_consumer_config(cls, hosts):
        config = cls.get_producer_config(hosts)
        config['default.topic.config'] = {'auto.offset.reset': 'smallest'}
        return config

    def _get_consumer(self, queue):
        """Create/get a consumer instance for the given topic/queue
        :rtype: Consumer
        """
        queue = self.sanitize_queue_name(queue)
        consumer = self._kafka_consumers.get(queue, None)
        if consumer is None:
            config = self.get_consumer_config(self._get_hosts())
            consumer = Consumer(config)
            consumer.subscribe([queue])
            self._kafka_consumers[queue] = consumer
        return consumer

    def _put(self, queue, message, **kwargs):
        """Put a message on the topic/queue"""
        queue = self.sanitize_queue_name(queue)
        producer = self._get_producer(queue)
        producer.produce(queue, dumps(message))

    def _get(self, queue, **kwargs):
        """Get a message from the topic/queue"""
        queue = self.sanitize_queue_name(queue)
        if 'celery' in queue:
            raise Empty()
        consumer = self._get_consumer(queue)
        try:
            message = consumer.poll(timeout=1000.0)  # timeout In milliseconds
        except:
            message = None
            print 'An unknown error occured, when polling queue = %s' % queue
            print traceback.format_exc()  # Will go to /var/log/supervisor/stdout

        if not message or message.error():
            if message and message.error().code() != KafkaError._PARTITION_EOF:
                print message.error()
            raise Empty()
        return self._unmarshall(message)

    def _unmarshall(self, message):
        """
        :type message: pykafka.protocol.Message
        :rtype: dict
        """
        return loads(message.value())

    def _purge(self, queue):
        """Purge all pending messages in the topic/queue, taken from the pykafka
        cli
        """
        return 0

    def _delete(self, queue, *args, **kwargs):
        """Delete a queue/topic"""
        # We will just let it go through. There is no API defined yet
        # for deleting a queue/topic, need to be done through kafka itself
        pass

    def _size(self, queue):
        """Gets the number of pending messages in the topic/queue"""
        return 0

    def _new_queue(self, queue, **kwargs):
        """Create a new queue if it does not exist"""
        # Just create a producer, the queue will be created automatically
        # Note: Please, please, please create the topic before hand,
        # preferably with high replication factor and loads of partitions
        queue = self.sanitize_queue_name(queue)
        self._get_producer(queue)

    def _has_queue(self, queue, **kwargs):
        """Check if a queue already exists"""
        return True  # Stubbed, since confluent_kafka will automatically create a topic

    def _get_hosts(self):
        conninfo = self.connection.client
        raw_hostname = conninfo.hostname
        config = parse_url(raw_hostname)
        parsed_hosts = [self._build_host_name(config)]
        if conninfo.alt:
            for alt_raw_hostname in conninfo.alt:
                parsed_hosts.append(self._build_host_name(alt_raw_hostname))
        return ','.join(parsed_hosts)

    @classmethod
    def parse_hostname(cls, raw_hostname, alt_hostname=None):
        raw_hosts = raw_hostname.split(',')
        parsed_hosts = []
        for raw_hostname in raw_hosts:
            config = parse_url(raw_hostname)
            host_name = cls._build_host_name(config)
            if host_name not in parsed_hosts:
                parsed_hosts.append(host_name)
        if alt_hostname:
            for alt_raw_hostname in alt_hostname:
                host_name = cls._build_host_name(alt_raw_hostname)
                if host_name not in parsed_hosts:
                    parsed_hosts.append(host_name)
        return ','.join(parsed_hosts)

    @classmethod
    def _build_host_name(cls, host_config):
        assert host_config['transport'] == 'pykafka', "Invalid kafka transport specified"
        assert host_config['hostname'], "Hostname missing for Kafka transport"
        assert host_config['port'], "No port specified for Kafka transport"
        return '{0}:{1}'.format(host_config['hostname'], int(host_config['port']))

    def close(self):
        super(Channel, self).close()

        for producer in self._kafka_producers.itervalues():
            producer.flush()

        self._kafka_producers = {}

        '''
        # Commented, as otherwise when this method is code is called
        # workers for across all the servers stop consuming message
        for consumer in self._kafka_consumers.itervalues():
            consumer.unsubscribe()
        '''

        self._kafka_consumers = {}


class Transport(virtual.Transport):
    Channel = Channel

    default_port = DEFAULT_PORT

    can_parse_url = True

    driver_type = 'kafka'
    driver_name = 'pykafka'

    def __init__(self, *args, **kwargs):
        if confluent_kafka is None:
            raise ImportError('The confluent_kafka library is not installed')

        super(Transport, self).__init__(*args, **kwargs)

    def driver_version(self):
        return confluent_kafka.version()[0]

    def establish_connection(self):
        return super(Transport, self).establish_connection()

    def close_connection(self, connection):
        return super(Transport, self).close_connection(connection)

PS: In my case generalizing on_requeue was proving hard, so I was just committing the offset & then re-queuing the message. Feel free to fix it 😄

[westurner]

IIUC, what is needed for a PR to add Kafka support to kombu:

• [ ] asyncio support
• [ ] tests pass (kombu, celery)
  
  
  
  • [ ] kombu tests
  
  
  • [ ] celery tests
  
  
  • [ ] docker config for Kafka and kombu/celery with Kafka config
  
  
• [ ] docs
  
  
  
  • [ ] kombu docs
    
    http://docs.celeryproject.org/projects/kombu/en/latest/introduction.html#transport-comparison
  
  
  • [ ] type (native || virtual)
  
  
  • [ ] direct
  
  
  • [ ] topic
  
  
  • [ ] fanout
  
  
  • [ ] priority
  
  
  • [ ] TTL
  
  
  • [ ] celery docs
    
    http://docs.celeryproject.org/en/latest/getting-started/brokers/
  
  
  • [ ] does it support monitoring?
  
  
  • [ ] does it support remote control?
  
  

[auvipy]

I left a comment on the PR

[kivo360]

Is this going to happen anytime in the next year? It would be cool to be able to switch to the Kafka back-end without changing our codebase.

[auvipy]

likely yes

[xavierhardy]

I'm also interested I could help with that.

[choobinejad]

+1... and available to assist if there are particular areas where the community needs assistance making Kafka a supported Celery broker.

[dmitrypol]

+1. Would be great to have official support

[micheller]

@auvipy
Hello!
Could you please specify what has to be done in order to add kafka support to celery?

[micheller]

Also folks,
anybody faced the

[2018-10-26 07:00:09,324: INFO/MainProcess] Closing connection to ::1:2181
[2018-10-26 07:00:09,324: INFO/MainProcess] Zookeeper session lost, state: CLOSED
[2018-10-26 07:00:09,335: INFO/MainProcess] RequestHandler.stop: about to flush requests queue
[2018-10-26 07:00:09,440: INFO/MainProcess] RequestHandler worker: exiting cleanly
[2018-10-26 07:00:09,443: INFO/MainProcess] Connecting to ::1:2181
[2018-10-26 07:00:09,452: INFO/MainProcess] Zookeeper connection established, state: CONNECTED
[2018-10-26 07:00:09,461: INFO/MainProcess] Closing connection to ::1:2181
[2018-10-26 07:00:09,462: INFO/MainProcess] Zookeeper session lost, state: CLOSED
[2018-10-26 07:00:09,470: INFO/MainProcess] RequestHandler.stop: about to flush requests queue

all the way down when running celery with @mahendra 's pykafka implementation?

[holmanb]

It looks like this conversation died in 2018. Is there anyone still working on this?

[auvipy]

not before celery 5 :) most probably

[alanjds]

I am following but not working. My actual projects does not use it, so will have no push to do code right now.

[joekendal]

Wow, 7 years still open even though the fork implemented it

[auvipy]

Wow, 7 years still open even though the fork implemented it

fund the effort for my time on implementing it on kombu 5.x

[westurner]

GitHub Sponsors may be available in your regions?
https://github.com/sponsors

(I have no need for this probably really useful functionality.)

[auvipy]

GitHub Sponsors may be available in your regions?
https://github.com/sponsors

(I have no need for this probably really useful functionality.)

that's not yet available right now. so I am planning to move somewhere in a more developed region soon. for those who want to speed up on this front can help by contributing to my time I can provide virtual US / EU Bank accounts.

[lightoyou]

What can we do to help you, to push this in main line ?
I think many people need this feature, and we are many who can help you to push this the official repository.

[alanjds]

or I can provide virtual US / EU Bank accounts.

@auvipy Could I suggest Gitcoin.co or some other cryptocurrency way? So easier to receive depending on where you live.

What can we do to help you, to push this in main line ?
I think many people need this feature, and we are many who can help you to push this the official repository.

@lightoyou On code, 1st thing is to test to see if existing solutions work. My branch might be working, yet is very old and never hit production. Kafka evolved a lot since them.

One could take it, update and make tests pass. This would be a huge improvement already.

[lightoyou]

Are you ok if I try to switch to https://github.com/dpkp/kafka-python or https://github.com/aio-libs/aiokafka
Maybe aiokafka cause it's use by Faust Streaming python library wich is also develop by ask ?

[auvipy]

Are you ok if I try to switch to https://github.com/dpkp/kafka-python or https://github.com/aio-libs/aiokafka
Maybe aiokafka cause it's use by Faust Streaming python library wich is also develop by ask ?

we are planning to convert core libraries to async-await as needed and move to aio based libs. so aiokafka could be a good playi ground in that regard as Proof of concept

[yuchen001]

would be happy to see it happen👀