# kafka package

## Subpackages

- kafka.consumer package
  - Submodules
  - kafka.consumer.base module
  - kafka.consumer.kafka module
  - kafka.consumer.multiprocess module
  - kafka.consumer.simple module
  - Module contents
- kafka.partitioner package
  - Submodules
  - kafka.partitioner.base module
  - kafka.partitioner.hashed module
  - kafka.partitioner.roundrobin module
  - Module contents
- kafka.producer package
  - Submodules
  - kafka.producer.base module
  - kafka.producer.keyed module
  - kafka.producer.simple module
  - Module contents

## Submodules

## kafka.cluster module

class `kafka.cluster.``ClusterMetadata`(**configs)

  Bases: `object`

  A class to manage kafka cluster metadata.

  This class does not perform any IO. It simply updates internal state given API responses (MetadataResponse, GroupCoordinatorResponse).

  Keyword Arguments:
    retry_backoff_ms (int): Milliseconds to backoff when retrying on
      errors. Default: 100.
    metadata_max_age_ms (int): The period of time in milliseconds after
      which we force a refresh of metadata even if we haven’t seen any partition leadership changes to proactively discover any new brokers or partitions. Default: 300000
    bootstrap_servers: ‘host[:port]’ string (or list of ‘host[:port]’
      strings) that the client should contact to bootstrap initial cluster metadata. This does not have to be the full node list. It just needs to have at least one broker that will respond to a Metadata API Request. Default port is 9092. If no servers are specified, will default to localhost:9092.
  `DEFAULT_CONFIG` = {'bootstrap_servers': [], 'metadata_max_age_ms': 300000, 'retry_backoff_ms': 100}
  `add_group_coordinator`(group, response)

    Update with metadata for a group coordinator

    Arguments:
      group (str): name of group from GroupCoordinatorRequest response (GroupCoordinatorResponse): broker response
    Returns:
      bool: True if metadata is updated, False on error
  `add_listener`(listener)

    Add a callback function to be called on each metadata update

  `available_partitions_for_topic`(topic)

    Return set of partitions with known leaders

    Arguments:
      topic (str): topic to check for partitions
    Returns:
      set: {partition (int), …} None if topic not found.
  `broker_metadata`(broker_id)

    Get BrokerMetadata

    Arguments:
      broker_id (int): node_id for a broker to check
    Returns:
      BrokerMetadata or None if not found
  `brokers`()

    Get all BrokerMetadata

    Returns:
      set: {BrokerMetadata, …}
  `coordinator_for_group`(group)

    Return node_id of group coordinator.

    Arguments:
      group (str): name of consumer group
    Returns:
      int: node_id for group coordinator None if the group does not exist.
  `failed_update`(exception)

    Update cluster state given a failed MetadataRequest.

  `is_bootstrap`(node_id)
  `leader_for_partition`(partition)

    Return node_id of leader, -1 unavailable, None if unknown.

  `partitions_for_broker`(broker_id)

    Return TopicPartitions for which the broker is a leader.

    Arguments:
      broker_id (int): node id for a broker
    Returns:
      set: {TopicPartition, …} None if the broker either has no partitions or does not exist.
  `partitions_for_topic`(topic)

    Return set of all partitions for topic (whether available or not)

    Arguments:
      topic (str): topic to check for partitions
    Returns:
      set: {partition (int), …}
  `refresh_backoff`()

    Return milliseconds to wait before attempting to retry after failure

  `remove_listener`(listener)

    Remove a previously added listener callback

  `request_update`()

    Flags metadata for update, return Future()

    Actual update must be handled separately. This method will only change the reported ttl()

    Returns:
      kafka.future.Future (value will be the cluster object after update)
  `topics`(exclude_internal_topics=True)

    Get set of known topics.

    Arguments:
      exclude_internal_topics (bool): Whether records from internal topics
        (such as offsets) should be exposed to the consumer. If set to True the only way to receive records from an internal topic is subscribing to it. Default True
    Returns:
      set: {topic (str), …}
  `ttl`()

    Milliseconds until metadata should be refreshed

  `update_metadata`(metadata)

    Update cluster state given a MetadataResponse.

    Arguments:
      metadata (MetadataResponse): broker response to a metadata request

    Returns: None

  `with_partitions`(partitions_to_add)

    Returns a copy of cluster metadata with partitions added

## kafka.client module

class `kafka.client.``SimpleClient`(hosts, client_id=b'kafka-python', timeout=120, correlation_id=0)

  Bases: `object`

  `CLIENT_ID` = b'kafka-python'
  `DEFAULT_SOCKET_TIMEOUT_SECONDS` = 120
  `close`()
  `copy`()

    Create an inactive copy of the client object, suitable for passing to a separate thread.

    Note that the copied connections are not initialized, so `reinit()` must be called on the returned copy.

  `ensure_topic_exists`(topic, timeout=30)
  `get_partition_ids_for_topic`(topic)
  `has_metadata_for_topic`(topic)
  `load_metadata_for_topics`(*topics, **kwargs)

    Fetch broker and topic-partition metadata from the server.

    Updates internal data: broker list, topic/partition list, and topic/partition -> broker map. This method should be called after receiving any error.

    Note: Exceptions will not be raised in a full refresh (i.e. no topic list). In this case, error codes will be logged as errors. Partition-level errors will also not be raised here (a single partition w/o a leader, for example).

    Arguments:
      *topics (optional): If a list of topics is provided,
        the metadata refresh will be limited to the specified topics only.
      ignore_leadernotavailable (bool): suppress LeaderNotAvailableError
        so that metadata is loaded correctly during auto-create. Default: False.
    Raises:
      UnknownTopicOrPartitionError: Raised for topics that do not exist,
        unless the broker is configured to auto-create topics.
      LeaderNotAvailableError: Raised for topics that do not exist yet,
        when the broker is configured to auto-create topics. Retry after a short backoff (topics/partitions are initializing).
  `reinit`()
  `reset_all_metadata`()
  `reset_topic_metadata`(*topics)
  `send_consumer_metadata_request`(payloads=(), fail_on_error=True, callback=None)
  `send_fetch_request`(payloads=(), fail_on_error=True, callback=None, max_wait_time=100, min_bytes=4096)

    Encode and send a FetchRequest

    Payloads are grouped by topic and partition so they can be pipelined to the same brokers.

  `send_list_offset_request`(payloads=(), fail_on_error=True, callback=None)
  `send_metadata_request`(payloads=(), fail_on_error=True, callback=None)
  `send_offset_commit_request`(group, payloads=(), fail_on_error=True, callback=None)
  `send_offset_fetch_request`(group, payloads=(), fail_on_error=True, callback=None)
  `send_offset_fetch_request_kafka`(group, payloads=(), fail_on_error=True, callback=None)
  `send_offset_request`(payloads=(), fail_on_error=True, callback=None)
  `send_produce_request`(payloads=(), acks=1, timeout=1000, fail_on_error=True, callback=None)

    Encode and send some ProduceRequests

    ProduceRequests will be grouped by (topic, partition) and then sent to a specific broker. Output is a list of responses in the same order as the list of payloads specified

    Arguments:
      payloads (list of ProduceRequest): produce requests to send to kafka
        ProduceRequest payloads must not contain duplicates for any topic-partition.
      acks (int, optional): how many acks the servers should receive from replica
        brokers before responding to the request. If it is 0, the server will not send any response. If it is 1, the server will wait until the data is written to the local log before sending a response. If it is -1, the server will wait until the message is committed by all in-sync replicas before sending a response. For any value > 1, the server will wait for this number of acks to occur (but the server will never wait for more acknowledgements than there are in-sync replicas). defaults to 1.
      timeout (int, optional): maximum time in milliseconds the server can
        await the receipt of the number of acks, defaults to 1000.
      fail_on_error (bool, optional): raise exceptions on connection and
        server response errors, defaults to True.
      callback (function, optional): instead of returning the ProduceResponse,
        first pass it through this function, defaults to None.
    Returns:
      list of ProduceResponses, or callback results if supplied, in the order of input payloads
  `topics`

## kafka.codec module

`kafka.codec.``gzip_decode`(payload)

`kafka.codec.``gzip_encode`(payload, compresslevel=None)

`kafka.codec.``has_gzip`()

`kafka.codec.``has_lz4`()

`kafka.codec.``has_snappy`()

`kafka.codec.``lz4_decode_old_kafka`(payload)

`kafka.codec.``lz4_encode`(payload, **kwargs)

`kafka.codec.``lz4_encode_old_kafka`(payload)

  Encode payload for 0.8/0.9 brokers – requires an incorrect header checksum.

`kafka.codec.``lz4f_decode`(payload)

  Decode payload using interoperable LZ4 framing. Requires Kafka >= 0.10

`kafka.codec.``snappy_decode`(payload)

`kafka.codec.``snappy_encode`(payload, xerial_compatible=True, xerial_blocksize=32768)

  Encodes the given data with snappy compression.

  If xerial_compatible is set then the stream is encoded in a fashion compatible with the xerial snappy library.

  The block size (xerial_blocksize) controls how frequent the blocking occurs 32k is the default in the xerial library.

  The format winds up being:

  > Header | Block1 len | Block1 data | Blockn len | Blockn data
  > 16 bytes | BE int32 | snappy bytes | BE int32 | snappy bytes

  It is important to note that the blocksize is the amount of uncompressed data presented to snappy at each block, whereas the blocklen is the number of bytes that will be present in the stream; so the length will always be <= blocksize.

## kafka.common module

## kafka.conn module

