)]}'
{"monasca_common/kafka/consumer.py":[{"author":{"_account_id":14517,"name":"Ryan Brandt","email":"ryan.brandt@hpe.com","username":"ryan-brandt"},"change_message_id":"51f90babf65d435f6ea721ec856db1e9c8e80d47","unresolved":false,"context_lines":[{"line_number":1,"context_line":"# Copyright (c) 2014 Hewlett-Packard Development Company, L.P."},{"line_number":2,"context_line":"#"},{"line_number":3,"context_line":"# Licensed under the Apache License, Version 2.0 (the \"License\");"},{"line_number":4,"context_line":"# you may not use this file except in compliance with the License."}],"source_content_type":"text/x-python","patch_set":1,"id":"3a29b11f_1ae036cc","line":1,"updated":"2015-10-22 22:40:04.000000000","message":"Update copyright","commit_id":"316e313935befc6367fb7a76fd51036677c8de5e"},{"author":{"_account_id":2419,"name":"Roland Hochmuth","email":"roland.hochmuth@hpe.com","username":"roland-hochmuth-s"},"change_message_id":"ab9f8b2e9448b2175cf68a2f890c0bbee41cde5a","unresolved":false,"context_lines":[{"line_number":14,"context_line":"# limitations under the License."},{"line_number":15,"context_line":""},{"line_number":16,"context_line":"import datetime"},{"line_number":17,"context_line":"import logging"},{"line_number":18,"context_line":"import threading"},{"line_number":19,"context_line":""},{"line_number":20,"context_line":"import kafka.client"}],"source_content_type":"text/x-python","patch_set":5,"id":"3a29b11f_9be173d3","line":17,"updated":"2015-10-26 00:25:44.000000000","message":"Should we use\n\nfrom oslo_log import log\n\nThis is what we are using in the monasca api.","commit_id":"8bd1f2ac2c10452f7b8a9df6f9c4465698eb9389"},{"author":{"_account_id":2419,"name":"Roland Hochmuth","email":"roland.hochmuth@hpe.com","username":"roland-hochmuth-s"},"change_message_id":"ab9f8b2e9448b2175cf68a2f890c0bbee41cde5a","unresolved":false,"context_lines":[{"line_number":15,"context_line":""},{"line_number":16,"context_line":"import datetime"},{"line_number":17,"context_line":"import logging"},{"line_number":18,"context_line":"import threading"},{"line_number":19,"context_line":""},{"line_number":20,"context_line":"import kafka.client"},{"line_number":21,"context_line":"import kafka.common"}],"source_content_type":"text/x-python","patch_set":5,"id":"3a29b11f_5bebdbb0","line":18,"updated":"2015-10-26 00:25:44.000000000","message":"Is there a special oslo library that we should be using for threading too?","commit_id":"8bd1f2ac2c10452f7b8a9df6f9c4465698eb9389"},{"author":{"_account_id":14273,"name":"Joe Keen","email":"joe.keen@hpe.com","username":"jkeen"},"change_message_id":"ce77b9465d5dd4932fc4bca7d4f2467b6b91f7a1","unresolved":false,"context_lines":[{"line_number":15,"context_line":""},{"line_number":16,"context_line":"import datetime"},{"line_number":17,"context_line":"import logging"},{"line_number":18,"context_line":"import threading"},{"line_number":19,"context_line":""},{"line_number":20,"context_line":"import kafka.client"},{"line_number":21,"context_line":"import kafka.common"}],"source_content_type":"text/x-python","patch_set":5,"id":"3a29b11f_3e404d07","line":18,"in_reply_to":"3a29b11f_5bebdbb0","updated":"2015-10-26 02:07:00.000000000","message":"There is a concurrency library from Oslo but we\u0027re handing a threading object to the Kazoo library to allow it to wake us up in the future.  I\u0027m not certain that the oslo concurrency library is compatible so I\u0027d prefer to stick with the one I\u0027ve tested for now.","commit_id":"8bd1f2ac2c10452f7b8a9df6f9c4465698eb9389"},{"author":{"_account_id":11809,"name":"Craig Bryant","email":"craig.bryant@hpe.com","username":"craig-bryant"},"change_message_id":"d6e22561a966c13ab8fd6489b8d485d4d2c93de8","unresolved":false,"context_lines":[{"line_number":28,"context_line":""},{"line_number":29,"context_line":""},{"line_number":30,"context_line":"class KafkaConsumer(object):"},{"line_number":31,"context_line":"    \"\"\"Read alarm from the alarm topic"},{"line_number":32,"context_line":"       No commit is being done until processing is finished"},{"line_number":33,"context_line":"    \"\"\""},{"line_number":34,"context_line":""}],"source_content_type":"text/x-python","patch_set":5,"id":"1a26ad4f_74b6f43e","line":31,"updated":"2015-10-27 19:15:19.000000000","message":"Is this reading an alarm or is this reading a generic object?","commit_id":"8bd1f2ac2c10452f7b8a9df6f9c4465698eb9389"},{"author":{"_account_id":14273,"name":"Joe Keen","email":"joe.keen@hpe.com","username":"jkeen"},"change_message_id":"958c9f762190b0b1e144db1f256ef45ee0794a1a","unresolved":false,"context_lines":[{"line_number":28,"context_line":""},{"line_number":29,"context_line":""},{"line_number":30,"context_line":"class KafkaConsumer(object):"},{"line_number":31,"context_line":"    \"\"\"Read alarm from the alarm topic"},{"line_number":32,"context_line":"       No commit is being done until processing is finished"},{"line_number":33,"context_line":"    \"\"\""},{"line_number":34,"context_line":""}],"source_content_type":"text/x-python","patch_set":5,"id":"1a26ad4f_9495d041","line":31,"in_reply_to":"1a26ad4f_74b6f43e","updated":"2015-10-27 19:19:09.000000000","message":"No, it\u0027s just a generic piece of data.  I thought I got all the component specific comments out of it.","commit_id":"8bd1f2ac2c10452f7b8a9df6f9c4465698eb9389"},{"author":{"_account_id":2419,"name":"Roland Hochmuth","email":"roland.hochmuth@hpe.com","username":"roland-hochmuth-s"},"change_message_id":"ab9f8b2e9448b2175cf68a2f890c0bbee41cde5a","unresolved":false,"context_lines":[{"line_number":37,"context_line":"                 group, topic,"},{"line_number":38,"context_line":"                 repartition_callback\u003dNone,"},{"line_number":39,"context_line":"                 timeout_callback\u003dNone,"},{"line_number":40,"context_line":"                 timeout_seconds\u003d30):"},{"line_number":41,"context_line":"        \"\"\"Init"},{"line_number":42,"context_line":"             kafka_url, group, topic - kafka connection details"},{"line_number":43,"context_line":"        \"\"\""}],"source_content_type":"text/x-python","patch_set":5,"id":"3a29b11f_bbc30f38","line":40,"updated":"2015-10-26 00:25:44.000000000","message":"Should timeout_seconds be made configurable, or minimally use a variable declaration such as \n\nDEFAULT_TIMEOUT_SECS \u003d 30","commit_id":"8bd1f2ac2c10452f7b8a9df6f9c4465698eb9389"},{"author":{"_account_id":14273,"name":"Joe Keen","email":"joe.keen@hpe.com","username":"jkeen"},"change_message_id":"ce77b9465d5dd4932fc4bca7d4f2467b6b91f7a1","unresolved":false,"context_lines":[{"line_number":37,"context_line":"                 group, topic,"},{"line_number":38,"context_line":"                 repartition_callback\u003dNone,"},{"line_number":39,"context_line":"                 timeout_callback\u003dNone,"},{"line_number":40,"context_line":"                 timeout_seconds\u003d30):"},{"line_number":41,"context_line":"        \"\"\"Init"},{"line_number":42,"context_line":"             kafka_url, group, topic - kafka connection details"},{"line_number":43,"context_line":"        \"\"\""}],"source_content_type":"text/x-python","patch_set":5,"id":"3a29b11f_5e5c29ed","line":40,"in_reply_to":"3a29b11f_bbc30f38","updated":"2015-10-26 02:07:00.000000000","message":"The timeout is configurable but I was planning to make it configurable at the level of the module that constructs this object.  So the API would not configure either a callback function or a timeout value while the Persister would configure both.\n\nWould you expect to see configuration information for this passed down from the module constructing these objects or would you expect these objects to look for their own config in whatever config file is currently loaded?","commit_id":"8bd1f2ac2c10452f7b8a9df6f9c4465698eb9389"},{"author":{"_account_id":11809,"name":"Craig Bryant","email":"craig.bryant@hpe.com","username":"craig-bryant"},"change_message_id":"d6e22561a966c13ab8fd6489b8d485d4d2c93de8","unresolved":false,"context_lines":[{"line_number":101,"context_line":""},{"line_number":102,"context_line":"                    yield message"},{"line_number":103,"context_line":""},{"line_number":104,"context_line":"                if self._timeout_callback:"},{"line_number":105,"context_line":"                    time_now \u003d datetime.datetime.now()"},{"line_number":106,"context_line":"                    time_delta \u003d time_now - timeout_start_time"},{"line_number":107,"context_line":"                    if time_delta.total_seconds() \u003e self.timeout_seconds:"}],"source_content_type":"text/x-python","patch_set":5,"id":"1a26ad4f_f4e1a4ff","line":104,"updated":"2015-10-27 19:15:19.000000000","message":"_timeout_callback is not a good name for this. It gets called after every batch of messages even if 1000 messages were read in .1 seconds","commit_id":"8bd1f2ac2c10452f7b8a9df6f9c4465698eb9389"},{"author":{"_account_id":11809,"name":"Craig Bryant","email":"craig.bryant@hpe.com","username":"craig-bryant"},"change_message_id":"9fbc620e5d28f9540d467c1e1a0daf51ea043ab0","unresolved":false,"context_lines":[{"line_number":101,"context_line":""},{"line_number":102,"context_line":"                    yield message"},{"line_number":103,"context_line":""},{"line_number":104,"context_line":"                if self._timeout_callback:"},{"line_number":105,"context_line":"                    time_now \u003d datetime.datetime.now()"},{"line_number":106,"context_line":"                    time_delta \u003d time_now - timeout_start_time"},{"line_number":107,"context_line":"                    if time_delta.total_seconds() \u003e self.timeout_seconds:"}],"source_content_type":"text/x-python","patch_set":5,"id":"1a26ad4f_94cab04e","line":104,"in_reply_to":"1a26ad4f_f4e1a4ff","updated":"2015-10-27 19:21:52.000000000","message":"No, I\u0027m wrong, it only gets called every self.timeout_seconds. But, it still isn\u0027t really a timeout, it is more just a periodic callback.","commit_id":"8bd1f2ac2c10452f7b8a9df6f9c4465698eb9389"},{"author":{"_account_id":2419,"name":"Roland Hochmuth","email":"roland.hochmuth@hpe.com","username":"roland-hochmuth-s"},"change_message_id":"ab9f8b2e9448b2175cf68a2f890c0bbee41cde5a","unresolved":false,"context_lines":[{"line_number":146,"context_line":"                elif self._set_partitioner.release:"},{"line_number":147,"context_line":"                    log.info(\"Releasing locks on partition set {} \""},{"line_number":148,"context_line":"                             \"for topic {}\".format(self._partitions,"},{"line_number":149,"context_line":"                                                   self._kafka_topic))"},{"line_number":150,"context_line":"                    self._set_partitioner.release_set()"},{"line_number":151,"context_line":""},{"line_number":152,"context_line":"                    self._partitions \u003d []"}],"source_content_type":"text/x-python","patch_set":5,"id":"3a29b11f_1ba00390","line":149,"updated":"2015-10-26 00:25:44.000000000","message":"Do you really want this as a log.info or should this be a log.debug?","commit_id":"8bd1f2ac2c10452f7b8a9df6f9c4465698eb9389"},{"author":{"_account_id":14273,"name":"Joe Keen","email":"joe.keen@hpe.com","username":"jkeen"},"change_message_id":"ce77b9465d5dd4932fc4bca7d4f2467b6b91f7a1","unresolved":false,"context_lines":[{"line_number":146,"context_line":"                elif self._set_partitioner.release:"},{"line_number":147,"context_line":"                    log.info(\"Releasing locks on partition set {} \""},{"line_number":148,"context_line":"                             \"for topic {}\".format(self._partitions,"},{"line_number":149,"context_line":"                                                   self._kafka_topic))"},{"line_number":150,"context_line":"                    self._set_partitioner.release_set()"},{"line_number":151,"context_line":""},{"line_number":152,"context_line":"                    self._partitions \u003d []"}],"source_content_type":"text/x-python","patch_set":5,"id":"3a29b11f_9ea5e1b6","line":149,"in_reply_to":"3a29b11f_1ba00390","updated":"2015-10-26 02:07:00.000000000","message":"I don\u0027t know.  These messages have been set to the info level in both the Persister and Notification system so far.  Do we have any guidelines for what we consider debug as opposed to info messages?","commit_id":"8bd1f2ac2c10452f7b8a9df6f9c4465698eb9389"},{"author":{"_account_id":2419,"name":"Roland Hochmuth","email":"roland.hochmuth@hpe.com","username":"roland-hochmuth-s"},"change_message_id":"ab9f8b2e9448b2175cf68a2f890c0bbee41cde5a","unresolved":false,"context_lines":[{"line_number":183,"context_line":"                        return"},{"line_number":184,"context_line":""},{"line_number":185,"context_line":"                elif self._set_partitioner.allocating:"},{"line_number":186,"context_line":"                    log.info(\"Waiting to acquire locks on partition set\")"},{"line_number":187,"context_line":"                    self._set_partitioner.wait_for_acquire()"},{"line_number":188,"context_line":""},{"line_number":189,"context_line":"        except Exception:"}],"source_content_type":"text/x-python","patch_set":5,"id":"3a29b11f_bbac6f8d","line":186,"updated":"2015-10-26 00:25:44.000000000","message":"Do you really want this as a log.info or should this be a log.debug?","commit_id":"8bd1f2ac2c10452f7b8a9df6f9c4465698eb9389"},{"author":{"_account_id":11809,"name":"Craig Bryant","email":"craig.bryant@hpe.com","username":"craig-bryant"},"change_message_id":"88ee50f4e3d92ee8b9d2e558e2cce0699f706bdb","unresolved":false,"context_lines":[{"line_number":101,"context_line":""},{"line_number":102,"context_line":"                if self._periodic_callback:"},{"line_number":103,"context_line":"                    time_now \u003d datetime.datetime.now()"},{"line_number":104,"context_line":"                    time_delta \u003d time_now - self._last_commit"},{"line_number":105,"context_line":"                    if time_delta.total_seconds() \u003e self._periodic_callback_rate:"},{"line_number":106,"context_line":"                        self._periodic_callback()"},{"line_number":107,"context_line":""}],"source_content_type":"text/x-python","patch_set":8,"id":"1a26ad4f_ff227400","line":104,"updated":"2015-10-30 20:44:56.000000000","message":"You seem to be assuming that commit will be called from periodic_callback(). That is the only way self._last_commit gets reset. That would be a bad assumption for a generic class.","commit_id":"c6574138fa185c0d0bbc6cacf4377da545d2439c"}],"monasca_common/kafka/producer.py":[{"author":{"_account_id":2419,"name":"Roland Hochmuth","email":"roland.hochmuth@hpe.com","username":"roland-hochmuth-s"},"change_message_id":"ab9f8b2e9448b2175cf68a2f890c0bbee41cde5a","unresolved":false,"context_lines":[{"line_number":34,"context_line":"            self._kafka,"},{"line_number":35,"context_line":"            async\u003dFalse,"},{"line_number":36,"context_line":"            req_acks\u003dkafka.producer.KeyedProducer.ACK_AFTER_LOCAL_WRITE,"},{"line_number":37,"context_line":"            ack_timeout\u003d2000)"},{"line_number":38,"context_line":""},{"line_number":39,"context_line":"    def publish(self, topic, messages):"},{"line_number":40,"context_line":"        \"\"\"Takes messages and puts them on the supplied kafka topic"}],"source_content_type":"text/x-python","patch_set":5,"id":"3a29b11f_db728b49","line":37,"updated":"2015-10-26 00:25:44.000000000","message":"Should timeout be configurable?","commit_id":"8bd1f2ac2c10452f7b8a9df6f9c4465698eb9389"},{"author":{"_account_id":14273,"name":"Joe Keen","email":"joe.keen@hpe.com","username":"jkeen"},"change_message_id":"ce77b9465d5dd4932fc4bca7d4f2467b6b91f7a1","unresolved":false,"context_lines":[{"line_number":34,"context_line":"            self._kafka,"},{"line_number":35,"context_line":"            async\u003dFalse,"},{"line_number":36,"context_line":"            req_acks\u003dkafka.producer.KeyedProducer.ACK_AFTER_LOCAL_WRITE,"},{"line_number":37,"context_line":"            ack_timeout\u003d2000)"},{"line_number":38,"context_line":""},{"line_number":39,"context_line":"    def publish(self, topic, messages):"},{"line_number":40,"context_line":"        \"\"\"Takes messages and puts them on the supplied kafka topic"}],"source_content_type":"text/x-python","patch_set":5,"id":"3a29b11f_be753d33","line":37,"in_reply_to":"3a29b11f_db728b49","updated":"2015-10-26 02:07:00.000000000","message":"I don\u0027t know.  Depending on what we\u0027re doing we may not even want acks but this seems to be the settings we\u0027re using most places at the moment.","commit_id":"8bd1f2ac2c10452f7b8a9df6f9c4465698eb9389"},{"author":{"_account_id":2419,"name":"Roland Hochmuth","email":"roland.hochmuth@hpe.com","username":"roland-hochmuth-s"},"change_message_id":"ab9f8b2e9448b2175cf68a2f890c0bbee41cde5a","unresolved":false,"context_lines":[{"line_number":54,"context_line":"        try:"},{"line_number":55,"context_line":"            key \u003d time.time() * 1000"},{"line_number":56,"context_line":"            partition \u003d self._producer._next_partition(topic, key)"},{"line_number":57,"context_line":"            self._producer.send_messages(topic, partition, *messages)"},{"line_number":58,"context_line":"        except Exception:"},{"line_number":59,"context_line":"            log.exception(\u0027Error publishing to {} topic.\u0027.format(topic))"},{"line_number":60,"context_line":"            raise"}],"source_content_type":"text/x-python","patch_set":5,"id":"3a29b11f_bb754f33","line":57,"updated":"2015-10-26 00:25:44.000000000","message":"Just some questions.\n\nIn the Java code we key messages, not purely based on time. The ideas is to get metrics with the same tenant, name and dimensions routed to the same consumer. This is required for use cases such as the Anomaly Engine.\n\nSee MetricService.java, but the code is as follows:\n\n  public void create(List\u003cMetric\u003e metrics, String tenantId, @Nullable String crossTenantId) {\n    String metricTenantId \u003d Strings.isNullOrEmpty(crossTenantId) ? tenantId : crossTenantId;\n    Builder\u003cString, Object\u003e metaBuilder \u003d\n        new ImmutableMap.Builder\u003cString, Object\u003e().put(\"tenantId\", metricTenantId).put(\"region\",\n            config.region);\n    ImmutableMap\u003cString, Object\u003e meta \u003d metaBuilder.build();\n\n    List\u003cKeyedMessage\u003cString, String\u003e\u003e keyedMessages \u003d new ArrayList\u003c\u003e(metrics.size());\n    for (Metric metric : metrics) {\n      MetricEnvelope envelope \u003d new MetricEnvelope(metric, meta);\n      keyedMessages.add(new KeyedMessage\u003c\u003e(config.metricsTopic, buildKey(metricTenantId, metric),\n                        MetricEnvelopes.toJson(envelope)));\n      metricMeter.mark();\n    }\n\n    producer.send(keyedMessages);\n    batchMeter.mark();\n  }\n\n  private String buildKey(String metricTenantId, Metric metric) {\n    final StringBuilder key \u003d new StringBuilder(metricTenantId);\n    key.append(metric.name);\n\n    // Dimensions are optional.\n    if (metric.dimensions !\u003d null \u0026\u0026 !metric.dimensions.isEmpty()) {\n\n      // Key must be the same for the same metric so sort the dimensions so they will be\n      // in a known order\n      for (final Map.Entry\u003cString, String\u003e dim : buildSortedDimSet(metric.dimensions)) {\n        key.append(dim.getKey());\n        key.append(dim.getValue());\n      }\n    }\n    String keyValue \u003d key.toString();\n    return keyValue;\n  }","commit_id":"8bd1f2ac2c10452f7b8a9df6f9c4465698eb9389"},{"author":{"_account_id":14273,"name":"Joe Keen","email":"joe.keen@hpe.com","username":"jkeen"},"change_message_id":"ce77b9465d5dd4932fc4bca7d4f2467b6b91f7a1","unresolved":false,"context_lines":[{"line_number":54,"context_line":"        try:"},{"line_number":55,"context_line":"            key \u003d time.time() * 1000"},{"line_number":56,"context_line":"            partition \u003d self._producer._next_partition(topic, key)"},{"line_number":57,"context_line":"            self._producer.send_messages(topic, partition, *messages)"},{"line_number":58,"context_line":"        except Exception:"},{"line_number":59,"context_line":"            log.exception(\u0027Error publishing to {} topic.\u0027.format(topic))"},{"line_number":60,"context_line":"            raise"}],"source_content_type":"text/x-python","patch_set":5,"id":"3a29b11f_9ebac1bf","line":57,"in_reply_to":"3a29b11f_bb754f33","updated":"2015-10-26 02:07:00.000000000","message":"If I\u0027m reading that correctly it\u0027s building a key for each individual message in a batch?  If it is then I\u0027m not certain that the Python version we\u0027re using right now has that capability.  It accepts a key for any message, or list of messages, that you write to Kafka but to get the performance we want we need to write as many as possible at once.\n\nThere have been quite a few improvements in the python kafka library since we froze on 0.9.2 so it\u0027s possible we can do what the Java library is doing with more recent versions.\n\nWould the current method of distribution be sufficient for the Threshold engine?  It was sufficient for the Notification engine and should be sufficient for the Persister.","commit_id":"8bd1f2ac2c10452f7b8a9df6f9c4465698eb9389"}]}