Repository: Shopify/sarama
Pull request #1545 (2019-11-20 18:17:07)

Fix possible faulty metrics in TestFuncProducing*
# Bugfix

Functional tests `TestFuncProducing*` can fail because of "faulty" metric recorded when one of Kafka brokers is not reachable.

This happened when building #1539 as seen from the following logs (https://travis-ci.org/Shopify/sarama/jobs/612081678#L12183-L12217):
```
=== 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
[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
[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)
[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 root cause is likely that broker number 1 stopped somehow and was used as first seed broker in the failed test case:
```
[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
[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
[sarama] 2019/11/14 21:59:11 Closed connection to broker localhost:9091
```
Such a metadata request failed not because of a connection refused but because the TCP connection was closed prematurely because we are using toxiproxy.
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 include all requests including the one to `localhost:9091` the minimum latency and response size were recorded with `0` instead of at least `1` hence the failure.

## Versions

| Sarama | Kafka | Go |
|--------|-------|----|
| bb74e49545653206693c03ce785366a3e928f43d | N/A | 1.13 |

## Changes

Do not check the `request-latency-in-ms` metric for the cluster as it could be 0 if one of the broker is down. Similarly the `response-size` metric for the cluster could be recorded with 0 bytes.
The respective broker metrics on the other hand are useful.

Also dump applicative logs from all brokers in case the tests (or something other build task) fail as it can be useful to identify the root cause.

## Testing done

Start all brokers using Vagrant to run functional tests:
```
$ vagrant up
Bringing machine 'default' up with 'virtualbox' provider...
...
    default: finished produceWithDifferentProducers
    default:
    default: simplest uncommitted message producer finished
```

Stop broker number 2:
```
$ vagrant ssh -c "pgrep -f kafkaServer.*kafka-9092"
11893
Connection to 127.0.0.1 closed.
$ vagrant ssh -c "sudo pkill -f kafkaServer.*kafka-9092"
Connection to 127.0.0.1 closed.
$ vagrant ssh -c "pgrep -f kafkaServer.*kafka-9092"
Connection to 127.0.0.1 closed.
```

Use broker number 2 and 3 as seed brokers when running tests to reproduce the issue (might need to be run a few time for broker number 2 to be used as the first seed broker):
```
$ KAFKA_PEERS=192.168.100.67:9092,192.168.100.67:9093 DEBUG=true go test -run=TestFuncProducingSnappy -v
[sarama] 2019/11/20 10:00:55 Using random seed: 1574272855832395000
=== RUN   TestFuncProducingSnappy
[sarama] 2019/11/20 10:00:55 Initializing new client
[sarama] 2019/11/20 10:00:55 ClientID is the default of 'sarama', you should consider setting it to something application-specific.
[sarama] 2019/11/20 10:00:55 ClientID is the default of 'sarama', you should consider setting it to something application-specific.
[sarama] 2019/11/20 10:00:55 client/metadata fetching metadata for all topics from broker 192.168.100.67:9092
[sarama] 2019/11/20 10:00:55 Connected to broker at 192.168.100.67:9092 (unregistered)
[sarama] 2019/11/20 10:00:55 client/metadata got error from broker -1 while fetching metadata: read tcp 192.168.100.1:62021->192.168.100.67:9092: read: connection reset by peer
[sarama] 2019/11/20 10:00:55 Closed connection to broker 192.168.100.67:9092
[sarama] 2019/11/20 10:00:55 ClientID is the default of 'sarama', you should consider setting it to something application-specific.
[sarama] 2019/11/20 10:00:55 client/metadata fetching metadata for all topics from broker 192.168.100.67:9093
[sarama] 2019/11/20 10:00:55 Connected to broker at 192.168.100.67:9093 (unregistered)
[sarama] 2019/11/20 10:00:55 client/brokers registered new broker #9091 at 192.168.100.67:9091
[sarama] 2019/11/20 10:00:55 client/brokers registered new broker #9094 at 192.168.100.67:9094
[sarama] 2019/11/20 10:00:55 client/brokers registered new broker #9093 at 192.168.100.67:9093
[sarama] 2019/11/20 10:00:55 client/brokers registered new broker #9095 at 192.168.100.67:9095
[sarama] 2019/11/20 10:00:55 Successfully initialized new client
[sarama] 2019/11/20 10:00:55 ClientID is the default of 'sarama', you should consider setting it to something application-specific.
[sarama] 2019/11/20 10:00:55 Connected to broker at 192.168.100.67:9094 (registered as #9094)
[sarama] 2019/11/20 10:00:55 producer/broker/9094 starting up
[sarama] 2019/11/20 10:00:55 producer/broker/9094 state change to [open] on test.1/0
[sarama] 2019/11/20 10:00:55 Producer shutting down.
[sarama] 2019/11/20 10:00:55 producer/broker/9094 input chan closed
[sarama] 2019/11/20 10:00:55 producer/broker/9094 shut down
[sarama] 2019/11/20 10:00:55 consumer/broker/9094 added subscription to test.1/0
[sarama] 2019/11/20 10:00:56 consumer/broker/9094 closed dead subscription to test.1/0
[sarama] 2019/11/20 10:00:56 Closing Client
[sarama] 2019/11/20 10:00:56 Closed connection to broker 192.168.100.67:9094
[sarama] 2019/11/20 10:00:56 Closed connection to broker 192.168.100.67:9093
--- FAIL: TestFuncProducingSnappy (0.39s)
    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
FAIL
exit status 1
FAIL    github.com/Shopify/sarama       0.539s
```

Same scenario with proposed patch:
```
$ KAFKA_PEERS=192.168.100.67:9092,192.168.100.67:9093 DEBUG=true go test -run=TestFuncProducingSnappy -v
[sarama] 2019/11/20 10:02:33 Using random seed: 1574272953410242000
=== RUN   TestFuncProducingSnappy
[sarama] 2019/11/20 10:02:33 Initializing new client
[sarama] 2019/11/20 10:02:33 ClientID is the default of 'sarama', you should consider setting it to something application-specific.
[sarama] 2019/11/20 10:02:33 ClientID is the default of 'sarama', you should consider setting it to something application-specific.
[sarama] 2019/11/20 10:02:33 client/metadata fetching metadata for all topics from broker 192.168.100.67:9092
[sarama] 2019/11/20 10:02:33 Connected to broker at 192.168.100.67:9092 (unregistered)
[sarama] 2019/11/20 10:02:33 client/metadata got error from broker -1 while fetching metadata: read tcp 192.168.100.1:62064->192.168.100.67:9092: read: connection reset by peer
[sarama] 2019/11/20 10:02:33 Closed connection to broker 192.168.100.67:9092
[sarama] 2019/11/20 10:02:33 ClientID is the default of 'sarama', you should consider setting it to something application-specific.
[sarama] 2019/11/20 10:02:33 client/metadata fetching metadata for all topics from broker 192.168.100.67:9093
[sarama] 2019/11/20 10:02:33 Connected to broker at 192.168.100.67:9093 (unregistered)
[sarama] 2019/11/20 10:02:33 client/brokers registered new broker #9091 at 192.168.100.67:9091
[sarama] 2019/11/20 10:02:33 client/brokers registered new broker #9094 at 192.168.100.67:9094
[sarama] 2019/11/20 10:02:33 client/brokers registered new broker #9093 at 192.168.100.67:9093
[sarama] 2019/11/20 10:02:33 client/brokers registered new broker #9095 at 192.168.100.67:9095
[sarama] 2019/11/20 10:02:33 Successfully initialized new client
[sarama] 2019/11/20 10:02:33 ClientID is the default of 'sarama', you should consider setting it to something application-specific.
[sarama] 2019/11/20 10:02:33 Connected to broker at 192.168.100.67:9094 (registered as #9094)
[sarama] 2019/11/20 10:02:33 producer/broker/9094 starting up
[sarama] 2019/11/20 10:02:33 producer/broker/9094 state change to [open] on test.1/0
[sarama] 2019/11/20 10:02:33 Producer shutting down.
[sarama] 2019/11/20 10:02:33 producer/broker/9094 input chan closed
[sarama] 2019/11/20 10:02:33 producer/broker/9094 shut down
[sarama] 2019/11/20 10:02:33 consumer/broker/9094 added subscription to test.1/0
[sarama] 2019/11/20 10:02:33 consumer/broker/9094 closed dead subscription to test.1/0
[sarama] 2019/11/20 10:02:33 Closing Client
[sarama] 2019/11/20 10:02:33 Closed connection to broker 192.168.100.67:9094
[sarama] 2019/11/20 10:02:33 Closed connection to broker 192.168.100.67:9093
--- PASS: TestFuncProducingSnappy (0.40s)
PASS
ok      github.com/Shopify/sarama       0.477s
```

And to test the dump of broker logs (note that `kafka-server-stop.sh` does not work properly because of [KAFKA-4931](https://issues.apache.org/jira/browse/KAFKA-4931) plus we have more than one broker running anyway):
```
$ vagrant ssh -c "sudo TRAVIS_TEST_RESULT=1 KAFKA_INSTALL_ROOT=/opt /vagrant/vagrant/halt_cluster.sh"
Dumping Kafka brokers server.log:
kafka-9091 [2019-11-19 22:43:16,810] WARN Client session timed out, have not heard from server in 7726650ms for sessionid 0x16e85405ddd0000 (org.apache.zookeeper.ClientCnxn)
kafka-9091 [2019-11-19 22:43:16,813] INFO Client session timed out, have not heard from server in 7726650ms for sessionid 0x16e85405ddd0000, closing socket connection and attempting reconnect (org.apache.zookeeper.ClientCnxn)
...
kafka-9095 [2019-11-20 18:03:52,000] INFO Expiring session 0x56e854052720003, timeout of 4000ms exceeded (org.apache.zookeeper.server.ZooKeeperServer)
kafka-9095 [2019-11-20 18:03:52,001] INFO Processed session termination for sessionid: 0x56e854052720003 (org.apache.zookeeper.server.PrepRequestProcessor)
+ for i in 1 2 3 4 5
++ expr 1 + 9090
+ KAFKA_PORT=9091
+ cd /opt/kafka-9091
+ bin/kafka-server-stop.sh
No kafka server to stop
Connection to 127.0.0.1 closed.
```

And in case of a successful build there are no broker logs:
```
$ vagrant ssh -c "sudo TRAVIS_TEST_RESULT=0 KAFKA_INSTALL_ROOT=/opt /vagrant/vagrant/halt_clust
er.sh"
+ for i in 1 2 3 4 5
++ expr 1 + 9090
+ KAFKA_PORT=9091
+ cd /opt/kafka-9091
+ bin/kafka-server-stop.sh
No kafka server to stop
Connection to 127.0.0.1 closed.
```


Patch:
diff --git a/config.go b/config.go
index e515e04..69c7161 100644
--- a/config.go
+++ b/config.go
@@ -338,8 +338,15 @@ type Config struct {
 		// offsets. This currently requires the manual use of an OffsetManager
 		// but will eventually be automated.
 		Offsets struct {
-			// How frequently to commit updated offsets. Defaults to 1s.
-			CommitInterval time.Duration
+			AutoCommit struct {
+				// Whether or not to auto-commit updated offsets back to the broker.
+				// (default enabled).
+				Enable bool
+
+				// How frequently to commit updated offsets. Ineffective unless
+				// auto-commit is enabled (default 1s)
+				Interval time.Duration
+			}
 
 			// The initial offset to use if no offset was previously committed.
 			// Should be OffsetNewest or OffsetOldest. Defaults to OffsetNewest.
@@ -423,7 +430,8 @@ func NewConfig() *Config {
 	c.Consumer.MaxWaitTime = 250 * time.Millisecond
 	c.Consumer.MaxProcessingTime = 100 * time.Millisecond
 	c.Consumer.Return.Errors = false
-	c.Consumer.Offsets.CommitInterval = 1 * time.Second
+	c.Consumer.Offsets.AutoCommit.Enable = true
+	c.Consumer.Offsets.AutoCommit.Interval = 1 * time.Second
 	c.Consumer.Offsets.Initial = OffsetNewest
 	c.Consumer.Offsets.Retry.Max = 3
 
@@ -650,7 +658,7 @@ func (c *Config) Validate() error {
 		return ConfigurationError("Consumer.MaxProcessingTime must be > 0")
 	case c.Consumer.Retry.Backoff < 0:
 		return ConfigurationError("Consumer.Retry.Backoff must be >= 0")
-	case c.Consumer.Offsets.CommitInterval <= 0:
+	case c.Consumer.Offsets.AutoCommit.Interval <= 0:
 		return ConfigurationError("Consumer.Offsets.CommitInterval must be > 0")
 	case c.Consumer.Offsets.Initial != OffsetOldest && c.Consumer.Offsets.Initial != OffsetNewest:
 		return ConfigurationError("Consumer.Offsets.Initial must be OffsetOldest or OffsetNewest")
diff --git a/offset_manager.go b/offset_manager.go
index 923972f..e40f429 100644
--- a/offset_manager.go
+++ b/offset_manager.go
@@ -58,7 +58,7 @@ func newOffsetManagerFromClient(group, memberID string, generation int32, client
 		client: client,
 		conf:   conf,
 		group:  group,
-		ticker: time.NewTicker(conf.Consumer.Offsets.CommitInterval),
+		ticker: time.NewTicker(conf.Consumer.Offsets.AutoCommit.Interval),
 		poms:   make(map[string]map[int32]*partitionOffsetManager),
 
 		memberID:   memberID,
@@ -233,7 +233,12 @@ func (om *offsetManager) mainLoop() {
 	}
 }
 
+// flushToBroker is ignored if auto-commit offsets is disabled
 func (om *offsetManager) flushToBroker() {
+	if !om.conf.Consumer.Offsets.AutoCommit.Enable {
+		return
+	}
+
 	req := om.constructRequest()
 	if req == nil {
 		return
diff --git a/vagrant/halt_cluster.sh b/vagrant/halt_cluster.sh
index f255bc3..e671c48 100755
--- a/vagrant/halt_cluster.sh
+++ b/vagrant/halt_cluster.sh
@@ -1,4 +1,14 @@
-#!/bin/sh
+#!/bin/bash
+
+# If the functional tests failed (or some other task) then
+# we might want to look into the broker logs
+if [ "$TRAVIS_TEST_RESULT" = "1" ]; then
+    echo "Dumping Kafka brokers server.log:"
+    for i in 1 2 3 4 5; do
+        KAFKA_PORT=`expr $i + 9090`
+        sed -e "s/^/kafka-${KAFKA_PORT} /" ${KAFKA_INSTALL_ROOT}/kafka-${KAFKA_PORT}/logs/server.log{.*,}
+    done
+fi
 
 set -ex
 


Test patch:
diff --git a/functional_producer_test.go b/functional_producer_test.go
index 91bf3b5..5acd3b4 100644
--- a/functional_producer_test.go
+++ b/functional_producer_test.go
@@ -214,7 +214,6 @@ func validateMetrics(t *testing.T, client Client) {
 	metricValidators.register(minCountMeterValidator("request-rate", 3))
 	metricValidators.register(minCountHistogramValidator("request-size", 3))
 	metricValidators.register(minValHistogramValidator("request-size", 1))
-	metricValidators.register(minValHistogramValidator("request-latency-in-ms", minRequestLatencyInMs))
 	// and at least 2 requests to the registered broker (offset + produces)
 	metricValidators.registerForBroker(broker, minCountMeterValidator("request-rate", 2))
 	metricValidators.registerForBroker(broker, minCountHistogramValidator("request-size", 2))
@@ -248,7 +247,6 @@ func validateMetrics(t *testing.T, client Client) {
 		// in exactly 2 global responses (metadata + offset)
 		metricValidators.register(countMeterValidator("response-rate", 2))
 		metricValidators.register(minCountHistogramValidator("response-size", 2))
-		metricValidators.register(minValHistogramValidator("response-size", 1))
 		// and exactly 1 offset response for the registered broker
 		metricValidators.registerForBroker(broker, countMeterValidator("response-rate", 1))
 		metricValidators.registerForBroker(broker, minCountHistogramValidator("response-size", 1))
@@ -257,7 +255,6 @@ func validateMetrics(t *testing.T, client Client) {
 		// in at least 3 global responses (metadata + offset + produces)
 		metricValidators.register(minCountMeterValidator("response-rate", 3))
 		metricValidators.register(minCountHistogramValidator("response-size", 3))
-		metricValidators.register(minValHistogramValidator("response-size", 1))
 		// and at least 2 for the registered broker
 		metricValidators.registerForBroker(broker, minCountMeterValidator("response-rate", 2))
 		metricValidators.registerForBroker(broker, minCountHistogramValidator("response-size", 2))
diff --git a/offset_manager_test.go b/offset_manager_test.go
index 5fb6b21..765db16 100644
--- a/offset_manager_test.go
+++ b/offset_manager_test.go
@@ -7,15 +7,14 @@ import (
 )
 
 func initOffsetManagerWithBackoffFunc(t *testing.T, retention time.Duration,
-	backoffFunc func(retries, maxRetries int) time.Duration) (om OffsetManager,
+	backoffFunc func(retries, maxRetries int) time.Duration, config *Config) (om OffsetManager,
 	testClient Client, broker, coordinator *MockBroker) {
 
-	config := NewConfig()
 	config.Metadata.Retry.Max = 1
 	if backoffFunc != nil {
 		config.Metadata.Retry.BackoffFunc = backoffFunc
 	}
-	config.Consumer.Offsets.CommitInterval = 1 * time.Millisecond
+	config.Consumer.Offsets.AutoCommit.Interval = 1 * time.Millisecond
 	config.Version = V0_9_0_0
 	if retention > 0 {
 		config.Consumer.Offsets.Retention = retention
@@ -52,7 +51,7 @@ func initOffsetManagerWithBackoffFunc(t *testing.T, retention time.Duration,
 
 func initOffsetManager(t *testing.T, retention time.Duration) (om OffsetManager,
 	testClient Client, broker, coordinator *MockBroker) {
-	return initOffsetManagerWithBackoffFunc(t, retention, nil)
+	return initOffsetManagerWithBackoffFunc(t, retention, nil, NewConfig())
 }
 
 func initPartitionOffsetManager(t *testing.T, om OffsetManager,
@@ -97,6 +96,82 @@ func TestNewOffsetManager(t *testing.T) {
 	}
 }
 
+var offsetsautocommitTestTable = []struct {
+	name   string
+	set    bool // if given will override default configuration for Consumer.Offsets.AutoCommit.Enable
+	enable bool
+}{
+	{
+		"AutoCommit (default)",
+		false, // use default
+		true,
+	},
+	{
+		"AutoCommit Enabled",
+		true,
+		true,
+	},
+	{
+		"AutoCommit Disabled",
+		true,
+		false,
+	},
+}
+
+func TestNewOffsetManagerOffsetsAutoCommit(t *testing.T) {
+	// Tests to validate configuration of `Consumer.Offsets.AutoCommit.Enable`
+	for _, tt := range offsetsautocommitTestTable {
+		t.Run(tt.name, func(t *testing.T) {
+
+			config := NewConfig()
+			if tt.set {
+				config.Consumer.Offsets.AutoCommit.Enable = tt.enable
+			}
+			om, testClient, broker, coordinator := initOffsetManagerWithBackoffFunc(t, 0, nil, config)
+			pom := initPartitionOffsetManager(t, om, coordinator, 5, "original_meta")
+
+			// Wait long enough for the test not to fail..
+			timeout := 50 * config.Consumer.Offsets.AutoCommit.Interval
+
+			called := make(chan none)
+
+			ocResponse := new(OffsetCommitResponse)
+			ocResponse.AddError("my_topic", 0, ErrNoError)
+			handler := func(req *request) (res encoder) {
+				close(called)
+				return ocResponse
+			}
+			coordinator.setHandler(handler)
+
+			// Should force an offset commit, if auto-commit is enabled.
+			expected := int64(1)
+			pom.ResetOffset(expected, "modified_meta")
+			_, _ = pom.NextOffset()
+
+			select {
+			case <-called:
+				// OffsetManager called on the wire.
+				if !config.Consumer.Offsets.AutoCommit.Enable {
+					t.Errorf("Received request for: %s when AutoCommit is disabled", tt.name)
+				}
+			case <-time.After(timeout):
+				// Timeout waiting for OffsetManager to call on the wire.
+				if config.Consumer.Offsets.AutoCommit.Enable {
+					t.Errorf("No request received for: %s after waiting for %v", tt.name, timeout)
+				}
+			}
+
+			broker.Close()
+			coordinator.Close()
+
+			// !! om must be closed before the pom so pom.release() is called before pom.Close()
+			safeClose(t, om)
+			safeClose(t, pom)
+			safeClose(t, testClient)
+		})
+	}
+}
+
 // Test recovery from ErrNotCoordinatorForConsumer
 // on first fetchInitialOffset call
 func TestOffsetManagerFetchInitialFail(t *testing.T) {
@@ -148,7 +223,7 @@ func TestOffsetManagerFetchInitialLoadInProgress(t *testing.T) {
 		atomic.AddInt32(&retryCount, 1)
 		return 0
 	}
-	om, testClient, broker, coordinator := initOffsetManagerWithBackoffFunc(t, 0, backoff)
+	om, testClient, broker, coordinator := initOffsetManagerWithBackoffFunc(t, 0, backoff, NewConfig())
 
 	// Error on first fetchInitialOffset call
 	responseBlock := OffsetFetchResponseBlock{