class `kafka.conn.``BrokerConnection`(host, port, afi, **configs)

  Bases: `object`

  Initialize a Kafka broker connection

  Keyword Arguments:
    client_id (str): a name for this client. This string is passed in
      each request to servers and can be used to identify specific server-side log entries that correspond to this client. Also submitted to GroupCoordinator for logging with respect to consumer group administration. Default: ‘kafka-python-{version}’
    reconnect_backoff_ms (int): The amount of time in milliseconds to
      wait before attempting to reconnect to a given host. Default: 50.
    reconnect_backoff_max_ms (int): The maximum amount of time in
      milliseconds to wait when reconnecting to a broker that has repeatedly failed to connect. If provided, the backoff per host will increase exponentially for each consecutive connection failure, up to this maximum. To avoid connection storms, a randomization factor of 0.2 will be applied to the backoff resulting in a random range between 20% below and 20% above the computed value. Default: 1000.
    request_timeout_ms (int): Client request timeout in milliseconds.
      Default: 30000.
    max_in_flight_requests_per_connection (int): Requests are pipelined
      to kafka brokers up to this number of maximum requests per broker connection. Default: 5.
    receive_buffer_bytes (int): The size of the TCP receive buffer
      (SO_RCVBUF) to use when reading data. Default: None (relies on system defaults). Java client defaults to 32768.
    send_buffer_bytes (int): The size of the TCP send buffer
      (SO_SNDBUF) to use when sending data. Default: None (relies on system defaults). Java client defaults to 131072.
    socket_options (list): List of tuple-arguments to socket.setsockopt
      to apply to broker connection sockets. Default: [(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)]
    security_protocol (str): Protocol used to communicate with brokers.
      Valid values are: PLAINTEXT, SSL, SASL_PLAINTEXT, SASL_SSL. Default: PLAINTEXT.
    ssl_context (ssl.SSLContext): pre-configured SSLContext for wrapping
      socket connections. If provided, all other ssl_* configurations will be ignored. Default: None.
    ssl_check_hostname (bool): flag to configure whether ssl handshake
      should verify that the certificate matches the brokers hostname. default: True.
    ssl_cafile (str): optional filename of ca file to use in certificate
      verification. default: None.
    ssl_certfile (str): optional filename of file in pem format containing
      the client certificate, as well as any ca certificates needed to establish the certificate’s authenticity. default: None.
    ssl_keyfile (str): optional filename containing the client private key.
      default: None.
    ssl_password (callable, str, bytes, bytearray): optional password or
      callable function that returns a password, for decrypting the client private key. Default: None.
    ssl_crlfile (str): optional filename containing the CRL to check for
      certificate expiration. By default, no CRL check is done. When providing a file, only the leaf certificate will be checked against this CRL. The CRL can only be checked with Python 3.4+ or 2.7.9+. default: None.
    ssl_ciphers (str): optionally set the available ciphers for ssl
      connections. It should be a string in the OpenSSL cipher list format. If no cipher can be selected (because compile-time options or other configuration forbids use of all the specified ciphers), an ssl.SSLError will be raised. See ssl.SSLContext.set_ciphers
    api_version (tuple): Specify which Kafka API version to use.
      Accepted values are: (0, 8, 0), (0, 8, 1), (0, 8, 2), (0, 9), (0, 10). Default: (0, 8, 2)
    api_version_auto_timeout_ms (int): number of milliseconds to throw a
      timeout exception from the constructor when checking the broker api version. Only applies if api_version is None
    selector (selectors.BaseSelector): Provide a specific selector
      implementation to use for I/O multiplexing. Default: selectors.DefaultSelector
    state_change_callback (callable): function to be called when the
      connection state changes from CONNECTING to CONNECTED etc.
    metrics (kafka.metrics.Metrics): Optionally provide a metrics
      instance for capturing network IO stats. Default: None.

    metric_group_prefix (str): Prefix for metric names. Default: ‘’ sasl_mechanism (str): Authentication mechanism when security_protocol

    > is configured for SASL_PLAINTEXT or SASL_SSL. Valid values are: PLAIN, GSSAPI, OAUTHBEARER.

    sasl_plain_username (str): username for sasl PLAIN authentication.
      Required if sasl_mechanism is PLAIN.
    sasl_plain_password (str): password for sasl PLAIN authentication.
      Required if sasl_mechanism is PLAIN.
    sasl_kerberos_service_name (str): Service name to include in GSSAPI
      sasl mechanism handshake. Default: ‘kafka’
    sasl_kerberos_domain_name (str): kerberos domain name to use in GSSAPI
      sasl mechanism handshake. Default: one of bootstrap servers
    sasl_oauth_token_provider (AbstractTokenProvider): OAuthBearer token provider
      instance. (See kafka.oauth.abstract). Default: None
  `DEFAULT_CONFIG` = {'api_version': (0, 8, 2), 'client_id': 'kafka-python-1.4.6', 'max_in_flight_requests_per_connection': 5, 'metric_group_prefix': '', 'metrics': None, 'node_id': 0, 'receive_buffer_bytes': None, 'reconnect_backoff_max_ms': 1000, 'reconnect_backoff_ms': 50, 'request_timeout_ms': 30000, 'sasl_kerberos_domain_name': None, 'sasl_kerberos_service_name': 'kafka', 'sasl_mechanism': None, 'sasl_oauth_token_provider': None, 'sasl_plain_password': None, 'sasl_plain_username': None, 'security_protocol': 'PLAINTEXT', 'selector': <class 'selectors.EpollSelector'>, 'send_buffer_bytes': None, 'sock_chunk_buffer_count': 1000, 'sock_chunk_bytes': 4096, 'socket_options': [(6, 1, 1)], 'ssl_cafile': None, 'ssl_certfile': None, 'ssl_check_hostname': True, 'ssl_ciphers': None, 'ssl_context': None, 'ssl_crlfile': None, 'ssl_keyfile': None, 'ssl_password': None, 'state_change_callback': <function BrokerConnection.<lambda>>}
  `SASL_MECHANISMS` = ('PLAIN', 'GSSAPI', 'OAUTHBEARER')
  `SECURITY_PROTOCOLS` = ('PLAINTEXT', 'SSL', 'SASL_PLAINTEXT', 'SASL_SSL')
  `blacked_out`()

    Return true if we are disconnected from the given node and can’t re-establish a connection yet

  `can_send_more`()

    Return True unless there are max_in_flight_requests_per_connection.

  `check_version`(timeout=2, strict=False, topics=[])

    Attempt to guess the broker version.

    Note: This is a blocking call.

    Returns: version tuple, i.e. (0, 10), (0, 9), (0, 8, 2), …

  `close`(error=None)

    Close socket and fail all in-flight-requests.

    Arguments:
      error (Exception, optional): pending in-flight-requests
        will be failed with this exception. Default: kafka.errors.KafkaConnectionError.
  `connect`()

    Attempt to connect and return ConnectionState

  `connect_blocking`(timeout=inf)
  `connected`()

    Return True iff socket is connected.

  `connecting`()

    Returns True if still connecting (this may encompass several different states, such as SSL handshake, authorization, etc).

  `connection_delay`()

    Return the number of milliseconds to wait, based on the connection state, before attempting to send data. When disconnected, this respects the reconnect backoff time. When connecting, returns 0 to allow non-blocking connect to finish. When connected, returns a very large number to handle slow/stalled connections.

  `disconnected`()

    Return True iff socket is closed

  `get_api_versions`()
  `recv`()

    Non-blocking network receive.

    Return list of (response, future) tuples

  `requests_timed_out`()
  `send`(request, blocking=True)

    Queue request for async network send, return Future()

  `send_pending_requests`()

    Can block on network if request is larger than send_buffer_bytes

class `kafka.conn.``BrokerConnectionMetrics`(metrics, metric_group_prefix, node_id)

  Bases: `object`

class `kafka.conn.``ConnectionStates`

  Bases: `object`

  `AUTHENTICATING` = '<authenticating>'
  `CONNECTED` = '<connected>'
  `CONNECTING` = '<connecting>'
  `DISCONNECTED` = '<disconnected>'
  `DISCONNECTING` = '<disconnecting>'
  `HANDSHAKE` = '<handshake>'

`kafka.conn.``collect_hosts`(hosts, randomize=True)

  Collects a comma-separated set of hosts (host:port) and optionally randomize the returned list.

`kafka.conn.``dns_lookup`(host, port, afi=<AddressFamily.AF_UNSPEC: 0>)

  Returns a list of getaddrinfo structs, optionally filtered to an afi (ipv4 / ipv6)

`kafka.conn.``get_ip_port_afi`(host_and_port_str)

  Parse the IP and port from a string in the format of:

  > - host_or_ip <- Can be either IPv4 address literal or hostname/fqdn
  > - host_or_ipv4:port <- Can be either IPv4 address literal or hostname/fqdn
  > - [host_or_ip] <- IPv6 address literal
  > - [host_or_ip]:port. <- IPv6 address literal

  Note

  IPv6 address literals with ports must be enclosed in brackets

  Note

  If the port is not specified, default will be returned.

  Returns: | tuple (host, port, afi), afi will be socket.AF_INET or socket.AF_INET6 or socket.AF_UNSPEC

`kafka.conn.``is_inet_4_or_6`(gai)

  Given a getaddrinfo struct, return True iff ipv4 or ipv6

## kafka.context module

Context manager to commit/rollback consumer offsets.

class `kafka.context.``OffsetCommitContext`(consumer)

  Bases: `object`

  Provides commit/rollback semantics around a SimpleConsumer.

  Usage assumes that auto_commit is disabled, that messages are consumed in batches, and that the consuming process will record its own successful processing of each message. Both the commit and rollback operations respect a “high-water mark” to ensure that last unsuccessfully processed message will be retried.

  Example:

      consumer = SimpleConsumer(client, group, topic, auto_commit=False)
      consumer.provide_partition_info()
      consumer.fetch_last_known_offsets()

      while some_condition:
          with OffsetCommitContext(consumer) as context:
              messages = consumer.get_messages(count, block=False)

              for partition, message in messages:
                  if can_process(message):
                      context.mark(partition, message.offset)
                  else:
                      break

              if not context:
                  sleep(delay)

  These semantics allow for deferred message processing (e.g. if can_process compares message time to clock time) and for repeated processing of the last unsuccessful message (until some external error is resolved).

  `commit`()

    Commit this context’s offsets:

    > - If the high-water mark has moved, commit up to and position the consumer at the high-water mark.
    > - Otherwise, reset to the consumer to the initial offsets.

  `commit_partition_offsets`(partition_offsets)

    Commit explicit partition/offset pairs.

  `handle_out_of_range`()

    Handle out of range condition by seeking to the beginning of valid ranges.

    This assumes that an out of range doesn’t happen by seeking past the end of valid ranges – which is far less likely.

  `mark`(partition, offset)

    Set the high-water mark in the current context.

    In order to know the current partition, it is helpful to initialize the consumer to provide partition info via:

        consumer.provide_partition_info()

  `rollback`()

    Rollback this context:

    > - Position the consumer at the initial offsets.

  `update_consumer_offsets`(partition_offsets)

    Update consumer offsets to explicit positions.

## kafka.protocol module

## kafka.util module

class `kafka.util.``Dict`

  Bases: `dict`

  Utility class to support passing weakrefs to dicts

  See: https://docs.python.org/2/library/weakref.html

class `kafka.util.``ReentrantTimer`(t, fn, *args, **kwargs)

  Bases: `object`

  A timer that can be restarted, unlike threading.Timer (although this uses threading.Timer)

  Arguments:

  > t: timer interval in milliseconds fn: a callable to invoke args: tuple of args to be passed to function kwargs: keyword arguments to be passed to function

  `start`()
  `stop`()

class `kafka.util.``WeakMethod`(object_dot_method)

  Bases: `object`

  Callable that weakly references a method and the object it is bound to. It is based on https://stackoverflow.com/a/24287465.

  Arguments:

  > object_dot_method: A bound instance method (i.e. ‘object.method’).

`kafka.util.``crc32`(data)

`kafka.util.``group_by_topic_and_partition`(tuples)

`kafka.util.``read_short_string`(data, cur)

`kafka.util.``relative_unpack`(fmt, data, cur)

`kafka.util.``try_method_on_system_exit`(obj, method, *args, **kwargs)

`kafka.util.``write_int_string`(s)

## Module contents

