Repository: Shopify/sarama
Pull request #1539 (2019-11-14 21:32:11)

Add requests-in-flight metric
# Feature

Add `requests-in-flight` metric similar to the Kafka Java client described as:

> The current number of in-flight requests awaiting a response.

This is an interesting metric to monitor for optimizing throughput when writing to a Kafka cluster on 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.

The related configuration in Sarama is `config.MaxOpenRequests` and defaults to `5` like the Java Kafka client (`max.in.flight.requests.per.connection`).

This configuration is actually not fully honoured when using the `AsyncProducer`, more on that below.

## Changes

- add `requests-in-flight` and `requests-in-flight-for-broker-<brokerID>` metrics to the broker
- move some `b.updateOutgoingCommunicationMetrics(...)` just after `b.write(...)` for consistency
- add `-max-open-requests` flag to `kafka-producer-performance`
- update `kafka-producer-performance` to monitor total requests in flight
- update unit tests
- document new metrics

## Testing done

Running the updated tests:
```
$ export KAFKA_PEERS=192.168.100.67:9091,192.168.100.67:9092,192.168.100.67:9093,192.168.100.67:9094,192.168.100.67:9095
$ go test ./...
ok      github.com/Shopify/sarama       207.703s
ok      github.com/Shopify/sarama/examples/http_server  0.057s
?       github.com/Shopify/sarama/examples/sasl_scram_client    [no test files]
ok      github.com/Shopify/sarama/mocks 0.056s
?       github.com/Shopify/sarama/tools/kafka-console-consumer  [no test files]
?       github.com/Shopify/sarama/tools/kafka-console-partitionconsumer [no test files]
?       github.com/Shopify/sarama/tools/kafka-console-producer  [no test files]
?       github.com/Shopify/sarama/tools/kafka-producer-performance      [no test files]
?       github.com/Shopify/sarama/tools/tls     [no test files]
```

Using the updated `kafka-producer-performance` to see how many total requests are in flight when writing to 5 brokers (with a network latency of ~70ms):
```
$ ./kafka-producer-performance \
  -brokers kafka:9093 \
  -security-protocol SSL \
  -topic topic \
  -message-load 1000000 \
  -message-size 1000 \
  -required-acks -1 \
  -flush-bytes 500000 \
  -flush-frequency 100ms \
  | sed 's/latency,.*,/latency,/'
449761 records sent, 96105.2 records/sec (91.65 MiB/sec ingress, 29.84 MiB/sec egress), 302.5 ms avg latency, 5 total req. in flight
692402 records sent, 63983.6 records/sec (61.02 MiB/sec ingress, 32.43 MiB/sec egress), 382.8 ms avg latency, 1 total req. in flight
981950 records sent, 66890.9 records/sec (63.79 MiB/sec ingress, 38.83 MiB/sec egress), 380.3 ms avg latency, 5 total req. in flight
1000000 records sent, 59960.3 records/sec (57.18 MiB/sec ingress, 36.58 MiB/sec egress), 379.7 ms avg latency, 0 total req. in flight
```

Note that the maximum number of requests in flight is `5` which corresponds to the number of brokers (the topic has 64 partitions spread mostly evenly between those 5 brokers).

And when writing to a single broker/topic partition:
```
$ ./kafka-producer-performance \
  -brokers kafka:9093 \
  -security-protocol SSL \
  -topic topic \
  -message-load 200000 \
  -message-size 1000 \
  -required-acks -1 \
  -flush-bytes 500000 \
  -flush-frequency 100ms \
  | sed 's/latency,.*,/latency,/'
24838 records sent, 9202.7 records/sec (8.78 MiB/sec ingress, 5.09 MiB/sec egress), 100.8 ms avg latency, 1 total req. in flight
76460 records sent, 9931.2 records/sec (9.47 MiB/sec ingress, 7.66 MiB/sec egress), 94.7 ms avg latency, 1 total req. in flight
131004 records sent, 10316.1 records/sec (9.84 MiB/sec ingress, 8.68 MiB/sec egress), 90.9 ms avg latency, 1 total req. in flight
187496 records sent, 10593.6 records/sec (10.10 MiB/sec ingress, 9.28 MiB/sec egress), 88.5 ms avg latency, 1 total req. in flight
200000 records sent, 10549.3 records/sec (10.06 MiB/sec ingress, 9.31 MiB/sec egress), 88.8 ms avg latency, 0 total req. in flight
```

