KafkaClient

class kafka.client.KafkaClient(**configs)

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!

add_topic(topic)

Add a topic to the list of topics tracked via metadata.

Parameters:topic (str) – topic to track Returns:resolves after metadata request/response Return type:Future

check_version(node_id=None, timeout=2, strict=False)

Attempt to guess a broker version

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)

Closes one or all broker connections.

Parameters:node_id (int, optional) – the id of the node to close

connected(node_id)

Return True iff the node_id is connected.

connection_delay(node_id)

Returns 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.

Parameters:node_id (int) – The id of the node to check Returns:The number of milliseconds to wait. Return type:int

in_flight_request_count(node_id=None)

Get the number of in-flight requests for a node or all nodes.

Parameters:node_id (int, optional) – a specific node to check. If unspecified, return the total for all nodes Returns:pending in-flight requests for the node, or all nodes if None Return type:int

is_disconnected(node_id)

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.

Parameters:node_id (int) – the id of the node to check Returns:True iff the node exists and is disconnected Return type:bool

is_ready(node_id)

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.

Parameters:node_id (int) – id of the node to check Returns:True if the node is ready and metadata is not refreshing Return type:bool

least_loaded_node()

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)

Try to read and write to sockets.

This method will also attempt to complete node connections, refresh stale metadata, and run previously-scheduled tasks.

Parameters:

• 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:

responses received (can be empty)

Return type:

list

ready(node_id)

Check whether a node is connected and ok to send more requests.

Parameters:node_id (int) – the id of the node to check Returns:True if we are ready to send to the given node Return type:bool

schedule(task, at)

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.

Parameters:

• task (callable) – task to be scheduled
• at (float or int) – epoch seconds when task should run

Returns:

resolves to result of task call, or exception if raised

Return type:

Future

send(node_id, request)

Send a request to a specific node.

Parameters:

• node_id (int) – destination node
• request (Struct) – request object (not-encoded)

Raises:

AssertionError – if node_id is not in current cluster metadata

Returns:

resolves to Response struct or Error

Return type:

Future

set_topics(topics)

Set specific topics to track for metadata.

Parameters:topics (list of str) – topics to check for metadata Returns:resolves after metadata request/response Return type:Future

unschedule(task)

Unschedule a task.

This will remove all instances of the task from the task queue. This is a no-op if the task is not scheduled.

Parameters:task (callable) – task to be unscheduled