coveralls: 
[![Coverage Status](https://coveralls.io/builds/5522917/badge)](https://coveralls.io/builds/5522917)

Changes Unknown when pulling **bb2548705a3be822be9e17ea6eb824061fc9fb8f on sock_send_bytes** into ** on master**.

coveralls: 
[![Coverage Status](https://coveralls.io/builds/5664292/badge)](https://coveralls.io/builds/5664292)

Changes Unknown when pulling **7af174fe0a6bcc5962a8c8008d66e0b3b05e5fc2 on sock_send_bytes** into ** on master**.

diff --git a/kafka/conn.py b/kafka/conn.py
index 2b82b6d..ffc839e 100644
--- a/kafka/conn.py
+++ b/kafka/conn.py
@@ -188,10 +188,12 @@ class BrokerConnection(object):
             # and send bytes asynchronously. For now, just block
             # sending each request payload
             self._sock.setblocking(True)
-            sent_bytes = self._sock.send(size)
-            assert sent_bytes == len(size)
-            sent_bytes = self._sock.send(message)
-            assert sent_bytes == len(message)
+            for data in (size, message):
+                total_sent = 0
+                while total_sent < len(data):
+                    sent_bytes = self._sock.send(data[total_sent:])
+                    total_sent += sent_bytes
+                assert total_sent == len(data)
             self._sock.setblocking(False)
         except (AssertionError, ConnectionError) as e:
             log.exception("Error sending %s to %s", request, self)
diff --git a/kafka/future.py b/kafka/future.py
index 06b8c3a..c7e0b14 100644
--- a/kafka/future.py
+++ b/kafka/future.py
@@ -15,10 +15,10 @@ class Future(object):
         self._errbacks = []
 
     def succeeded(self):
-        return self.is_done and not self.exception
+        return self.is_done and not bool(self.exception)
 
     def failed(self):
-        return self.is_done and self.exception
+        return self.is_done and bool(self.exception)
 
     def retriable(self):
         try:

kafka.common.ConnectionError on big messages + gevent
i'm getting kafka.common.ConnectionError trying to send big message. Code below

```python
from gevent.monkey import patch_all; patch_all()
from kafka import KafkaProducer

producer = KafkaProducer(bootstrap_servers=xxxxxxxx,
                         buffer_memory=10 * 1024 * 1024,
                         max_request_size=10 * 1024 * 1024,)

producer.send('test', 'a' * 1024 * 1024 * 3, ).get(timeout=60)
producer.flush()
```

causing this

```
2016-03-23 11:50:58,147 - kafka.conn - ERROR - Error sending ProduceRequest(required_acks=1, timeout=30000, topics=[(topic='test2', partitions=[(1, <_io.BytesIO object at 0x7fe64edc2950>)])]) to <BrokerConnection host=127.0.0.1 port=9093>
Traceback (most recent call last):
  File "/usr/local/lib/python2.7/dist-packages/kafka/conn.py", line 187, in send
    assert sent_bytes == len(message)
AssertionError
2016-03-23 11:50:58,150 - kafka.producer.sender - DEBUG - Error sending produce request to node 2: 
2016-03-23 11:50:58,150 - kafka.producer.record_accumulator - DEBUG - Produced messages to topic-partition TopicPartition(topic='test2', partition=1) with base offset -1 and error .
2016-03-23 11:50:58,150 - kafka.client - DEBUG - Initializing connection to node 2 for metadata request
Traceback (most recent call last):
  File "test_producer.py", line 15, in <module>
    producer.send('test2', 'a' * 1024 * 1024 * 3, ).get(timeout=60)
  File "/usr/local/lib/python2.7/dist-packages/kafka/producer/future.py", line 50, in get
    raise self.exception # pylint: disable-msg=raising-bad-type
kafka.common.ConnectionError

```
this works well if i comment out **patch_all()** line

Fixing kafka.conn.BrokerConnection.send() method solved it, but i'm not sure about side effects.

```python
def send()
.....
.....
        size = Int32.encode(len(message))
        try:
            self._sock.setblocking(True)
            sent_bytes = self._sock.send(size)
            assert sent_bytes == len(size)
            total_sent = 0
            while total_sent < len(message):  # sending in loop
                sent_bytes = self._sock.send(message[total_sent:])
                assert sent_bytes
                total_sent += sent_bytes
            self._sock.setblocking(False)
        except (AssertionError, ConnectionError) as e:
            log.exception("Error sending %s to %s", request, self)
....
```
Any chances to have similar fix in master branch?
diff --git a/test/test_conn.py b/test/test_conn.py
index d394f74..5432ebd 100644
--- a/test/test_conn.py
+++ b/test/test_conn.py
@@ -2,12 +2,15 @@
 from __future__ import absolute_import
 
 from errno import EALREADY, EINPROGRESS, EISCONN, ECONNRESET
-import socket
 import time
 
 import pytest
 
 from kafka.conn import BrokerConnection, ConnectionStates
+from kafka.protocol.api import RequestHeader
+from kafka.protocol.metadata import MetadataRequest
+
+import kafka.common as Errors
 
 
 @pytest.fixture
@@ -20,6 +23,7 @@ def socket(mocker):
 
 @pytest.fixture
 def conn(socket):
+    from socket import AF_INET
     conn = BrokerConnection('localhost', 9092, socket.AF_INET)
     return conn
 
@@ -61,22 +65,111 @@ def test_connect_timeout(socket, conn):
 
 
 def test_blacked_out(conn):
-    assert not conn.blacked_out()
+    assert conn.blacked_out() is False
     conn.last_attempt = time.time()
-    assert conn.blacked_out()
+    assert conn.blacked_out() is True
 
 
 def test_connected(conn):
-    assert not conn.connected()
+    assert conn.connected() is False
     conn.state = ConnectionStates.CONNECTED
-    assert conn.connected()
+    assert conn.connected() is True
 
 
 def test_connecting(conn):
-    assert not conn.connecting()
+    assert conn.connecting() is False
+    conn.state = ConnectionStates.CONNECTING
+    assert conn.connecting() is True
+    conn.state = ConnectionStates.CONNECTED
+    assert conn.connecting() is False
+
+
+def test_send_disconnected(conn):
+    conn.state = ConnectionStates.DISCONNECTED
+    f = conn.send('foobar')
+    assert f.failed() is True
+    assert isinstance(f.exception, Errors.ConnectionError)
+
+
+def test_send_connecting(conn):
     conn.state = ConnectionStates.CONNECTING
-    assert conn.connecting()
+    f = conn.send('foobar')
+    assert f.failed() is True
+    assert isinstance(f.exception, Errors.NodeNotReadyError)
+
+
+def test_send_max_ifr(conn):
     conn.state = ConnectionStates.CONNECTED
-    assert not conn.connecting()
+    max_ifrs = conn.config['max_in_flight_requests_per_connection']
+    for _ in range(max_ifrs):
+        conn.in_flight_requests.append('foo')
+    f = conn.send('foobar')
+    assert f.failed() is True
+    assert isinstance(f.exception, Errors.TooManyInFlightRequests)
+
+
+def test_send_no_response(socket, conn):
+    conn.connect()
+    assert conn.state is ConnectionStates.CONNECTED
+    req = MetadataRequest([])
+    header = RequestHeader(req, client_id=conn.config['client_id'])
+    payload_bytes = len(header.encode()) + len(req.encode())
+    third = payload_bytes // 3
+    remainder = payload_bytes % 3
+    socket.send.side_effect = [4, third, third, third, remainder]
+
+    assert len(conn.in_flight_requests) == 0
+    f = conn.send(req, expect_response=False)
+    assert f.succeeded() is True
+    assert f.value is None
+    assert len(conn.in_flight_requests) == 0
+
+
+def test_send_response(socket, conn):
+    conn.connect()
+    assert conn.state is ConnectionStates.CONNECTED
+    req = MetadataRequest([])
+    header = RequestHeader(req, client_id=conn.config['client_id'])
+    payload_bytes = len(header.encode()) + len(req.encode())
+    third = payload_bytes // 3
+    remainder = payload_bytes % 3
+    socket.send.side_effect = [4, third, third, third, remainder]
+
+    assert len(conn.in_flight_requests) == 0
+    f = conn.send(req)
+    assert f.is_done is False
+    assert len(conn.in_flight_requests) == 1
+
+
+def test_send_error(socket, conn):
+    conn.connect()
+    assert conn.state is ConnectionStates.CONNECTED
+    req = MetadataRequest([])
+    header = RequestHeader(req, client_id=conn.config['client_id'])
+    try:
+        error = ConnectionError
+    except NameError:
+        from socket import error
+    socket.send.side_effect = error
+    f = conn.send(req)
+    assert f.failed() is True
+    assert isinstance(f.exception, Errors.ConnectionError)
+    assert socket.close.call_count == 1
+    assert conn.state is ConnectionStates.DISCONNECTED
+
+
+def test_can_send_more(conn):
+    assert conn.can_send_more() is True
+    max_ifrs = conn.config['max_in_flight_requests_per_connection']
+    for _ in range(max_ifrs):
+        assert conn.can_send_more() is True
+        conn.in_flight_requests.append('foo')
+    assert conn.can_send_more() is False
+
+
+def test_recv(socket, conn):
+    pass # TODO
+
 
-# TODO: test_send, test_recv, test_can_send_more, test_close
+def test_close(conn):
+    pass # TODO

pip install -e .
parse_log_pytest
requirements.txt
pytest pytest-cov pytest-catchlog pytest-pylint pytest-sugar pytest-mock mock python-snappy lz4tools xxhash
apt-get update apt-get install -y libsnappy-dev
3.5
pytest --no-header -rA --tb=line --color=no -p no:cacheprovider -W ignore::DeprecationWarning
astroid==2.11.7
attrs==22.2.0
certifi==2021.5.30
coverage==6.2
cramjam==2.5.0
dill==0.3.4
importlib-metadata==4.8.3
iniconfig==1.1.1
isort==5.10.1
-e git+https://github.com/dpkp/kafka-python.git@d81963a919fa8161c94b5bef5e6de0697b91c4a6#egg=kafka_python
lazy-object-proxy==1.7.1
lz4tools==1.3.1.2
mccabe==0.7.0
mock==5.2.0
packaging==21.3
platformdirs==2.4.0
pluggy==1.0.0
py==1.11.0
pylint==2.13.9
pyparsing==3.1.4
pytest==7.0.1
pytest-catchlog==1.2.2
pytest-cov==4.0.0
pytest-mock==3.6.1
pytest-pylint==0.18.0
pytest-sugar==0.9.6
python-snappy==0.7.3
six==1.17.0
termcolor==1.1.0
toml==0.10.2
tomli==1.2.3
typed-ast==1.5.5
typing_extensions==4.1.1
wrapt==1.16.0
xxhash==3.2.0
zipp==3.6.0

name: kafka-python
channels:
  - defaults
  - https://repo.anaconda.com/pkgs/main
  - https://repo.anaconda.com/pkgs/r
  - conda-forge
dependencies:
  - _libgcc_mutex=0.1=main
  - _openmp_mutex=5.1=1_gnu
  - ca-certificates=2025.2.25=h06a4308_0
  - certifi=2021.5.30=py36h06a4308_0
  - ld_impl_linux-64=2.40=h12ee557_0
  - libffi=3.3=he6710b0_2
  - libgcc-ng=11.2.0=h1234567_1
  - libgomp=11.2.0=h1234567_1
  - libstdcxx-ng=11.2.0=h1234567_1
  - ncurses=6.4=h6a678d5_0
  - openssl=1.1.1w=h7f8727e_0
  - pip=21.2.2=py36h06a4308_0
  - python=3.6.13=h12debd9_1
  - readline=8.2=h5eee18b_0
  - setuptools=58.0.4=py36h06a4308_0
  - sqlite=3.45.3=h5eee18b_0
  - tk=8.6.14=h39e8969_0
  - wheel=0.37.1=pyhd3eb1b0_0
  - xz=5.6.4=h5eee18b_1
  - zlib=1.2.13=h5eee18b_1
  - pip:
      - astroid==2.11.7
      - attrs==22.2.0
      - coverage==6.2
      - cramjam==2.5.0
      - dill==0.3.4
      - importlib-metadata==4.8.3
      - iniconfig==1.1.1
      - isort==5.10.1
      - lazy-object-proxy==1.7.1
      - lz4tools==1.3.1.2
      - mccabe==0.7.0
      - mock==5.2.0
      - packaging==21.3
      - platformdirs==2.4.0
      - pluggy==1.0.0
      - py==1.11.0
      - pylint==2.13.9
      - pyparsing==3.1.4
      - pytest==7.0.1
      - pytest-catchlog==1.2.2
      - pytest-cov==4.0.0
      - pytest-mock==3.6.1
      - pytest-pylint==0.18.0
      - pytest-sugar==0.9.6
      - python-snappy==0.7.3
      - six==1.17.0
      - termcolor==1.1.0
      - toml==0.10.2
      - tomli==1.2.3
      - typed-ast==1.5.5
      - typing-extensions==4.1.1
      - wrapt==1.16.0
      - xxhash==3.2.0
      - zipp==3.6.0
prefix: /opt/conda/envs/kafka-python

test/test_conn.py::test_send_disconnected test/test_conn.py::test_send_connecting test/test_conn.py::test_send_max_ifr test/test_conn.py::test_send_no_response test/test_conn.py::test_send_response test/test_conn.py::test_send_error
test/test_conn.py::test_connect[states0] test/test_conn.py::test_connect[states1] test/test_conn.py::test_connect[states2] test/test_conn.py::test_connect[states3] test/test_conn.py::test_connect[states4] test/test_conn.py::test_connect_timeout test/test_conn.py::test_blacked_out test/test_conn.py::test_connected test/test_conn.py::test_connecting test/test_conn.py::test_can_send_more test/test_conn.py::test_recv test/test_conn.py::test_close
kafka/future.py kafka/conn.py
kafka/future.py kafka/conn.py