As you can see the number of requests in flight seems to never go above `1` per broker despite having `config.MaxOpenRequests = 5`.

This can be explained by a few things:
- the metric exposed by the `kafka-producer-performance` is captured every 5 seconds and there might not be much requests in flight at that time (could even be `0` when accumulating records or building a `ProduceRequest` prior to sending it)
- the request might be acknowledged fairly fast (hence the scenario with a ~70ms network latency and using `-required-acks=-1`)
- the `AsyncProducer` only allows a single `ProduceRequest` in flight at a time to a broker

The last point can be seen when looking at the following section of the `brokerProducer` where the "bridge" goroutine builds and forwards `ProduceRequest`s one at a time:
https://github.com/Shopify/sarama/blob/675b0b1ff204c259877004140a540d6adf38db17/async_producer.go#L660-L674
and the fact that the `Produce` receiver on a `Broker` is blocking till a response is received or an error occurs:
https://github.com/Shopify/sarama/blob/675b0b1ff204c259877004140a540d6adf38db17/broker.go#L336-L355

I have another contribution pending to address that limitation without creating more go routines that results in better throughput in such scenario.
Nevertheless having such metric is one way to check how many requests are awaiting a response to a given broker or to the whole cluster.

d1egoaz: @slaunay this looks amazing! looking forward to see the final results!
slaunay: Looking into why the build failed, it seems that toxiproxy fails to establish connection to the target broker `localhost:29091` when running `TestFuncProducingSnappy` in configuration `Go: 1.11.x,KAFKA_VERSION=2.2.1,KAFKA_SCALA_VERSION=2.12` (it passed in all the other configurations).