class `kafka.``KafkaAdminClient`(**configs)

  Bases: `object`

  A class for administering the Kafka cluster.

  Warning:
    This is an unstable interface that was recently added and is subject to change without warning. In particular, many methods currently return raw protocol tuples. In future releases, we plan to make these into nicer, more pythonic objects. Unfortunately, this will likely break those interfaces.

  The KafkaAdminClient class will negotiate for the latest version of each message protocol format supported by both the kafka-python client library and the Kafka broker. Usage of optional fields from protocol versions that are not supported by the broker will result in IncompatibleBrokerVersion exceptions.

  Use of this class requires a minimum broker version >= 0.10.0.0.

  Keyword Arguments:
    bootstrap_servers: ‘host[:port]’ string (or list of ‘host[:port]’
      strings) that the consumer should contact to bootstrap initial cluster metadata. This does not have to be the full node list. It just needs to have at least one broker that will respond to a Metadata API Request. Default port is 9092. If no servers are specified, will default to localhost:9092.
    client_id (str): a name for this client. This string is passed in
      each request to servers and can be used to identify specific server-side log entries that correspond to this client. Also submitted to GroupCoordinator for logging with respect to consumer group administration. Default: ‘kafka-python-{version}’
    reconnect_backoff_ms (int): The amount of time in milliseconds to
      wait before attempting to reconnect to a given host. Default: 50.
    reconnect_backoff_max_ms (int): The maximum amount of time in
      milliseconds to wait when reconnecting to a broker that has repeatedly failed to connect. If provided, the backoff per host will increase exponentially for each consecutive connection failure, up to this maximum. To avoid connection storms, a randomization factor of 0.2 will be applied to the backoff resulting in a random range between 20% below and 20% above the computed value. Default: 1000.
    request_timeout_ms (int): Client request timeout in milliseconds.
      Default: 30000.
    connections_max_idle_ms: Close idle connections after the number of
      milliseconds specified by this config. The broker closes idle connections after connections.max.idle.ms, so this avoids hitting unexpected socket disconnected errors on the client. Default: 540000
    retry_backoff_ms (int): Milliseconds to backoff when retrying on
      errors. Default: 100.
    max_in_flight_requests_per_connection (int): Requests are pipelined
      to kafka brokers up to this number of maximum requests per broker connection. Default: 5.
    receive_buffer_bytes (int): The size of the TCP receive buffer
      (SO_RCVBUF) to use when reading data. Default: None (relies on system defaults). Java client defaults to 32768.
    send_buffer_bytes (int): The size of the TCP send buffer
      (SO_SNDBUF) to use when sending data. Default: None (relies on system defaults). Java client defaults to 131072.
    socket_options (list): List of tuple-arguments to socket.setsockopt
      to apply to broker connection sockets. Default: [(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)]
    metadata_max_age_ms (int): The period of time in milliseconds after
      which we force a refresh of metadata even if we haven’t seen any partition leadership changes to proactively discover any new brokers or partitions. Default: 300000
    security_protocol (str): Protocol used to communicate with brokers.
      Valid values are: PLAINTEXT, SSL. Default: PLAINTEXT.
    ssl_context (ssl.SSLContext): Pre-configured SSLContext for wrapping
      socket connections. If provided, all other ssl_* configurations will be ignored. Default: None.
    ssl_check_hostname (bool): Flag to configure whether SSL handshake
      should verify that the certificate matches the broker’s hostname. Default: True.
    ssl_cafile (str): Optional filename of CA file to use in certificate
      veriication. Default: None.
    ssl_certfile (str): Optional filename of file in PEM format containing
      the client certificate, as well as any CA certificates needed to establish the certificate’s authenticity. Default: None.
    ssl_keyfile (str): Optional filename containing the client private key.
      Default: None.
    ssl_password (str): Optional password to be used when loading the
      certificate chain. Default: None.
    ssl_crlfile (str): Optional filename containing the CRL to check for
      certificate expiration. By default, no CRL check is done. When providing a file, only the leaf certificate will be checked against this CRL. The CRL can only be checked with Python 3.4+ or 2.7.9+. Default: None.
    api_version (tuple): Specify which Kafka API version to use. If set
      to None, KafkaClient will attempt to infer the broker version by probing various APIs. Example: (0, 10, 2). Default: None
    api_version_auto_timeout_ms (int): number of milliseconds to throw a
      timeout exception from the constructor when checking the broker api version. Only applies if api_version is None
    selector (selectors.BaseSelector): Provide a specific selector
      implementation to use for I/O multiplexing. Default: selectors.DefaultSelector
    metrics (kafka.metrics.Metrics): Optionally provide a metrics
      instance for capturing network IO stats. Default: None.

    metric_group_prefix (str): Prefix for metric names. Default: ‘’ sasl_mechanism (str): Authentication mechanism when security_protocol

    > is configured for SASL_PLAINTEXT or SASL_SSL. Valid values are: PLAIN, GSSAPI, OAUTHBEARER.

    sasl_plain_username (str): username for sasl PLAIN authentication.
      Required if sasl_mechanism is PLAIN.
    sasl_plain_password (str): password for sasl PLAIN authentication.
      Required if sasl_mechanism is PLAIN.
    sasl_kerberos_service_name (str): Service name to include in GSSAPI
      sasl mechanism handshake. Default: ‘kafka’
    sasl_oauth_token_provider (AbstractTokenProvider): OAuthBearer token provider
      instance. (See kafka.oauth.abstract). Default: None
  `DEFAULT_CONFIG` = {'api_version': None, 'api_version_auto_timeout_ms': 2000, 'bootstrap_servers': 'localhost', 'client_id': 'kafka-python-1.4.6', 'connections_max_idle_ms': 540000, 'max_in_flight_requests_per_connection': 5, 'metadata_max_age_ms': 300000, 'metric_reporters': [], 'metrics_num_samples': 2, 'metrics_sample_window_ms': 30000, 'receive_buffer_bytes': None, 'reconnect_backoff_max_ms': 1000, 'reconnect_backoff_ms': 50, 'request_timeout_ms': 30000, 'retry_backoff_ms': 100, 'sasl_kerberos_service_name': 'kafka', 'sasl_mechanism': None, 'sasl_oauth_token_provider': None, 'sasl_plain_password': None, 'sasl_plain_username': None, 'security_protocol': 'PLAINTEXT', 'selector': <class 'selectors.EpollSelector'>, 'send_buffer_bytes': None, 'sock_chunk_buffer_count': 1000, 'sock_chunk_bytes': 4096, 'socket_options': [(6, 1, 1)], 'ssl_cafile': None, 'ssl_certfile': None, 'ssl_check_hostname': True, 'ssl_context': None, 'ssl_crlfile': None, 'ssl_keyfile': None, 'ssl_password': None}
  `alter_configs`(config_resources)

    Alter configuration parameters of one or more Kafka resources.

    Warning:
      This is currently broken for BROKER resources because those must be sent to that specific broker, versus this always picks the least-loaded node. See the comment in the source code for details. We would happily accept a PR fixing this.

    Parameters: | config_resources – A list of ConfigResource objects.
    Returns: | Appropriate version of AlterConfigsResponse class.

  `close`()

    Close the KafkaAdminClient connection to the Kafka broker.

  `create_partitions`(topic_partitions, timeout_ms=None, validate_only=False)

    Create additional partitions for an existing topic.

    Parameters:

    - topic_partitions – A map of topic name strings to NewPartition objects.
    - timeout_ms – Milliseconds to wait for new partitions to be created before the broker returns.
    - validate_only – If True, don’t actually create new partitions. Default: False

    Returns: | Appropriate version of CreatePartitionsResponse class.

  `create_topics`(new_topics, timeout_ms=None, validate_only=False)

    Create new topics in the cluster.

    Parameters:

    - new_topics – A list of NewTopic objects.
    - timeout_ms – Milliseconds to wait for new topics to be created before the broker returns.
    - validate_only – If True, don’t actually create new topics. Not supported by all versions. Default: False

    Returns: | Appropriate version of CreateTopicResponse class.

  `delete_topics`(topics, timeout_ms=None)

    Delete topics from the cluster.

    Parameters:

    - topics – A list of topic name strings.
    - timeout_ms – Milliseconds to wait for topics to be deleted before the broker returns.

    Returns: | Appropriate version of DeleteTopicsResponse class.

  `describe_configs`(config_resources, include_synonyms=False)

    Fetch configuration parameters for one or more Kafka resources.

    Parameters:

    - config_resources – An list of ConfigResource objects. Any keys in ConfigResource.configs dict will be used to filter the result. Setting the configs dict to None will get all values. An empty dict will get zero values (as per Kafka protocol).
    - include_synonyms – If True, return synonyms in response. Not supported by all versions. Default: False.

    Returns: | Appropriate version of DescribeConfigsResponse class.

  `describe_consumer_groups`(group_ids, group_coordinator_id=None)

    Describe a set of consumer groups.

    Any errors are immediately raised.

    Parameters:

    - group_ids – A list of consumer group IDs. These are typically the group names as strings.
    - group_coordinator_id – The node_id of the groups’ coordinator broker. If set to None, it will query the cluster for each group to find that group’s coordinator. Explicitly specifying this can be useful for avoiding extra network round trips if you already know the group coordinator. This is only useful when all the group_ids have the same coordinator, otherwise it will error. Default: None.

    Returns: | A list of group descriptions. For now the group descriptions are the raw results from the DescribeGroupsResponse. Long-term, we plan to change this to return namedtuples as well as decoding the partition assignments.

  `list_consumer_group_offsets`(group_id, group_coordinator_id=None, partitions=None)

    Fetch Consumer Group Offsets.

    Note: This does not verify that the group_id or partitions actually exist in the cluster.

    As soon as any error is encountered, it is immediately raised.

    Parameters:

    - group_id – The consumer group id name for which to fetch offsets.
    - group_coordinator_id – The node_id of the group’s coordinator broker. If set to None, will query the cluster to find the group coordinator. Explicitly specifying this can be useful to prevent that extra network round trip if you already know the group coordinator. Default: None.
    - partitions – A list of TopicPartitions for which to fetch offsets. On brokers >= 0.10.2, this can be set to None to fetch all known offsets for the consumer group. Default: None.

    Return dictionary:
    | A dictionary with TopicPartition keys and OffsetAndMetada values. Partitions that are not specified and for which the group_id does not have a recorded offset are omitted. An offset value of -1 indicates the group_id has no offset for that TopicPartition. A -1 can only happen for partitions that are explicitly specified.

  `list_consumer_groups`(broker_ids=None)

    List all consumer groups known to the cluster.

    This returns a list of Consumer Group tuples. The tuples are composed of the consumer group name and the consumer group protocol type.

    Only consumer groups that store their offsets in Kafka are returned. The protocol type will be an empty string for groups created using Kafka < 0.9 APIs because, although they store their offsets in Kafka, they don’t use Kafka for group coordination. For groups created using Kafka >= 0.9, the protocol type will typically be “consumer”.

    As soon as any error is encountered, it is immediately raised.

    Parameters: | broker_ids – A list of broker node_ids to query for consumer groups. If set to None, will query all brokers in the cluster. Explicitly specifying broker(s) can be useful for determining which consumer groups are coordinated by those broker(s). Default: None
    Return list: | List of tuples of Consumer Groups.

    Raises:

    - GroupCoordinatorNotAvailableError – The coordinator is not available, so cannot process requests.
    - GroupLoadInProgressError – The coordinator is loading and hence can’t process requests.

