KafkaClient¶
-
class
kafka.client.
KafkaClient
(**configs)[source]¶ A network client for asynchronous request/response network I/O.
This is an internal class used to implement the user-facing producer and consumer clients.
This class is not thread-safe!
- Attributes:
- cluster (ClusterMetadata): Local cache of cluster metadata, retrived
- via MetadataRequests during
poll()
.
- 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.
- request_timeout_ms (int): Client request timeout in milliseconds.
- Default: 40000.
- 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 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.
- 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. For the full list of supported versions, see KafkaClient.API_VERSIONS. 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): string picking sasl mechanism when security_protocol
is SASL_PLAINTEXT or SASL_SSL. Currently only PLAIN is supported. Default: None- sasl_plain_username (str): username for sasl PLAIN authentication.
- Default: None
- sasl_plain_password (str): password for sasl PLAIN authentication.
- Default: None
-
add_topic
(topic)[source]¶ Add a topic to the list of topics tracked via metadata.
- Arguments:
- topic (str): topic to track
- Returns:
- Future: resolves after metadata request/response
-
check_version
(node_id=None, timeout=2, strict=False)[source]¶ Attempt to guess the version of a Kafka broker.
- Note: It is possible that this method blocks longer than the
- specified timeout. This can happen if the entire cluster is down and the client enters a bootstrap backoff sleep. This is only possible if node_id is None.
Returns: version tuple, i.e. (0, 10), (0, 9), (0, 8, 2), …
- Raises:
- NodeNotReadyError (if node_id is provided) NoBrokersAvailable (if node_id is None) UnrecognizedBrokerVersion: please file bug if seen! AssertionError (if strict=True): please file bug if seen!
-
close
(node_id=None)[source]¶ Close one or all broker connections.
- Arguments:
- node_id (int, optional): the id of the node to close
-
connection_delay
(node_id)[source]¶ 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.
- Arguments:
- node_id (int): The id of the node to check
- Returns:
- int: The number of milliseconds to wait.
-
in_flight_request_count
(node_id=None)[source]¶ Get the number of in-flight requests for a node or all nodes.
- Arguments:
- node_id (int, optional): a specific node to check. If unspecified,
- return the total for all nodes
- Returns:
- int: pending in-flight requests for the node, or all nodes if None
-
is_disconnected
(node_id)[source]¶ Check whether the node connection has been disconnected or failed.
A disconnected node has either been closed or has failed. Connection failures are usually transient and can be resumed in the next ready() call, but there are cases where transient failures need to be caught and re-acted upon.
- Arguments:
- node_id (int): the id of the node to check
- Returns:
- bool: True iff the node exists and is disconnected
-
is_ready
(node_id, metadata_priority=True)[source]¶ Check whether a node is ready to send more requests.
In addition to connection-level checks, this method also is used to block additional requests from being sent during a metadata refresh.
- Arguments:
node_id (int): id of the node to check metadata_priority (bool): Mark node as not-ready if a metadata
refresh is required. Default: True- Returns:
- bool: True if the node is ready and metadata is not refreshing
-
least_loaded_node
()[source]¶ Choose the node with fewest outstanding requests, with fallbacks.
This method will prefer a node with an existing connection, but will potentially choose a node for which we don’t yet have a connection if all existing connections are in use. This method will never choose a node that was disconnected within the reconnect backoff period. If all else fails, the method will attempt to bootstrap again using the bootstrap_servers list.
- Returns:
- node_id or None if no suitable node was found
-
poll
(timeout_ms=None, future=None, sleep=True, delayed_tasks=True)[source]¶ Try to read and write to sockets.
This method will also attempt to complete node connections, refresh stale metadata, and run previously-scheduled tasks.
- Arguments:
- timeout_ms (int, optional): maximum amount of time to wait (in ms)
- for at least one response. Must be non-negative. The actual timeout will be the minimum of timeout, request timeout and metadata timeout. Default: request_timeout_ms
future (Future, optional): if provided, blocks until future.is_done sleep (bool): if True and there is nothing to do (no connections
or requests in flight), will sleep for duration timeout before returning empty results. Default: False.- Returns:
- list: responses received (can be empty)
-
ready
(node_id, metadata_priority=True)[source]¶ Check whether a node is connected and ok to send more requests.
- Arguments:
node_id (int): the id of the node to check metadata_priority (bool): Mark node as not-ready if a metadata
refresh is required. Default: True- Returns:
- bool: True if we are ready to send to the given node
-
schedule
(task, at)[source]¶ Schedule a new task to be executed at the given time.
This is “best-effort” scheduling and should only be used for coarse synchronization. A task cannot be scheduled for multiple times simultaneously; any previously scheduled instance of the same task will be cancelled.
- Arguments:
- task (callable): task to be scheduled at (float or int): epoch seconds when task should run
- Returns:
- Future: resolves to result of task call, or exception if raised
-
send
(node_id, request)[source]¶ Send a request to a specific node.
- Arguments:
- node_id (int): destination node request (Struct): request object (not-encoded)
- Raises:
- AssertionError: if node_id is not in current cluster metadata
- Returns:
- Future: resolves to Response struct or Error