retry on franz consumer group session errors (#154)

* retry on franz consumer group session errors * address feedback * Update README.md Co-authored-by: Lovro Mažgon <[email protected]> * remove unused retry leader error config * remove retryLeaderErrors from tests * Update source/franz.go Co-authored-by: Lovro Mažgon <[email protected]> * used -typed flag in franz client mock --------- Co-authored-by: Lovro Mažgon <[email protected]> (cherry picked from commit 3eb3b3a)
ConduitIO · Jul 12, 2024 · 8e7bc74 · 8e7bc74
1 parent fae5b08
commit 8e7bc74
Show file tree

Hide file tree

Showing 13 changed files with 351 additions and 45 deletions.
diff --git a/Makefile b/Makefile
@@ -27,6 +27,11 @@ test: test-kafka test-redpanda
 generate:
 	go generate ./...
 
+.PHONY: fmt
+fmt: ## Format Go files using gofumpt and gci.
+	gofumpt -l -w .
+	gci write --skip-generated  .
+
 .PHONY: lint
 lint:
 	golangci-lint run

diff --git a/README.md b/README.md
@@ -43,6 +43,7 @@ A source is getting associated with a consumer group ID the first time the `Read
 | `saslMechanism`      | SASL mechanism to be used. Possible values: PLAIN, SCRAM-SHA-256, SCRAM-SHA-512. If empty, authentication won't be performed.                                                                                | false    |                           |
 | `saslUsername`       | SASL username. If provided, a password needs to be provided too.                                                                                                                                             | false    |                           |
 | `saslPassword`       | SASL password. If provided, a username needs to be provided too.                                                                                                                                             | false    |                           |
+| `retryGroupJoinErrors`       | determines whether the connector will continually retry on group join errors                                                                                                                                              | false    | `true` |
 
 ## Destination
 

diff --git a/common/config.go b/common/config.go
@@ -23,10 +23,8 @@ import (
 	"github.com/twmb/franz-go/pkg/kgo"
 )
 
-var (
-	// TODO make the timeout configurable
-	connectionTimeout = time.Second * 10
-)
+// TODO make the timeout configurable
+var connectionTimeout = time.Second * 10
 
 // Config contains common configuration parameters.
 type Config struct {
@@ -78,6 +76,7 @@ func (c Config) TryDial(ctx context.Context) error {
 		case <-ctx.Done():
 			return err
 		case <-time.After(time.Second):
+			sdk.Logger(ctx).Warn().Msg("failed to dial broker, trying again...")
 			// ping again
 		}
 	}

diff --git a/source/config.go b/source/config.go
@@ -39,6 +39,8 @@ type Config struct {
 	ReadFromBeginning bool `json:"readFromBeginning"`
 	// GroupID defines the consumer group id.
 	GroupID string `json:"groupID"`
+	// RetryGroupJoinErrors determines whether the connector will continually retry on group join errors.
+	RetryGroupJoinErrors bool `json:"retryGroupJoinErrors" default:"true"`
 }
 
 // Validate executes manual validations beyond what is defined in struct tags.

diff --git a/source/config_test.go b/source/config_test.go
@@ -30,34 +30,35 @@ func TestConfig_ValidateTopics(t *testing.T) {
 		name    string
 		cfg     Config
 		wantErr string
-	}{{
-		name: `one of "topic" and "topics" should be provided.`,
-		cfg: Config{
-			Topics: []string{},
-			Topic:  "",
+	}{
+		{
+			name: `one of "topic" and "topics" should be provided.`,
+			cfg: Config{
+				Topics: []string{},
+				Topic:  "",
+			},
+			wantErr: `required parameter missing: "topics"`,
+		}, {
+			name: "invalid, only provide one.",
+			cfg: Config{
+				Topics: []string{"topic2"},
+				Topic:  "topic1",
+			},
+			wantErr: `can't provide both "topic" and "topics" parameters, "topic" is deprecated and will be removed, use the "topics" parameter instead`,
+		}, {
+			name: "valid with warning, will be deprecated soon",
+			cfg: Config{
+				Topics: []string{},
+				Topic:  "topic1",
+			},
+			wantErr: "",
+		}, {
+			name: "valid",
+			cfg: Config{
+				Topics: []string{"topic1"},
+			},
+			wantErr: "",
 		},
