Repository: Shopify/sarama (go)

Issue:
[Bug] Infinity loop in Consume() method
##### Versions

| Sarama | Kafka | Go |
|--------|-------|----|
| v1.34.1 | - | 1.18 |

##### Configuration

What configuration values are you using for Sarama and Kafka?
``` go
        // default reader configuration
        readerOptions := make([]kafka.ReaderOption, 0)
        readerOptions = append(readerOptions, kafka.WithReaderTlSConfig(nil))
	readerOptions = append(readerOptions, kafka.WithReaderSaslConfig(user, password, kafka.SaslAlgorithm("SHA512")))

	reader, err := kafka.NewReader(brokers, consumerGroup, readerOptions...)
```

##### Logs

<details><summary>logs: CLICK ME</summary>
<p>

```
[Sarama] 2023/01/25 15:27:40 client/coordinator retrying after 250ms... (3 attempts remaining)
[Sarama] 2023/01/25 15:27:40 client/coordinator requesting coordinator for consumergroup kafka-demo from <ip>
[Sarama] 2023/01/25 15:27:40 client was not authorized to access group kafka-demo while attempting to find coordinator
[Sarama] 2023/01/25 15:27:40 client/coordinator retrying after 250ms... (2 attempts remaining)
[Sarama] 2023/01/25 15:27:40 client/coordinator requesting coordinator for consumergroup kafka-demo from <ip>
[Sarama] 2023/01/25 15:27:41 client was not authorized to access group kafka-demo while attempting to find coordinator
[Sarama] 2023/01/25 15:27:41 client/coordinator retrying after 250ms... (1 attempts remaining)
[Sarama] 2023/01/25 15:27:41 client/coordinator requesting coordinator for consumergroup kafka-demo from <ip>
[Sarama] 2023/01/25 15:27:41 client was not authorized to access group kafka-demo while attempting to find coordinator
[Sarama] 2023/01/25 15:27:43 client/coordinator requesting coordinator for consumergroup kafka-demo from <ip>
[Sarama] 2023/01/25 15:27:43 client was not authorized to access group kafka-demo while attempting to find coordinator
[Sarama] 2023/01/25 15:27:43 client/coordinator retrying after 250ms... (3 attempts remaining)

...infinity loop...
```

</p>
</details>

##### Problem Description
If the client is not authorized to access the consumer group, the Consume() method is expected to return an error, but this does not happen. Instead, the Consume() method freezes. 

Suggested error location: from the analysis of the sarama code, it can be seen that the code execution is endlessly looping on the `func (c *consumer Group) retryNewSession()` method (file `consumer_group.go`). Inside this method, a recursive construction is performed, which in this case leads to an infinite attempt to update the session in 3 attempts (default value: 3). After three attempts, an error is expected to return, but instead, the retryNewSession() method is recursively called again, which again fails with an error.

Pull request:
fix(consumer): don't retry FindCoordinator forever
If the consumer group hit an error finding the coordinator whilst setting up a session it would end up retrying forever because the retry count didn't get decremented and there was no exit of the loop. Fix that up and add a unittest to cover the scenario

Fixes #2426

Patch:
diff --git a/consumer_group.go b/consumer_group.go
index ecdbcfa..140d9f7 100644
--- a/consumer_group.go
+++ b/consumer_group.go
@@ -252,7 +252,10 @@ func (c *consumerGroup) retryNewSession(ctx context.Context, topics []string, ha
 	if refreshCoordinator {
 		err := c.client.RefreshCoordinator(c.groupID)
 		if err != nil {
-			return c.retryNewSession(ctx, topics, handler, retries, true)
+			if retries <= 0 {
+				return nil, err
+			}
+			return c.retryNewSession(ctx, topics, handler, retries-1, true)
 		}
 	}
 


Test patch:
diff --git a/consumer_group_test.go b/consumer_group_test.go
index 15a49c5..c93017b 100644
--- a/consumer_group_test.go
+++ b/consumer_group_test.go
@@ -6,6 +6,8 @@ import (
 	"sync"
 	"testing"
 	"time"
+
+	assert "github.com/stretchr/testify/require"
 )
 
 type handler struct {
@@ -200,3 +202,46 @@ outerFor:
 
 	cancel()
 }
+
+// TestConsumerGroupSessionDoesNotRetryForever ensures that an error fetching
+// the coordinator decrements the retry attempts and doesn't end up retrying
+// forever
+func TestConsumerGroupSessionDoesNotRetryForever(t *testing.T) {
+	config := NewTestConfig()
+	config.ClientID = t.Name()
+	config.Version = V2_0_0_0
+	config.Consumer.Return.Errors = true
+	config.Consumer.Group.Rebalance.Retry.Max = 1
+	config.Consumer.Group.Rebalance.Retry.Backoff = 0
+
+	broker0 := NewMockBroker(t, 0)
+
+	broker0.SetHandlerByMap(map[string]MockResponse{
+		"MetadataRequest": NewMockMetadataResponse(t).
+			SetBroker(broker0.Addr(), broker0.BrokerID()).
+			SetLeader("my-topic", 0, broker0.BrokerID()),
+		"FindCoordinatorRequest": NewMockFindCoordinatorResponse(t).
+			SetError(CoordinatorGroup, "my-group", ErrGroupAuthorizationFailed),
+	})
+
+	group, err := NewConsumerGroup([]string{broker0.Addr()}, "my-group", config)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer func() { _ = group.Close() }()
+
+	ctx, cancel := context.WithCancel(context.Background())
+	h := &handler{t, cancel}
+
+	var wg sync.WaitGroup
+	wg.Add(1)
+
+	go func() {
+		topics := []string{"my-topic"}
+		err := group.Consume(ctx, topics, h)
+		assert.Error(t, err)
+		wg.Done()
+	}()
+
+	wg.Wait()
+}