Here are the logs from the failed build (
https://travis-ci.org/Shopify/sarama/jobs/612081678#L12183-L12217) with some annotations:
```
=== RUN   TestFuncProducingSnappy
[sarama] 2019/11/14 21:59:11 Initializing new client
[sarama] 2019/11/14 21:59:11 ClientID is the default of 'sarama', you should consider setting it to something application-specific.
[sarama] 2019/11/14 21:59:11 ClientID is the default of 'sarama', you should consider setting it to something application-specific.
[sarama] 2019/11/14 21:59:11 client/metadata fetching metadata for all topics from broker localhost:9091
INFO[0376] Accepted client                               client=127.0.0.1:49120 name=kafka1 proxy=[::]:9091 upstream=localhost:29091
ERRO[0376] Unable to open connection to upstream         client=127.0.0.1:49120 name=kafka1 proxy=[::]:9091 upstream=localhost:29091 # <-- bad proxy
[sarama] 2019/11/14 21:59:11 Connected to broker at localhost:9091 (unregistered)
[sarama] 2019/11/14 21:59:11 client/metadata got error from broker -1 while fetching metadata: EOF # <-- read on a closed connection from toxiproxy
[sarama] 2019/11/14 21:59:11 Closed connection to broker localhost:9091
[sarama] 2019/11/14 21:59:11 ClientID is the default of 'sarama', you should consider setting it to something application-specific.
[sarama] 2019/11/14 21:59:11 client/metadata fetching metadata for all topics from broker localhost:9092
INFO[0376] Accepted client                               client=127.0.0.1:32824 name=kafka2 proxy=[::]:9092 upstream=localhost:29092
[sarama] 2019/11/14 21:59:11 Connected to broker at localhost:9092 (unregistered) # <-- fallback to the next seed broker
[sarama] 2019/11/14 21:59:11 client/brokers registered new broker #9093 at localhost:9093
[sarama] 2019/11/14 21:59:11 client/brokers registered new broker #9092 at localhost:9092
[sarama] 2019/11/14 21:59:11 client/brokers registered new broker #9094 at localhost:9094
[sarama] 2019/11/14 21:59:11 client/brokers registered new broker #9095 at localhost:9095
[sarama] 2019/11/14 21:59:11 Successfully initialized new client
[sarama] 2019/11/14 21:59:11 ClientID is the default of 'sarama', you should consider setting it to something application-specific.
INFO[0376] Accepted client                               client=127.0.0.1:48224 name=kafka5 proxy=[::]:9095 upstream=localhost:29095
[sarama] 2019/11/14 21:59:11 Connected to broker at localhost:9095 (registered as #9095)
[sarama] 2019/11/14 21:59:11 producer/broker/9095 starting up
[sarama] 2019/11/14 21:59:11 producer/broker/9095 state change to [open] on test.1/0
[sarama] 2019/11/14 21:59:12 Producer shutting down.
[sarama] 2019/11/14 21:59:12 producer/broker/9095 input chan closed
[sarama] 2019/11/14 21:59:12 producer/broker/9095 shut down
[sarama] 2019/11/14 21:59:12 consumer/broker/9095 added subscription to test.1/0
[sarama] 2019/11/14 21:59:12 consumer/broker/9095 closed dead subscription to test.1/0
[sarama] 2019/11/14 21:59:12 Closing Client
WARN[0377] Source terminated                             bytes=18356 err=read tcp 127.0.0.1:49024->127.0.0.1:29095: use of closed network connection name=kafka5
[sarama] 2019/11/14 21:59:12 Closed connection to broker localhost:9095
[sarama] 2019/11/14 21:59:12 Closed connection to broker localhost:9092
WARN[0377] Source terminated                             bytes=7360 err=read tcp 127.0.0.1:59590->127.0.0.1:29092: use of closed network connection name=kafka2
--- FAIL: TestFuncProducingSnappy (1.03s)
    metrics_test.go:160: Expected histogram metric 'request-latency-in-ms' min >= 10, got 0
    metrics_test.go:160: Expected histogram metric 'response-size' min >= 1, got 0
```


The failed initial metadata request led to a connection to a different broker `localhost:29092` and eventually to `127.0.0.1:29095` for producing.

In that test case the metrics are validated against the cluster (i.e. all brokers) and a specific one `localhost:9095` (leader for partition `0` of topic `test.1`).
But when sending the metadata request to `localhost:9091`, I believe the `connect` and `write`  on the TCP socket worked but the `read` failed immediately with no bytes read (meaning a `request-latency-in-ms` metric for that broker was recorded with `0 ms` and a `response-size` metric was recorded with `0 B`).
Because the cluster metrics includes broker `localhost:9091` the minimum latency and response size were `0` hence the failure.

Looking into that build, it seems that there are plenty of those errors:
```
[sarama] ... Connected to broker at localhost:9091 (unregistered)
[sarama] ... client/metadata got error from broker -1 while fetching metadata: EOF
[sarama] ... Closed connection to broker localhost:9091
```

My guess is that broker number 1 stopped somehow (there is no `server.log` to prove that) and unfortunately was used as first seed broker in this test case.
This would not be an issue if we were connecting directly to the broker as it would fail and we would not record metrics but when using toxiproxy the connection can still be established and the first write is successful (probably because any upstream connection is accepted but the downstream connection is only established after the first write).

Note that this failure is not related to the current PR but something that probably happens from time to time so I will submit a separate PR to address it.
bai: 👋 I'm about to cut a new release when kafka 2.4.0 is released in a few days. Any further thoughts on this PR?
slaunay: You are probably asking the community but I would love to this integrated as I would then be able to propose some changes to fully honour `MaxOpenRequests` when using the `AsyncProducer`.

The changes in `kafka-producer-performance` will allow to easily see the `requests-in-flight-for-broker-B` metric go above 1 and the throughput increase when producing on a high latency network link (typically across regions).

I believe it will be a great addition to improve performance.
slaunay: Last build failed with configuration `Go: 1.12.x KAFKA_VERSION=2.3.0 KAFKA_SCALA_VERSION=2.12`:
```
=== RUN   TestAsyncProducerRetryWithReferenceOpen
[sarama] 2019/12/03 17:13:00 *** mockbroker/1 listening on 127.0.0.1:40597
[sarama] 2019/12/03 17:13:00 *** mockbroker/2 listening on 127.0.0.1:42303
[sarama] 2019/12/03 17:13:00 Initializing new client
[sarama] 2019/12/03 17:13:00 ClientID is the default of 'sarama', you should consider setting it to something application-specific.
[sarama] 2019/12/03 17:13:00 ClientID is the default of 'sarama', you should consider setting it to something application-specific.
[sarama] 2019/12/03 17:13:00 client/metadata fetching metadata for all topics from broker 127.0.0.1:40597
[sarama] 2019/12/03 17:13:00 Connected to broker at 127.0.0.1:40597 (unregistered)
[sarama] 2019/12/03 17:13:00 *** mockbroker/1/0: connection opened
[sarama] 2019/12/03 17:13:00 *** mockbroker/1/0: served &{0 sarama 0xc0004331a0} -> &{0 0 [0xc0003f4700] <nil> 0 [0xc0000a7300]}
[sarama] 2019/12/03 17:13:00 client/brokers registered new broker #2 at 127.0.0.1:42303
[sarama] 2019/12/03 17:13:00 Successfully initialized new client
[sarama] 2019/12/03 17:13:00 ClientID is the default of 'sarama', you should consider setting it to something application-specific.
[sarama] 2019/12/03 17:13:00 Connected to broker at 127.0.0.1:42303 (registered as #2)
[sarama] 2019/12/03 17:13:00 *** mockbroker/2/0: connection opened
[sarama] 2019/12/03 17:13:00 producer/broker/2 starting up
[sarama] 2019/12/03 17:13:00 producer/broker/2 state change to [open] on my_topic/0
[sarama] 2019/12/03 17:13:00 *** mockbroker/2/0: served &{0 sarama 0xc0000af4c0} -> &{map[my_topic:map[0:0xc000381f80]] 0 0s}
[sarama] 2019/12/03 17:13:00 producer/broker/2 state change to [open] on my_topic/1
[sarama] 2019/12/03 17:13:00 *** mockbroker/2/0: served &{1 sarama 0xc0000af700} -> &{map[my_topic:map[1:0xc000433680]] 0 0s}
[sarama] 2019/12/03 17:13:00 *** mockbroker/2/0: invalid request: err=read tcp 127.0.0.1:42303->127.0.0.1:47188: use of closed network connection, ([]uint8) <nil>
[sarama] 2019/12/03 17:13:00 *** mockbroker/2/0: connection closed, err=<nil>
[sarama] 2019/12/03 17:13:00 *** mockbroker/2: listener closed, err=accept tcp 127.0.0.1:42303: use of closed network connection
--- FAIL: TestAsyncProducerRetryWithReferenceOpen (0.01s)
    mockbroker.go:366: listen tcp 127.0.0.1:42303: bind: address already in use
```


It seems that there is a race condition between closing the TCP listener and creating a new one on the same port:
https://github.com/Shopify/sarama/blob/c82acaf7831341c57794be79578431c927540840/async_producer_test.go#L729-L731

Not sure why though as there is some synchronization in place to ensure that `b.listener.Close()` is called before `*MockBroker.Close()` finishes using the `stopper` channel: 
https://github.com/Shopify/sarama/blob/c82acaf7831341c57794be79578431c927540840/mockbroker.go#L131-L166

It is possible but unlikely that another test or process has reused that port before it is used on the new `MockBroker`.

I have not seen that failure before but I do not believe it is related to this PR.
bai: It's just a flaky test, I restarted the affected build.

Patch:
diff --git a/broker.go b/broker.go
index de83987..d27ebd2 100644
--- a/broker.go
+++ b/broker.go
@@ -40,6 +40,7 @@ type Broker struct {
 	outgoingByteRate       metrics.Meter
 	responseRate           metrics.Meter
 	responseSize           metrics.Histogram
+	requestsInFlight       metrics.Counter
 	brokerIncomingByteRate metrics.Meter
 	brokerRequestRate      metrics.Meter
 	brokerRequestSize      metrics.Histogram
@@ -47,6 +48,7 @@ type Broker struct {
 	brokerOutgoingByteRate metrics.Meter
 	brokerResponseRate     metrics.Meter
 	brokerResponseSize     metrics.Histogram
+	brokerRequestsInFlight metrics.Counter
 
 	kerberosAuthenticator GSSAPIKerberosAuth
 }
@@ -182,6 +184,7 @@ func (b *Broker) Open(conf *Config) error {
 		b.outgoingByteRate = metrics.GetOrRegisterMeter("outgoing-byte-rate", conf.MetricRegistry)
 		b.responseRate = metrics.GetOrRegisterMeter("response-rate", conf.MetricRegistry)
 		b.responseSize = getOrRegisterHistogram("response-size", conf.MetricRegistry)
+		b.requestsInFlight = metrics.GetOrRegisterCounter("requests-in-flight", conf.MetricRegistry)
 		// Do not gather metrics for seeded broker (only used during bootstrap) because they share
 		// the same id (-1) and are already exposed through the global metrics above
 		if b.id >= 0 {
@@ -712,16 +715,19 @@ func (b *Broker) send(rb protocolBody, promiseResponse bool) (*responsePromise,
 	}
 
 	requestTime := time.Now()
+	// Will be decremented in responseReceiver (except error or request with NoResponse)
+	b.addRequestInFlightMetrics(1)
 	bytes, err := b.write(buf)
 	b.updateOutgoingCommunicationMetrics(bytes)
 	if err != nil {
+		b.addRequestInFlightMetrics(-1)
 		return nil, err
 	}
 	b.correlationID++
 
 	if !promiseResponse {
 		// Record request latency without the response
-		b.updateRequestLatencyMetrics(time.Since(requestTime))
+		b.updateRequestLatencyAndInFlightMetrics(time.Since(requestTime))
 		return nil, nil
 	}
 
@@ -816,6 +822,9 @@ func (b *Broker) responseReceiver() {
 
 	for response := range b.responses {
 		if dead != nil {
+			// This was previously incremented in send() and
+			// we are not calling updateIncomingCommunicationMetrics()
+			b.addRequestInFlightMetrics(-1)
 			response.errors <- dead
 			continue
 		}
@@ -891,9 +900,12 @@ func (b *Broker) sendAndReceiveSASLHandshake(saslType SASLMechanism, version int
 	}
 
 	requestTime := time.Now()
+	// Will be decremented in updateIncomingCommunicationMetrics (except error)
+	b.addRequestInFlightMetrics(1)
 	bytes, err := b.write(buf)
 	b.updateOutgoingCommunicationMetrics(bytes)
 	if err != nil {
+		b.addRequestInFlightMetrics(-1)
 		Logger.Printf("Failed to send SASL handshake %s: %s\n", b.addr, err.Error())
 		return err
 	}
@@ -902,6 +914,7 @@ func (b *Broker) sendAndReceiveSASLHandshake(saslType SASLMechanism, version int
 	header := make([]byte, 8) // response header
 	_, err = b.readFull(header)
 	if err != nil {
+		b.addRequestInFlightMetrics(-1)
 		Logger.Printf("Failed to read SASL handshake header : %s\n", err.Error())
 		return err
 	}
@@ -910,6 +923,7 @@ func (b *Broker) sendAndReceiveSASLHandshake(saslType SASLMechanism, version int
 	payload := make([]byte, length-4)
 	n, err := b.readFull(payload)
 	if err != nil {
+		b.addRequestInFlightMetrics(-1)
 		Logger.Printf("Failed to read SASL handshake payload : %s\n", err.Error())
 		return err
 	}
@@ -981,9 +995,12 @@ func (b *Broker) sendAndReceiveV0SASLPlainAuth() error {
 	copy(authBytes[4:], []byte(b.conf.Net.SASL.AuthIdentity+"\x00"+b.conf.Net.SASL.User+"\x00"+b.conf.Net.SASL.Password))
 
 	requestTime := time.Now()
+	// Will be decremented in updateIncomingCommunicationMetrics (except error)
+	b.addRequestInFlightMetrics(1)
 	bytesWritten, err := b.write(authBytes)
 	b.updateOutgoingCommunicationMetrics(bytesWritten)
 	if err != nil {
+		b.addRequestInFlightMetrics(-1)
 		Logger.Printf("Failed to write SASL auth header to broker %s: %s\n", b.addr, err.Error())
 		return err
 	}
@@ -1008,11 +1025,13 @@ func (b *Broker) sendAndReceiveV1SASLPlainAuth() error {
 
 	requestTime := time.Now()
 
+	// Will be decremented in updateIncomingCommunicationMetrics (except error)
+	b.addRequestInFlightMetrics(1)
 	bytesWritten, err := b.sendSASLPlainAuthClientResponse(correlationID)
-
 	b.updateOutgoingCommunicationMetrics(bytesWritten)
 
 	if err != nil {
+		b.addRequestInFlightMetrics(-1)
 		Logger.Printf("Failed to write SASL auth header to broker %s: %s\n", b.addr, err.Error())
 		return err
 	}
@@ -1066,14 +1085,17 @@ func (b *Broker) sendAndReceiveSASLOAuth(provider AccessTokenProvider) error {
 // rejected.
 func (b *Broker) sendClientMessage(message []byte) (bool, error) {
 	requestTime := time.Now()
+	// Will be decremented in updateIncomingCommunicationMetrics (except error)
+	b.addRequestInFlightMetrics(1)
 	correlationID := b.correlationID
 
 	bytesWritten, err := b.sendSASLOAuthBearerClientMessage(message, correlationID)
+	b.updateOutgoingCommunicationMetrics(bytesWritten)
 	if err != nil {
+		b.addRequestInFlightMetrics(-1)
 		return false, err
 	}
 
-	b.updateOutgoingCommunicationMetrics(bytesWritten)
 	b.correlationID++
 
 	res := &SaslAuthenticateResponse{}
@@ -1108,17 +1130,21 @@ func (b *Broker) sendAndReceiveSASLSCRAMv1() error {
 
 	for !scramClient.Done() {
 		requestTime := time.Now()
+		// Will be decremented in updateIncomingCommunicationMetrics (except error)
+		b.addRequestInFlightMetrics(1)
 		correlationID := b.correlationID
 		bytesWritten, err := b.sendSaslAuthenticateRequest(correlationID, []byte(msg))
+		b.updateOutgoingCommunicationMetrics(bytesWritten)
 		if err != nil {
+			b.addRequestInFlightMetrics(-1)
 			Logger.Printf("Failed to write SASL auth header to broker %s: %s\n", b.addr, err.Error())
 			return err
 		}
 
-		b.updateOutgoingCommunicationMetrics(bytesWritten)
 		b.correlationID++
 		challenge, err := b.receiveSaslAuthenticateResponse(correlationID)
 		if err != nil {
+			b.addRequestInFlightMetrics(-1)
 			Logger.Printf("Failed to read response while authenticating with SASL to broker %s: %s\n", b.addr, err.Error())
 			return err
 		}
@@ -1271,7 +1297,7 @@ func (b *Broker) receiveSASLServerResponse(res *SaslAuthenticateResponse, correl
 }
 
 func (b *Broker) updateIncomingCommunicationMetrics(bytes int, requestLatency time.Duration) {
-	b.updateRequestLatencyMetrics(requestLatency)
+	b.updateRequestLatencyAndInFlightMetrics(requestLatency)
 	b.responseRate.Mark(1)
 
 	if b.brokerResponseRate != nil {
@@ -1290,13 +1316,22 @@ func (b *Broker) updateIncomingCommunicationMetrics(bytes int, requestLatency ti
 	}
 }
 
-func (b *Broker) updateRequestLatencyMetrics(requestLatency time.Duration) {
+func (b *Broker) updateRequestLatencyAndInFlightMetrics(requestLatency time.Duration) {
 	requestLatencyInMs := int64(requestLatency / time.Millisecond)
 	b.requestLatency.Update(requestLatencyInMs)
 
 	if b.brokerRequestLatency != nil {
 		b.brokerRequestLatency.Update(requestLatencyInMs)
 	}
+
+	b.addRequestInFlightMetrics(-1)
+}
+
+func (b *Broker) addRequestInFlightMetrics(i int64) {
+	b.requestsInFlight.Inc(i)
+	if b.brokerRequestsInFlight != nil {
+		b.brokerRequestsInFlight.Inc(i)
+	}
 }
 
 func (b *Broker) updateOutgoingCommunicationMetrics(bytes int) {
@@ -1325,6 +1360,7 @@ func (b *Broker) registerMetrics() {
 	b.brokerOutgoingByteRate = b.registerMeter("outgoing-byte-rate")
 	b.brokerResponseRate = b.registerMeter("response-rate")
 	b.brokerResponseSize = b.registerHistogram("response-size")
+	b.brokerRequestsInFlight = b.registerCounter("requests-in-flight")
 }
 
 func (b *Broker) unregisterMetrics() {
@@ -1344,3 +1380,9 @@ func (b *Broker) registerHistogram(name string) metrics.Histogram {
 	b.registeredMetrics = append(b.registeredMetrics, nameForBroker)
 	return getOrRegisterHistogram(nameForBroker, b.conf.MetricRegistry)
 }
+
+func (b *Broker) registerCounter(name string) metrics.Counter {
+	nameForBroker := getMetricNameForBroker(name, b)
+	b.registeredMetrics = append(b.registeredMetrics, nameForBroker)
+	return metrics.GetOrRegisterCounter(nameForBroker, b.conf.MetricRegistry)
+}
diff --git a/sarama.go b/sarama.go
index 1e0277a..48f362d 100644
--- a/sarama.go
+++ b/sarama.go
@@ -39,6 +39,10 @@ Broker related metrics:
 	| response-rate-for-broker-<broker-id>         | meter      | Responses/second received from a given broker                 |
 	| response-size                                | histogram  | Distribution of the response size in bytes for all brokers    |
 	| response-size-for-broker-<broker-id>         | histogram  | Distribution of the response size in bytes for a given broker |
+	| requests-in-flight                           | counter    | The current number of in-flight requests awaiting a response  |
+	|                                              |            | for all brokers                                               |
+	| requests-in-flight-for-broker-<broker-id>    | counter    | The current number of in-flight requests awaiting a response  |
+	|                                              |            | for a given broker                                            |
 	+----------------------------------------------+------------+---------------------------------------------------------------+
 
 Note that we do not gather specific metrics for seed brokers but they are part of the "all brokers" metrics.
diff --git a/tools/kafka-producer-performance/main.go b/tools/kafka-producer-performance/main.go
index 7bf1531..e1a8163 100644
--- a/tools/kafka-producer-performance/main.go
+++ b/tools/kafka-producer-performance/main.go
@@ -80,6 +80,11 @@ var (
 		0,
 		"The maximum number of messages to send per second (0 for no limit).",
 	)
+	maxOpenRequests = flag.Int(
+		"max-open-requests",
+		5,
+		"The maximum number of unacknowledged requests the client will send on a single connection before blocking (default: 5).",
+	)
 	maxMessageBytes = flag.Int(
 		"max-message-bytes",
 		1000000,
@@ -248,6 +253,7 @@ func main() {
 
 	config := sarama.NewConfig()
 
+	config.Net.MaxOpenRequests = *maxOpenRequests
 	config.Producer.MaxMessageBytes = *maxMessageBytes
 	config.Producer.RequiredAcks = sarama.RequiredAcks(*requiredAcks)
 	config.Producer.Timeout = *timeout
@@ -430,17 +436,20 @@ func printMetrics(w io.Writer, r metrics.Registry) {
 	recordSendRateMetric := r.Get("record-send-rate")
 	requestLatencyMetric := r.Get("request-latency-in-ms")
 	outgoingByteRateMetric := r.Get("outgoing-byte-rate")
+	requestsInFlightMetric := r.Get("requests-in-flight")
 
-	if recordSendRateMetric == nil || requestLatencyMetric == nil || outgoingByteRateMetric == nil {
+	if recordSendRateMetric == nil || requestLatencyMetric == nil || outgoingByteRateMetric == nil ||
+		requestsInFlightMetric == nil {
 		return
 	}
 	recordSendRate := recordSendRateMetric.(metrics.Meter).Snapshot()
 	requestLatency := requestLatencyMetric.(metrics.Histogram).Snapshot()
 	requestLatencyPercentiles := requestLatency.Percentiles([]float64{0.5, 0.75, 0.95, 0.99, 0.999})
 	outgoingByteRate := outgoingByteRateMetric.(metrics.Meter).Snapshot()
+	requestsInFlight := requestsInFlightMetric.(metrics.Counter).Count()
 	fmt.Fprintf(w, "%d records sent, %.1f records/sec (%.2f MiB/sec ingress, %.2f MiB/sec egress), "+
 		"%.1f ms avg latency, %.1f ms stddev, %.1f ms 50th, %.1f ms 75th, "+
-		"%.1f ms 95th, %.1f ms 99th, %.1f ms 99.9th\n",
+		"%.1f ms 95th, %.1f ms 99th, %.1f ms 99.9th, %d total req. in flight\n",
 		recordSendRate.Count(),
 		recordSendRate.RateMean(),
 		recordSendRate.RateMean()*float64(*messageSize)/1024/1024,
@@ -452,6 +461,7 @@ func printMetrics(w io.Writer, r metrics.Registry) {
 		requestLatencyPercentiles[2],
 		requestLatencyPercentiles[3],
 		requestLatencyPercentiles[4],
+		requestsInFlight,
 	)
 }
 


Test patch:
diff --git a/broker_test.go b/broker_test.go
index 224971a..4c759ea 100644
--- a/broker_test.go
+++ b/broker_test.go
@@ -217,6 +217,7 @@ func TestSASLOAuthBearer(t *testing.T) {
 		broker.responseSize = metrics.NilHistogram{}
 		broker.responseRate = metrics.NilMeter{}
 		broker.requestLatency = metrics.NilHistogram{}
+		broker.requestsInFlight = metrics.NilCounter{}
 
 		conf := NewConfig()
 		conf.Net.SASL.Mechanism = SASLTypeOAuth
@@ -335,6 +336,7 @@ func TestSASLSCRAMSHAXXX(t *testing.T) {
 		broker.responseSize = metrics.NilHistogram{}
 		broker.responseRate = metrics.NilMeter{}
 		broker.requestLatency = metrics.NilHistogram{}
+		broker.requestsInFlight = metrics.NilCounter{}
 
 		mockSASLAuthResponse := NewMockSaslAuthenticateResponse(t).SetAuthBytes([]byte(test.scramChallengeResp))
 		mockSASLHandshakeResponse := NewMockSaslHandshakeResponse(t).SetEnabledMechanisms([]string{SASLTypeSCRAMSHA256, SASLTypeSCRAMSHA512})
@@ -453,6 +455,7 @@ func TestSASLPlainAuth(t *testing.T) {
 		broker.responseSize = metrics.NilHistogram{}
 		broker.responseRate = metrics.NilMeter{}
 		broker.requestLatency = metrics.NilHistogram{}
+		broker.requestsInFlight = metrics.NilCounter{}
 
 		conf := NewConfig()
 		conf.Net.SASL.Mechanism = SASLTypePlaintext
@@ -536,6 +539,7 @@ func TestSASLReadTimeout(t *testing.T) {
 		broker.responseSize = metrics.NilHistogram{}
 		broker.responseRate = metrics.NilMeter{}
 		broker.requestLatency = metrics.NilHistogram{}
+		broker.requestsInFlight = metrics.NilCounter{}
 	}
 
 	conf := NewConfig()
@@ -626,6 +630,7 @@ func TestGSSAPIKerberosAuth_Authorize(t *testing.T) {
 		broker.responseSize = metrics.NilHistogram{}
 		broker.responseRate = metrics.NilMeter{}
 		broker.requestLatency = metrics.NilHistogram{}
+		broker.requestsInFlight = metrics.NilCounter{}
 		conf := NewConfig()
 		conf.Net.SASL.Mechanism = SASLTypeGSSAPI
 		conf.Net.SASL.GSSAPI.ServiceName = "kafka"
@@ -990,6 +995,9 @@ func validateBrokerMetrics(t *testing.T, broker *Broker, mockBrokerMetrics broke
 	metricValidators.registerForAllBrokers(broker, countHistogramValidator("request-size", 1))
 	metricValidators.registerForAllBrokers(broker, minMaxHistogramValidator("request-size", mockBrokerBytesRead, mockBrokerBytesRead))
 
+	// Check that there is no more requests in flight
+	metricValidators.registerForAllBrokers(broker, counterValidator("requests-in-flight", 0))
+
 	// Run the validators
 	metricValidators.run(t, broker.conf.MetricRegistry)
 }
diff --git a/functional_producer_test.go b/functional_producer_test.go
index 311de0a..1fa0ba1 100644
--- a/functional_producer_test.go
+++ b/functional_producer_test.go
@@ -260,6 +260,9 @@ func validateMetrics(t *testing.T, client Client) {
 		metricValidators.registerForBroker(broker, minValHistogramValidator("response-size", 1))
 	}
 
+	// There should be no requests in flight anymore
+	metricValidators.registerForAllBrokers(broker, counterValidator("requests-in-flight", 0))
+
 	// Run the validators
 	metricValidators.run(t, client.Config().MetricRegistry)
 }
diff --git a/metrics_test.go b/metrics_test.go
index 789c0ff..7572f5b 100644
--- a/metrics_test.go
+++ b/metrics_test.go
@@ -170,3 +170,19 @@ func maxValHistogramValidator(name string, maxMax int) *metricValidator {
 		}
 	})
 }
+
+func counterValidator(name string, expectedCount int) *metricValidator {
+	return &metricValidator{
+		name: name,
+		validator: func(t *testing.T, metric interface{}) {
+			if counter, ok := metric.(metrics.Counter); !ok {
+				t.Errorf("Expected counter metric for '%s', got %T", name, metric)
+			} else {
+				count := counter.Count()
+				if count != int64(expectedCount) {
+					t.Errorf("Expected counter metric '%s' count = %d, got %d", name, expectedCount, count)
+				}
+			}
+		},
+	}
+}