# Path: kiel/exc.py
# class NoOffsetsError(KielError):
#     """
#     Error raised when requests fetching offsets fail fatally.
#     """
#     pass
# 
# Path: kiel/protocol/fetch.py
# class PartitionRequest(Part):
# class TopicRequest(Part):
# class FetchRequest(Request):
# class PartitionResponse(Part):
# class TopicResponse(Part):
# class FetchResponse(Response):
# 
# Path: kiel/protocol/errors.py
# class Errors(object):
# 
# Path: kiel/constants.py
# CONSUMER_REPLICA_ID = -1
# 
# ERROR_CODES = {
#     0: "no_error",
#     -1: "unknown",
#     1: "offset_out_of_range",
#     2: "invalid_message",
#     3: "unknown_topic_or_partition",
#     4: "invalid_message_size",
#     5: "leader_not_available",
#     6: "not_partition_leader",
#     7: "request_timed_out",
#     8: "broker_not_available",
#     9: "replica_not_available",
#     10: "message_size_too_large",
#     11: "stale_controller_epoch",
#     12: "offset_metadata_too_large",
#     14: "offsets_load_in_progress",
#     15: "coordinator_not_available",
#     16: "not_coordinator",
#     17: "invalid_topic",
#     18: "record_list_too_large",
#     19: "not_enough_replicas",
#     20: "not_enough_replicas_after_append",
#     21: "invalid_required_acks",
#     22: "illegal_generation",
#     23: "inconsistent_group_protocol",
#     24: "invalid_group_id",
#     25: "unknown_member_id",
#     26: "invalid_session_timeout",
#     27: "rebalance_in_progress",
#     28: "invalid_commit_offset_size",
#     29: "topic_authorization_failed",
#     30: "group_authorization_failed",
#     31: "cluster_authorization_failed",
# }
# 
# Path: kiel/clients/client.py
# class Client(object):
#     """
#     Base class for all client classes.
# 
#     Handles basic cluster management and request sending.
#     """
#     def __init__(self, brokers):
#         super(Client, self).__init__()
# 
#         self.cluster = Cluster(brokers)
# 
#         self.heal_cluster = False
#         self.closing = False
# 
#     @gen.coroutine
#     def connect(self):
#         """
#         Starts the underlying cluster, connecting and gathering metadata.
#         """
#         yield self.cluster.start()
# 
#     @gen.coroutine
#     def close(self):
#         """
#         Marks a client as closing and winds down connections.
# 
#         Calls the ``wind_down()`` coroutine that subclasses must implement.
#         """
#         self.closing = True
# 
#         yield self.wind_down()
# 
#         self.cluster.stop()
# 
#     @gen.coroutine
#     def wind_down(self):
#         """
#         Cleanup method left to subclasses to define.
# 
#         Called by ``close()``, should clean up any subclass-specific resources.
#         """
#         raise NotImplementedError
# 
#     @gen.coroutine
#     def send(self, request_by_broker):
#         """
#         Sends a dict of requests keyed on broker ID and handles responses.
# 
#         Returns a dictionary of the results of
#         ``handle_<response.api>_response`` method calls, keyed to the
#         corresponding broker ID.
# 
#         Raises ``UnhandledResponseError`` if the client subclass does not have
#         a ``handle_<response.api>_response`` method available to handle an
#         incoming response object.
# 
#         If an error occurs in a response, the ``heal_cluster`` flag is set
#         and the ``heal()`` method on the cluster is called after processing
#         each response.
# 
#         Responses are handled in the order they come in, but this method does
#         not yield a value until all responses are handled.
#         """
#         iterator = gen.WaitIterator(**{
#             str(broker_id): self.cluster[broker_id].send(request)
#             for broker_id, request in six.iteritems(request_by_broker)
#         })
# 
#         results = {}
#         while not iterator.done():
#             try:
#                 response = yield iterator.next()
#             except BrokerConnectionError as e:
#                 log.info("Connection to %s:%s lost", e.host, e.port)
#                 self.heal_cluster = True
#                 continue
#             except iostream.StreamClosedError:
#                 log.info("Connection to broker lost.")
#                 continue
#             except Exception:
#                 log.exception("Error sending request.")
#                 self.heal_cluster = True
#                 continue
# 
#             handler = getattr(self, "handle_%s_response" % response.api, None)
#             if handler is None:
#                 raise UnhandledResponseError(response.api)
# 
#             result = yield gen.maybe_future(handler(response))
#             results[int(iterator.current_index)] = result
# 
#         if self.heal_cluster:
#             yield self.cluster.heal()
#             self.heal_cluster = False
# 
#         raise gen.Return(results)


import collections
import logging
import json
import socket
import six
from tornado import gen
from kiel.exc import NoOffsetsError
from kiel.protocol import fetch, errors
from kiel.constants import CONSUMER_REPLICA_ID, ERROR_CODES
from .client import Client
        """
        Subclass coroutine function for setting values in ``self.offsets``.

        Kafka offers a simple "offset" api as well as a more involved set
        of offset fetch and commit apis.  Determining which ones to use and
        how is left to the subclasses.
        """
        raise NotImplementedError

    @gen.coroutine
    def consume(self, topic, start=None):
        """
        Fetches from a given topics returns a list of deserialized values.

        If the given topic is not known to have synced offsets, a call to
        `determine_offsets()` is made first.

        If a topic is unknown entirely the cluster's ``heal()`` method is
        called and the check retried.

        Since error codes and deserialization are taken care of by
        `handle_fetch_response` this method merely yields to wait on the
        deserialized results and returns a flattened list.
        """
        if self.closing:
            return

        if topic not in self.synced_offsets:
            try:
                yield self.determine_offsets(topic, start)

