from __future__ import absolute_import

import logging

from kafka import KafkaProducer
from django.utils.functional import cached_property

from sentry import quotas
from sentry.models import Organization
from sentry.eventstream.base import EventStream
from sentry.utils import json
from sentry.utils.pubsub import QueuedPublisher

logger = logging.getLogger(__name__)


# Beware! Changing this, or the message format/fields themselves requires
# consideration of all downstream consumers.
# Version 1 format: (1, '(insert|delete)', {...event json...}, {...state for post-processing...})
EVENT_PROTOCOL_VERSION = 1


class KafkaPublisher(object):
    def __init__(self, connection):
        self.connection = connection or {}

    @cached_property
    def client(self):
        return KafkaProducer(**self.connection)

    def publish(self, topic, value, key=None):
        return self.client.send(topic, key=key, value=value)


class KafkaEventStream(EventStream):
    def __init__(self, publish_topic='events', sync=False, connection=None, **options):
        self.publish_topic = publish_topic
        self.pubsub = KafkaPublisher(connection)
        if not sync:
            self.pubsub = QueuedPublisher(self.pubsub)

    def publish(self, group, event, is_new, is_sample, is_regression, is_new_group_environment, primary_hash, skip_consume=False):
        project = event.project
        retention_days = quotas.get_event_retention(
            organization=Organization(project.organization_id)
        )

        try:
            key = '%s:%s' % (event.project_id, event.event_id)
            value = (EVENT_PROTOCOL_VERSION, 'insert', {
                'group_id': event.group_id,
                'event_id': event.event_id,
                'organization_id': project.organization_id,
                'project_id': event.project_id,
                'message': event.message,
                'platform': event.platform,
                'datetime': event.datetime,
                'data': event.data.data,
                'primary_hash': primary_hash,
                'retention_days': retention_days,
            }, {
                'is_new': is_new,
                'is_sample': is_sample,
                'is_regression': is_regression,
                'is_new_group_environment': is_new_group_environment,
            })

            self.pubsub.publish(self.publish_topic, key=key.encode('utf-8'), value=json.dumps(value))
        except Exception as error:
            logger.warning('Could not publish event: %s', error, exc_info=True)
            raise

@@ -54,16 +54,26 @@
 g%0A%0Afrom 
+confluent_
 kafka im
@@ -77,21 +77,16 @@
  import 
-Kafka
 Producer
@@ -285,56 +285,8 @@
 json
-%0Afrom sentry.utils.pubsub import QueuedPublisher
 %0A%0Alo
@@ -583,24 +583,31 @@
 afka
-Publisher(object
+EventStream(EventStream
 ):%0A 
@@ -632,202 +632,114 @@
 lf, 
-connection):%0A        self.connection = connection or %7B%7D%0A%0A    @cached_property%0A    def client(self):%0A        return KafkaProducer(**self.connection)%0A%0A    def publish(self, topic, value, key=
+publish_topic='events', producer_configuration=None, **options):%0A        if producer_configuration is 
 None
-)
 :%0A  
@@ -748,187 +748,40 @@
     
-return self.client.send(topic, key=key, value=value)%0A%0A%0Aclass KafkaEventStream(EventStream):%0A    def __init__(self, publish_topic='events', sync=False, connection=None, **options):
+    producer_configuration = %7B%7D%0A
 %0A   
@@ -838,117 +838,284 @@
 lf.p
-ubsub = KafkaPublisher(connec
+roducer_configuration = producer_configura
 tion
-)
+%0A
 %0A    
-    if not sync:%0A            self.pubsub = QueuedPublisher(self.pubsub
+@cached_property%0A    def producer(self):%0A        return Producer(self.producer_configuration)%0A%0A    def delivery_callback(self, error, message):%0A        logger.warning('Could not publish event (error: %25s): %25r', error, message
 )%0A%0A 
@@ -1403,16 +1403,665 @@
      )%0A%0A
+        # Polling the producer is required to ensure callbacks are fired. This%0A        # means that the latency between a message being delivered (or failing%0A        # to be delivered) and the corresponding callback being fired is%0A        # roughly the same as the duration of time that passes between publish%0A        # calls. If this ends up being too high, the publisher should be moved%0A        # into a background thread that can poll more frequently without%0A        # interfering with request handling. (This does %60poll%60 does not act as%0A        # a heartbeat for the purposes of any sort of session expiration.)%0A        self.producer.poll(0.0)%0A%0A
         
@@ -2872,17 +2872,16 @@
      %7D)%0A
-%0A
         
@@ -2894,22 +2894,41 @@
 lf.p
-ubsub.publish(
+roducer.produce(%0A                
 self
@@ -2942,16 +2942,32 @@
 h_topic,
+%0A               
  key=key
@@ -2983,16 +2983,32 @@
 utf-8'),
+%0A               
  value=j
@@ -3023,16 +3023,82 @@
 s(value)
+,%0A                on_delivery=self.delivery_callback,%0A            
 )%0A      

Use confluent-kafka (#9094)