Repository: Shopify/sarama
Pull request #2094 (2021-12-24 17:24:24)

Support request pipelining in AsyncProducer
# Enhancement

Although there is a `Config.Net.MaxOpenRequests` parameter, the `AsyncProducer` uses at most one in-flight produce request when producing to a single Kafka broker.

As shown in #1539 a while back, using request pipelining increases throughput when writing to a Kafka cluster over a high latency network link and something that is recommended when implementing the Kafka protocol as described in
[their network section](https://kafka.apache.org/protocol#protocol_network):

> The broker's request processing allows only a single in-flight request per
> connection in order to guarantee this ordering. Note that clients can (and
> ideally should) use non-blocking IO to implement request pipelining and
> achieve higher throughput. i.e., clients can send requests even while
> awaiting responses for preceding requests since the outstanding requests
> will be buffered in the underlying OS socket buffer.

`MaxOpenRequests` defaults to 5 but if you were to execute the following code:
```go
for {
  // Blocks till we get a response from the broker or an error is returned
  res, err := broker.Produce(req)
  // Check for error or else use response
}
```
then you would only see a single in-flight `Produce` request.
That is, to send the next request you must first wait for the `responseReceiver` Go routine to read the current response bytes and return it to calling Go routine.

Obviously, if it takes a while to read such response because of network latency or slow replication, the calling Go routine will seat idle instead of sending more records therefore reducing the throughput (more on that in the `Testing done`
section).

Unfortunately this is how the `AsyncProducer` currently works, at least how the `brokerProducer` "bridge" Go routine processes batches (i.e. `*produceSet`):
https://github.com/Shopify/sarama/blob/635bcf350a7b8a92b4da0aacd288ee09311e673d/async_producer.go#L693-L707

By adding a new `Broker.AyncProduce` receiver and using it in the `AsyncProducer`, we can achieve proper pipelining of `Produce` requests.

The new receiver relies on a callback to eventually receive the response:
```go
for {
  // Blocks once we reach MaxOpenRequests in-flight requests
  err := broker.AsyncProduce(req, func(res *ProduceResponse, err error) {
    // Check for response error or else use response
  })
  // Check for request error
}
```

The same could technically be done with `Broker.Produce` and N go routines:
```go
for {
  go func() {
    res, err := broker.Produce(req)
    // Check for error or else use response
  }
}
```
but it gets tricky to get the back-pressure behavior from the internal `Brokers.responses` channel and avoid creating too many Go routines.

With the callback the back-pressure is propagated up and we can still use a channel to handle the responses asynchronously if necessary: 
```go
type responseOrError struct {
  res *ProduceResponse
  err error
}

c := make(chan responseOrError)

// Handle future responses asynchronously
go func() {
  for resOrErr := range c {
    // Check for response error or else use response
  }
}

for {
  // Blocks once we reach MaxOpenRequests in-flight requests
  err := broker.AsyncProduce(req, func(res *ProduceResponse, err error) {
    // Pass the response or error to a separate Go routine to unblock
    // the Broker.responseReceiver Go routine
    c <- responseOrError{res, err}
  })
  // Check for request error
}
```

Returning or else passing a channel to `Broker.AsyncProduce` is another option but I believe it leads to more complex and less flexible code.

As this is a non trivial behavior change in the `AsyncProducer`, I introduced a new configuration property to make that feature opt-in.

I do believe that `Produce` request pipelining should be the default (like in the Java client) but an application that cares about ordering might see records persisted out of order depending on how it is configured which can be considered a breaking change.

## Changes

- introduce `Broker.AsyncProduce` with a callback to have a mostly non blocking way to produce to a single broker
- refactor `Broker.send` and `Broker.responseReceiver` to support callback
- factorize throttleTime metrics into `Broker.updateThrottleMetric`
- add `Config.Produce.Pipeline` parameter (defaults to `false` for backward compatibility)
- honor `MaxOpenRequests` when using `AsyncProducer` and `conf.Produce.Pipeline` is enabled
- add unit and functional tests
- add `-enable-pipeline` flag to `kafka-producer-performance` for benchmarking

## Testing done

The new unit and functional tests should provide good coverage for the new changes as they touch a critical part of the `Broker` and the `AsyncProducer` logic.

But for a real end to end and performance test I am using the `kafka-producer-performance` tool with a new `-enable-pipeline` to show the impact of request pipelining when it is the most useful.

For that, I am running the tool from an EC2 instance in AWS `us-west-2` and producing to:
- a local Kafka cluster in the same AZ with a RTT of `~1` ms
- a remote Kafka cluster in AWS `us-east-1` with a RTT of `~70` ms
- a remote Kafka cluster in AWS `eu-central-1` with a RTT of `~140` ms

Here are some details about the EC2 instance and the command line used as the baseline:
```
$ lsb_release -d
Description:    Ubuntu 18.04.2 LTS
$ ec2metadata --instance-type
c5d.xlarge
$ sysctl net.ipv4.tcp_wmem
net.ipv4.tcp_wmem = 4096        16384   67108864
$ ./kafka-producer-performance-pipeline \
  -brokers <broker>:9093 \
  -security-protocol SSL -tls-ca-certs ca.pem -tls-client-cert cert.pem -tls-client-key key.pem \
  -topic test-topic-for-intg \
  -message-load 1000000 \       # 1M records
  -message-size 1000 \          # 1 kB record payload
  -required-acks -1 \           # Wait for 2 ISRs out of 4 replicas
  -flush-bytes 500000 \
  -flush-frequency 100ms
```

Here are the performance results for each remote target cluster with the respective extra flags:

| Target cluster | Extra flags                                                   | RTT (ms) | Max records/s | Max throughput (MiB/s) | Max requests in flight |
| -------------- | ------------------------------------------------------------- | -------: | ------------: | ---------------------: | ---------------------: |
| `us-west-2`    | `-partitioner manual`<br>`-partition 0`                       | `1`      | `55,417.7`    | `35.87`                | `1`                    |
| `us-west-2`    | `-partitioner manual`<br>`-partition 0`<br>`-enable-pipeline` | `1`      | `63,645.8`    | `41.70`                | `6`                    |
| `us-east-1`    | `-partitioner manual`<br>`-partition 0`                       | `70`     | `10,758.9`    | `9.74`                 | `1`                    |
| `us-east-1`    | `-partitioner manual`<br>`-partition 0`<br>`-enable-pipeline` | `70`     | `64,800.0`    | `36.12`                | `6`                    |
| `eu-central-1` | `-partitioner manual`<br>`-partition 0`                       | `140`    | `5,494.0`     | `5.14`                 | `1`                    |
| `eu-central-1` | `-partitioner manual`<br>`-partition 0`<br>`-enable-pipeline`<br>`-max-open-requests 1` | `140` | `11,371.9` | `10.10` | `2`               |
| `eu-central-1` | `-partitioner manual`<br>`-partition 0`<br>`-enable-pipeline` | `140`    | `32,999.4`    | `25.47`                | `6`                    |
| `eu-central-1` | default `roundrobin` partitioner<br>on 5 Brokers/128 partitions  | `140`    | `119,782.4`   | `59.19`                | `5`                    |
| `eu-central-1` | default `roundrobin` partitioner<br>on 5 Brokers/128 partitions<br>plus `-enable-pipeline` | `140` | `133,124.7` | `65.88` | `16`             |


Here is the detailed output of `kafka-producer-performance` when producing to `eu-central-1` over a single topic partition (`-partitioner manual -partition 0`).

<details>
<summary>Request pipelining disabled (current behavior) details</summary>

```
7306 records sent, 2945.7 records/sec (2.81 MiB/sec ingress, 0.73 MiB/sec egress), 343.9 ms avg latency, 335.1 ms stddev, 161.5 ms 50th, 412.2 ms 75th, 1192.0 ms 95th, 1192.0 ms 99th, 1192.0 ms 99.9th, 1 total req. in flight
32630 records sent, 4362.6 records/sec (4.16 MiB/sec ingress, 2.15 MiB/sec egress), 230.4 ms avg latency, 182.3 ms stddev, 161.0 ms 50th, 305.0 ms 75th, 634.0 ms 95th, 1192.0 ms 99th, 1192.0 ms 99.9th, 1 total req. in flight
56980 records sent, 4565.8 records/sec (4.35 MiB/sec ingress, 2.81 MiB/sec egress), 214.0 ms avg latency, 144.6 ms stddev, 161.0 ms 50th, 175.0 ms 75th, 314.0 ms 95th, 1192.0 ms 99th, 1192.0 ms 99.9th, 1 total req. in flight
85226 records sent, 4875.6 records/sec (4.65 MiB/sec ingress, 3.36 MiB/sec egress), 200.8 ms avg latency, 120.6 ms stddev, 162.0 ms 50th, 175.0 ms 75th, 306.6 ms 95th, 1192.0 ms 99th, 1192.0 ms 99.9th, 1 total req. in flight
113472 records sent, 5047.8 records/sec (4.81 MiB/sec ingress, 3.72 MiB/sec egress), 192.5 ms avg latency, 106.4 ms stddev, 161.0 ms 50th, 169.0 ms 75th, 306.0 ms 95th, 1058.1 ms 99th, 1192.0 ms 99.9th, 1 total req. in flight
139770 records sent, 5086.2 records/sec (4.85 MiB/sec ingress, 3.93 MiB/sec egress), 190.5 ms avg latency, 98.1 ms stddev, 161.0 ms 50th, 170.5 ms 75th, 306.0 ms 95th, 857.2 ms 99th, 1192.0 ms 99.9th, 1 total req. in flight
168016 records sent, 5173.0 records/sec (4.93 MiB/sec ingress, 4.13 MiB/sec egress), 187.4 ms avg latency, 91.0 ms stddev, 161.0 ms 50th, 168.5 ms 75th, 306.0 ms 95th, 641.4 ms 99th, 1192.0 ms 99.9th, 1 total req. in flight
196262 records sent, 5236.5 records/sec (4.99 MiB/sec ingress, 4.28 MiB/sec egress), 185.1 ms avg latency, 85.5 ms stddev, 161.0 ms 50th, 165.0 ms 75th, 305.9 ms 95th, 445.7 ms 99th, 1192.0 ms 99.9th, 1 total req. in flight
222560 records sent, 5239.2 records/sec (5.00 MiB/sec ingress, 4.37 MiB/sec egress), 184.4 ms avg latency, 81.9 ms stddev, 161.0 ms 50th, 165.0 ms 75th, 306.0 ms 95th, 425.2 ms 99th, 1192.0 ms 99.9th, 1 total req. in flight
251780 records sent, 5302.9 records/sec (5.06 MiB/sec ingress, 4.49 MiB/sec egress), 182.3 ms avg latency, 77.7 ms stddev, 161.0 ms 50th, 164.0 ms 75th, 306.0 ms 95th, 402.4 ms 99th, 1192.0 ms 99.9th, 1 total req. in flight
281000 records sent, 5354.4 records/sec (5.11 MiB/sec ingress, 4.60 MiB/sec egress), 180.2 ms avg latency, 73.8 ms stddev, 161.0 ms 50th, 164.0 ms 75th, 305.0 ms 95th, 379.6 ms 99th, 1192.0 ms 99.9th, 1 total req. in flight
307298 records sent, 5346.2 records/sec (5.10 MiB/sec ingress, 4.64 MiB/sec egress), 180.5 ms avg latency, 72.2 ms stddev, 161.0 ms 50th, 164.0 ms 75th, 305.0 ms 95th, 362.1 ms 99th, 1192.0 ms 99.9th, 1 total req. in flight
336518 records sent, 5386.1 records/sec (5.14 MiB/sec ingress, 4.72 MiB/sec egress), 179.4 ms avg latency, 69.5 ms stddev, 161.0 ms 50th, 164.0 ms 75th, 305.0 ms 95th, 344.7 ms 99th, 1192.0 ms 99.9th, 1 total req. in flight
363790 records sent, 5391.1 records/sec (5.14 MiB/sec ingress, 4.76 MiB/sec egress), 178.7 ms avg latency, 67.4 ms stddev, 161.0 ms 50th, 164.0 ms 75th, 305.0 ms 95th, 328.5 ms 99th, 1192.0 ms 99.9th, 1 total req. in flight
391062 records sent, 5395.4 records/sec (5.15 MiB/sec ingress, 4.79 MiB/sec egress), 178.7 ms avg latency, 65.8 ms stddev, 161.0 ms 50th, 164.0 ms 75th, 305.0 ms 95th, 314.0 ms 99th, 1192.0 ms 99.9th, 1 total req. in flight
420282 records sent, 5424.4 records/sec (5.17 MiB/sec ingress, 4.85 MiB/sec egress), 177.8 ms avg latency, 63.9 ms stddev, 161.0 ms 50th, 164.0 ms 75th, 305.0 ms 95th, 314.0 ms 99th, 1192.0 ms 99.9th, 1 total req. in flight
448528 records sent, 5438.1 records/sec (5.19 MiB/sec ingress, 4.89 MiB/sec egress), 177.2 ms avg latency, 62.2 ms stddev, 161.0 ms 50th, 163.0 ms 75th, 305.0 ms 95th, 314.0 ms 99th, 1192.0 ms 99.9th, 1 total req. in flight
476774 records sent, 5450.1 records/sec (5.20 MiB/sec ingress, 4.92 MiB/sec egress), 176.9 ms avg latency, 61.0 ms stddev, 161.0 ms 50th, 163.0 ms 75th, 305.0 ms 95th, 314.0 ms 99th, 1192.0 ms 99.9th, 1 total req. in flight
504046 records sent, 5450.3 records/sec (5.20 MiB/sec ingress, 4.94 MiB/sec egress), 176.7 ms avg latency, 60.0 ms stddev, 161.0 ms 50th, 163.0 ms 75th, 305.0 ms 95th, 313.6 ms 99th, 1192.0 ms 99.9th, 1 total req. in flight
532292 records sent, 5460.5 records/sec (5.21 MiB/sec ingress, 4.97 MiB/sec egress), 176.5 ms avg latency, 59.0 ms stddev, 161.0 ms 50th, 163.0 ms 75th, 305.0 ms 95th, 313.0 ms 99th, 1192.0 ms 99.9th, 1 total req. in flight
556642 records sent, 5431.7 records/sec (5.18 MiB/sec ingress, 4.96 MiB/sec egress), 177.3 ms avg latency, 59.2 ms stddev, 161.0 ms 50th, 163.8 ms 75th, 305.0 ms 95th, 318.0 ms 99th, 1192.0 ms 99.9th, 1 total req. in flight
584888 records sent, 5441.9 records/sec (5.19 MiB/sec ingress, 4.98 MiB/sec egress), 176.9 ms avg latency, 58.1 ms stddev, 161.0 ms 50th, 163.0 ms 75th, 305.0 ms 95th, 314.0 ms 99th, 1192.0 ms 99.9th, 1 total req. in flight
614108 records sent, 5459.7 records/sec (5.21 MiB/sec ingress, 5.01 MiB/sec egress), 176.4 ms avg latency, 57.0 ms stddev, 161.0 ms 50th, 163.0 ms 75th, 305.0 ms 95th, 314.0 ms 99th, 1192.0 ms 99.9th, 1 total req. in flight
643328 records sent, 5476.1 records/sec (5.22 MiB/sec ingress, 5.04 MiB/sec egress), 175.8 ms avg latency, 55.8 ms stddev, 161.0 ms 50th, 163.0 ms 75th, 305.0 ms 95th, 314.0 ms 99th, 1192.0 ms 99.9th, 1 total req. in flight
670600 records sent, 5475.2 records/sec (5.22 MiB/sec ingress, 5.05 MiB/sec egress), 175.9 ms avg latency, 55.4 ms stddev, 161.0 ms 50th, 163.0 ms 75th, 305.0 ms 95th, 314.0 ms 99th, 1192.0 ms 99.9th, 1 total req. in flight
699820 records sent, 5489.6 records/sec (5.24 MiB/sec ingress, 5.08 MiB/sec egress), 175.3 ms avg latency, 54.3 ms stddev, 161.0 ms 50th, 163.0 ms 75th, 305.0 ms 95th, 313.6 ms 99th, 1192.0 ms 99.9th, 1 total req. in flight
725144 records sent, 5473.6 records/sec (5.22 MiB/sec ingress, 5.07 MiB/sec egress), 175.8 ms avg latency, 54.5 ms stddev, 161.0 ms 50th, 163.0 ms 75th, 305.0 ms 95th, 314.0 ms 99th, 1192.0 ms 99.9th, 1 total req. in flight
752416 records sent, 5472.9 records/sec (5.22 MiB/sec ingress, 5.08 MiB/sec egress), 175.9 ms avg latency, 54.2 ms stddev, 161.0 ms 50th, 163.0 ms 75th, 305.0 ms 95th, 314.0 ms 99th, 1192.0 ms 99.9th, 1 total req. in flight
779688 records sent, 5472.3 records/sec (5.22 MiB/sec ingress, 5.09 MiB/sec egress), 175.9 ms avg latency, 53.9 ms stddev, 161.0 ms 50th, 163.0 ms 75th, 305.0 ms 95th, 314.0 ms 99th, 1192.0 ms 99.9th, 1 total req. in flight
808908 records sent, 5484.9 records/sec (5.23 MiB/sec ingress, 5.11 MiB/sec egress), 175.6 ms avg latency, 53.1 ms stddev, 161.0 ms 50th, 163.0 ms 75th, 305.0 ms 95th, 314.0 ms 99th, 1192.0 ms 99.9th, 1 total req. in flight
837154 records sent, 5490.3 records/sec (5.24 MiB/sec ingress, 5.13 MiB/sec egress), 175.3 ms avg latency, 52.5 ms stddev, 161.0 ms 50th, 163.0 ms 75th, 305.0 ms 95th, 314.0 ms 99th, 1192.0 ms 99.9th, 1 total req. in flight
865400 records sent, 5495.3 records/sec (5.24 MiB/sec ingress, 5.14 MiB/sec egress), 175.2 ms avg latency, 52.0 ms stddev, 161.0 ms 50th, 163.0 ms 75th, 305.0 ms 95th, 314.0 ms 99th, 1192.0 ms 99.9th, 1 total req. in flight
892672 records sent, 5494.0 records/sec (5.24 MiB/sec ingress, 5.14 MiB/sec egress), 175.3 ms avg latency, 51.8 ms stddev, 161.0 ms 50th, 163.0 ms 75th, 305.0 ms 95th, 314.0 ms 99th, 1192.0 ms 99.9th, 1 total req. in flight
918970 records sent, 5487.1 records/sec (5.23 MiB/sec ingress, 5.14 MiB/sec egress), 175.4 ms avg latency, 51.7 ms stddev, 161.0 ms 50th, 163.0 ms 75th, 305.0 ms 95th, 314.5 ms 99th, 1192.0 ms 99.9th, 1 total req. in flight
947216 records sent, 5491.8 records/sec (5.24 MiB/sec ingress, 5.15 MiB/sec egress), 175.4 ms avg latency, 51.3 ms stddev, 161.0 ms 50th, 163.0 ms 75th, 305.0 ms 95th, 314.3 ms 99th, 1192.0 ms 99.9th, 1 total req. in flight
972540 records sent, 5479.7 records/sec (5.23 MiB/sec ingress, 5.15 MiB/sec egress), 175.8 ms avg latency, 51.5 ms stddev, 161.0 ms 50th, 163.0 ms 75th, 305.0 ms 95th, 314.0 ms 99th, 1192.0 ms 99.9th, 1 total req. in flight
999812 records sent, 5479.0 records/sec (5.23 MiB/sec ingress, 5.15 MiB/sec egress), 175.7 ms avg latency, 51.2 ms stddev, 161.0 ms 50th, 163.0 ms 75th, 305.0 ms 95th, 314.0 ms 99th, 1171.2 ms 99.9th, 1 total req. in flight
1000000 records sent, 5478.7 records/sec (5.22 MiB/sec ingress, 5.15 MiB/sec egress), 175.4 ms avg latency, 50.5 ms stddev, 161.0 ms 50th, 163.0 ms 75th, 305.0 ms 95th, 314.0 ms 99th, 1168.3 ms 99.9th, 0 total req. in flight
```
</details>

<details>
<summary>Request pipelining enabled (-enable-pipeline -max-open-requests 1) details</summary>

 ```
 1462 records sent, 1304.2 records/sec (1.24 MiB/sec ingress, 0.05 MiB/sec egress), 453.0 ms avg latency, 0.0 ms stddev, 453.0 ms 50th, 453.0 ms 75th, 453.0 ms 95th, 453.0 ms 99th, 453.0 ms 99.9th, 2 total req. in flight
 54058 records sent, 8831.9 records/sec (8.42 MiB/sec ingress, 3.56 MiB/sec egress), 207.8 ms avg latency, 172.2 ms stddev, 162.0 ms 50th, 167.0 ms 75th, 482.6 ms 95th, 1348.0 ms 99th, 1348.0 ms 99.9th, 2 total req. in flight
 111524 records sent, 10028.3 records/sec (9.56 MiB/sec ingress, 5.50 MiB/sec egress), 186.5 ms avg latency, 122.0 ms stddev, 162.0 ms 50th, 164.0 ms 75th, 288.2 ms 95th, 1235.9 ms 99th, 1348.0 ms 99.9th, 2 total req. in flight
 169964 records sent, 10542.8 records/sec (10.05 MiB/sec ingress, 6.69 MiB/sec egress), 179.1 ms avg latency, 99.4 ms stddev, 162.0 ms 50th, 164.0 ms 75th, 240.8 ms 95th, 787.8 ms 99th, 1348.0 ms 99.9th, 2 total req. in flight
 228404 records sent, 10814.0 records/sec (10.31 MiB/sec ingress, 7.49 MiB/sec egress), 175.2 ms avg latency, 86.0 ms stddev, 162.0 ms 50th, 164.0 ms 75th, 183.0 ms 95th, 549.2 ms 99th, 1348.0 ms 99.9th, 2 total req. in flight
 285870 records sent, 10944.2 records/sec (10.44 MiB/sec ingress, 8.03 MiB/sec egress), 173.3 ms avg latency, 77.3 ms stddev, 162.0 ms 50th, 163.0 ms 75th, 181.3 ms 95th, 461.9 ms 99th, 1348.0 ms 99.9th, 2 total req. in flight
 344310 records sent, 11063.7 records/sec (10.55 MiB/sec ingress, 8.46 MiB/sec egress), 171.5 ms avg latency, 70.6 ms stddev, 162.0 ms 50th, 163.0 ms 75th, 180.3 ms 95th, 367.7 ms 99th, 1348.0 ms 99.9th, 2 total req. in flight
 401776 records sent, 11123.2 records/sec (10.61 MiB/sec ingress, 8.77 MiB/sec egress), 170.9 ms avg latency, 65.6 ms stddev, 162.0 ms 50th, 163.0 ms 75th, 181.0 ms 95th, 294.9 ms 99th, 1348.0 ms 99.9th, 2 total req. in flight
 460216 records sent, 11191.8 records/sec (10.67 MiB/sec ingress, 9.03 MiB/sec egress), 170.0 ms avg latency, 61.4 ms stddev, 162.0 ms 50th, 163.0 ms 75th, 180.0 ms 95th, 294.3 ms 99th, 1348.0 ms 99.9th, 2 total req. in flight
 517682 records sent, 11224.5 records/sec (10.70 MiB/sec ingress, 9.24 MiB/sec egress), 169.5 ms avg latency, 58.1 ms stddev, 162.0 ms 50th, 163.0 ms 75th, 179.4 ms 95th, 292.4 ms 99th, 1348.0 ms 99.9th, 2 total req. in flight
 576122 records sent, 11269.7 records/sec (10.75 MiB/sec ingress, 9.42 MiB/sec egress), 168.8 ms avg latency, 55.1 ms stddev, 162.0 ms 50th, 163.0 ms 75th, 179.0 ms 95th, 289.4 ms 99th, 1348.0 ms 99.9th, 2 total req. in flight
 634562 records sent, 11307.1 records/sec (10.78 MiB/sec ingress, 9.58 MiB/sec egress), 168.2 ms avg latency, 52.6 ms stddev, 162.0 ms 50th, 163.0 ms 75th, 178.0 ms 95th, 288.5 ms 99th, 1348.0 ms 99.9th, 2 total req. in flight
 693002 records sent, 11338.3 records/sec (10.81 MiB/sec ingress, 9.71 MiB/sec egress), 168.2 ms avg latency, 50.4 ms stddev, 162.0 ms 50th, 163.0 ms 75th, 180.0 ms 95th, 287.9 ms 99th, 1348.0 ms 99.9th, 2 total req. in flight
 751442 records sent, 11364.8 records/sec (10.84 MiB/sec ingress, 9.82 MiB/sec egress), 168.0 ms avg latency, 48.4 ms stddev, 162.0 ms 50th, 163.0 ms 75th, 180.0 ms 95th, 287.3 ms 99th, 1348.0 ms 99.9th, 2 total req. in flight
 807934 records sent, 11360.1 records/sec (10.83 MiB/sec ingress, 9.90 MiB/sec egress), 168.0 ms avg latency, 47.1 ms stddev, 162.0 ms 50th, 163.0 ms 75th, 180.0 ms 95th, 287.7 ms 99th, 1348.0 ms 99.9th, 2 total req. in flight
 864426 records sent, 11355.9 records/sec (10.83 MiB/sec ingress, 9.97 MiB/sec egress), 168.0 ms avg latency, 45.9 ms stddev, 162.0 ms 50th, 163.0 ms 75th, 180.0 ms 95th, 288.1 ms 99th, 1348.0 ms 99.9th, 2 total req. in flight
 922866 records sent, 11376.4 records/sec (10.85 MiB/sec ingress, 10.05 MiB/sec egress), 167.8 ms avg latency, 44.5 ms stddev, 162.0 ms 50th, 163.0 ms 75th, 180.0 ms 95th, 287.5 ms 99th, 1348.0 ms 99.9th, 2 total req. in flight
 979358 records sent, 11371.9 records/sec (10.85 MiB/sec ingress, 10.10 MiB/sec egress), 167.8 ms avg latency, 43.6 ms stddev, 162.0 ms 50th, 163.0 ms 75th, 180.0 ms 95th, 287.9 ms 99th, 1343.5 ms 99.9th, 2 total req. in flight
 1000000 records sent, 11375.8 records/sec (10.85 MiB/sec ingress, 10.13 MiB/sec egress), 167.4 ms avg latency, 42.2 ms stddev, 162.0 ms 50th, 163.0 ms 75th, 180.0 ms 95th, 286.7 ms 99th, 1326.3 ms 99.9th, 0 total req. in flight
```
</details>

<details>
<summary>Request pipelining enabled (-enable-pipeline [-max-open-requests 5]) details</summary>

```
36526 records sent, 14647.2 records/sec (13.97 MiB/sec ingress, 3.63 MiB/sec egress), 255.9 ms avg latency, 191.9 ms stddev, 174.0 ms 50th, 314.0 ms 75th, 675.4 ms 95th, 1185.0 ms 99th, 1185.0 ms 99.9th, 6 total req. in flight
211846 records sent, 28270.6 records/sec (26.96 MiB/sec ingress, 13.96 MiB/sec egress), 178.6 ms avg latency, 83.0 ms stddev, 161.0 ms 50th, 164.0 ms 75th, 241.6 ms 95th, 456.7 ms 99th, 1185.0 ms 99.9th, 6 total req. in flight
390088 records sent, 31223.1 records/sec (29.78 MiB/sec ingress, 19.23 MiB/sec egress), 171.2 ms avg latency, 61.6 ms stddev, 161.0 ms 50th, 162.0 ms 75th, 194.6 ms 95th, 450.1 ms 99th, 1185.0 ms 99.9th, 6 total req. in flight
551772 records sent, 31541.5 records/sec (30.08 MiB/sec ingress, 21.73 MiB/sec egress), 173.2 ms avg latency, 52.5 ms stddev, 161.0 ms 50th, 174.2 ms 75th, 200.0 ms 95th, 372.0 ms 99th, 1185.0 ms 99.9th, 6 total req. in flight
728066 records sent, 32368.0 records/sec (30.87 MiB/sec ingress, 23.87 MiB/sec egress), 170.6 ms avg latency, 46.0 ms stddev, 161.0 ms 50th, 167.0 ms 75th, 199.0 ms 95th, 317.8 ms 99th, 1185.0 ms 99.9th, 6 total req. in flight
907282 records sent, 32999.4 records/sec (31.47 MiB/sec ingress, 25.47 MiB/sec egress), 169.0 ms avg latency, 41.3 ms stddev, 161.0 ms 50th, 165.0 ms 75th, 197.0 ms 95th, 311.0 ms 99th, 1185.0 ms 99.9th, 6 total req. in flight
1000000 records sent, 33098.5 records/sec (31.57 MiB/sec ingress, 26.04 MiB/sec egress), 168.2 ms avg latency, 38.4 ms stddev, 161.0 ms 50th, 164.0 ms 75th, 196.0 ms 95th, 216.1 ms 99th, 1163.9 ms 99.9th, 0 total req. in flight
```
</details>

There is an oddity with using `-max-open-requests` as it is reported as `2` when set to `1` and `6` when set to `5` (default).

I thought it was because the `requests-in-flight` metric is increased before writing to the socket but the effective rate limiting is done later when writing to the `Brokers.responses` channel:
https://github.com/Shopify/sarama/blob/635bcf350a7b8a92b4da0aacd288ee09311e673d/broker.go#L830-L848

I think there might actually be room for one more in-flight request even if the channel is buffered with a `N-1` capacity. That would explain the 2x throughput improvement when using `-max-open-requests 1` as we would be able to
pipeline up to 2 `Produce` requests.

Anyway, based on those numbers the throughput performance improvement ranges from `11%` to `395%`.

When writing to multiple topic partitions like when using the `roundrobin` partitioner (i.e. writing to more than one broker) the throughput increases because multiple `brokerProducer`s are active.

If you have Kafka producers distributed around the world and a few centralized Kafka clusters like we do, writing to multiple brokers and using request pipeline can greatly improve the throughput.
The same applies if you want to mirror records between two Kafka clusters located in remote data centers.

Having a constant stream of bytes written to a TCP connection is key to increase the TCP window and therefore the throughput especially on high latency network links.

We have been using this specific enhancement in our log shipper successfully for a few years in production.
With other improvements, we are able to use Sarama to push up to 1M records/s per instance (131 Bytes records in average) and produce overall ~6M records/s during peak hours across our 37 data centers.

So thanks a lot for creating and maintaining this Go library and merry Christmas 🎄!

slaunay: I updated `async_producer.go` to use `AsyncProduce` and remove the `Pipeline` configuration bits.

Here are up to date performance results using `kafka-producer-performance` against our infrastructure with the following respective extra flags (same baseline as before):
| Target cluster | Extra flags                                                   | RTT (ms) | Max records/s | Max throughput (MiB/s) | Max requests in flight |
| -------------- | ------------------------------------------------------------- | -------: | ------------: | ---------------------: | ---------------------: |
| `us-west-2`    | `-partitioner manual`<br>`-partition 0`                       | `1`      | `83,533.7`    | `52.49`                | `6`                    |
| `us-east-1`    | `-partitioner manual`<br>`-partition 0`                       | `70`     | `67,800.0`    | `45.08`                | `6`                    |
| `eu-central-1` | `-partitioner manual`<br>`-partition 0`                       | `140`    | `32,652.3`    | `24.19`                | `6`                    |
| `eu-central-1` | `-partitioner manual`<br>`-partition 0`<br>`-max-open-requests 1` | `140`| `11,462.7`    | `10.13`                | `2`                    |
| `eu-central-1` | `-partitioner manual`<br>`-partition 0`<br>`-required-acks 0`                       | `140`    | `51,136.0`    | `36.47`                | `1`                    |
| `eu-central-1` | default `roundrobin` partitioner<br>on 5 Brokers/128 partitions  | `140` | `177,886.2`   | `75.70`                | `19`                   |

The results are consistent with the initial changes.
I added an entry for using `RequiredAcks = NoResponse` which is technically close the max throughput over a single TCP connection when we do not have to wait for the broker to acknowledge the `Produce` request (minus TCP ACKs).

I made sure most of the changes have close to full coverage but I have been running into flaky test from `functional_consumer_test.go`.
Sometimes the docker containers are not ready to accept traffic when producing records leading to tons of goroutines stuck in retry.
I believe ordering might not always be honoured when producing to the `test.1` topic:
https://github.com/Shopify/sarama/blob/fba46c96597c8aa81be2bec7ccfb54b7c2740eeb/functional_consumer_test.go#L217-L233

On top of using more than one in flight request now (in case one fails but others succeed), there is no guarantee that the `SyncProducer` will process each calls to `SendMessage` in order when using multiple goroutines.
The same probably applies to appending to the `producedMessages` slice using the `producedMessagesMu` lock.
Also calling `t.Errorf` will not work as expected as it must be called from the goroutine running the test.

Using a single synchronous call to `SendMessages` or else using the `AsyncProducer` (and waiting for all successes) would address the ordering issue (and also speed up the ingestion) if all `Produce` requests are succeeding.

Patch:
diff --git a/async_producer.go b/async_producer.go
index f842d6b..2ca9712 100644
--- a/async_producer.go
+++ b/async_producer.go
@@ -670,20 +670,12 @@ func (pp *partitionProducer) updateLeader() error {
 	})
 }
 