-		wantErr: `required parameter missing: "topics"`,
-	}, {
-		name: "invalid, only provide one.",
-		cfg: Config{
-			Topics: []string{"topic2"},
-			Topic:  "topic1",
-		},
-		wantErr: `can't provide both "topic" and "topics" parameters, "topic" is deprecated and will be removed, use the "topics" parameter instead`,
-	}, {
-		name: "valid with warning, will be deprecated soon",
-		cfg: Config{
-			Topics: []string{},
-			Topic:  "topic1",
-		},
-		wantErr: "",
-	}, {
-		name: "valid",
-		cfg: Config{
-			Topics: []string{"topic1"},
-		},
-		wantErr: "",
-	},
 	}
 
 	for _, tc := range testCases {

diff --git a/source/franz.go b/source/franz.go
@@ -12,23 +12,36 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+//go:generate mockgen -typed -destination franz_mock.go -package source -mock_names=Client=MockClient . Client
+
 package source
 
 import (
 	"context"
 	"errors"
 	"fmt"
+	"strings"
 	"sync"
 
 	sdk "github.com/conduitio/conduit-connector-sdk"
 	"github.com/twmb/franz-go/pkg/kgo"
 )
 
 type FranzConsumer struct {
-	client *kgo.Client
+	client Client
 	acker  *batchAcker
 
 	iter *kgo.FetchesRecordIter
+
+	retryGroupJoinErrors bool
+}
+
+// Client is a franz-go kafka client.
+type Client interface {
+	Close()
+	CommitRecords(ctx context.Context, rs ...*kgo.Record) error
+	OptValue(opt any) any
+	PollFetches(ctx context.Context) kgo.Fetches
 }
 
 var _ Consumer = (*FranzConsumer)(nil)
@@ -53,16 +66,24 @@ func NewFranzConsumer(ctx context.Context, cfg Config) (*FranzConsumer, error) {
 	}
 
 	return &FranzConsumer{
-		client: cl,
-		acker:  newBatchAcker(cl, 1000),
-		iter:   &kgo.FetchesRecordIter{}, // empty iterator is done
+		client:               cl,
+		acker:                newBatchAcker(cl, 1000),
+		iter:                 &kgo.FetchesRecordIter{}, // empty iterator is done
+		retryGroupJoinErrors: cfg.RetryGroupJoinErrors,
 	}, nil
 }
 
 func (c *FranzConsumer) Consume(ctx context.Context) (*Record, error) {
 	for c.iter.Done() {
 		fetches := c.client.PollFetches(ctx)
 		if err := fetches.Err(); err != nil {
+			var errGroupSession *kgo.ErrGroupSession
+			if c.retryGroupJoinErrors &&
+				(errors.As(err, &errGroupSession) || strings.Contains(err.Error(), "unable to join group session")) {
+				sdk.Logger(ctx).Warn().Err(err).Msgf("group session error, retrying")
+				return nil, sdk.ErrBackoffRetry
+			}
+
 			return nil, err
 		}
 		if fetches.Empty() {
@@ -92,7 +113,7 @@ func (c *FranzConsumer) Close(ctx context.Context) error {
 
 // batchAcker commits acks in batches.
 type batchAcker struct {
-	client *kgo.Client
+	client Client
 
 	batchSize     int
 	curBatchIndex int
@@ -101,7 +122,7 @@ type batchAcker struct {
 	m       sync.Mutex
 }
 
-func newBatchAcker(client *kgo.Client, batchSize int) *batchAcker {
+func newBatchAcker(client Client, batchSize int) *batchAcker {
 	return &batchAcker{
 		client:        client,
 		batchSize:     batchSize,

diff --git a/source/franz_integration_test.go b/source/franz_integration_test.go
@@ -28,7 +28,7 @@ func TestFranzConsumer_Consume_FromBeginning(t *testing.T) {
 	is := is.New(t)
 	ctx := context.Background()
 
-	cfg := test.ParseConfigMap[Config](t, test.SourceConfigMap(t, false))
+	cfg := test.ParseConfigMap[Config](t, test.SourceConfigMap(t, false, false))
 	cfg.ReadFromBeginning = true
 
 	records := test.GenerateFranzRecords(1, 6)
@@ -56,7 +56,7 @@ func TestFranzConsumer_Consume_LastOffset(t *testing.T) {
 	is := is.New(t)
 	ctx := context.Background()
 
-	cfg := test.ParseConfigMap[Config](t, test.SourceConfigMap(t, false))
+	cfg := test.ParseConfigMap[Config](t, test.SourceConfigMap(t, false, false))
 	cfg.ReadFromBeginning = false
 
 	records := test.GenerateFranzRecords(1, 6)
@@ -93,7 +93,7 @@ func TestFranzConsumer_Consume_MultipleTopics(t *testing.T) {
 	is := is.New(t)
 	ctx := context.Background()
 
-	cfg := test.ParseConfigMap[Config](t, test.SourceConfigMap(t, true))
+	cfg := test.ParseConfigMap[Config](t, test.SourceConfigMap(t, true, false))
 	cfg.ReadFromBeginning = true
 
 	records := test.GenerateFranzRecords(1, 6)