class `kafka.``KafkaConsumer`(*topics, **configs)

  Bases: `object`

  Consume records from a Kafka cluster.

  The consumer will transparently handle the failure of servers in the Kafka cluster, and adapt as topic-partitions are created or migrate between brokers. It also interacts with the assigned kafka Group Coordinator node to allow multiple consumers to load balance consumption of topics (requires kafka >= 0.9.0.0).

  The consumer is not thread safe and should not be shared across threads.

  Arguments:
    *topics (str): optional list of topics to subscribe to. If not set,
      call `subscribe()` or `assign()` before consuming records.
  Keyword Arguments:
    bootstrap_servers: ‘host[:port]’ string (or list of ‘host[:port]’
      strings) that the consumer should contact to bootstrap initial cluster metadata. This does not have to be the full node list. It just needs to have at least one broker that will respond to a Metadata API Request. Default port is 9092. If no servers are specified, will default to localhost:9092.
    client_id (str): A name for this client. This string is passed in
      each request to servers and can be used to identify specific server-side log entries that correspond to this client. Also submitted to GroupCoordinator for logging with respect to consumer group administration. Default: ‘kafka-python-{version}’
    group_id (str or None): The name of the consumer group to join for dynamic
      partition assignment (if enabled), and to use for fetching and committing offsets. If None, auto-partition assignment (via group coordinator) and offset commits are disabled. Default: None
    key_deserializer (callable): Any callable that takes a
      raw message key and returns a deserialized key.
    value_deserializer (callable): Any callable that takes a
      raw message value and returns a deserialized value.
    fetch_min_bytes (int): Minimum amount of data the server should
      return for a fetch request, otherwise wait up to fetch_max_wait_ms for more data to accumulate. Default: 1.
    fetch_max_wait_ms (int): The maximum amount of time in milliseconds
      the server will block before answering the fetch request if there isn’t sufficient data to immediately satisfy the requirement given by fetch_min_bytes. Default: 500.
    fetch_max_bytes (int): The maximum amount of data the server should
      return for a fetch request. This is not an absolute maximum, if the first message in the first non-empty partition of the fetch is larger than this value, the message will still be returned to ensure that the consumer can make progress. NOTE: consumer performs fetches to multiple brokers in parallel so memory usage will depend on the number of brokers containing partitions for the topic. Supported Kafka version >= 0.10.1.0. Default: 52428800 (50 MB).
    max_partition_fetch_bytes (int): The maximum amount of data
      per-partition the server will return. The maximum total memory used for a request = #partitions * max_partition_fetch_bytes. This size must be at least as large as the maximum message size the server allows or else it is possible for the producer to send messages larger than the consumer can fetch. If that happens, the consumer can get stuck trying to fetch a large message on a certain partition. Default: 1048576.
    request_timeout_ms (int): Client request timeout in milliseconds.
      Default: 305000.
    retry_backoff_ms (int): Milliseconds to backoff when retrying on
      errors. Default: 100.
    reconnect_backoff_ms (int): The amount of time in milliseconds to
      wait before attempting to reconnect to a given host. Default: 50.
    reconnect_backoff_max_ms (int): The maximum amount of time in
      milliseconds to wait when reconnecting to a broker that has repeatedly failed to connect. If provided, the backoff per host will increase exponentially for each consecutive connection failure, up to this maximum. To avoid connection storms, a randomization factor of 0.2 will be applied to the backoff resulting in a random range between 20% below and 20% above the computed value. Default: 1000.
    max_in_flight_requests_per_connection (int): Requests are pipelined
      to kafka brokers up to this number of maximum requests per broker connection. Default: 5.
    auto_offset_reset (str): A policy for resetting offsets on
      OffsetOutOfRange errors: ‘earliest’ will move to the oldest available message, ‘latest’ will move to the most recent. Any other value will raise the exception. Default: ‘latest’.
    enable_auto_commit (bool): If True , the consumer’s offset will be
      periodically committed in the background. Default: True.
    auto_commit_interval_ms (int): Number of milliseconds between automatic
      offset commits, if enable_auto_commit is True. Default: 5000.
    default_offset_commit_callback (callable): Called as
      callback(offsets, response) response will be either an Exception or an OffsetCommitResponse struct. This callback can be used to trigger custom actions when a commit request completes.
    check_crcs (bool): Automatically check the CRC32 of the records
      consumed. This ensures no on-the-wire or on-disk corruption to the messages occurred. This check adds some overhead, so it may be disabled in cases seeking extreme performance. Default: True
    metadata_max_age_ms (int): The period of time in milliseconds after
      which we force a refresh of metadata, even if we haven’t seen any partition leadership changes to proactively discover any new brokers or partitions. Default: 300000
    partition_assignment_strategy (list): List of objects to use to
      distribute partition ownership amongst consumer instances when group management is used. Default: [RangePartitionAssignor, RoundRobinPartitionAssignor]
    max_poll_records (int): The maximum number of records returned in a
      single call to `poll()`. Default: 500
    max_poll_interval_ms (int): The maximum delay between invocations of
      `poll()` when using consumer group management. This places an upper bound on the amount of time that the consumer can be idle before fetching more records. If `poll()` is not called before expiration of this timeout, then the consumer is considered failed and the group will rebalance in order to reassign the partitions to another member. Default 300000
    session_timeout_ms (int): The timeout used to detect failures when
      using Kafka’s group management facilities. The consumer sends periodic heartbeats to indicate its liveness to the broker. If no heartbeats are received by the broker before the expiration of this session timeout, then the broker will remove this consumer from the group and initiate a rebalance. Note that the value must be in the allowable range as configured in the broker configuration by group.min.session.timeout.ms and group.max.session.timeout.ms. Default: 10000
    heartbeat_interval_ms (int): The expected time in milliseconds
      between heartbeats to the consumer coordinator when using Kafka’s group management facilities. Heartbeats are used to ensure that the consumer’s session stays active and to facilitate rebalancing when new consumers join or leave the group. The value must be set lower than session_timeout_ms, but typically should be set no higher than 1/3 of that value. It can be adjusted even lower to control the expected time for normal rebalances. Default: 3000
    receive_buffer_bytes (int): The size of the TCP receive buffer
      (SO_RCVBUF) to use when reading data. Default: None (relies on system defaults). The java client defaults to 32768.
    send_buffer_bytes (int): The size of the TCP send buffer
      (SO_SNDBUF) to use when sending data. Default: None (relies on system defaults). The java client defaults to 131072.
    socket_options (list): List of tuple-arguments to socket.setsockopt
      to apply to broker connection sockets. Default: [(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)]
    consumer_timeout_ms (int): number of milliseconds to block during
      message iteration before raising StopIteration (i.e., ending the iterator). Default block forever [float(‘inf’)].
    security_protocol (str): Protocol used to communicate with brokers.
      Valid values are: PLAINTEXT, SSL. Default: PLAINTEXT.
    ssl_context (ssl.SSLContext): Pre-configured SSLContext for wrapping
      socket connections. If provided, all other ssl_* configurations will be ignored. Default: None.
    ssl_check_hostname (bool): Flag to configure whether ssl handshake
      should verify that the certificate matches the brokers hostname. Default: True.
    ssl_cafile (str): Optional filename of ca file to use in certificate
      verification. Default: None.
    ssl_certfile (str): Optional filename of file in pem format containing
      the client certificate, as well as any ca certificates needed to establish the certificate’s authenticity. Default: None.
    ssl_keyfile (str): Optional filename containing the client private key.
      Default: None.
    ssl_password (str): Optional password to be used when loading the
      certificate chain. Default: None.
    ssl_crlfile (str): Optional filename containing the CRL to check for
      certificate expiration. By default, no CRL check is done. When providing a file, only the leaf certificate will be checked against this CRL. The CRL can only be checked with Python 3.4+ or 2.7.9+. Default: None.
    ssl_ciphers (str): optionally set the available ciphers for ssl
      connections. It should be a string in the OpenSSL cipher list format. If no cipher can be selected (because compile-time options or other configuration forbids use of all the specified ciphers), an ssl.SSLError will be raised. See ssl.SSLContext.set_ciphers
    api_version (tuple): Specify which Kafka API version to use. If set to

      None, the client will attempt to infer the broker version by probing various APIs. Different versions enable different functionality.

      Examples:
        (0, 9) enables full group coordination features with automatic
          partition assignment and rebalancing,
        (0, 8, 2) enables kafka-storage offset commits with manual
          partition assignment only,
        (0, 8, 1) enables zookeeper-storage offset commits with manual
          partition assignment only,
        (0, 8, 0) enables basic functionality but requires manual
          partition assignment and offset management.

      Default: None

    api_version_auto_timeout_ms (int): number of milliseconds to throw a
      timeout exception from the constructor when checking the broker api version. Only applies if api_version set to ‘auto’
    connections_max_idle_ms: Close idle connections after the number of
      milliseconds specified by this config. The broker closes idle connections after connections.max.idle.ms, so this avoids hitting unexpected socket disconnected errors on the client. Default: 540000
    metric_reporters (list): A list of classes to use as metrics reporters.
      Implementing the AbstractMetricsReporter interface allows plugging in classes that will be notified of new metric creation. Default: []
    metrics_num_samples (int): The number of samples maintained to compute
      metrics. Default: 2
    metrics_sample_window_ms (int): The maximum age in milliseconds of
      samples used to compute metrics. Default: 30000
    selector (selectors.BaseSelector): Provide a specific selector
      implementation to use for I/O multiplexing. Default: selectors.DefaultSelector
    exclude_internal_topics (bool): Whether records from internal topics
      (such as offsets) should be exposed to the consumer. If set to True the only way to receive records from an internal topic is subscribing to it. Requires 0.10+ Default: True
    sasl_mechanism (str): Authentication mechanism when security_protocol
      is configured for SASL_PLAINTEXT or SASL_SSL. Valid values are: PLAIN, GSSAPI, OAUTHBEARER.
    sasl_plain_username (str): Username for sasl PLAIN authentication.
      Required if sasl_mechanism is PLAIN.
    sasl_plain_password (str): Password for sasl PLAIN authentication.
      Required if sasl_mechanism is PLAIN.
    sasl_kerberos_service_name (str): Service name to include in GSSAPI
      sasl mechanism handshake. Default: ‘kafka’
    sasl_kerberos_domain_name (str): kerberos domain name to use in GSSAPI
      sasl mechanism handshake. Default: one of bootstrap servers
    sasl_oauth_token_provider (AbstractTokenProvider): OAuthBearer token provider
      instance. (See kafka.oauth.abstract). Default: None
  Note:
    Configuration parameters are described in more detail at https://kafka.apache.org/documentation/#newconsumerconfigs
  `DEFAULT_CONFIG` = {'api_version': None, 'api_version_auto_timeout_ms': 2000, 'auto_commit_interval_ms': 5000, 'auto_offset_reset': 'latest', 'bootstrap_servers': 'localhost', 'check_crcs': True, 'client_id': 'kafka-python-1.4.6', 'connections_max_idle_ms': 540000, 'consumer_timeout_ms': inf, 'default_offset_commit_callback': <function KafkaConsumer.<lambda>>, 'enable_auto_commit': True, 'exclude_internal_topics': True, 'fetch_max_bytes': 52428800, 'fetch_max_wait_ms': 500, 'fetch_min_bytes': 1, 'group_id': None, 'heartbeat_interval_ms': 3000, 'key_deserializer': None, 'max_in_flight_requests_per_connection': 5, 'max_partition_fetch_bytes': 1048576, 'max_poll_interval_ms': 300000, 'max_poll_records': 500, 'metadata_max_age_ms': 300000, 'metric_group_prefix': 'consumer', 'metric_reporters': [], 'metrics_num_samples': 2, 'metrics_sample_window_ms': 30000, 'partition_assignment_strategy': (<class 'kafka.coordinator.assignors.range.RangePartitionAssignor'>, <class 'kafka.coordinator.assignors.roundrobin.RoundRobinPartitionAssignor'>), 'receive_buffer_bytes': None, 'reconnect_backoff_max_ms': 1000, 'reconnect_backoff_ms': 50, 'request_timeout_ms': 305000, 'retry_backoff_ms': 100, 'sasl_kerberos_domain_name': None, 'sasl_kerberos_service_name': 'kafka', 'sasl_mechanism': None, 'sasl_oauth_token_provider': None, 'sasl_plain_password': None, 'sasl_plain_username': None, 'security_protocol': 'PLAINTEXT', 'selector': <class 'selectors.EpollSelector'>, 'send_buffer_bytes': None, 'session_timeout_ms': 10000, 'sock_chunk_buffer_count': 1000, 'sock_chunk_bytes': 4096, 'socket_options': [(6, 1, 1)], 'ssl_cafile': None, 'ssl_certfile': None, 'ssl_check_hostname': True, 'ssl_ciphers': None, 'ssl_context': None, 'ssl_crlfile': None, 'ssl_keyfile': None, 'ssl_password': None, 'value_deserializer': None}
  `DEFAULT_SESSION_TIMEOUT_MS_0_9` = 30000
  `assign`(partitions)

    Manually assign a list of TopicPartitions to this consumer.

    Arguments:
      partitions (list of TopicPartition): Assignment for this instance.
    Raises:
      IllegalStateError: If consumer has already called `subscribe()`.
    Warning:
      It is not possible to use both manual partition assignment with `assign()` and group assignment with `subscribe()`.
    Note:
      This interface does not support incremental assignment and will replace the previous assignment (if there was one).
    Note:
      Manual topic assignment through this method does not use the consumer’s group management functionality. As such, there will be no rebalance operation triggered when group membership or cluster and topic metadata change.
  `assignment`()

    Get the TopicPartitions currently assigned to this consumer.

    If partitions were directly assigned using `assign()`, then this will simply return the same partitions that were previously assigned. If topics were subscribed using `subscribe()`, then this will give the set of topic partitions currently assigned to the consumer (which may be None if the assignment hasn’t happened yet, or if the partitions are in the process of being reassigned).

    Returns:
      set: {TopicPartition, …}
  `beginning_offsets`(partitions)

    Get the first offset for the given partitions.

    This method does not change the current consumer position of the partitions.

    Note:
      This method may block indefinitely if the partition does not exist.
    Arguments:
      partitions (list): List of TopicPartition instances to fetch
        offsets for.
    Returns:
      `{TopicPartition: int}`: The earliest available offsets for the given partitions.
    Raises:
      UnsupportedVersionError: If the broker does not support looking
        up the offsets by timestamp.

      KafkaTimeoutError: If fetch failed in request_timeout_ms.

  `close`(autocommit=True)

    Close the consumer, waiting indefinitely for any needed cleanup.

    Keyword Arguments:
      autocommit (bool): If auto-commit is configured for this consumer,
        this optional flag causes the consumer to attempt to commit any pending consumed offsets prior to close. Default: True
  `commit`(offsets=None)

    Commit offsets to kafka, blocking until success or error.

    This commits offsets only to Kafka. The offsets committed using this API will be used on the first fetch after every rebalance and also on startup. As such, if you need to store offsets in anything other than Kafka, this API should not be used. To avoid re-processing the last message read if a consumer is restarted, the committed offset should be the next message your application should consume, i.e.: last_offset + 1.

    Blocks until either the commit succeeds or an unrecoverable error is encountered (in which case it is thrown to the caller).

    Currently only supports kafka-topic offset storage (not zookeeper).

    Arguments:
      offsets (dict, optional): {TopicPartition: OffsetAndMetadata} dict
        to commit with the configured group_id. Defaults to currently consumed offsets for all subscribed partitions.
  `commit_async`(offsets=None, callback=None)

    Commit offsets to kafka asynchronously, optionally firing callback.

    This commits offsets only to Kafka. The offsets committed using this API will be used on the first fetch after every rebalance and also on startup. As such, if you need to store offsets in anything other than Kafka, this API should not be used. To avoid re-processing the last message read if a consumer is restarted, the committed offset should be the next message your application should consume, i.e.: last_offset + 1.

    This is an asynchronous call and will not block. Any errors encountered are either passed to the callback (if provided) or discarded.

    Arguments:
      offsets (dict, optional): {TopicPartition: OffsetAndMetadata} dict
        to commit with the configured group_id. Defaults to currently consumed offsets for all subscribed partitions.
      callback (callable, optional): Called as callback(offsets, response)
        with response as either an Exception or an OffsetCommitResponse struct. This callback can be used to trigger custom actions when a commit request completes.
    Returns:
      kafka.future.Future
  `committed`(partition)

    Get the last committed offset for the given partition.

    This offset will be used as the position for the consumer in the event of a failure.

    This call may block to do a remote call if the partition in question isn’t assigned to this consumer or if the consumer hasn’t yet initialized its cache of committed offsets.

    Arguments:
      partition (TopicPartition): The partition to check.
    Returns:
      The last committed offset, or None if there was no prior commit.
  `configure`(**configs)
  `end_offsets`(partitions)

    Get the last offset for the given partitions. The last offset of a partition is the offset of the upcoming message, i.e. the offset of the last available message + 1.

    This method does not change the current consumer position of the partitions.

    Note:
      This method may block indefinitely if the partition does not exist.
    Arguments:
      partitions (list): List of TopicPartition instances to fetch
        offsets for.
    Returns:
      `{TopicPartition: int}`: The end offsets for the given partitions.
    Raises:
      UnsupportedVersionError: If the broker does not support looking
        up the offsets by timestamp.

      KafkaTimeoutError: If fetch failed in request_timeout_ms

  `fetch_messages`()
  `get_partition_offsets`(topic, partition, request_time_ms, max_num_offsets)
  `highwater`(partition)

    Last known highwater offset for a partition.

    A highwater offset is the offset that will be assigned to the next message that is produced. It may be useful for calculating lag, by comparing with the reported position. Note that both position and highwater refer to the next offset – i.e., highwater offset is one greater than the newest available message.

    Highwater offsets are returned in FetchResponse messages, so will not be available if no FetchRequests have been sent for this partition yet.

    Arguments:
      partition (TopicPartition): Partition to check
    Returns:
      int or None: Offset if available
  `metrics`(raw=False)

    Get metrics on consumer performance.

    This is ported from the Java Consumer, for details see: https://kafka.apache.org/documentation/#new_consumer_monitoring

    Warning:
      This is an unstable interface. It may change in future releases without warning.
  `offsets`(group=None)
  `offsets_for_times`(timestamps)

    Look up the offsets for the given partitions by timestamp. The returned offset for each partition is the earliest offset whose timestamp is greater than or equal to the given timestamp in the corresponding partition.

    This is a blocking call. The consumer does not have to be assigned the partitions.

    If the message format version in a partition is before 0.10.0, i.e. the messages do not have timestamps, `None` will be returned for that partition. `None` will also be returned for the partition if there are no messages in it.

    Note:
      This method may block indefinitely if the partition does not exist.
    Arguments:
      timestamps (dict): `{TopicPartition: int}` mapping from partition
        to the timestamp to look up. Unit should be milliseconds since beginning of the epoch (midnight Jan 1, 1970 (UTC))
    Returns:
      `{TopicPartition: OffsetAndTimestamp}`: mapping from partition to the timestamp and offset of the first message with timestamp greater than or equal to the target timestamp.
    Raises:

      ValueError: If the target timestamp is negative UnsupportedVersionError: If the broker does not support looking

      > up the offsets by timestamp.

      KafkaTimeoutError: If fetch failed in request_timeout_ms

  `partitions_for_topic`(topic)

    Get metadata about the partitions for a given topic.

    Arguments:
      topic (str): Topic to check.
    Returns:
      set: Partition ids
  `pause`(*partitions)

    Suspend fetching from the requested partitions.

    Future calls to `poll()` will not return any records from these partitions until they have been resumed using `resume()`.

    Note: This method does not affect partition subscription. In particular, it does not cause a group rebalance when automatic assignment is used.

    Arguments:
      *partitions (TopicPartition): Partitions to pause.
  `paused`()

    Get the partitions that were previously paused using `pause()`.

    Returns:
      set: {partition (TopicPartition), …}
  `poll`(timeout_ms=0, max_records=None)

    Fetch data from assigned topics / partitions.

    Records are fetched and returned in batches by topic-partition. On each poll, consumer will try to use the last consumed offset as the starting offset and fetch sequentially. The last consumed offset can be manually set through `seek()` or automatically set as the last committed offset for the subscribed list of partitions.

    Incompatible with iterator interface – use one or the other, not both.

    Arguments:
      timeout_ms (int, optional): Milliseconds spent waiting in poll if
        data is not available in the buffer. If 0, returns immediately with any records that are available currently in the buffer, else returns empty. Must not be negative. Default: 0
      max_records (int, optional): The maximum number of records returned
        in a single call to `poll()`. Default: Inherit value from max_poll_records.
    Returns:
      dict: Topic to list of records since the last fetch for the
        subscribed list of topics and partitions.
  `position`(partition)

    Get the offset of the next record that will be fetched

    Arguments:
      partition (TopicPartition): Partition to check
    Returns:
      int: Offset
  `resume`(*partitions)

    Resume fetching from the specified (paused) partitions.

    Arguments:
      *partitions (TopicPartition): Partitions to resume.
  `seek`(partition, offset)

    Manually specify the fetch offset for a TopicPartition.

    Overrides the fetch offsets that the consumer will use on the next `poll()`. If this API is invoked for the same partition more than once, the latest offset will be used on the next `poll()`.

    Note: You may lose data if this API is arbitrarily used in the middle of consumption to reset the fetch offsets.

    Arguments:
      partition (TopicPartition): Partition for seek operation offset (int): Message offset in partition
    Raises:
      AssertionError: If offset is not an int >= 0; or if partition is not
        currently assigned.
  `seek_to_beginning`(*partitions)

    Seek to the oldest available offset for partitions.

    Arguments:
      *partitions: Optionally provide specific TopicPartitions, otherwise
        default to all assigned partitions.
    Raises:
      AssertionError: If any partition is not currently assigned, or if
        no partitions are assigned.
  `seek_to_end`(*partitions)

    Seek to the most recent available offset for partitions.

    Arguments:
      *partitions: Optionally provide specific TopicPartitions, otherwise
        default to all assigned partitions.
    Raises:
      AssertionError: If any partition is not currently assigned, or if
        no partitions are assigned.
  `set_topic_partitions`(*topics)
  `subscribe`(topics=(), pattern=None, listener=None)

    Subscribe to a list of topics, or a topic regex pattern.

    Partitions will be dynamically assigned via a group coordinator. Topic subscriptions are not incremental: this list will replace the current assignment (if there is one).

    This method is incompatible with `assign()`.

    Arguments:

      topics (list): List of topics for subscription. pattern (str): Pattern to match available topics. You must provide

      > either topics or pattern, but not both.

      listener (ConsumerRebalanceListener): Optionally include listener

        callback, which will be called before and after each rebalance operation.

        As part of group management, the consumer will keep track of the list of consumers that belong to a particular group and will trigger a rebalance operation if one of the following events trigger:

        - Number of partitions change for any of the subscribed topics
        - Topic is created or deleted
        - An existing member of the consumer group dies
        - A new member is added to the consumer group

        When any of these events are triggered, the provided listener will be invoked first to indicate that the consumer’s assignment has been revoked, and then again when the new assignment has been received. Note that this listener will immediately override any listener set in a previous call to subscribe. It is guaranteed, however, that the partitions revoked/assigned through this interface are from topics subscribed in this call.

    Raises:
      IllegalStateError: If called after previously calling
        `assign()`.

      AssertionError: If neither topics or pattern is provided. TypeError: If listener is not a ConsumerRebalanceListener.

  `subscription`()

    Get the current topic subscription.

    Returns:
      set: {topic, …}
  `task_done`(message)
  `topics`()

    Get all topics the user is authorized to view.

    Returns:
      set: topics
  `unsubscribe`()

    Unsubscribe from all topics and clear all assigned partitions.

