# 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
            msg for messageset in results.values() for msg in messageset
            if messageset
        ])

    def handle_fetch_response(self, response):
        """
        Handler for responses from the message "fetch" api.

        Messages returned with the "no error" code are deserialized and
        collected, the full resulting list is returned.

        A retriable error code will cause the cluster "heal" flag to be set.

        An error indicating that the offset used for the partition was out
        of range will cause the offending topic's offsets to be redetermined
        on the next call to `consume()`.

        .. note::
          This class and its subclasses assume that fetch requests are made
          on one topic at a time, so this handler only deals with the first
          topic returned.
        """
        messages = []

        # we only fetch one topic so we can assume only one comes back
        topic = response.topics[0].name
        for partition in response.topics[0].partitions:
            code = partition.error_code
            if code == errors.no_error:
                messages.extend(self.deserialize_messages(topic, partition))

# 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

            msg for messageset in results.values() for msg in messageset
            if messageset
        ])

    def handle_fetch_response(self, response):
        """
        Handler for responses from the message "fetch" api.

        Messages returned with the "no error" code are deserialized and
        collected, the full resulting list is returned.

        A retriable error code will cause the cluster "heal" flag to be set.

        An error indicating that the offset used for the partition was out
        of range will cause the offending topic's offsets to be redetermined
        on the next call to `consume()`.

        .. note::
          This class and its subclasses assume that fetch requests are made
          on one topic at a time, so this handler only deals with the first
          topic returned.
        """
        messages = []

        # we only fetch one topic so we can assume only one comes back
        topic = response.topics[0].name
        for partition in response.topics[0].partitions:
            code = partition.error_code
            if code == errors.no_error:
                messages.extend(self.deserialize_messages(topic, partition))

            elif code in errors.retriable: