)]}'
{"oslo_messaging/_drivers/impl_kafka.py":[{"author":{"_account_id":2813,"name":"Mehdi Abaakouk","email":"sileht@sileht.net","username":"sileht"},"change_message_id":"e2f45bf59ce9cf4746ba772d0ca59544fdd6e6e8","unresolved":false,"context_lines":[{"line_number":31,"context_line":"PURPOSE_LISTEN \u003d \u0027listen\u0027"},{"line_number":32,"context_line":""},{"line_number":33,"context_line":"kafka_opts \u003d ["},{"line_number":34,"context_line":"    cfg.StrOpt(\u0027kafka_host\u0027, default\u003d\u0027localhost\u0027,"},{"line_number":35,"context_line":"               help\u003d\u0027Kafka broker Port\u0027),"},{"line_number":36,"context_line":""},{"line_number":37,"context_line":"    cfg.IntOpt(\u0027kafka_port\u0027, default\u003d9092,"}],"source_content_type":"text/x-python","patch_set":6,"id":"da20952f_eb8ca2d8","line":34,"updated":"2015-09-07 07:44:32.000000000","message":"This options should come from the url passed to driver not from oslo.config.","commit_id":"0f5601007874d802b9703c233e20f3a43a52f7f5"},{"author":{"_account_id":2813,"name":"Mehdi Abaakouk","email":"sileht@sileht.net","username":"sileht"},"change_message_id":"e2f45bf59ce9cf4746ba772d0ca59544fdd6e6e8","unresolved":false,"context_lines":[{"line_number":34,"context_line":"    cfg.StrOpt(\u0027kafka_host\u0027, default\u003d\u0027localhost\u0027,"},{"line_number":35,"context_line":"               help\u003d\u0027Kafka broker Port\u0027),"},{"line_number":36,"context_line":""},{"line_number":37,"context_line":"    cfg.IntOpt(\u0027kafka_port\u0027, default\u003d9092,"},{"line_number":38,"context_line":"               help\u003d\u0027Kafka broker Port\u0027),"},{"line_number":39,"context_line":""},{"line_number":40,"context_line":"    cfg.StrOpt(\u0027kafka_consumer_group\u0027, default\u003d\u0027openstack\u0027,"}],"source_content_type":"text/x-python","patch_set":6,"id":"da20952f_8bbe4663","line":37,"updated":"2015-09-07 07:44:32.000000000","message":"This options should come from the url passed to driver not from oslo.config.","commit_id":"0f5601007874d802b9703c233e20f3a43a52f7f5"},{"author":{"_account_id":2813,"name":"Mehdi Abaakouk","email":"sileht@sileht.net","username":"sileht"},"change_message_id":"e2f45bf59ce9cf4746ba772d0ca59544fdd6e6e8","unresolved":false,"context_lines":[{"line_number":38,"context_line":"               help\u003d\u0027Kafka broker Port\u0027),"},{"line_number":39,"context_line":""},{"line_number":40,"context_line":"    cfg.StrOpt(\u0027kafka_consumer_group\u0027, default\u003d\u0027openstack\u0027,"},{"line_number":41,"context_line":"               help\u003d\u0027Consumer group for Kafka subscriber\u0027),"},{"line_number":42,"context_line":""},{"line_number":43,"context_line":"    cfg.IntOpt(\u0027kafka_max_queue_size\u0027, default\u003d1024,"},{"line_number":44,"context_line":"               help\u003d\u0027Max queue size\u0027),"}],"source_content_type":"text/x-python","patch_set":6,"id":"da20952f_2bae1a2e","line":41,"updated":"2015-09-07 07:44:32.000000000","message":"This looks like a \"exchange\", this options should come from the url passed to driver not from oslo.config.","commit_id":"0f5601007874d802b9703c233e20f3a43a52f7f5"},{"author":{"_account_id":2813,"name":"Mehdi Abaakouk","email":"sileht@sileht.net","username":"sileht"},"change_message_id":"e2f45bf59ce9cf4746ba772d0ca59544fdd6e6e8","unresolved":false,"context_lines":[{"line_number":84,"context_line":"    return context_dict"},{"line_number":85,"context_line":""},{"line_number":86,"context_line":""},{"line_number":87,"context_line":"class ConnectionPool(pool.Pool):"},{"line_number":88,"context_line":"    \"\"\"Connection pool for Kafka\"\"\""},{"line_number":89,"context_line":""},{"line_number":90,"context_line":"    def __init__(self, conf, pool_size, url, connection_cls):"}],"source_content_type":"text/x-python","patch_set":6,"id":"da20952f_8bd2c65a","line":87,"updated":"2015-09-07 07:44:32.000000000","message":"Why have this code been duplicated?","commit_id":"0f5601007874d802b9703c233e20f3a43a52f7f5"},{"author":{"_account_id":2813,"name":"Mehdi Abaakouk","email":"sileht@sileht.net","username":"sileht"},"change_message_id":"e2f45bf59ce9cf4746ba772d0ca59544fdd6e6e8","unresolved":false,"context_lines":[{"line_number":104,"context_line":"            item.close()"},{"line_number":105,"context_line":""},{"line_number":106,"context_line":""},{"line_number":107,"context_line":"class ConnectionContext(object):"},{"line_number":108,"context_line":"    def __init__(self, connection_pool, purpose):"},{"line_number":109,"context_line":"        \"\"\"Create a new connection, or get one from the pool.\"\"\""},{"line_number":110,"context_line":"        self.connection \u003d None"}],"source_content_type":"text/x-python","patch_set":6,"id":"da20952f_4bc82ea8","line":107,"updated":"2015-09-07 07:44:32.000000000","message":"Why have this code been duplicated?","commit_id":"0f5601007874d802b9703c233e20f3a43a52f7f5"},{"author":{"_account_id":2813,"name":"Mehdi Abaakouk","email":"sileht@sileht.net","username":"sileht"},"change_message_id":"e2f45bf59ce9cf4746ba772d0ca59544fdd6e6e8","unresolved":false,"context_lines":[{"line_number":175,"context_line":"        driver_conf \u003d conf.oslo_messaging_kafka"},{"line_number":176,"context_line":""},{"line_number":177,"context_line":"        self.conf \u003d conf"},{"line_number":178,"context_line":"        self.purpose \u003d purpose"},{"line_number":179,"context_line":"        self.local_queue \u003d []"},{"line_number":180,"context_line":"        self.kafka_client \u003d None"},{"line_number":181,"context_line":"        self.kafka_producer \u003d None"}],"source_content_type":"text/x-python","patch_set":6,"id":"da20952f_6b109200","line":178,"updated":"2015-09-07 07:44:32.000000000","message":"purpose is never used","commit_id":"0f5601007874d802b9703c233e20f3a43a52f7f5"},{"author":{"_account_id":2813,"name":"Mehdi Abaakouk","email":"sileht@sileht.net","username":"sileht"},"change_message_id":"e2f45bf59ce9cf4746ba772d0ca59544fdd6e6e8","unresolved":false,"context_lines":[{"line_number":185,"context_line":"        self.max_queue_length \u003d driver_conf.kafka_max_queue_size"},{"line_number":186,"context_line":"        self.max_retry \u003d driver_conf.kafka_max_queue_size"},{"line_number":187,"context_line":"        self.policy \u003d driver_conf.kafka_send_policy"},{"line_number":188,"context_line":"        self.max_retry \u003d driver_conf.kafka_max_retry"},{"line_number":189,"context_line":"        self.consumer_timeout_ms \u003d driver_conf.kafka_consumer_timeout_ms"},{"line_number":190,"context_line":"        self.url \u003d url"},{"line_number":191,"context_line":"        self._parse_url()"}],"source_content_type":"text/x-python","patch_set":6,"id":"da20952f_6b9ad263","line":188,"updated":"2015-09-07 07:44:32.000000000","message":"Why did you have two options for retry? Also try is something that come from public API, that shouldn\u0027t be a oslo.config option.","commit_id":"0f5601007874d802b9703c233e20f3a43a52f7f5"},{"author":{"_account_id":2813,"name":"Mehdi Abaakouk","email":"sileht@sileht.net","username":"sileht"},"change_message_id":"e2f45bf59ce9cf4746ba772d0ca59544fdd6e6e8","unresolved":false,"context_lines":[{"line_number":197,"context_line":""},{"line_number":198,"context_line":"        try:"},{"line_number":199,"context_line":"            self._get_kafka_client()"},{"line_number":200,"context_line":"        except Exception as e:"},{"line_number":201,"context_line":"            LOG.exception(_LE(\"Kafka Broker is not available: %s\"), e)"},{"line_number":202,"context_line":""},{"line_number":203,"context_line":"    def _parse_url(self):"}],"source_content_type":"text/x-python","patch_set":6,"id":"da20952f_8bc96628","line":200,"updated":"2015-09-07 07:44:32.000000000","message":"Can you use more precise exception handling.","commit_id":"0f5601007874d802b9703c233e20f3a43a52f7f5"},{"author":{"_account_id":2813,"name":"Mehdi Abaakouk","email":"sileht@sileht.net","username":"sileht"},"change_message_id":"e2f45bf59ce9cf4746ba772d0ca59544fdd6e6e8","unresolved":false,"context_lines":[{"line_number":208,"context_line":"            self.host \u003d driver_conf.kafka_host"},{"line_number":209,"context_line":"        try:"},{"line_number":210,"context_line":"            self.port \u003d self.url.hosts[0].port or driver_conf.kafka_port"},{"line_number":211,"context_line":"        except Exception:"},{"line_number":212,"context_line":"            self.port \u003d driver_conf.kafka_port"},{"line_number":213,"context_line":""},{"line_number":214,"context_line":"    def notify_send(self, topic, ctxt, message):"}],"source_content_type":"text/x-python","patch_set":6,"id":"da20952f_eb47c2a0","line":211,"updated":"2015-09-07 07:44:32.000000000","message":"Can you use more precise exception handling.","commit_id":"0f5601007874d802b9703c233e20f3a43a52f7f5"},{"author":{"_account_id":2813,"name":"Mehdi Abaakouk","email":"sileht@sileht.net","username":"sileht"},"change_message_id":"e2f45bf59ce9cf4746ba772d0ca59544fdd6e6e8","unresolved":false,"context_lines":[{"line_number":218,"context_line":"        :param message: messages for publishing"},{"line_number":219,"context_line":"        \"\"\""},{"line_number":220,"context_line":"        pack_context_with_message(ctxt, message)"},{"line_number":221,"context_line":"        self.local_queue.append({\"topic\": topic, \"message\": message})"},{"line_number":222,"context_line":""},{"line_number":223,"context_line":"        try:"},{"line_number":224,"context_line":"            self._check_kafka_connection()"}],"source_content_type":"text/x-python","patch_set":6,"id":"da20952f_ebd4a2e9","line":221,"updated":"2015-09-07 07:44:32.000000000","message":"Why did you create this queue ? I don\u0027t really like to have an driver that create an in-memory queue of messages. For me, this is something that the application that consume the lib should deal with, not the driver.","commit_id":"0f5601007874d802b9703c233e20f3a43a52f7f5"},{"author":{"_account_id":2813,"name":"Mehdi Abaakouk","email":"sileht@sileht.net","username":"sileht"},"change_message_id":"e2f45bf59ce9cf4746ba772d0ca59544fdd6e6e8","unresolved":false,"context_lines":[{"line_number":234,"context_line":"    def flush(self):"},{"line_number":235,"context_line":"        \"\"\"flush the queue of this publisher.\"\"\""},{"line_number":236,"context_line":"        queue \u003d self.local_queue"},{"line_number":237,"context_line":"        self.local_queue \u003d self._process_queue(queue)"},{"line_number":238,"context_line":""},{"line_number":239,"context_line":"        if self.policy \u003d\u003d \u0027queue\u0027:"},{"line_number":240,"context_line":"            self._check_queue_length()"}],"source_content_type":"text/x-python","patch_set":6,"id":"da20952f_8bbc06dd","line":237,"updated":"2015-09-07 07:44:32.000000000","message":"This is not thread safe, messages can be added into self.local_queue during process_queue proceed and will be lost during this assignment.","commit_id":"0f5601007874d802b9703c233e20f3a43a52f7f5"},{"author":{"_account_id":2813,"name":"Mehdi Abaakouk","email":"sileht@sileht.net","username":"sileht"},"change_message_id":"e2f45bf59ce9cf4746ba772d0ca59544fdd6e6e8","unresolved":false,"context_lines":[{"line_number":286,"context_line":"        This would be helpful when we wants to control the velocity of"},{"line_number":287,"context_line":"        subscription."},{"line_number":288,"context_line":"        \"\"\""},{"line_number":289,"context_line":"        messages \u003d self._retrieve()"},{"line_number":290,"context_line":"        return messages"},{"line_number":291,"context_line":""},{"line_number":292,"context_line":"    def _retrieve(self):"}],"source_content_type":"text/x-python","patch_set":6,"id":"da20952f_6b6e5201","line":289,"updated":"2015-09-07 07:44:32.000000000","message":"This variable assignement is useless.","commit_id":"0f5601007874d802b9703c233e20f3a43a52f7f5"},{"author":{"_account_id":2813,"name":"Mehdi Abaakouk","email":"sileht@sileht.net","username":"sileht"},"change_message_id":"e2f45bf59ce9cf4746ba772d0ca59544fdd6e6e8","unresolved":false,"context_lines":[{"line_number":290,"context_line":"        return messages"},{"line_number":291,"context_line":""},{"line_number":292,"context_line":"    def _retrieve(self):"},{"line_number":293,"context_line":"        return self.kafka_subscriber.fetch_messages()"},{"line_number":294,"context_line":""},{"line_number":295,"context_line":"    def reset(self):"},{"line_number":296,"context_line":"        \"\"\"Reset a connection so it can be used again.\"\"\""}],"source_content_type":"text/x-python","patch_set":6,"id":"da20952f_0b7a36bd","line":293,"updated":"2015-09-07 07:44:32.000000000","message":"This can go in retreive method.","commit_id":"0f5601007874d802b9703c233e20f3a43a52f7f5"},{"author":{"_account_id":2813,"name":"Mehdi Abaakouk","email":"sileht@sileht.net","username":"sileht"},"change_message_id":"e2f45bf59ce9cf4746ba772d0ca59544fdd6e6e8","unresolved":false,"context_lines":[{"line_number":296,"context_line":"        \"\"\"Reset a connection so it can be used again.\"\"\""},{"line_number":297,"context_line":"        if self.kafka_client:"},{"line_number":298,"context_line":"            self.kafka_client.close()"},{"line_number":299,"context_line":"        self.kafka_producer \u003d None"},{"line_number":300,"context_line":"        self.kafka_subscriber \u003d None"},{"line_number":301,"context_line":"        self.local_queue \u003d []"},{"line_number":302,"context_line":""}],"source_content_type":"text/x-python","patch_set":6,"id":"da20952f_ab9a2a12","line":299,"updated":"2015-09-07 07:44:32.000000000","message":"Because kafka_client is never set to None, kafka_producer will never got a kafka.SimpleProducer object again...","commit_id":"0f5601007874d802b9703c233e20f3a43a52f7f5"},{"author":{"_account_id":2813,"name":"Mehdi Abaakouk","email":"sileht@sileht.net","username":"sileht"},"change_message_id":"e2f45bf59ce9cf4746ba772d0ca59544fdd6e6e8","unresolved":false,"context_lines":[{"line_number":318,"context_line":"    def declare_topic_consumer(self, topics):"},{"line_number":319,"context_line":"        self._get_kafka_consumer(topics)"},{"line_number":320,"context_line":""},{"line_number":321,"context_line":"    def _get_kafka_consumer(self, topics):"},{"line_number":322,"context_line":"        self.kafka_subscriber \u003d kafka.KafkaConsumer("},{"line_number":323,"context_line":"            topics, group_id\u003dself.group,"},{"line_number":324,"context_line":"            metadata_broker_list\u003d[\"%s:%s\" % (self.host, str(self.port))],"}],"source_content_type":"text/x-python","patch_set":6,"id":"da20952f_ab178aa1","line":321,"updated":"2015-09-07 07:44:32.000000000","message":"This method is useless, you can move the code into declare_topic_consumer.","commit_id":"0f5601007874d802b9703c233e20f3a43a52f7f5"},{"author":{"_account_id":2813,"name":"Mehdi Abaakouk","email":"sileht@sileht.net","username":"sileht"},"change_message_id":"e2f45bf59ce9cf4746ba772d0ca59544fdd6e6e8","unresolved":false,"context_lines":[{"line_number":336,"context_line":"        try:"},{"line_number":337,"context_line":"            self._get_kafka_client()"},{"line_number":338,"context_line":"        except Exception as e:"},{"line_number":339,"context_line":"            raise Exception(\"Kafka Broker is not available: %s\" % e)"},{"line_number":340,"context_line":""},{"line_number":341,"context_line":""},{"line_number":342,"context_line":"class KafkaMessage(base.IncomingMessage):"}],"source_content_type":"text/x-python","patch_set":6,"id":"da20952f_abcc6a35","line":339,"updated":"2015-09-07 07:44:32.000000000","message":"This exception handling looks useless.","commit_id":"0f5601007874d802b9703c233e20f3a43a52f7f5"},{"author":{"_account_id":2813,"name":"Mehdi Abaakouk","email":"sileht@sileht.net","username":"sileht"},"change_message_id":"e2f45bf59ce9cf4746ba772d0ca59544fdd6e6e8","unresolved":false,"context_lines":[{"line_number":339,"context_line":"            raise Exception(\"Kafka Broker is not available: %s\" % e)"},{"line_number":340,"context_line":""},{"line_number":341,"context_line":""},{"line_number":342,"context_line":"class KafkaMessage(base.IncomingMessage):"},{"line_number":343,"context_line":""},{"line_number":344,"context_line":"    def __init__(self, listener, ctxt, message):"},{"line_number":345,"context_line":"        super(KafkaMessage, self).__init__(listener, ctxt, message)"}],"source_content_type":"text/x-python","patch_set":6,"id":"da20952f_cb8e7e78","line":342,"updated":"2015-09-07 07:44:32.000000000","message":"This class is useless.","commit_id":"0f5601007874d802b9703c233e20f3a43a52f7f5"},{"author":{"_account_id":12850,"name":"Komei Shimamura","email":"komei.t.f@gmail.com","username":"kshimamu"},"change_message_id":"01e7695a83a4edc98593b4ce4b25f3c70f042f7c","unresolved":false,"context_lines":[{"line_number":339,"context_line":"            raise Exception(\"Kafka Broker is not available: %s\" % e)"},{"line_number":340,"context_line":""},{"line_number":341,"context_line":""},{"line_number":342,"context_line":"class KafkaMessage(base.IncomingMessage):"},{"line_number":343,"context_line":""},{"line_number":344,"context_line":"    def __init__(self, listener, ctxt, message):"},{"line_number":345,"context_line":"        super(KafkaMessage, self).__init__(listener, ctxt, message)"}],"source_content_type":"text/x-python","patch_set":6,"id":"9a1a9d01_f2c9bcad","line":342,"in_reply_to":"da20952f_cb8e7e78","updated":"2015-09-24 20:10:25.000000000","message":"We have small question about this. The reason for using this class is following the other drivers such as impl_qpid.py, impl_rabbit.py, .... \n\nProbably, instead of removing this class, defining this class and using this class for abstracting the messages would be helpful for supporting easy extending the new functionality in the future. or do we still not need this class..?","commit_id":"0f5601007874d802b9703c233e20f3a43a52f7f5"},{"author":{"_account_id":2813,"name":"Mehdi Abaakouk","email":"sileht@sileht.net","username":"sileht"},"change_message_id":"e2f45bf59ce9cf4746ba772d0ca59544fdd6e6e8","unresolved":false,"context_lines":[{"line_number":373,"context_line":"                    ctxt \u003d unpack_context_from_message(message)"},{"line_number":374,"context_line":"                    self.incoming.append(KafkaMessage("},{"line_number":375,"context_line":"                        listener\u003dself, ctxt\u003dctxt, message\u003dmessage))"},{"line_number":376,"context_line":"            except Exception:  # should write some exception class for Timeout"},{"line_number":377,"context_line":"                return None"},{"line_number":378,"context_line":""},{"line_number":379,"context_line":"    def stop(self):"}],"source_content_type":"text/x-python","patch_set":6,"id":"da20952f_0b61d604","line":376,"updated":"2015-09-07 07:44:32.000000000","message":"You really can\u0027t hide all exceptions like this.","commit_id":"0f5601007874d802b9703c233e20f3a43a52f7f5"},{"author":{"_account_id":2813,"name":"Mehdi Abaakouk","email":"sileht@sileht.net","username":"sileht"},"change_message_id":"e2f45bf59ce9cf4746ba772d0ca59544fdd6e6e8","unresolved":false,"context_lines":[{"line_number":420,"context_line":"        raise NotImplementedError("},{"line_number":421,"context_line":"            \u0027The RPC implementation for Kafka is not implemented\u0027)"},{"line_number":422,"context_line":""},{"line_number":423,"context_line":"    def send_notification(self, target, ctxt, message, version\u003dNone):"},{"line_number":424,"context_line":"        try:"},{"line_number":425,"context_line":"            with self._get_connection(purpose\u003dPURPOSE_SEND) as conn:"},{"line_number":426,"context_line":"                conn.notify_send(target.topic, ctxt, message)"}],"source_content_type":"text/x-python","patch_set":6,"id":"da20952f_0b98f676","line":423,"updated":"2015-09-07 07:44:32.000000000","message":"The method signature is incorrect.","commit_id":"0f5601007874d802b9703c233e20f3a43a52f7f5"},{"author":{"_account_id":2813,"name":"Mehdi Abaakouk","email":"sileht@sileht.net","username":"sileht"},"change_message_id":"e2f45bf59ce9cf4746ba772d0ca59544fdd6e6e8","unresolved":false,"context_lines":[{"line_number":424,"context_line":"        try:"},{"line_number":425,"context_line":"            with self._get_connection(purpose\u003dPURPOSE_SEND) as conn:"},{"line_number":426,"context_line":"                conn.notify_send(target.topic, ctxt, message)"},{"line_number":427,"context_line":"        except Exception as e:"},{"line_number":428,"context_line":"            raise e"},{"line_number":429,"context_line":""},{"line_number":430,"context_line":"    def listen(self, target):"}],"source_content_type":"text/x-python","patch_set":6,"id":"da20952f_4b802ef7","line":427,"updated":"2015-09-07 07:44:32.000000000","message":"This try..catch is useless","commit_id":"0f5601007874d802b9703c233e20f3a43a52f7f5"},{"author":{"_account_id":1297,"name":"Joshua Harlow","email":"jxharlow@godaddy.com","username":"harlowja"},"change_message_id":"0720498ed6881dc3ba90311e2f637ae8980d7415","unresolved":false,"context_lines":[{"line_number":70,"context_line":"    context_dict \u003d {}"},{"line_number":71,"context_line":"    for key in list(msg.keys()):"},{"line_number":72,"context_line":"        key \u003d six.text_type(key)"},{"line_number":73,"context_line":"        if key.startswith(\u0027_context_\u0027):"},{"line_number":74,"context_line":"            value \u003d msg.pop(key)"},{"line_number":75,"context_line":"            context_dict[key[9:]] \u003d value"},{"line_number":76,"context_line":"    return context_dict"}],"source_content_type":"text/x-python","patch_set":7,"id":"9a1a9d01_412386a7","line":73,"updated":"2015-09-29 00:14:28.000000000","message":"Is there anyway this can just be nested instead of having special keys?\n\n real_message \u003d {\n    \"message\": {}, \n    \"context\": {}\n }\n\nThen u don\u0027t need special keys at all.","commit_id":"4f71a8d5103554b4bf03de2f80203dc84889a0af"},{"author":{"_account_id":12850,"name":"Komei Shimamura","email":"komei.t.f@gmail.com","username":"kshimamu"},"change_message_id":"b93f8ee3256d90b0af9c5c3538e217331029f3cb","unresolved":false,"context_lines":[{"line_number":70,"context_line":"    context_dict \u003d {}"},{"line_number":71,"context_line":"    for key in list(msg.keys()):"},{"line_number":72,"context_line":"        key \u003d six.text_type(key)"},{"line_number":73,"context_line":"        if key.startswith(\u0027_context_\u0027):"},{"line_number":74,"context_line":"            value \u003d msg.pop(key)"},{"line_number":75,"context_line":"            context_dict[key[9:]] \u003d value"},{"line_number":76,"context_line":"    return context_dict"}],"source_content_type":"text/x-python","patch_set":7,"id":"9a1a9d01_66333b15","line":73,"in_reply_to":"9a1a9d01_412386a7","updated":"2015-09-29 04:09:04.000000000","message":"yeah, It\u0027s definitely good idea! Will change this. Thanks!","commit_id":"4f71a8d5103554b4bf03de2f80203dc84889a0af"},{"author":{"_account_id":1297,"name":"Joshua Harlow","email":"jxharlow@godaddy.com","username":"harlowja"},"change_message_id":"fd92ed27511c2f9a0092ff51244a488ddaacab76","unresolved":false,"context_lines":[{"line_number":128,"context_line":""},{"line_number":129,"context_line":"    def _send_and_retry(self, message, topic):"},{"line_number":130,"context_line":"        current_retry \u003d 0"},{"line_number":131,"context_line":"        while message:"},{"line_number":132,"context_line":"            try:"},{"line_number":133,"context_line":"                if not isinstance(message, str):"},{"line_number":134,"context_line":"                    message \u003d jsonutils.dumps(message)"}],"source_content_type":"text/x-python","patch_set":7,"id":"9a1a9d01_97f277d1","line":131,"updated":"2015-09-26 16:30:51.000000000","message":"Should this be while message is not None:\n\nIs an empty string a valid message?","commit_id":"4f71a8d5103554b4bf03de2f80203dc84889a0af"},{"author":{"_account_id":12850,"name":"Komei Shimamura","email":"komei.t.f@gmail.com","username":"kshimamu"},"change_message_id":"be7912cb9985e9621ed93a3a074902ef15a56ff8","unresolved":false,"context_lines":[{"line_number":128,"context_line":""},{"line_number":129,"context_line":"    def _send_and_retry(self, message, topic):"},{"line_number":130,"context_line":"        current_retry \u003d 0"},{"line_number":131,"context_line":"        while message:"},{"line_number":132,"context_line":"            try:"},{"line_number":133,"context_line":"                if not isinstance(message, str):"},{"line_number":134,"context_line":"                    message \u003d jsonutils.dumps(message)"}],"source_content_type":"text/x-python","patch_set":7,"id":"9a1a9d01_92ba8522","line":131,"in_reply_to":"9a1a9d01_97f277d1","updated":"2015-09-26 18:11:16.000000000","message":"Thanks, will change.\ncurrently empty string isn\u0027t sent to kafka server, but if the message have the corresponding ctxt, this message(L131) isn\u0027t be empty (probably this is okay..?)","commit_id":"4f71a8d5103554b4bf03de2f80203dc84889a0af"},{"author":{"_account_id":1297,"name":"Joshua Harlow","email":"jxharlow@godaddy.com","username":"harlowja"},"change_message_id":"fd92ed27511c2f9a0092ff51244a488ddaacab76","unresolved":false,"context_lines":[{"line_number":130,"context_line":"        current_retry \u003d 0"},{"line_number":131,"context_line":"        while message:"},{"line_number":132,"context_line":"            try:"},{"line_number":133,"context_line":"                if not isinstance(message, str):"},{"line_number":134,"context_line":"                    message \u003d jsonutils.dumps(message)"},{"line_number":135,"context_line":"                self._send(message, topic)"},{"line_number":136,"context_line":"                message \u003d None"}],"source_content_type":"text/x-python","patch_set":7,"id":"9a1a9d01_b7ed3bf1","line":133,"updated":"2015-09-26 16:30:51.000000000","message":"Should this conversion happen outside the loop instead of inside it?\n\nSeems like something better suited for happening before the loop instead of inside it.","commit_id":"4f71a8d5103554b4bf03de2f80203dc84889a0af"},{"author":{"_account_id":12850,"name":"Komei Shimamura","email":"komei.t.f@gmail.com","username":"kshimamu"},"change_message_id":"be7912cb9985e9621ed93a3a074902ef15a56ff8","unresolved":false,"context_lines":[{"line_number":130,"context_line":"        current_retry \u003d 0"},{"line_number":131,"context_line":"        while message:"},{"line_number":132,"context_line":"            try:"},{"line_number":133,"context_line":"                if not isinstance(message, str):"},{"line_number":134,"context_line":"                    message \u003d jsonutils.dumps(message)"},{"line_number":135,"context_line":"                self._send(message, topic)"},{"line_number":136,"context_line":"                message \u003d None"}],"source_content_type":"text/x-python","patch_set":7,"id":"9a1a9d01_b2b54932","line":133,"in_reply_to":"9a1a9d01_b7ed3bf1","updated":"2015-09-26 18:11:16.000000000","message":"oh yes, thanks!!","commit_id":"4f71a8d5103554b4bf03de2f80203dc84889a0af"},{"author":{"_account_id":1297,"name":"Joshua Harlow","email":"jxharlow@godaddy.com","username":"harlowja"},"change_message_id":"fd92ed27511c2f9a0092ff51244a488ddaacab76","unresolved":false,"context_lines":[{"line_number":194,"context_line":"                \"%s:%s\" % (self.host, str(self.port)))"},{"line_number":195,"context_line":"            self.kafka_producer \u003d kafka.SimpleProducer(self.kafka_client)"},{"line_number":196,"context_line":"        except KafkaError:"},{"line_number":197,"context_line":"            self.kafka_client \u003d None"},{"line_number":198,"context_line":""},{"line_number":199,"context_line":"    def declare_topic_consumer(self, topic, group):"},{"line_number":200,"context_line":"        subscriber \u003d kafka.KafkaConsumer("}],"source_content_type":"text/x-python","patch_set":7,"id":"9a1a9d01_57bb7fe5","line":197,"updated":"2015-09-26 16:30:51.000000000","message":"Should this exception/error be logged at least?","commit_id":"4f71a8d5103554b4bf03de2f80203dc84889a0af"},{"author":{"_account_id":12850,"name":"Komei Shimamura","email":"komei.t.f@gmail.com","username":"kshimamu"},"change_message_id":"be7912cb9985e9621ed93a3a074902ef15a56ff8","unresolved":false,"context_lines":[{"line_number":194,"context_line":"                \"%s:%s\" % (self.host, str(self.port)))"},{"line_number":195,"context_line":"            self.kafka_producer \u003d kafka.SimpleProducer(self.kafka_client)"},{"line_number":196,"context_line":"        except KafkaError:"},{"line_number":197,"context_line":"            self.kafka_client \u003d None"},{"line_number":198,"context_line":""},{"line_number":199,"context_line":"    def declare_topic_consumer(self, topic, group):"},{"line_number":200,"context_line":"        subscriber \u003d kafka.KafkaConsumer("}],"source_content_type":"text/x-python","patch_set":7,"id":"9a1a9d01_f72173b5","line":197,"in_reply_to":"9a1a9d01_57bb7fe5","updated":"2015-09-26 18:11:16.000000000","message":"Thanks for the review, \nWill write LOG.exception(_LE(\u0027some message\u0027))","commit_id":"4f71a8d5103554b4bf03de2f80203dc84889a0af"},{"author":{"_account_id":1297,"name":"Joshua Harlow","email":"jxharlow@godaddy.com","username":"harlowja"},"change_message_id":"fd92ed27511c2f9a0092ff51244a488ddaacab76","unresolved":false,"context_lines":[{"line_number":230,"context_line":"        while not self._stopped.is_set():"},{"line_number":231,"context_line":"            if len(self.incoming_queue) \u003e 0:"},{"line_number":232,"context_line":"                return self.incoming_queue.pop(0)"},{"line_number":233,"context_line":"            try:"},{"line_number":234,"context_line":"                messages \u003d self.conn.retrieve(timeout\u003dtimeout)"},{"line_number":235,"context_line":"                if messages is None:"},{"line_number":236,"context_line":"                    continue"}],"source_content_type":"text/x-python","patch_set":7,"id":"9a1a9d01_b7c45b5f","line":233,"updated":"2015-09-26 16:30:51.000000000","message":"Doesnt this timeout need to decrease, if for example the timeout is 5 seconds, then \u0027retrieve\u0027 blocks for 4 seconds, then returns messages as None, then the loop will continue, but the timeout should be 1 seconds ( 5 - 4), not 5 again. Right?","commit_id":"4f71a8d5103554b4bf03de2f80203dc84889a0af"},{"author":{"_account_id":12850,"name":"Komei Shimamura","email":"komei.t.f@gmail.com","username":"kshimamu"},"change_message_id":"be7912cb9985e9621ed93a3a074902ef15a56ff8","unresolved":false,"context_lines":[{"line_number":230,"context_line":"        while not self._stopped.is_set():"},{"line_number":231,"context_line":"            if len(self.incoming_queue) \u003e 0:"},{"line_number":232,"context_line":"                return self.incoming_queue.pop(0)"},{"line_number":233,"context_line":"            try:"},{"line_number":234,"context_line":"                messages \u003d self.conn.retrieve(timeout\u003dtimeout)"},{"line_number":235,"context_line":"                if messages is None:"},{"line_number":236,"context_line":"                    continue"}],"source_content_type":"text/x-python","patch_set":7,"id":"9a1a9d01_f247411b","line":233,"in_reply_to":"9a1a9d01_b7c45b5f","updated":"2015-09-26 18:11:16.000000000","message":"I see, I need to change the logic here. Thanks!!","commit_id":"4f71a8d5103554b4bf03de2f80203dc84889a0af"},{"author":{"_account_id":1297,"name":"Joshua Harlow","email":"jxharlow@godaddy.com","username":"harlowja"},"change_message_id":"fd92ed27511c2f9a0092ff51244a488ddaacab76","unresolved":false,"context_lines":[{"line_number":241,"context_line":"                    self.incoming_queue.append(KafkaMessage("},{"line_number":242,"context_line":"                        listener\u003dself, ctxt\u003dctxt, message\u003dmessage))"},{"line_number":243,"context_line":"            except KafkaError as e:"},{"line_number":244,"context_line":"                # Timeout is also handled"},{"line_number":245,"context_line":"                LOG.warn(_LW(\u0027A KafkaListener Error: %s\u0027), e)"},{"line_number":246,"context_line":"                return None"},{"line_number":247,"context_line":""}],"source_content_type":"text/x-python","patch_set":7,"id":"9a1a9d01_d7cf8f7d","line":244,"updated":"2015-09-26 16:30:51.000000000","message":"I don\u0027t think an expected timeout should cause a warning message, is there a better way to handle this?","commit_id":"4f71a8d5103554b4bf03de2f80203dc84889a0af"},{"author":{"_account_id":12850,"name":"Komei Shimamura","email":"komei.t.f@gmail.com","username":"kshimamu"},"change_message_id":"be7912cb9985e9621ed93a3a074902ef15a56ff8","unresolved":false,"context_lines":[{"line_number":241,"context_line":"                    self.incoming_queue.append(KafkaMessage("},{"line_number":242,"context_line":"                        listener\u003dself, ctxt\u003dctxt, message\u003dmessage))"},{"line_number":243,"context_line":"            except KafkaError as e:"},{"line_number":244,"context_line":"                # Timeout is also handled"},{"line_number":245,"context_line":"                LOG.warn(_LW(\u0027A KafkaListener Error: %s\u0027), e)"},{"line_number":246,"context_line":"                return None"},{"line_number":247,"context_line":""}],"source_content_type":"text/x-python","patch_set":7,"id":"9a1a9d01_f2d98184","line":244,"in_reply_to":"9a1a9d01_d7cf8f7d","updated":"2015-09-26 18:11:16.000000000","message":"I\u0027ve just aware of that kafkaComsumer.fetch_messages() doesn\u0027t raise errors when it reaches timeout, but just return None. so this comment is my wrong, sorry. will remove this.","commit_id":"4f71a8d5103554b4bf03de2f80203dc84889a0af"},{"author":{"_account_id":2813,"name":"Mehdi Abaakouk","email":"sileht@sileht.net","username":"sileht"},"change_message_id":"49ba5313877f45503ea9cb24da96e1b387e86420","unresolved":false,"context_lines":[{"line_number":38,"context_line":"    cfg.IntOpt(\u0027kafka_default_port\u0027, default\u003d9092,"},{"line_number":39,"context_line":"               help\u003d\u0027Default Kafka broker Port\u0027),"},{"line_number":40,"context_line":""},{"line_number":41,"context_line":"    cfg.IntOpt(\u0027kafka_max_retry\u0027, default\u003d10,"},{"line_number":42,"context_line":"               help\u003d\u0027Max retry time\u0027),"},{"line_number":43,"context_line":""},{"line_number":44,"context_line":"    cfg.IntOpt(\u0027kafka_max_fetch_bytes\u0027, default\u003d1024 * 1024,"}],"source_content_type":"text/x-python","patch_set":15,"id":"9a8ffd7b_057d94b0","line":41,"updated":"2015-11-26 15:09:24.000000000","message":"This should not exists the number of retry can be passed trought the API, parameter \u0027retry\u0027 of \u0027send_notificiation\u0027 of the driver.","commit_id":"cd5f60b1be31ce4c5274dbde3051467b16e783f3"},{"author":{"_account_id":2813,"name":"Mehdi Abaakouk","email":"sileht@sileht.net","username":"sileht"},"change_message_id":"49ba5313877f45503ea9cb24da96e1b387e86420","unresolved":false,"context_lines":[{"line_number":162,"context_line":"                continue"},{"line_number":163,"context_line":""},{"line_number":164,"context_line":"            consumer \u003d self.consumers.pop(0)"},{"line_number":165,"context_line":"            self.consumers.append(consumer)"},{"line_number":166,"context_line":"            consumer._config[\u0027fetch_wait_max_ms\u0027] \u003d poll_timeout * 1000.0"},{"line_number":167,"context_line":""},{"line_number":168,"context_line":"            # consumers always return generator"}],"source_content_type":"text/x-python","patch_set":15,"id":"9a8ffd7b_3acd5366","line":165,"updated":"2015-11-26 15:09:24.000000000","message":"Do we do roundrobin against all consumers ?","commit_id":"cd5f60b1be31ce4c5274dbde3051467b16e783f3"},{"author":{"_account_id":2813,"name":"Mehdi Abaakouk","email":"sileht@sileht.net","username":"sileht"},"change_message_id":"49ba5313877f45503ea9cb24da96e1b387e86420","unresolved":false,"context_lines":[{"line_number":163,"context_line":""},{"line_number":164,"context_line":"            consumer \u003d self.consumers.pop(0)"},{"line_number":165,"context_line":"            self.consumers.append(consumer)"},{"line_number":166,"context_line":"            consumer._config[\u0027fetch_wait_max_ms\u0027] \u003d poll_timeout * 1000.0"},{"line_number":167,"context_line":""},{"line_number":168,"context_line":"            # consumers always return generator"},{"line_number":169,"context_line":"            messages \u003d consumer.fetch_messages()"}],"source_content_type":"text/x-python","patch_set":15,"id":"9a8ffd7b_3ac79397","line":166,"updated":"2015-11-26 15:09:24.000000000","message":"We must not use private API of the kafka client lib. I guess we can use consumer.configure() instead.","commit_id":"cd5f60b1be31ce4c5274dbde3051467b16e783f3"},{"author":{"_account_id":2813,"name":"Mehdi Abaakouk","email":"sileht@sileht.net","username":"sileht"},"change_message_id":"49ba5313877f45503ea9cb24da96e1b387e86420","unresolved":false,"context_lines":[{"line_number":168,"context_line":"            # consumers always return generator"},{"line_number":169,"context_line":"            messages \u003d consumer.fetch_messages()"},{"line_number":170,"context_line":"            # generator to list object"},{"line_number":171,"context_line":"            message_list \u003d [message for message in messages]"},{"line_number":172,"context_line":""},{"line_number":173,"context_line":"            if len(message_list) \u003d\u003d 0:"},{"line_number":174,"context_line":"                poll_timeout \u003d timer.check_return("}],"source_content_type":"text/x-python","patch_set":15,"id":"9a8ffd7b_1a472f4c","line":171,"updated":"2015-11-26 15:09:24.000000000","message":"You can just use :\n\n  messages \u003d list(consumer.fetch_messages())","commit_id":"cd5f60b1be31ce4c5274dbde3051467b16e783f3"},{"author":{"_account_id":2813,"name":"Mehdi Abaakouk","email":"sileht@sileht.net","username":"sileht"},"change_message_id":"49ba5313877f45503ea9cb24da96e1b387e86420","unresolved":false,"context_lines":[{"line_number":170,"context_line":"            # generator to list object"},{"line_number":171,"context_line":"            message_list \u003d [message for message in messages]"},{"line_number":172,"context_line":""},{"line_number":173,"context_line":"            if len(message_list) \u003d\u003d 0:"},{"line_number":174,"context_line":"                poll_timeout \u003d timer.check_return("},{"line_number":175,"context_line":"                    _raise_timeout, maximum\u003dself.consumer_timeout)"},{"line_number":176,"context_line":"                continue"}],"source_content_type":"text/x-python","patch_set":15,"id":"9a8ffd7b_fa602bd1","line":173,"updated":"2015-11-26 15:09:24.000000000","message":"No need to compute the length empty list:\n\n    if not messages:\n        ...","commit_id":"cd5f60b1be31ce4c5274dbde3051467b16e783f3"},{"author":{"_account_id":2813,"name":"Mehdi Abaakouk","email":"sileht@sileht.net","username":"sileht"},"change_message_id":"49ba5313877f45503ea9cb24da96e1b387e86420","unresolved":false,"context_lines":[{"line_number":227,"context_line":"            metadata_broker_list\u003d[\"%s:%s\" % (self.host, str(self.port))],"},{"line_number":228,"context_line":"            # auto_commit_enable\u003dself.auto_commit,"},{"line_number":229,"context_line":"            fetch_message_max_bytes\u003dself.fetch_messages_max_bytes,"},{"line_number":230,"context_line":"            fetch_wait_max_ms\u003dself.consumer_timeout * 1000.0)"},{"line_number":231,"context_line":"        self.consumers.append(subscriber)"},{"line_number":232,"context_line":""},{"line_number":233,"context_line":""}],"source_content_type":"text/x-python","patch_set":15,"id":"9a8ffd7b_3a70f30a","line":230,"updated":"2015-11-26 15:09:24.000000000","message":"fetch_wait_max_ms  seems always overrided L166.","commit_id":"cd5f60b1be31ce4c5274dbde3051467b16e783f3"},{"author":{"_account_id":2813,"name":"Mehdi Abaakouk","email":"sileht@sileht.net","username":"sileht"},"change_message_id":"49ba5313877f45503ea9cb24da96e1b387e86420","unresolved":false,"context_lines":[{"line_number":264,"context_line":"                    self.incoming_queue.append(OsloKafkaMessage("},{"line_number":265,"context_line":"                        listener\u003dself, ctxt\u003dmessage[\u0027context\u0027],"},{"line_number":266,"context_line":"                        message\u003dmessage[\u0027message\u0027]))"},{"line_number":267,"context_line":"            except (KafkaError, driver_common.Timeout):"},{"line_number":268,"context_line":"                return None"},{"line_number":269,"context_line":""},{"line_number":270,"context_line":"    def stop(self):"}],"source_content_type":"text/x-python","patch_set":15,"id":"9a8ffd7b_9aa23f4a","line":267,"updated":"2015-11-26 15:09:24.000000000","message":"We cannot hide all KafkaError without logging.\n\nUnexpected expection should not be catched. The executor does expected to get None in case of unexpected behavior.\n\nNone should be returned only if the listener is stopped or the timetout reached.","commit_id":"cd5f60b1be31ce4c5274dbde3051467b16e783f3"},{"author":{"_account_id":2813,"name":"Mehdi Abaakouk","email":"sileht@sileht.net","username":"sileht"},"change_message_id":"49ba5313877f45503ea9cb24da96e1b387e86420","unresolved":false,"context_lines":[{"line_number":311,"context_line":"        raise NotImplementedError("},{"line_number":312,"context_line":"            \u0027The RPC implementation for Kafka is not implemented\u0027)"},{"line_number":313,"context_line":""},{"line_number":314,"context_line":"    def send_notification(self, target, ctxt, message, version, retry\u003dNone):"},{"line_number":315,"context_line":"        with self._get_connection(purpose\u003dPURPOSE_SEND) as conn:"},{"line_number":316,"context_line":"            conn.notify_send(target.topic, ctxt, message)"},{"line_number":317,"context_line":""}],"source_content_type":"text/x-python","patch_set":15,"id":"9a8ffd7b_0554b41e","line":314,"updated":"2015-11-26 15:09:24.000000000","message":"retry should not be ignored.","commit_id":"cd5f60b1be31ce4c5274dbde3051467b16e783f3"},{"author":{"_account_id":2813,"name":"Mehdi Abaakouk","email":"sileht@sileht.net","username":"sileht"},"change_message_id":"49ba5313877f45503ea9cb24da96e1b387e86420","unresolved":false,"context_lines":[{"line_number":320,"context_line":"            \u0027The RPC implementation for Kafka is not implemented\u0027)"},{"line_number":321,"context_line":""},{"line_number":322,"context_line":"    def listen_for_notifications("},{"line_number":323,"context_line":"            self, targets_and_priorities, connections_pool\u003dNone):"},{"line_number":324,"context_line":"        conn \u003d self._get_connection(purpose\u003dPURPOSE_LISTEN)"},{"line_number":325,"context_line":""},{"line_number":326,"context_line":"        for target, priority in targets_and_priorities:"}],"source_content_type":"text/x-python","patch_set":15,"id":"9a8ffd7b_65b4a081","line":323,"updated":"2015-11-26 15:09:24.000000000","message":"The method signature is wrong \u0027connections_pool\u0027 should be just \u0027pool\u0027","commit_id":"cd5f60b1be31ce4c5274dbde3051467b16e783f3"},{"author":{"_account_id":2813,"name":"Mehdi Abaakouk","email":"sileht@sileht.net","username":"sileht"},"change_message_id":"49ba5313877f45503ea9cb24da96e1b387e86420","unresolved":false,"context_lines":[{"line_number":324,"context_line":"        conn \u003d self._get_connection(purpose\u003dPURPOSE_LISTEN)"},{"line_number":325,"context_line":""},{"line_number":326,"context_line":"        for target, priority in targets_and_priorities:"},{"line_number":327,"context_line":"            conn.declare_topic_consumer(target.topic, target.exchange)"},{"line_number":328,"context_line":""},{"line_number":329,"context_line":"        listener \u003d KafkaListener(self, conn)"},{"line_number":330,"context_line":"        return listener"}],"source_content_type":"text/x-python","patch_set":15,"id":"9a8ffd7b_653900f2","line":327,"updated":"2015-11-26 15:09:24.000000000","message":"Why don\u0027t you create only one KafkaConsumer per listener that consume all topics at once ?\n\nThe current design have a issue with the order of messages delivery. You create a Consumer per topic/exchange and read X bytes from the first consumer and then X bytes from the next consumer. That means reading a random number of messages of each consumers. That looks weird to me. Kafka allows to create one Consumer that read all topics.\n\nAlso you use the exchange as kafka \"group_id\" but that doesn\u0027t really match the kafka meaning of group.\n\nFor me, the kafka group_id is the \u0027pool\u0027 parameter of this method and the kafka topic is something like \u0027target.exchange + target.topic\u0027.\n\n\nAlso if you do this you can make the Listener API completely sync and can remove the Listener.incoming_queue. The listener.poll() will be mapped the only consumer get_messages() method.","commit_id":"cd5f60b1be31ce4c5274dbde3051467b16e783f3"},{"author":{"_account_id":12850,"name":"Komei Shimamura","email":"komei.t.f@gmail.com","username":"kshimamu"},"change_message_id":"298774474873e60ac0f5e85207e3f24d3e257b55","unresolved":false,"context_lines":[{"line_number":324,"context_line":"        conn \u003d self._get_connection(purpose\u003dPURPOSE_LISTEN)"},{"line_number":325,"context_line":""},{"line_number":326,"context_line":"        for target, priority in targets_and_priorities:"},{"line_number":327,"context_line":"            conn.declare_topic_consumer(target.topic, target.exchange)"},{"line_number":328,"context_line":""},{"line_number":329,"context_line":"        listener \u003d KafkaListener(self, conn)"},{"line_number":330,"context_line":"        return listener"}],"source_content_type":"text/x-python","patch_set":15,"id":"9a8ffd7b_f50eec8a","line":327,"in_reply_to":"9a8ffd7b_653900f2","updated":"2015-11-26 23:49:58.000000000","message":"Thanks for the helpful comments. Will reflect to the next patch.\n\nThe reason I wrote this was that I concerned the case different target.exchanges for the same topic are selected. but using (target.exchange + target.topic) for the topic can solve my concern and also using pool for the group_id looks better for matching the kafka group. (one thing i\u0027m not sure is whether we can use string for pool arg.)\n\nThanks!","commit_id":"cd5f60b1be31ce4c5274dbde3051467b16e783f3"},{"author":{"_account_id":2813,"name":"Mehdi Abaakouk","email":"sileht@sileht.net","username":"sileht"},"change_message_id":"f1d71388e1a45c1b9a80d3fe6864f1718fa70418","unresolved":false,"context_lines":[{"line_number":261,"context_line":"                for msg in messages:"},{"line_number":262,"context_line":"                    message \u003d msg.value"},{"line_number":263,"context_line":"                    if isinstance(message, str):"},{"line_number":264,"context_line":"                        message \u003d jsonutils.loads(message)"},{"line_number":265,"context_line":"                    self.incoming_queue.append(OsloKafkaMessage("},{"line_number":266,"context_line":"                        listener\u003dself, ctxt\u003dmessage[\u0027context\u0027],"},{"line_number":267,"context_line":"                        message\u003dmessage[\u0027message\u0027]))"}],"source_content_type":"text/x-python","patch_set":17,"id":"9a8ffd7b_8e718a78","line":264,"updated":"2015-11-27 07:43:58.000000000","message":"Why the json loads is conditional ?","commit_id":"326104f69dee732287f1b1bff054422ee7ff2b78"},{"author":{"_account_id":12850,"name":"Komei Shimamura","email":"komei.t.f@gmail.com","username":"kshimamu"},"change_message_id":"621e91e5c984cb029e5f18ae5682b1c63f1beadf","unresolved":false,"context_lines":[{"line_number":261,"context_line":"                for msg in messages:"},{"line_number":262,"context_line":"                    message \u003d msg.value"},{"line_number":263,"context_line":"                    if isinstance(message, str):"},{"line_number":264,"context_line":"                        message \u003d jsonutils.loads(message)"},{"line_number":265,"context_line":"                    self.incoming_queue.append(OsloKafkaMessage("},{"line_number":266,"context_line":"                        listener\u003dself, ctxt\u003dmessage[\u0027context\u0027],"},{"line_number":267,"context_line":"                        message\u003dmessage[\u0027message\u0027]))"}],"source_content_type":"text/x-python","patch_set":17,"id":"9a8ffd7b_ce0eb28b","line":264,"in_reply_to":"9a8ffd7b_8e718a78","updated":"2015-11-27 07:50:43.000000000","message":"Actually this is not needed if user uses send_notification() for kafka broker, will remove this soon","commit_id":"326104f69dee732287f1b1bff054422ee7ff2b78"},{"author":{"_account_id":5638,"name":"Davanum Srinivas","email":"davanum@gmail.com","username":"dims-v"},"change_message_id":"58fdbd57aabc8a81933d2c0e140f7b2c3e4c368c","unresolved":false,"context_lines":[{"line_number":13,"context_line":"# under the License."},{"line_number":14,"context_line":"import threading"},{"line_number":15,"context_line":""},{"line_number":16,"context_line":"from oslo_messaging._drivers import amqp"},{"line_number":17,"context_line":"from oslo_messaging._drivers import base"},{"line_number":18,"context_line":"from oslo_messaging._drivers import common as driver_common"},{"line_number":19,"context_line":"from oslo_messaging._i18n import _LE"}],"source_content_type":"text/x-python","patch_set":18,"id":"9a8ffd7b_3c4d019a","line":16,"updated":"2015-11-29 11:56:58.000000000","message":"Hmm, should we be moving ConnectionContent from amqp to a common spot?","commit_id":"e8a6e080f55538d91d10902df6e51d7d41d1de67"},{"author":{"_account_id":12850,"name":"Komei Shimamura","email":"komei.t.f@gmail.com","username":"kshimamu"},"change_message_id":"eca5de6c888e213849769e0d75d2c73c67e714ff","unresolved":false,"context_lines":[{"line_number":13,"context_line":"# under the License."},{"line_number":14,"context_line":"import threading"},{"line_number":15,"context_line":""},{"line_number":16,"context_line":"from oslo_messaging._drivers import amqp"},{"line_number":17,"context_line":"from oslo_messaging._drivers import base"},{"line_number":18,"context_line":"from oslo_messaging._drivers import common as driver_common"},{"line_number":19,"context_line":"from oslo_messaging._i18n import _LE"}],"source_content_type":"text/x-python","patch_set":18,"id":"9a8ffd7b_c2b2d04b","line":16,"in_reply_to":"9a8ffd7b_3c4d019a","updated":"2015-11-29 13:22:50.000000000","message":"Thanks you for the comment. For me, personally yes. amqp.ConnectionContent may be useful by other future drivers as well.","commit_id":"e8a6e080f55538d91d10902df6e51d7d41d1de67"}],"requirements.txt":[{"author":{"_account_id":2813,"name":"Mehdi Abaakouk","email":"sileht@sileht.net","username":"sileht"},"change_message_id":"e2f45bf59ce9cf4746ba772d0ca59544fdd6e6e8","unresolved":false,"context_lines":[{"line_number":35,"context_line":"kombu\u003e\u003d3.0.7"},{"line_number":36,"context_line":""},{"line_number":37,"context_line":"# kafka driver is provided for notification usage"},{"line_number":38,"context_line":"kafka-python\u003e\u003d0.9.3"},{"line_number":39,"context_line":""},{"line_number":40,"context_line":"# middleware"},{"line_number":41,"context_line":"oslo.middleware\u003e\u003d2.8.0 # Apache-2.0"}],"source_content_type":"text/plain","patch_set":6,"id":"da20952f_ebffc2ff","line":38,"updated":"2015-09-07 07:44:32.000000000","message":"you need to update global requirement first if you want to raise the minimun version.\n\nWhy does 0.9.2 is required for this change ?","commit_id":"0f5601007874d802b9703c233e20f3a43a52f7f5"},{"author":{"_account_id":13290,"name":"Oleksii Zamiatin","email":"ozamiatin@mirantis.com","username":"ozamiatin"},"change_message_id":"3b4afb6eae37acc1a1f71732ee0136df7fb6a2e8","unresolved":false,"context_lines":[{"line_number":35,"context_line":"kombu\u003e\u003d3.0.7"},{"line_number":36,"context_line":""},{"line_number":37,"context_line":"# kafka driver is provided for notification usage"},{"line_number":38,"context_line":"kafka-python\u003e\u003d0.9.2"},{"line_number":39,"context_line":""},{"line_number":40,"context_line":"# middleware"},{"line_number":41,"context_line":"oslo.middleware\u003e\u003d2.9.0 # Apache-2.0"}],"source_content_type":"text/plain","patch_set":12,"id":"ba8a016a_5a53b039","line":38,"updated":"2015-11-23 15:35:54.000000000","message":"Please put this into test-requirements.txt and make sure that unit-tests can pass (skip kafka tests) if package is not present. We do the same for amqp and zmq.","commit_id":"13ba37bb488f1c1a5f37b9725f11ba00c57f93bc"},{"author":{"_account_id":12850,"name":"Komei Shimamura","email":"komei.t.f@gmail.com","username":"kshimamu"},"change_message_id":"d519c4ee90db11f45719595c5c8048693c8bd1b1","unresolved":false,"context_lines":[{"line_number":35,"context_line":"kombu\u003e\u003d3.0.7"},{"line_number":36,"context_line":""},{"line_number":37,"context_line":"# kafka driver is provided for notification usage"},{"line_number":38,"context_line":"kafka-python\u003e\u003d0.9.2"},{"line_number":39,"context_line":""},{"line_number":40,"context_line":"# middleware"},{"line_number":41,"context_line":"oslo.middleware\u003e\u003d2.9.0 # Apache-2.0"}],"source_content_type":"text/plain","patch_set":12,"id":"ba8a016a_b51f4d5a","line":38,"in_reply_to":"ba8a016a_5a53b039","updated":"2015-11-23 15:37:28.000000000","message":"Thanks for the comment, will try soon","commit_id":"13ba37bb488f1c1a5f37b9725f11ba00c57f93bc"},{"author":{"_account_id":13290,"name":"Oleksii Zamiatin","email":"ozamiatin@mirantis.com","username":"ozamiatin"},"change_message_id":"79936b4f5bdf546ccde09917db25e5f3093d9ab0","unresolved":false,"context_lines":[{"line_number":37,"context_line":"kombu\u003e\u003d3.0.7"},{"line_number":38,"context_line":""},{"line_number":39,"context_line":"# kafka driver is provided for notification usage"},{"line_number":40,"context_line":"kafka-python\u003e\u003d0.9.2 # Apache-2.0"},{"line_number":41,"context_line":""},{"line_number":42,"context_line":"# middleware"},{"line_number":43,"context_line":"oslo.middleware\u003e\u003d3.0.0 # Apache-2.0"}],"source_content_type":"text/plain","patch_set":19,"id":"9a8ffd7b_124bf4b1","line":40,"updated":"2015-12-01 11:10:51.000000000","message":"Why we move this back from test-requirements?","commit_id":"3843217659b0e37574bcc6267a7553e2d88e5af7"},{"author":{"_account_id":12850,"name":"Komei Shimamura","email":"komei.t.f@gmail.com","username":"kshimamu"},"change_message_id":"8d5f1e68b9f95bcd24df08b8f0f69374d2c327cf","unresolved":false,"context_lines":[{"line_number":37,"context_line":"kombu\u003e\u003d3.0.7"},{"line_number":38,"context_line":""},{"line_number":39,"context_line":"# kafka driver is provided for notification usage"},{"line_number":40,"context_line":"kafka-python\u003e\u003d0.9.2 # Apache-2.0"},{"line_number":41,"context_line":""},{"line_number":42,"context_line":"# middleware"},{"line_number":43,"context_line":"oslo.middleware\u003e\u003d3.0.0 # Apache-2.0"}],"source_content_type":"text/plain","patch_set":19,"id":"9a8ffd7b_35793d30","line":40,"in_reply_to":"9a8ffd7b_124bf4b1","updated":"2015-12-01 12:00:23.000000000","message":"Oh,.. I will put this line to test-requirements soon.","commit_id":"3843217659b0e37574bcc6267a7553e2d88e5af7"}]}