summaryrefslogtreecommitdiff
path: root/kafka/client_async.py
Commit message (Collapse)AuthorAgeFilesLines
* Do not call state_change_callback with lock (#1775)Dana Powers2019-04-021-8/+8
|
* Dont treat popped conn.close() as failure in state change callback (#1773)Dana Powers2019-04-011-3/+10
|
* Avoid race condition on client._conns in send() (#1772)Dana Powers2019-03-311-2/+3
| | | There was a very small possibility that between checking `self._can_send_request(node_id)` and grabbing the connection object via `self._conns[node_id]` that the connection could get closed / recycled / removed from _conns and cause a KeyError. This PR should prevent such a KeyError. In the case where the connection is disconnected by the time we call send(), we should expect conn.send() simply to fail the request.
* lock client.check_version (#1771)Dana Powers2019-03-311-0/+5
|
* Dont wakeup during maybe_refresh_metadata -- it is only called by poll() (#1769)Dana Powers2019-03-301-4/+4
|
* Send pending requests before waiting for responses (#1762)Dana Powers2019-03-271-2/+4
|
* Dont do client wakeup when sending from sender thread (#1761)Dana Powers2019-03-241-4/+7
|
* Update sasl configuration docstringsDana Powers2019-03-231-5/+5
|
* Support SASL OAuthBearer Authentication (#1750)Phong Pham2019-03-221-1/+4
|
* Maintain shadow cluster metadata for bootstrapping (#1753)Dana Powers2019-03-211-25/+10
|
* Allow configuration of SSL Ciphers (#1755)Dana Powers2019-03-211-0/+6
|
* Retry bootstrapping after backoff when necessary (#1736)Dana Powers2019-03-141-83/+66
| | | | | | | The current client attempts to bootstrap once during initialization, but if it fails there is no second attempt and the client will be inoperable. This can happen, for example, if an entire cluster is down at the time a long-running client starts execution. This commit attempts to fix this by removing the synchronous bootstrapping from `KafkaClient` init, and instead merges bootstrap metadata with the cluster metadata. The Java client uses a similar approach. This allows us to continue falling back to bootstrap data when necessary throughout the life of a long-running consumer or producer. Fix #1670
* Minor updates to client_async.pyDana Powers2019-03-131-4/+4
|
* Recheck connecting nodes sooner when refreshing metadata (#1737)Dana Powers2019-03-131-3/+1
|
* Do network connections and writes in KafkaClient.poll() (#1729)Dana Powers2019-03-081-16/+43
| | | | | | * Add BrokerConnection.send_pending_requests to support async network sends * Send network requests during KafkaClient.poll() rather than in KafkaClient.send() * Dont acquire lock during KafkaClient.send if node is connected / ready * Move all network connection IO into KafkaClient.poll()
* Do not require client lock for read-only operations (#1730)Dana Powers2019-03-061-50/+50
| | | In an effort to reduce the surface area of lock coordination, and thereby hopefully reduce lock contention, I think we can remove locking from the read-only KafkaClient methods: connected, is_disconnected, in_flight_request_count, and least_loaded_node . Given that the read data could change after the lock is released but before the caller uses it, the value of acquiring a lock here does not seem high to me.
* Break KafkaClient poll if closedDana Powers2018-11-201-0/+2
|
* Be explicit with tuples for %s formattingJeff Widman2018-11-181-1/+1
| | | | Fix #1633
* Use TypeError for invalid typeJeff Widman2018-11-171-1/+1
|
* set socket timeout for the wake_w (#1577)flaneur2018-11-101-0/+5
|
* (Attempt to) Fix deadlock between consumer and heartbeat (#1628)Dana Powers2018-11-101-3/+1
|
* Add KafkaAdmin classRichard Lee2018-10-241-0/+16
| | | | | | Requires cluster version > 0.10.0.0, and uses new wire protocol classes to do many things via broker connection that previously needed to be done directly in zookeeper.
* add kerberos domain name config for gssapi sasl mechanism handshake (#1542)the-sea2018-08-311-0/+3
|
* add support for smaller topic metadata fetch during bootstrap (#1541)Ning Xie2018-08-311-3/+10
|
* Document connections_max_idle_msJeff Widman2018-06-261-0/+5
| | | | | This was added in #1068 but never documented. Fix #1497
* Stop shadowing `ConnectionError`Jeff Widman2018-05-231-1/+1
| | | | | | | | | | In Python3, `ConnectionError` is a native exception. So rename our custom one to `KafkaConnectionError` to prevent accidentally shadowing the native one. Note that there are still valid uses of `ConnectionError` in this code. They already expect a native Python3 `ConnectionError`, and also already handle the Python2 compatibility issues.
* Minor doc capitalization cleanupJeff Widman2018-04-241-15/+15
|
* Fix KafkaConsumer docstring for request_timeout_ms default (#1459)Dana Powers2018-03-231-2/+2
|
* Do not validate api_version against known versions (#1434)Dana Powers2018-03-101-16/+1
|
* Add BrokerConnection.connect_blocking() (#1411)Dana Powers2018-03-091-5/+1
|
* Check timeout type in KafkaClient constructorAndre Araujo2018-02-211-0/+2
| | | | | | | | | | If a future was passed as the only positional parameter it would be assigned to the "timeout_ms" parameter erroneously. This mistake would not raise any exception but would lead to odd behaviour later, what could make it extremely difficult to troubleshoot. Adding a type check ensures that an exception is raise earlier to notify the user about the problem.
* Fix pending completion IndexError bug caused by multiple threads (#1372)Dana Powers2018-02-081-2/+8
|
* use absolute imports everywhere (#1362)Kevin Tindall2018-02-061-12/+12
|
* Remove assertion with side effect (#1348)Buğra Gedik2018-01-231-2/+2
|
* Read all available socket bytes (#1332)Dana Powers2018-01-101-0/+2
| | | | * Recv all available network bytes before parsing * Add experimental support for configuring socket chunking parameters
* Improve KafkaConsumer cleanup (#1339)Dana Powers2018-01-101-8/+17
|
* KAFKA-3888 Use background thread to process consumer heartbeats (#1266)Dana Powers2017-12-211-283/+182
|
* use python standard max value (#1303)lukeWx2017-12-071-1/+1
|
* Fixup for PR 1264 -- required to propagate configuration to BrokerConnectionDana Powers2017-10-211-0/+3
|
* Merge pull request #1258 from dpkp/pending_completionsTaras Voinarovskyi2017-10-211-7/+25
|\ | | | | Move callback processing from BrokerConnection to KafkaClient
| * Move callback processing from BrokerConnection to KafkaClientpending_completionsDana Powers2017-10-151-7/+25
| |
* | Explicitly check for None rather than falseyJeff Widman2017-10-191-3/+3
|/ | | | Be pedantic about checking for identity rather than equality to avoid issues like #1237 / 411bc08f214b7afc36f11bde2047096c06467088
* Remove a few unused imports (#1188)James Lamb2017-08-291-1/+2
| | | | | * Removed a few unused imports * Added note on socketpair monkey-path
* BrokerConnection receive bytes pipe (#1032)Dana Powers2017-08-151-13/+3
|
* Drop unused sleep kwarg to poll (#1177)Dana Powers2017-08-151-9/+3
|
* Select on sockets to avoid busy polling during bootstrap (#1175)Dana Powers2017-08-131-0/+2
|
* KIP-144: Exponential backoff for broker reconnections (#1124)Dana Powers2017-06-191-19/+14
|
* Fixup for #1085 -- only check for changed metadata on disconnected nodesDana Powers2017-06-181-17/+23
|
* Deal with brokers that disappear, reappear with different IP address (#1085)Mike Fischer2017-06-181-1/+12
| | | | | | | | | | | | | | | | | | | | When KafkaClient connects to a broker in _maybe_connect, it inserts into self._conns a BrokerConnection configured with the current host/port for that node. The BrokerConnection remains there forever, though, so if the broker's IP or host ever changes, KafkaClient has no way to deal with this. The fix is to compare the latest metadata with the current node's connection, and if the host/IP has changed, decommission the old connection and allow a new one to be created. There's also a common race condition on broker startup where the initial metadata request sometimes returns an empty list of brokers, but subsequent requests behave normally. So, we must deal with broker being None here. This change is conservative in that it doesn't remove the connection from self._conns unless the new broker metadata contains an entry for that same node with a new IP/port.
* Timeout idle connections via connections_max_idle_ms (#1068)Dana Powers2017-04-101-1/+94
|