class `kafka.``KafkaProducer`(**configs)

  Bases: `object`

  A Kafka client that publishes records to the Kafka cluster.

  The producer is thread safe and sharing a single producer instance across threads will generally be faster than having multiple instances.

  The producer consists of a pool of buffer space that holds records that haven’t yet been transmitted to the server as well as a background I/O thread that is responsible for turning these records into requests and transmitting them to the cluster.

  `send()` is asynchronous. When called it adds the record to a buffer of pending record sends and immediately returns. This allows the producer to batch together individual records for efficiency.

  The ‘acks’ config controls the criteria under which requests are considered complete. The “all” setting will result in blocking on the full commit of the record, the slowest but most durable setting.

  If the request fails, the producer can automatically retry, unless ‘retries’ is configured to 0. Enabling retries also opens up the possibility of duplicates (see the documentation on message delivery semantics for details: https://kafka.apache.org/documentation.html#semantics ).

  The producer maintains buffers of unsent records for each partition. These buffers are of a size specified by the ‘batch_size’ config. Making this larger can result in more batching, but requires more memory (since we will generally have one of these buffers for each active partition).

  By default a buffer is available to send immediately even if there is additional unused space in the buffer. However if you want to reduce the number of requests you can set ‘linger_ms’ to something greater than 0. This will instruct the producer to wait up to that number of milliseconds before sending a request in hope that more records will arrive to fill up the same batch. This is analogous to Nagle’s algorithm in TCP. Note that records that arrive close together in time will generally batch together even with linger_ms=0 so under heavy load batching will occur regardless of the linger configuration; however setting this to something larger than 0 can lead to fewer, more efficient requests when not under maximal load at the cost of a small amount of latency.

  The buffer_memory controls the total amount of memory available to the producer for buffering. If records are sent faster than they can be transmitted to the server then this buffer space will be exhausted. When the buffer space is exhausted additional send calls will block.

  The key_serializer and value_serializer instruct how to turn the key and value objects the user provides into bytes.

  Keyword Arguments:
    bootstrap_servers: ‘host[:port]’ string (or list of ‘host[:port]’
      strings) that the producer should contact to bootstrap initial cluster metadata. This does not have to be the full node list. It just needs to have at least one broker that will respond to a Metadata API Request. Default port is 9092. If no servers are specified, will default to localhost:9092.
    client_id (str): a name for this client. This string is passed in
      each request to servers and can be used to identify specific server-side log entries that correspond to this client. Default: ‘kafka-python-producer-#’ (appended with a unique number per instance)
    key_serializer (callable): used to convert user-supplied keys to bytes
      If not None, called as f(key), should return bytes. Default: None.
    value_serializer (callable): used to convert user-supplied message
      values to bytes. If not None, called as f(value), should return bytes. Default: None.
    acks (0, 1, ‘all’): The number of acknowledgments the producer requires

      the leader to have received before considering a request complete. This controls the durability of records that are sent. The following settings are common:

      0: Producer will not wait for any acknowledgment from the server.
        The message will immediately be added to the socket buffer and considered sent. No guarantee can be made that the server has received the record in this case, and the retries configuration will not take effect (as the client won’t generally know of any failures). The offset given back for each record will always be set to -1.
      1: Wait for leader to write the record to its local log only.
        Broker will respond without awaiting full acknowledgement from all followers. In this case should the leader fail immediately after acknowledging the record but before the followers have replicated it then the record will be lost.
      all: Wait for the full set of in-sync replicas to write the record.
        This guarantees that the record will not be lost as long as at least one in-sync replica remains alive. This is the strongest available guarantee.

      If unset, defaults to acks=1.

    compression_type (str): The compression type for all data generated by
      the producer. Valid values are ‘gzip’, ‘snappy’, ‘lz4’, or None. Compression is of full batches of data, so the efficacy of batching will also impact the compression ratio (more batching means better compression). Default: None.
    retries (int): Setting a value greater than zero will cause the client
      to resend any record whose send fails with a potentially transient error. Note that this retry is no different than if the client resent the record upon receiving the error. Allowing retries without setting max_in_flight_requests_per_connection to 1 will potentially change the ordering of records because if two batches are sent to a single partition, and the first fails and is retried but the second succeeds, then the records in the second batch may appear first. Default: 0.
    batch_size (int): Requests sent to brokers will contain multiple
      batches, one for each partition with data available to be sent. A small batch size will make batching less common and may reduce throughput (a batch size of zero will disable batching entirely). Default: 16384
    linger_ms (int): The producer groups together any records that arrive
      in between request transmissions into a single batched request. Normally this occurs only under load when records arrive faster than they can be sent out. However in some circumstances the client may want to reduce the number of requests even under moderate load. This setting accomplishes this by adding a small amount of artificial delay; that is, rather than immediately sending out a record the producer will wait for up to the given delay to allow other records to be sent so that the sends can be batched together. This can be thought of as analogous to Nagle’s algorithm in TCP. This setting gives the upper bound on the delay for batching: once we get batch_size worth of records for a partition it will be sent immediately regardless of this setting, however if we have fewer than this many bytes accumulated for this partition we will ‘linger’ for the specified time waiting for more records to show up. This setting defaults to 0 (i.e. no delay). Setting linger_ms=5 would have the effect of reducing the number of requests sent but would add up to 5ms of latency to records sent in the absense of load. Default: 0.
    partitioner (callable): Callable used to determine which partition
      each message is assigned to. Called (after key serialization): partitioner(key_bytes, all_partitions, available_partitions). The default partitioner implementation hashes each non-None key using the same murmur2 algorithm as the java client so that messages with the same key are assigned to the same partition. When a key is None, the message is delivered to a random partition (filtered to partitions with available leaders only, if possible).
    buffer_memory (int): The total bytes of memory the producer should use
      to buffer records waiting to be sent to the server. If records are sent faster than they can be delivered to the server the producer will block up to max_block_ms, raising an exception on timeout. In the current implementation, this setting is an approximation. Default: 33554432 (32MB)
    connections_max_idle_ms: Close idle connections after the number of
      milliseconds specified by this config. The broker closes idle connections after connections.max.idle.ms, so this avoids hitting unexpected socket disconnected errors on the client. Default: 540000
    max_block_ms (int): Number of milliseconds to block during
      `send()` and `partitions_for()`. These methods can be blocked either because the buffer is full or metadata unavailable. Blocking in the user-supplied serializers or partitioner will not be counted against this timeout. Default: 60000.
    max_request_size (int): The maximum size of a request. This is also
      effectively a cap on the maximum record size. Note that the server has its own cap on record size which may be different from this. This setting will limit the number of record batches the producer will send in a single request to avoid sending huge requests. Default: 1048576.
    metadata_max_age_ms (int): The period of time in milliseconds after
      which we force a refresh of metadata even if we haven’t seen any partition leadership changes to proactively discover any new brokers or partitions. Default: 300000
    retry_backoff_ms (int): Milliseconds to backoff when retrying on
      errors. Default: 100.
    request_timeout_ms (int): Client request timeout in milliseconds.
      Default: 30000.
    receive_buffer_bytes (int): The size of the TCP receive buffer
      (SO_RCVBUF) to use when reading data. Default: None (relies on system defaults). Java client defaults to 32768.
    send_buffer_bytes (int): The size of the TCP send buffer
      (SO_SNDBUF) to use when sending data. Default: None (relies on system defaults). Java client defaults to 131072.
    socket_options (list): List of tuple-arguments to socket.setsockopt
      to apply to broker connection sockets. Default: [(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)]
    reconnect_backoff_ms (int): The amount of time in milliseconds to
      wait before attempting to reconnect to a given host. Default: 50.
    reconnect_backoff_max_ms (int): The maximum amount of time in
      milliseconds to wait when reconnecting to a broker that has repeatedly failed to connect. If provided, the backoff per host will increase exponentially for each consecutive connection failure, up to this maximum. To avoid connection storms, a randomization factor of 0.2 will be applied to the backoff resulting in a random range between 20% below and 20% above the computed value. Default: 1000.
    max_in_flight_requests_per_connection (int): Requests are pipelined
      to kafka brokers up to this number of maximum requests per broker connection. Note that if this setting is set to be greater than 1 and there are failed sends, there is a risk of message re-ordering due to retries (i.e., if retries are enabled). Default: 5.
    security_protocol (str): Protocol used to communicate with brokers.
      Valid values are: PLAINTEXT, SSL, SASL_PLAINTEXT, SASL_SSL. Default: PLAINTEXT.
    ssl_context (ssl.SSLContext): pre-configured SSLContext for wrapping
      socket connections. If provided, all other ssl_* configurations will be ignored. Default: None.
    ssl_check_hostname (bool): flag to configure whether ssl handshake
      should verify that the certificate matches the brokers hostname. default: true.
    ssl_cafile (str): optional filename of ca file to use in certificate
      veriication. default: none.
    ssl_certfile (str): optional filename of file in pem format containing
      the client certificate, as well as any ca certificates needed to establish the certificate’s authenticity. default: none.
    ssl_keyfile (str): optional filename containing the client private key.
      default: none.
    ssl_password (str): optional password to be used when loading the
      certificate chain. default: none.
    ssl_crlfile (str): optional filename containing the CRL to check for
      certificate expiration. By default, no CRL check is done. When providing a file, only the leaf certificate will be checked against this CRL. The CRL can only be checked with Python 3.4+ or 2.7.9+. default: none.
    ssl_ciphers (str): optionally set the available ciphers for ssl
      connections. It should be a string in the OpenSSL cipher list format. If no cipher can be selected (because compile-time options or other configuration forbids use of all the specified ciphers), an ssl.SSLError will be raised. See ssl.SSLContext.set_ciphers
    api_version (tuple): Specify which Kafka API version to use. If set to
      None, the client will attempt to infer the broker version by probing various APIs. Example: (0, 10, 2). Default: None
    api_version_auto_timeout_ms (int): number of milliseconds to throw a
      timeout exception from the constructor when checking the broker api version. Only applies if api_version set to ‘auto’
    metric_reporters (list): A list of classes to use as metrics reporters.
      Implementing the AbstractMetricsReporter interface allows plugging in classes that will be notified of new metric creation. Default: []
    metrics_num_samples (int): The number of samples maintained to compute
      metrics. Default: 2
    metrics_sample_window_ms (int): The maximum age in milliseconds of
      samples used to compute metrics. Default: 30000
    selector (selectors.BaseSelector): Provide a specific selector
      implementation to use for I/O multiplexing. Default: selectors.DefaultSelector
    sasl_mechanism (str): Authentication mechanism when security_protocol
      is configured for SASL_PLAINTEXT or SASL_SSL. Valid values are: PLAIN, GSSAPI, OAUTHBEARER.
    sasl_plain_username (str): username for sasl PLAIN authentication.
      Required if sasl_mechanism is PLAIN.
    sasl_plain_password (str): password for sasl PLAIN authentication.
      Required if sasl_mechanism is PLAIN.
    sasl_kerberos_service_name (str): Service name to include in GSSAPI
      sasl mechanism handshake. Default: ‘kafka’
    sasl_kerberos_domain_name (str): kerberos domain name to use in GSSAPI
      sasl mechanism handshake. Default: one of bootstrap servers
    sasl_oauth_token_provider (AbstractTokenProvider): OAuthBearer token provider
      instance. (See kafka.oauth.abstract). Default: None
  Note:
    Configuration parameters are described in more detail at https://kafka.apache.org/0100/configuration.html#producerconfigs
  `DEFAULT_CONFIG` = {'acks': 1, 'api_version': None, 'api_version_auto_timeout_ms': 2000, 'batch_size': 16384, 'bootstrap_servers': 'localhost', 'bootstrap_topics_filter': {}, 'buffer_memory': 33554432, 'client_id': None, 'compression_type': None, 'connections_max_idle_ms': 540000, 'key_serializer': None, 'linger_ms': 0, 'max_block_ms': 60000, 'max_in_flight_requests_per_connection': 5, 'max_request_size': 1048576, 'metadata_max_age_ms': 300000, 'metric_reporters': [], 'metrics_num_samples': 2, 'metrics_sample_window_ms': 30000, 'partitioner': <kafka.partitioner.default.DefaultPartitioner object>, 'receive_buffer_bytes': None, 'reconnect_backoff_max_ms': 1000, 'reconnect_backoff_ms': 50, 'request_timeout_ms': 30000, 'retries': 0, 'retry_backoff_ms': 100, 'sasl_kerberos_domain_name': None, 'sasl_kerberos_service_name': 'kafka', 'sasl_mechanism': None, 'sasl_oauth_token_provider': None, 'sasl_plain_password': None, 'sasl_plain_username': None, 'security_protocol': 'PLAINTEXT', 'selector': <class 'selectors.EpollSelector'>, 'send_buffer_bytes': None, 'sock_chunk_buffer_count': 1000, 'sock_chunk_bytes': 4096, 'socket_options': [(6, 1, 1)], 'ssl_cafile': None, 'ssl_certfile': None, 'ssl_check_hostname': True, 'ssl_ciphers': None, 'ssl_context': None, 'ssl_crlfile': None, 'ssl_keyfile': None, 'ssl_password': None, 'value_serializer': None}
  `close`(timeout=None)

    Close this producer.

    Arguments:
      timeout (float, optional): timeout in seconds to wait for completion.
  `flush`(timeout=None)

    Invoking this method makes all buffered records immediately available to send (even if linger_ms is greater than 0) and blocks on the completion of the requests associated with these records. The post-condition of `flush()` is that any previously sent record will have completed (e.g. Future.is_done() == True). A request is considered completed when either it is successfully acknowledged according to the ‘acks’ configuration for the producer, or it results in an error.

    Other threads can continue sending messages while one thread is blocked waiting for a flush call to complete; however, no guarantee is made about the completion of messages sent after the flush call begins.

    Arguments:
      timeout (float, optional): timeout in seconds to wait for completion.
    Raises:
      KafkaTimeoutError: failure to flush buffered records within the
        provided timeout
  `metrics`(raw=False)

    Get metrics on producer performance.

    This is ported from the Java Producer, for details see: https://kafka.apache.org/documentation/#producer_monitoring

    Warning:
      This is an unstable interface. It may change in future releases without warning.
  `partitions_for`(topic)

    Returns set of all known partitions for the topic.

  `send`(topic, value=None, key=None, headers=None, partition=None, timestamp_ms=None)

    Publish a message to a topic.

    Arguments:

      topic (str): topic where the message will be published value (optional): message value. Must be type bytes, or be

      > serializable to bytes via configured value_serializer. If value is None, key is required and message acts as a ‘delete’. See kafka compaction documentation for more details: https://kafka.apache.org/documentation.html#compaction (compaction requires kafka >= 0.8.1)

      partition (int, optional): optionally specify a partition. If not
        set, the partition will be selected using the configured ‘partitioner’.
      key (optional): a key to associate with the message. Can be used to
        determine which partition to send the message to. If partition is None (and producer’s partitioner config is left as default), then messages with the same key will be delivered to the same partition (but if key is None, partition is chosen randomly). Must be type bytes, or be serializable to bytes via configured key_serializer.
      headers (optional): a list of header key value pairs. List items
        are tuples of str key and bytes value.
      timestamp_ms (int, optional): epoch milliseconds (from Jan 1 1970 UTC)
        to use as the message timestamp. Defaults to current time.
    Returns:
      FutureRecordMetadata: resolves to RecordMetadata
    Raises:
      KafkaTimeoutError: if unable to fetch topic metadata, or unable
        to obtain memory buffer prior to configured max_block_ms

class `kafka.``KafkaClient`(*args, **kwargs)

  Bases: `kafka.client.SimpleClient`

class `kafka.``BrokerConnection`(host, port, afi, **configs)

  Bases: `object`

  Initialize a Kafka broker connection

  Keyword Arguments:
    client_id (str): a name for this client. This string is passed in
      each request to servers and can be used to identify specific server-side log entries that correspond to this client. Also submitted to GroupCoordinator for logging with respect to consumer group administration. Default: ‘kafka-python-{version}’
    reconnect_backoff_ms (int): The amount of time in milliseconds to
      wait before attempting to reconnect to a given host. Default: 50.
    reconnect_backoff_max_ms (int): The maximum amount of time in
      milliseconds to wait when reconnecting to a broker that has repeatedly failed to connect. If provided, the backoff per host will increase exponentially for each consecutive connection failure, up to this maximum. To avoid connection storms, a randomization factor of 0.2 will be applied to the backoff resulting in a random range between 20% below and 20% above the computed value. Default: 1000.
    request_timeout_ms (int): Client request timeout in milliseconds.
      Default: 30000.
    max_in_flight_requests_per_connection (int): Requests are pipelined
      to kafka brokers up to this number of maximum requests per broker connection. Default: 5.
    receive_buffer_bytes (int): The size of the TCP receive buffer
      (SO_RCVBUF) to use when reading data. Default: None (relies on system defaults). Java client defaults to 32768.
    send_buffer_bytes (int): The size of the TCP send buffer
      (SO_SNDBUF) to use when sending data. Default: None (relies on system defaults). Java client defaults to 131072.
    socket_options (list): List of tuple-arguments to socket.setsockopt
      to apply to broker connection sockets. Default: [(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)]
    security_protocol (str): Protocol used to communicate with brokers.
      Valid values are: PLAINTEXT, SSL, SASL_PLAINTEXT, SASL_SSL. Default: PLAINTEXT.
    ssl_context (ssl.SSLContext): pre-configured SSLContext for wrapping
      socket connections. If provided, all other ssl_* configurations will be ignored. Default: None.
    ssl_check_hostname (bool): flag to configure whether ssl handshake
      should verify that the certificate matches the brokers hostname. default: True.
    ssl_cafile (str): optional filename of ca file to use in certificate
      verification. default: None.
    ssl_certfile (str): optional filename of file in pem format containing
      the client certificate, as well as any ca certificates needed to establish the certificate’s authenticity. default: None.
    ssl_keyfile (str): optional filename containing the client private key.
      default: None.
    ssl_password (callable, str, bytes, bytearray): optional password or
      callable function that returns a password, for decrypting the client private key. Default: None.
    ssl_crlfile (str): optional filename containing the CRL to check for
      certificate expiration. By default, no CRL check is done. When providing a file, only the leaf certificate will be checked against this CRL. The CRL can only be checked with Python 3.4+ or 2.7.9+. default: None.
    ssl_ciphers (str): optionally set the available ciphers for ssl
      connections. It should be a string in the OpenSSL cipher list format. If no cipher can be selected (because compile-time options or other configuration forbids use of all the specified ciphers), an ssl.SSLError will be raised. See ssl.SSLContext.set_ciphers
    api_version (tuple): Specify which Kafka API version to use.
      Accepted values are: (0, 8, 0), (0, 8, 1), (0, 8, 2), (0, 9), (0, 10). Default: (0, 8, 2)
    api_version_auto_timeout_ms (int): number of milliseconds to throw a
      timeout exception from the constructor when checking the broker api version. Only applies if api_version is None
    selector (selectors.BaseSelector): Provide a specific selector
      implementation to use for I/O multiplexing. Default: selectors.DefaultSelector
    state_change_callback (callable): function to be called when the
      connection state changes from CONNECTING to CONNECTED etc.
    metrics (kafka.metrics.Metrics): Optionally provide a metrics
      instance for capturing network IO stats. Default: None.

    metric_group_prefix (str): Prefix for metric names. Default: ‘’ sasl_mechanism (str): Authentication mechanism when security_protocol

    > is configured for SASL_PLAINTEXT or SASL_SSL. Valid values are: PLAIN, GSSAPI, OAUTHBEARER.

    sasl_plain_username (str): username for sasl PLAIN authentication.
      Required if sasl_mechanism is PLAIN.
    sasl_plain_password (str): password for sasl PLAIN authentication.
      Required if sasl_mechanism is PLAIN.
    sasl_kerberos_service_name (str): Service name to include in GSSAPI
      sasl mechanism handshake. Default: ‘kafka’
    sasl_kerberos_domain_name (str): kerberos domain name to use in GSSAPI
      sasl mechanism handshake. Default: one of bootstrap servers
    sasl_oauth_token_provider (AbstractTokenProvider): OAuthBearer token provider
      instance. (See kafka.oauth.abstract). Default: None
  `DEFAULT_CONFIG` = {'api_version': (0, 8, 2), 'client_id': 'kafka-python-1.4.6', 'max_in_flight_requests_per_connection': 5, 'metric_group_prefix': '', 'metrics': None, 'node_id': 0, 'receive_buffer_bytes': None, 'reconnect_backoff_max_ms': 1000, 'reconnect_backoff_ms': 50, 'request_timeout_ms': 30000, 'sasl_kerberos_domain_name': None, 'sasl_kerberos_service_name': 'kafka', 'sasl_mechanism': None, 'sasl_oauth_token_provider': None, 'sasl_plain_password': None, 'sasl_plain_username': None, 'security_protocol': 'PLAINTEXT', 'selector': <class 'selectors.EpollSelector'>, 'send_buffer_bytes': None, 'sock_chunk_buffer_count': 1000, 'sock_chunk_bytes': 4096, 'socket_options': [(6, 1, 1)], 'ssl_cafile': None, 'ssl_certfile': None, 'ssl_check_hostname': True, 'ssl_ciphers': None, 'ssl_context': None, 'ssl_crlfile': None, 'ssl_keyfile': None, 'ssl_password': None, 'state_change_callback': <function BrokerConnection.<lambda>>}
  `SASL_MECHANISMS` = ('PLAIN', 'GSSAPI', 'OAUTHBEARER')
  `SECURITY_PROTOCOLS` = ('PLAINTEXT', 'SSL', 'SASL_PLAINTEXT', 'SASL_SSL')
  `blacked_out`()

    Return true if we are disconnected from the given node and can’t re-establish a connection yet

  `can_send_more`()

    Return True unless there are max_in_flight_requests_per_connection.

  `check_version`(timeout=2, strict=False, topics=[])

    Attempt to guess the broker version.

    Note: This is a blocking call.

    Returns: version tuple, i.e. (0, 10), (0, 9), (0, 8, 2), …

  `close`(error=None)

    Close socket and fail all in-flight-requests.

    Arguments:
      error (Exception, optional): pending in-flight-requests
        will be failed with this exception. Default: kafka.errors.KafkaConnectionError.
  `connect`()

    Attempt to connect and return ConnectionState

  `connect_blocking`(timeout=inf)
  `connected`()

    Return True iff socket is connected.

  `connecting`()

    Returns True if still connecting (this may encompass several different states, such as SSL handshake, authorization, etc).

  `connection_delay`()

    Return the number of milliseconds to wait, based on the connection state, before attempting to send data. When disconnected, this respects the reconnect backoff time. When connecting, returns 0 to allow non-blocking connect to finish. When connected, returns a very large number to handle slow/stalled connections.

  `disconnected`()

    Return True iff socket is closed

  `get_api_versions`()
  `recv`()

    Non-blocking network receive.

    Return list of (response, future) tuples

  `requests_timed_out`()
  `send`(request, blocking=True)

    Queue request for async network send, return Future()

  `send_pending_requests`()

    Can block on network if request is larger than send_buffer_bytes

class `kafka.``SimpleClient`(hosts, client_id=b'kafka-python', timeout=120, correlation_id=0)

  Bases: `object`

  `CLIENT_ID` = b'kafka-python'
  `DEFAULT_SOCKET_TIMEOUT_SECONDS` = 120
  `close`()
  `copy`()

    Create an inactive copy of the client object, suitable for passing to a separate thread.

    Note that the copied connections are not initialized, so `reinit()` must be called on the returned copy.

  `ensure_topic_exists`(topic, timeout=30)
  `get_partition_ids_for_topic`(topic)
  `has_metadata_for_topic`(topic)
  `load_metadata_for_topics`(*topics, **kwargs)

    Fetch broker and topic-partition metadata from the server.

    Updates internal data: broker list, topic/partition list, and topic/partition -> broker map. This method should be called after receiving any error.

    Note: Exceptions will not be raised in a full refresh (i.e. no topic list). In this case, error codes will be logged as errors. Partition-level errors will also not be raised here (a single partition w/o a leader, for example).

    Arguments:
      *topics (optional): If a list of topics is provided,
        the metadata refresh will be limited to the specified topics only.
      ignore_leadernotavailable (bool): suppress LeaderNotAvailableError
        so that metadata is loaded correctly during auto-create. Default: False.
    Raises:
      UnknownTopicOrPartitionError: Raised for topics that do not exist,
        unless the broker is configured to auto-create topics.
      LeaderNotAvailableError: Raised for topics that do not exist yet,
        when the broker is configured to auto-create topics. Retry after a short backoff (topics/partitions are initializing).
  `reinit`()
  `reset_all_metadata`()
  `reset_topic_metadata`(*topics)
  `send_consumer_metadata_request`(payloads=(), fail_on_error=True, callback=None)
  `send_fetch_request`(payloads=(), fail_on_error=True, callback=None, max_wait_time=100, min_bytes=4096)

    Encode and send a FetchRequest

    Payloads are grouped by topic and partition so they can be pipelined to the same brokers.

  `send_list_offset_request`(payloads=(), fail_on_error=True, callback=None)
  `send_metadata_request`(payloads=(), fail_on_error=True, callback=None)
  `send_offset_commit_request`(group, payloads=(), fail_on_error=True, callback=None)
  `send_offset_fetch_request`(group, payloads=(), fail_on_error=True, callback=None)
  `send_offset_fetch_request_kafka`(group, payloads=(), fail_on_error=True, callback=None)
  `send_offset_request`(payloads=(), fail_on_error=True, callback=None)
  `send_produce_request`(payloads=(), acks=1, timeout=1000, fail_on_error=True, callback=None)

    Encode and send some ProduceRequests

    ProduceRequests will be grouped by (topic, partition) and then sent to a specific broker. Output is a list of responses in the same order as the list of payloads specified

    Arguments:
      payloads (list of ProduceRequest): produce requests to send to kafka
        ProduceRequest payloads must not contain duplicates for any topic-partition.
      acks (int, optional): how many acks the servers should receive from replica
        brokers before responding to the request. If it is 0, the server will not send any response. If it is 1, the server will wait until the data is written to the local log before sending a response. If it is -1, the server will wait until the message is committed by all in-sync replicas before sending a response. For any value > 1, the server will wait for this number of acks to occur (but the server will never wait for more acknowledgements than there are in-sync replicas). defaults to 1.
      timeout (int, optional): maximum time in milliseconds the server can
        await the receipt of the number of acks, defaults to 1000.
      fail_on_error (bool, optional): raise exceptions on connection and
        server response errors, defaults to True.
      callback (function, optional): instead of returning the ProduceResponse,
        first pass it through this function, defaults to None.
    Returns:
      list of ProduceResponses, or callback results if supplied, in the order of input payloads
  `topics`

class `kafka.``SimpleProducer`(*args, **kwargs)

  Bases: `kafka.producer.base.Producer`

  A simple, round-robin producer.

  See Producer class for Base Arguments

  Additional Arguments:
    random_start (bool, optional): randomize the initial partition which
      the first message block will be published to, otherwise if false, the first message block will always publish to partition 0 before cycling through each partition, defaults to True.
  `send_messages`(topic, *msg)

    Helper method to send produce requests.

    Note that msg type must be encoded to bytes by user. Passing unicode message will not work, for example you should encode before calling send_messages via something like unicode_message.encode(‘utf-8’) All messages will set the message ‘key’ to None.

    Arguments:
      topic (str): name of topic for produce request partition (int): partition number for produce request *msg (bytes): one or more message payloads
    Returns:
      ResponseRequest returned by server
    Raises:
      FailedPayloadsError: low-level connection error, can be caused by
        networking failures, or a malformed request.
      KafkaUnavailableError: all known brokers are down when attempting
        to refresh metadata.
      LeaderNotAvailableError: topic or partition is initializing or
        a broker failed and leadership election is in progress.
      NotLeaderForPartitionError: metadata is out of sync; the broker
        that the request was sent to is not the leader for the topic or partition.
      UnknownTopicOrPartitionError: the topic or partition has not
        been created yet and auto-creation is not available.
      AsyncProducerQueueFull: in async mode, if too many messages are
        unsent and remain in the internal queue.

class `kafka.``KeyedProducer`(*args, **kwargs)

  Bases: `kafka.producer.base.Producer`

  A producer which distributes messages to partitions based on the key

  See Producer class for Arguments

  Additional Arguments:
    partitioner: A partitioner class that will be used to get the partition
      to send the message to. Must be derived from Partitioner. Defaults to HashedPartitioner.
  `send`(topic, key, msg)
  `send_messages`(topic, key, *msg)

    Helper method to send produce requests.

    Note that msg type must be encoded to bytes by user. Passing unicode message will not work, for example you should encode before calling send_messages via something like unicode_message.encode(‘utf-8’) All messages will set the message ‘key’ to None.

    Arguments:
      topic (str): name of topic for produce request partition (int): partition number for produce request *msg (bytes): one or more message payloads
    Returns:
      ResponseRequest returned by server
    Raises:
      FailedPayloadsError: low-level connection error, can be caused by
        networking failures, or a malformed request.
      KafkaUnavailableError: all known brokers are down when attempting
        to refresh metadata.
      LeaderNotAvailableError: topic or partition is initializing or
        a broker failed and leadership election is in progress.
      NotLeaderForPartitionError: metadata is out of sync; the broker
        that the request was sent to is not the leader for the topic or partition.
      UnknownTopicOrPartitionError: the topic or partition has not
        been created yet and auto-creation is not available.
      AsyncProducerQueueFull: in async mode, if too many messages are
        unsent and remain in the internal queue.

class `kafka.``RoundRobinPartitioner`(partitions=None)

  Bases: `kafka.partitioner.base.Partitioner`

  `partition`(key, all_partitions=None, available_partitions=None)

`kafka.``HashedPartitioner`

  alias of `kafka.partitioner.hashed.LegacyPartitioner`

`kafka.``create_message`(payload, key=None)

  Construct a Message

  Arguments:
    payload: bytes, the payload to send to Kafka key: bytes, a key used for partition routing (optional)

`kafka.``create_gzip_message`(payloads, key=None, compresslevel=None)

  Construct a Gzipped Message containing multiple Messages

  The given payloads will be encoded, compressed, and sent as a single atomic message to Kafka.

  Arguments:
    payloads: list(bytes), a list of payload to send be sent to Kafka key: bytes, a key used for partition routing (optional)

`kafka.``create_snappy_message`(payloads, key=None)

  Construct a Snappy Message containing multiple Messages

  The given payloads will be encoded, compressed, and sent as a single atomic message to Kafka.

  Arguments:
    payloads: list(bytes), a list of payload to send be sent to Kafka key: bytes, a key used for partition routing (optional)

class `kafka.``SimpleConsumer`(client, group, topic, auto_commit=True, partitions=None, auto_commit_every_n=100, auto_commit_every_t=5000, fetch_size_bytes=4096, buffer_size=4096, max_buffer_size=32768, iter_timeout=None, auto_offset_reset='largest')

  Bases: `kafka.consumer.base.Consumer`

  A simple consumer implementation that consumes all/specified partitions for a topic

  Arguments:

    client: a connected SimpleClient group: a name for this consumer, used for offset storage and must be unique

    > If you are connecting to a server that does not support offset commit/fetch (any prior to 0.8.1.1), then you must set this to None

    topic: the topic to consume

  Keyword Arguments:

    partitions: An optional list of partitions to consume the data from

    auto_commit: default True. Whether or not to auto commit the offsets

    auto_commit_every_n: default 100. How many messages to consume
      before a commit
    auto_commit_every_t: default 5000. How much time (in milliseconds) to
      wait before commit

    fetch_size_bytes: number of bytes to request in a FetchRequest

    buffer_size: default 4K. Initial number of bytes to tell kafka we
      have available. This will double as needed.
    max_buffer_size: default 16K. Max number of bytes to tell kafka we have
      available. None means no limit.
    iter_timeout: default None. How much time (in seconds) to wait for a
      message in the iterator before exiting. None means no timeout, so it will wait forever.
    auto_offset_reset: default largest. Reset partition offsets upon
      OffsetOutOfRangeError. Valid values are largest and smallest. Otherwise, do not reset the offsets and raise OffsetOutOfRangeError.

  Auto commit details: If both auto_commit_every_n and auto_commit_every_t are set, they will reset one another when one is triggered. These triggers simply call the commit method on this class. A manual call to commit will also reset these triggers

  `get_message`(block=True, timeout=0.1, get_partition_info=None)
  `get_messages`(count=1, block=True, timeout=0.1)

    Fetch the specified number of messages

    Keyword Arguments:

      count: Indicates the maximum number of messages to be fetched block: If True, the API will block till all messages are fetched.

      > If block is a positive integer the API will block until that many messages are fetched.

      timeout: When blocking is requested the function will block for
        the specified time (in seconds) until count messages is fetched. If None, it will block forever.
  `reset_partition_offset`(partition)

    Update offsets using auto_offset_reset policy (smallest|largest)

    Arguments:
      partition (int): the partition for which offsets should be updated

    Returns: Updated offset on success, None on failure

  `seek`(offset, whence=None, partition=None)

    Alter the current offset in the consumer, similar to fseek

    Arguments:

      offset: how much to modify the offset whence: where to modify it from, default is None

      > - None is an absolute offset
      > - 0 is relative to the earliest available offset (head)
      > - 1 is relative to the current offset
      > - 2 is relative to the latest known offset (tail)

      partition: modify which partition, default is None.
        If partition is None, would modify all partitions.

class `kafka.``MultiProcessConsumer`(client, group, topic, partitions=None, auto_commit=True, auto_commit_every_n=100, auto_commit_every_t=5000, num_procs=1, partitions_per_proc=0, **simple_consumer_options)

  Bases: `kafka.consumer.base.Consumer`

  A consumer implementation that consumes partitions for a topic in parallel using multiple processes

  Arguments:

    client: a connected SimpleClient group: a name for this consumer, used for offset storage and must be unique

    > If you are connecting to a server that does not support offset commit/fetch (any prior to 0.8.1.1), then you must set this to None

    topic: the topic to consume

  Keyword Arguments:

    partitions: An optional list of partitions to consume the data from auto_commit: default True. Whether or not to auto commit the offsets auto_commit_every_n: default 100. How many messages to consume

    > before a commit

    auto_commit_every_t: default 5000. How much time (in milliseconds) to
      wait before commit
    num_procs: Number of processes to start for consuming messages.
      The available partitions will be divided among these processes
    partitions_per_proc: Number of partitions to be allocated per process
      (overrides num_procs)

  Auto commit details: If both auto_commit_every_n and auto_commit_every_t are set, they will reset one another when one is triggered. These triggers simply call the commit method on this class. A manual call to commit will also reset these triggers

  `get_messages`(count=1, block=True, timeout=10)

    Fetch the specified number of messages

    Keyword Arguments:

      count: Indicates the maximum number of messages to be fetched block: If True, the API will block till all messages are fetched.

      > If block is a positive integer the API will block until that many messages are fetched.

      timeout: When blocking is requested the function will block for
        the specified time (in seconds) until count messages is fetched. If None, it will block forever.
  `stop`()

class `kafka.``ConsumerRebalanceListener`

  Bases: `object`

  A callback interface that the user can implement to trigger custom actions when the set of partitions assigned to the consumer changes.

  This is applicable when the consumer is having Kafka auto-manage group membership. If the consumer’s directly assign partitions, those partitions will never be reassigned and this callback is not applicable.

  When Kafka is managing the group membership, a partition re-assignment will be triggered any time the members of the group changes or the subscription of the members changes. This can occur when processes die, new process instances are added or old instances come back to life after failure. Rebalances can also be triggered by changes affecting the subscribed topics (e.g. when then number of partitions is administratively adjusted).

  There are many uses for this functionality. One common use is saving offsets in a custom store. By saving offsets in the on_partitions_revoked(), call we can ensure that any time partition assignment changes the offset gets saved.

  Another use is flushing out any kind of cache of intermediate results the consumer may be keeping. For example, consider a case where the consumer is subscribed to a topic containing user page views, and the goal is to count the number of page views per users for each five minute window. Let’s say the topic is partitioned by the user id so that all events for a particular user will go to a single consumer instance. The consumer can keep in memory a running tally of actions per user and only flush these out to a remote data store when its cache gets too big. However if a partition is reassigned it may want to automatically trigger a flush of this cache, before the new owner takes over consumption.

  This callback will execute in the user thread as part of the Consumer.poll() whenever partition assignment changes.

  It is guaranteed that all consumer processes will invoke on_partitions_revoked() prior to any process invoking on_partitions_assigned(). So if offsets or other state is saved in the on_partitions_revoked() call, it should be saved by the time the process taking over that partition has their on_partitions_assigned() callback called to load the state.

  `on_partitions_assigned`(assigned)

    A callback method the user can implement to provide handling of customized offsets on completion of a successful partition re-assignment. This method will be called after an offset re-assignment completes and before the consumer starts fetching data.

    It is guaranteed that all the processes in a consumer group will execute their on_partitions_revoked() callback before any instance executes its on_partitions_assigned() callback.

    Arguments:
      assigned (list of TopicPartition): the partitions assigned to the
        consumer (may include partitions that were previously assigned)
  `on_partitions_revoked`(revoked)

    A callback method the user can implement to provide handling of offset commits to a customized store on the start of a rebalance operation. This method will be called before a rebalance operation starts and after the consumer stops fetching data. It is recommended that offsets should be committed in this callback to either Kafka or a custom offset store to prevent duplicate data.

    NOTE: This method is only called before rebalances. It is not called prior to KafkaConsumer.close()

    Arguments:
      revoked (list of TopicPartition): the partitions that were assigned
        to the consumer on the last rebalance