# Path: kiel/exc.py
# class NoOffsetsError(KielError):
#     """
#     Error raised when requests fetching offsets fail fatally.
#     """
#     pass
# 
# Path: kiel/protocol/fetch.py
# class PartitionRequest(Part):
# class TopicRequest(Part):
# class FetchRequest(Request):
# class PartitionResponse(Part):
# class TopicResponse(Part):
# class FetchResponse(Response):
# 
# Path: kiel/protocol/errors.py
# class Errors(object):
# 
# Path: kiel/constants.py
# CONSUMER_REPLICA_ID = -1
# 
# ERROR_CODES = {
#     0: "no_error",
#     -1: "unknown",
#     1: "offset_out_of_range",
#     2: "invalid_message",
#     3: "unknown_topic_or_partition",
#     4: "invalid_message_size",
#     5: "leader_not_available",
#     6: "not_partition_leader",
#     7: "request_timed_out",
#     8: "broker_not_available",
#     9: "replica_not_available",
#     10: "message_size_too_large",
#     11: "stale_controller_epoch",
#     12: "offset_metadata_too_large",
#     14: "offsets_load_in_progress",
#     15: "coordinator_not_available",
#     16: "not_coordinator",
#     17: "invalid_topic",
#     18: "record_list_too_large",
#     19: "not_enough_replicas",
#     20: "not_enough_replicas_after_append",
#     21: "invalid_required_acks",
#     22: "illegal_generation",
#     23: "inconsistent_group_protocol",
#     24: "invalid_group_id",
#     25: "unknown_member_id",
#     26: "invalid_session_timeout",
#     27: "rebalance_in_progress",
#     28: "invalid_commit_offset_size",
#     29: "topic_authorization_failed",
#     30: "group_authorization_failed",
#     31: "cluster_authorization_failed",
# }
# 
# Path: kiel/clients/client.py
# class Client(object):
#     """
#     Base class for all client classes.
# 
#     Handles basic cluster management and request sending.
#     """
#     def __init__(self, brokers):
#         super(Client, self).__init__()
# 
#         self.cluster = Cluster(brokers)
# 
#         self.heal_cluster = False
#         self.closing = False
# 
#     @gen.coroutine
#     def connect(self):
#         """
#         Starts the underlying cluster, connecting and gathering metadata.
#         """
#         yield self.cluster.start()
# 
#     @gen.coroutine
#     def close(self):
#         """
#         Marks a client as closing and winds down connections.
# 
#         Calls the ``wind_down()`` coroutine that subclasses must implement.
#         """
#         self.closing = True
# 
#         yield self.wind_down()
# 
#         self.cluster.stop()
# 
#     @gen.coroutine
#     def wind_down(self):
#         """
#         Cleanup method left to subclasses to define.
# 
#         Called by ``close()``, should clean up any subclass-specific resources.
#         """
#         raise NotImplementedError
# 
#     @gen.coroutine
#     def send(self, request_by_broker):
#         """
#         Sends a dict of requests keyed on broker ID and handles responses.
# 
#         Returns a dictionary of the results of
#         ``handle_<response.api>_response`` method calls, keyed to the
#         corresponding broker ID.
# 
#         Raises ``UnhandledResponseError`` if the client subclass does not have
#         a ``handle_<response.api>_response`` method available to handle an
#         incoming response object.
# 
#         If an error occurs in a response, the ``heal_cluster`` flag is set
#         and the ``heal()`` method on the cluster is called after processing
#         each response.
# 
#         Responses are handled in the order they come in, but this method does
#         not yield a value until all responses are handled.
#         """
#         iterator = gen.WaitIterator(**{
#             str(broker_id): self.cluster[broker_id].send(request)
#             for broker_id, request in six.iteritems(request_by_broker)
#         })
# 
#         results = {}
#         while not iterator.done():
#             try:
#                 response = yield iterator.next()
#             except BrokerConnectionError as e:
#                 log.info("Connection to %s:%s lost", e.host, e.port)
#                 self.heal_cluster = True
#                 continue
#             except iostream.StreamClosedError:
#                 log.info("Connection to broker lost.")
#                 continue
#             except Exception:
#                 log.exception("Error sending request.")
#                 self.heal_cluster = True
#                 continue
# 
#             handler = getattr(self, "handle_%s_response" % response.api, None)
#             if handler is None:
#                 raise UnhandledResponseError(response.api)
# 
#             result = yield gen.maybe_future(handler(response))
#             results[int(iterator.current_index)] = result
# 
#         if self.heal_cluster:
#             yield self.cluster.heal()
#             self.heal_cluster = False
# 
#         raise gen.Return(results)


# Path: kiel/clients/consumer.py
import collections
import logging
import json
import socket
import six
from tornado import gen
from kiel.exc import NoOffsetsError
from kiel.protocol import fetch, errors
from kiel.constants import CONSUMER_REPLICA_ID, ERROR_CODES
from .client import Client

        """
        Subclass coroutine function for setting values in ``self.offsets``.

        Kafka offers a simple "offset" api as well as a more involved set
        of offset fetch and commit apis.  Determining which ones to use and
        how is left to the subclasses.
        """
        raise NotImplementedError

    @gen.coroutine
    def consume(self, topic, start=None):
        """
        Fetches from a given topics returns a list of deserialized values.

        If the given topic is not known to have synced offsets, a call to
        `determine_offsets()` is made first.

        If a topic is unknown entirely the cluster's ``heal()`` method is
        called and the check retried.

        Since error codes and deserialization are taken care of by
        `handle_fetch_response` this method merely yields to wait on the
        deserialized results and returns a flattened list.
        """
        if self.closing:
            return

        if topic not in self.synced_offsets:
            try:
                yield self.determine_offsets(topic, start)

            except NoOffsetsError: