# BrokerConnection

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

  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 backoff/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. Once the maximum is reached, reconnection attempts will continue periodically with this fixed rate. 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, SCRAM-SHA-256, SCRAM-SHA-512.

    sasl_plain_username (str): username for sasl PLAIN and SCRAM authentication.

      Required if sasl_mechanism is PLAIN or one of the SCRAM mechanisms.

    sasl_plain_password (str): password for sasl PLAIN and SCRAM authentication.

      Required if sasl_mechanism is PLAIN or one of the SCRAM mechanisms.

    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

  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

  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 or connected, returns a very large number to handle slow/stalled connections.

  disconnected()

    Return True iff socket is closed

  recv()

    Non-blocking network receive.

    Return list of (response, future) tuples

  send(request, blocking=True)

    Queue request for async network send, return Future()

  send_pending_requests()

    Attempts to send pending requests messages via blocking IO If all requests have been sent, return True Otherwise, if the socket is blocked and there are more bytes to send, return False.

  send_pending_requests_v2()

    Attempts to send pending requests messages via non-blocking IO If all requests have been sent, return True Otherwise, if the socket is blocked and there are more bytes to send, return False.