-type pendingResponse struct {
-	set      *produceSet
-	version  int16
-	promise  *responsePromise
-	response *ProduceResponse
-}
-
 // one per broker; also constructs an associated flusher
 func (p *asyncProducer) newBrokerProducer(broker *Broker) *brokerProducer {
 	var (
 		input     = make(chan *ProducerMessage)
 		bridge    = make(chan *produceSet)
 		responses = make(chan *brokerProducerResponse)
-		pendings  = make(chan *pendingResponse, p.conf.Net.MaxOpenRequests-1)
 	)
 
 	bp := &brokerProducer{
@@ -700,54 +692,36 @@ func (p *asyncProducer) newBrokerProducer(broker *Broker) *brokerProducer {
 
 	// minimal bridge to make the network response `select`able
 	go withRecover(func() {
-		defer close(pendings)
 		for set := range bridge {
-			set := set
-			var err error
-			var response *ProduceResponse
-			var promise *responsePromise
-
 			request := set.buildRequest()
-			if request.RequiredAcks != NoResponse {
-				response = new(ProduceResponse)
-			}
 
-			responseHeaderVersion := int16(-1)
-			if response != nil {
-				responseHeaderVersion = response.headerVersion()
-			}
-
-			promise, err = broker.send(request, response != nil, responseHeaderVersion)
-
-			// return quickly if failed or ackMode: NoResponse
-			if err != nil || promise == nil {
-				responses <- &brokerProducerResponse{
-					set: set,
-					err: err,
-					res: response,
+			// Capture the current set to forward in the callback
+			sendResponse := func(set *produceSet) ProduceCallback {
+				return func(response *ProduceResponse, err error) {
+					responses <- &brokerProducerResponse{
+						set: set,
+						err: err,
+						res: response,
+					}
 				}
-				continue
-			}
-			pending := &pendingResponse{set: set, version: request.version(), response: response, promise: promise}
-			pendings <- pending
-		}
-	})
+			}(set)
 
-	go withRecover(func() {
-		defer close(responses)
-		for pending := range pendings {
-			var err error
-			select {
-			case buf := <-pending.promise.packets:
-				err = versionedDecode(buf, pending.response, pending.version)
-			case err = <-pending.promise.errors:
+			// Use AsyncProduce vs Produce to not block waiting for the response
+			// so that we can pipeline multiple produce requests and achieve higher throughput, see:
+			// https://kafka.apache.org/protocol#protocol_network
+			err := broker.AsyncProduce(request, sendResponse)
+			if err != nil {
+				// Request failed to be sent
+				sendResponse(nil, err)
+				continue
 			}
-			responses <- &brokerProducerResponse{
-				set: pending.set,
-				err: err,
-				res: pending.response,
+			// Callback is not called when using NoResponse
+			if p.conf.Producer.RequiredAcks == NoResponse {
+				// Provide the expected nil response
+				sendResponse(nil, nil)
 			}
 		}
+		close(responses)
 	})
 
 	if p.conf.Producer.Retry.Max <= 0 {
diff --git a/broker.go b/broker.go
index ed92a66..80e824e 100644
--- a/broker.go
+++ b/broker.go
@@ -28,7 +28,7 @@ type Broker struct {
 	connErr       error
 	lock          sync.Mutex
 	opened        int32
-	responses     chan responsePromise
+	responses     chan *responsePromise
 	done          chan bool
 
 	registeredMetrics []string
@@ -121,10 +121,25 @@ type responsePromise struct {
 	requestTime   time.Time
 	correlationID int32
 	headerVersion int16
+	handler       func([]byte, error)
 	packets       chan []byte
 	errors        chan error
 }
 
+func (p *responsePromise) handle(packets []byte, err error) {
+	// Use callback when provided
+	if p.handler != nil {
+		p.handler(packets, err)
+		return
+	}
+	// Otherwise fallback to using channels
+	if err != nil {
+		p.errors <- err
+		return
+	}
+	p.packets <- packets
+}
+
 // NewBroker creates and returns a Broker targeting the given host:port address.
 // This does not attempt to actually connect, you have to call Open() for that.
 func NewBroker(addr string) *Broker {
@@ -219,7 +234,7 @@ func (b *Broker) Open(conf *Config) error {
 		}
 
 		b.done = make(chan bool)
-		b.responses = make(chan responsePromise, b.conf.Net.MaxOpenRequests-1)
+		b.responses = make(chan *responsePromise, b.conf.Net.MaxOpenRequests-1)
 
 		if b.id >= 0 {
 			DebugLogger.Printf("Connected to broker at %s (registered as #%d)\n", b.addr, b.id)
@@ -342,7 +357,55 @@ func (b *Broker) GetAvailableOffsets(request *OffsetRequest) (*OffsetResponse, e
 	return response, nil
 }
 
-// Produce returns a produce response or error
+// ProduceCallback function is called once the produce response has been parsed
+// or could not be read.
+type ProduceCallback func(*ProduceResponse, error)
+
+// AsyncProduce sends a produce request and eventually call the provided callback
+// with a produce response or an error.
+//
+// Waiting for the response is generally not blocking on the contrary to using Produce.
+// If the maximum number of in flight request configured is reached then
+// the request will be blocked till a previous response is received.
+//
+// When configured with RequiredAcks == NoResponse, the callback will not be invoked.
+// If an error is returned because the request could not be sent then the callback
+// will not be invoked either.
+func (b *Broker) AsyncProduce(request *ProduceRequest, cb ProduceCallback) error {
+	needAcks := request.RequiredAcks != NoResponse
+	// Use a nil promise when no acks is required
+	var promise *responsePromise
+
+	if needAcks {
+		// Create ProduceResponse early to provide the header version
+		res := new(ProduceResponse)
+		promise = &responsePromise{
+			headerVersion: res.headerVersion(),
+			// Packets will be converted to a ProduceResponse in the responseReceiver goroutine
+			handler: func(packets []byte, err error) {
+				if err != nil {
+					// Failed request
+					cb(nil, err)
+					return
+				}
+
+				if err := versionedDecode(packets, res, request.version()); err != nil {
+					// Malformed response
+					cb(nil, err)
+					return
+				}
+
+				// Wellformed response
+				b.updateThrottleMetric(res.ThrottleTime)
+				cb(res, nil)
+			},
+		}
+	}
+
+	return b.sendWithPromise(request, promise)
+}
+
+//Produce returns a produce response or error
 func (b *Broker) Produce(request *ProduceRequest) (*ProduceResponse, error) {
 	var (
 		response *ProduceResponse
@@ -354,15 +417,7 @@ func (b *Broker) Produce(request *ProduceRequest) (*ProduceResponse, error) {
 	} else {
 		response = new(ProduceResponse)
 		err = b.sendAndReceive(request, response)
-		if response.ThrottleTime != time.Duration(0) {
-			DebugLogger.Printf(
-				"producer/broker/%d ProduceResponse throttled %v\n",
-				b.ID(), response.ThrottleTime)
-			if b.brokerThrottleTime != nil {
-				throttleTimeInMs := int64(response.ThrottleTime / time.Millisecond)
-				b.brokerThrottleTime.Update(throttleTimeInMs)
-			}
-		}
+		b.updateThrottleMetric(response.ThrottleTime)
 	}
 
 	if err != nil {
@@ -807,24 +862,43 @@ func (b *Broker) write(buf []byte) (n int, err error) {
 }
 
 func (b *Broker) send(rb protocolBody, promiseResponse bool, responseHeaderVersion int16) (*responsePromise, error) {
+	var promise *responsePromise
+	if promiseResponse {
+		// Packets or error will be sent to the following channels
+		// once the response is received
+		promise = &responsePromise{
+			headerVersion: responseHeaderVersion,
+			packets:       make(chan []byte),
+			errors:        make(chan error),
+		}
+	}
+
+	if err := b.sendWithPromise(rb, promise); err != nil {
+		return nil, err
+	}
+
+	return promise, nil
+}
+
+func (b *Broker) sendWithPromise(rb protocolBody, promise *responsePromise) error {
 	b.lock.Lock()
 	defer b.lock.Unlock()
 
 	if b.conn == nil {
 		if b.connErr != nil {
-			return nil, b.connErr
+			return b.connErr
 		}
-		return nil, ErrNotConnected
+		return ErrNotConnected
 	}
 
 	if !b.conf.Version.IsAtLeast(rb.requiredVersion()) {
-		return nil, ErrUnsupportedVersion
+		return ErrUnsupportedVersion
 	}
 
 	req := &request{correlationID: b.correlationID, clientID: b.conf.ClientID, body: rb}
 	buf, err := encode(req, b.conf.MetricRegistry)
 	if err != nil {
-		return nil, err
+		return err
 	}
 
 	requestTime := time.Now()
@@ -834,20 +908,21 @@ func (b *Broker) send(rb protocolBody, promiseResponse bool, responseHeaderVersi
 	b.updateOutgoingCommunicationMetrics(bytes)
 	if err != nil {
 		b.addRequestInFlightMetrics(-1)
-		return nil, err
+		return err
 	}
 	b.correlationID++
 
-	if !promiseResponse {
+	if promise == nil {
 		// Record request latency without the response
 		b.updateRequestLatencyAndInFlightMetrics(time.Since(requestTime))
-		return nil, nil
+		return nil
 	}
 
-	promise := responsePromise{requestTime, req.correlationID, responseHeaderVersion, make(chan []byte), make(chan error)}
+	promise.requestTime = requestTime
+	promise.correlationID = req.correlationID
 	b.responses <- promise
 
-	return &promise, nil
+	return nil
 }
 
 func (b *Broker) sendAndReceive(req protocolBody, res protocolBody) error {
@@ -942,7 +1017,7 @@ func (b *Broker) responseReceiver() {
 			// This was previously incremented in send() and
 			// we are not calling updateIncomingCommunicationMetrics()
 			b.addRequestInFlightMetrics(-1)
-			response.errors <- dead
+			response.handle(nil, dead)
 			continue
 		}
 
@@ -954,7 +1029,7 @@ func (b *Broker) responseReceiver() {
 		if err != nil {
 			b.updateIncomingCommunicationMetrics(bytesReadHeader, requestLatency)
 			dead = err
-			response.errors <- err
+			response.handle(nil, err)
 			continue
 		}
 
@@ -963,7 +1038,7 @@ func (b *Broker) responseReceiver() {
 		if err != nil {
 			b.updateIncomingCommunicationMetrics(bytesReadHeader, requestLatency)
 			dead = err
-			response.errors <- err
+			response.handle(nil, err)
 			continue
 		}
 		if decodedHeader.correlationID != response.correlationID {
@@ -971,7 +1046,7 @@ func (b *Broker) responseReceiver() {
 			// TODO if decoded ID < cur ID, discard until we catch up
 			// TODO if decoded ID > cur ID, save it so when cur ID catches up we have a response
 			dead = PacketDecodingError{fmt.Sprintf("correlation ID didn't match, wanted %d, got %d", response.correlationID, decodedHeader.correlationID)}
-			response.errors <- dead
+			response.handle(nil, dead)
 			continue
 		}
 
@@ -980,11 +1055,11 @@ func (b *Broker) responseReceiver() {
 		b.updateIncomingCommunicationMetrics(bytesReadHeader+bytesReadBody, requestLatency)
 		if err != nil {
 			dead = err
-			response.errors <- err
+			response.handle(nil, err)
 			continue
 		}
 
-		response.packets <- buf
+		response.handle(buf, nil)
 	}
 	close(b.done)
 }
@@ -1545,6 +1620,18 @@ func (b *Broker) updateOutgoingCommunicationMetrics(bytes int) {
 	}
 }
 
+func (b *Broker) updateThrottleMetric(throttleTime time.Duration) {
+	if throttleTime != time.Duration(0) {
+		DebugLogger.Printf(
+			"producer/broker/%d ProduceResponse throttled %v\n",
+			b.ID(), throttleTime)
+		if b.brokerThrottleTime != nil {
+			throttleTimeInMs := int64(throttleTime / time.Millisecond)
+			b.brokerThrottleTime.Update(throttleTimeInMs)
+		}
+	}
+}
+
 func (b *Broker) registerMetrics() {
 	b.brokerIncomingByteRate = b.registerMeter("incoming-byte-rate")
 	b.brokerRequestRate = b.registerMeter("request-rate")
diff --git a/config.go b/config.go
index 3f86f1f..ef8bf8d 100644
--- a/config.go
+++ b/config.go
@@ -38,6 +38,9 @@ type Config struct {
 	Net struct {
 		// How many outstanding requests a connection is allowed to have before
 		// sending on it blocks (default 5).
+		// Throughput can improve but message ordering is not guaranteed if Producer.Idempotent is disabled, see:
+		// https://kafka.apache.org/protocol#protocol_network
+		// https://kafka.apache.org/28/documentation.html#producerconfigs_max.in.flight.requests.per.connection
 		MaxOpenRequests int
 
 		// All three of the below configurations are similar to the


Test patch:
diff --git a/broker_test.go b/broker_test.go
index 1afa0c8..a9e7da6 100644
--- a/broker_test.go
+++ b/broker_test.go
@@ -78,6 +78,38 @@ func TestBrokerAccessors(t *testing.T) {
 	}
 }
 
+type produceResponsePromise struct {
+	c chan produceResOrError
+}
+
+type produceResOrError struct {
+	res *ProduceResponse
+	err error
+}
+
+func newProduceResponsePromise() produceResponsePromise {
+	return produceResponsePromise{
+		c: make(chan produceResOrError, 0),
+	}
+}
+
+func (p produceResponsePromise) callback(res *ProduceResponse, err error) {
+	if err != nil {
+		p.c <- produceResOrError{
+			err: err,
+		}
+		return
+	}
+	p.c <- produceResOrError{
+		res: res,
+	}
+}
+
+func (p produceResponsePromise) Get() (*ProduceResponse, error) {
+	resOrError := <-p.c
+	return resOrError.res, resOrError.err
+}
+
 func TestSimpleBrokerCommunication(t *testing.T) {
 	for _, tt := range brokerTestTable {
 		t.Run(tt.name, func(t *testing.T) {
@@ -118,6 +150,47 @@ func TestSimpleBrokerCommunication(t *testing.T) {
 	}
 }
 
+func TestBrokerFailedRequest(t *testing.T) {
+	for _, tt := range brokerFailedReqTestTable {
+		t.Run(tt.name, func(t *testing.T) {
+			t.Logf("Testing broker communication for %s", tt.name)
+			mb := NewMockBroker(t, 0)
+			if !tt.stopBroker {
+				mb.Returns(&mockEncoder{tt.response})
+			}
+			broker := NewBroker(mb.Addr())
+			// Stop the broker before calling the runner to purposefully
+			// make the request fail right away, the port will be closed
+			// and should not be reused right away
+			if tt.stopBroker {
+				t.Log("Closing broker:", mb.Addr())
+				mb.Close()
+			}
+			conf := NewTestConfig()
+			conf.ApiVersionsRequest = false
+			conf.Version = tt.version
+			// Tune read timeout to speed up some test cases
+			conf.Net.ReadTimeout = 1 * time.Second
+			err := broker.Open(conf)
+			if err != nil {
+				t.Fatal(err)
+			}
+			tt.runner(t, broker)
+			if !tt.stopBroker {
+				mb.Close()
+			}
+			err = broker.Close()
+			if err != nil {
+				if tt.stopBroker && err == ErrNotConnected {
+					// We expect the broker to not close properly
+					return
+				}
+				t.Error(err)
+			}
+		})
+	}
+}
+
 var ErrTokenFailure = errors.New("Failure generating token")
 
 type TokenProvider struct {
@@ -804,6 +877,20 @@ var brokerTestTable = []struct {
 		},
 	},
 
+	{
+		V0_10_0_0,
+		"ProduceRequest (NoResponse) using AsyncProduce",
+		[]byte{},
+		func(t *testing.T, broker *Broker) {
+			request := ProduceRequest{}
+			request.RequiredAcks = NoResponse
+			err := broker.AsyncProduce(&request, nil)
+			if err != nil {
+				t.Error(err)
+			}
+		},
+	},
+
 	{
 		V0_10_0_0,
 		"ProduceRequest (WaitForLocal)",
@@ -821,6 +908,28 @@ var brokerTestTable = []struct {
 		},
 	},
 
+	{
+		V0_10_0_0,
+		"ProduceRequest (WaitForLocal) using AsyncProduce",
+		[]byte{0x00, 0x00, 0x00, 0x00},
+		func(t *testing.T, broker *Broker) {
+			request := ProduceRequest{}
+			request.RequiredAcks = WaitForLocal
+			produceResPromise := newProduceResponsePromise()
+			err := broker.AsyncProduce(&request, produceResPromise.callback)
+			if err != nil {
+				t.Error(err)
+			}
+			response, err := produceResPromise.Get()
+			if err != nil {
+				t.Error(err)
+			}
+			if response == nil {
+				t.Error("Produce request without NoResponse got no response!")
+			}
+		},
+	},
+
 	{
 		V0_10_0_0,
 		"FetchRequest",
@@ -1030,6 +1139,93 @@ var brokerTestTable = []struct {
 	},
 }
 
+// We are testing the handling of failed request or corrupt responses.
+var brokerFailedReqTestTable = []struct {
+	version    KafkaVersion
+	name       string
+	stopBroker bool
+	response   []byte
+	runner     func(*testing.T, *Broker)
+}{
+	{
+		version:    V0_10_0_0,
+		name:       "ProduceRequest (NoResponse) using AsyncProduce and stopped broker",
+		stopBroker: true,
+		runner: func(t *testing.T, broker *Broker) {
+			request := ProduceRequest{}
+			request.RequiredAcks = NoResponse
+			err := broker.AsyncProduce(&request, nil)
+			if err == nil {
+				t.Fatal("Expected a non nil error because broker is not listening")
+			}
+			t.Log("Got error:", err)
+		},
+	},
+
+	{
+		version:    V0_10_0_0,
+		name:       "ProduceRequest (WaitForLocal) using AsyncProduce and stopped broker",
+		stopBroker: true,
+		runner: func(t *testing.T, broker *Broker) {
+			request := ProduceRequest{}
+			request.RequiredAcks = WaitForLocal
+			err := broker.AsyncProduce(&request, nil)
+			if err == nil {
+				t.Fatal("Expected a non nil error because broker is not listening")
+			}
+			t.Log("Got error:", err)
+		},
+	},
+
+	{
+		version: V0_10_0_0,
+		name:    "ProduceRequest (WaitForLocal) using AsyncProduce and no response",
+		// A nil response means the mock broker will ignore the request leading to a read timeout
+		response: nil,
+		runner: func(t *testing.T, broker *Broker) {
+			request := ProduceRequest{}
+			request.RequiredAcks = WaitForLocal
+			produceResPromise := newProduceResponsePromise()
+			err := broker.AsyncProduce(&request, produceResPromise.callback)
+			if err != nil {
+				t.Error(err)
+			}
+			response, err := produceResPromise.Get()
+			if err == nil {
+				t.Fatal("Expected a non nil error because broker is not listening")
+			}
+			t.Log("Got error:", err)
+			if response != nil {
+				t.Error("Produce request should have failed, got response:", response)
+			}
+		},
+	},
+
+	{
+		version: V0_10_0_0,
+		name:    "ProduceRequest (WaitForLocal) using AsyncProduce and corrupt response",
+		// Corrupt response (3 bytes vs 4)
+		response: []byte{0x00, 0x00, 0x00},
+		runner: func(t *testing.T, broker *Broker) {
+			request := ProduceRequest{}
+			request.RequiredAcks = WaitForLocal
+			produceResPromise := newProduceResponsePromise()
+			err := broker.AsyncProduce(&request, produceResPromise.callback)
+			if err != nil {
+				t.Error(err)
+			}
+			response, err := produceResPromise.Get()
+			if err == nil {
+				t.Fatal(err)
+			}
+			t.Log("Got error:", err)
+			if response != nil {
+				t.Error("Produce request should have failed, got response:", response)
+			}
+		},
+	},
+}
+
 func validateBrokerMetrics(t *testing.T, broker *Broker, mockBrokerMetrics brokerMetrics) {
 	metricValidators := newMetricValidators()
 	mockBrokerBytesRead := mockBrokerMetrics.bytesRead