fix: move RetryCount default to config.Validate() across all drivers

drivers/db2/internal/config.go

@@ -83,6 +83,13 @@ func (c *Config) Validate() error {
 		c.MaxThreads = constants.DefaultThreadCount
 	}
 
+	if c.RetryCount < 0 {
+		return fmt.Errorf("retry count is required")
+	}
+	if c.RetryCount == 0 {
+		c.RetryCount = constants.DefaultRetryCount
+	}
+
 	if c.SSLConfiguration == nil {
 		c.SSLConfiguration = &utils.SSLConfig{
 			Mode: "disable",

drivers/db2/internal/config_test.go

@@ -110,6 +110,19 @@ func TestConfig_Validate(t *testing.T) {
 			},
 			expectErr: true,
 		},
+		{
+			// Rejects config with a negative retry count.
+			name: "invalid config - negative retry count",
+			config: &Config{
+				Host:       "db2-host",
+				Port:       50000,
+				Database:   "testdb",
+				Username:   "db2inst1",
+				Password:   "secret1234",
+				RetryCount: -1,
+			},
+			expectErr: true,
+		},
 	}
 
 	for _, tt := range tests {

drivers/db2/internal/db2.go

@@ -13,7 +13,6 @@ import (
 	"github.com/datazip-inc/olake/drivers/abstract"
 	"github.com/datazip-inc/olake/pkg/jdbc"
 	"github.com/datazip-inc/olake/types"
-	"github.com/datazip-inc/olake/utils"
 	"github.com/datazip-inc/olake/utils/logger"
 	"github.com/jmoiron/sqlx"
 	"golang.org/x/crypto/ssh"
@@ -79,7 +78,6 @@ func (d *DB2) Setup(ctx context.Context) error {
 	}
 
 	d.client = client
-	d.config.RetryCount = utils.Ternary(d.config.RetryCount <= 0, 1, d.config.RetryCount+1).(int)
 	return nil
 }
 

drivers/kafka/internal/config.go

@@ -69,7 +69,10 @@ func (c *Config) Validate() error {
 		c.MaxThreads = constants.DefaultThreadCount
 	}
 
-	if c.RetryCount <= 0 {
+	if c.RetryCount < 0 {
+		return fmt.Errorf("retry count is required")
+	}
+	if c.RetryCount == 0 {
 		c.RetryCount = constants.DefaultRetryCount
 	}
 

drivers/kafka/internal/kafka.go

@@ -130,10 +130,6 @@ func (k *Kafka) Setup(ctx context.Context) error {
 		logger.Infof("initialized schema registry client for endpoint: %s", k.config.SchemaRegistry.Endpoint)
 	}
 
-	// TODO: Avoid modifying the configured/default retry count during initialization across all drivers.
-	// check for default backoff count
-	k.config.RetryCount = utils.Ternary(k.config.RetryCount <= 0, 1, k.config.RetryCount+1).(int)
-
 	return nil
 }
 

drivers/mongodb/internal/backfill.go

@@ -78,7 +78,7 @@ func (m *Mongo) GetOrSplitChunks(ctx context.Context, pool *destination.WriterPo
 	// Generate and update chunks
 	var retryErr error
 	var chunksArray []types.Chunk
-	err = utils.RetryOnBackoff(ctx, m.config.RetryCount, constants.DefaultRetryTimeout, func(ctx context.Context) error {
+	err = utils.RetryOnBackoff(ctx, m.MaxRetries(), constants.DefaultRetryTimeout, func(ctx context.Context) error {
 		chunksArray, retryErr = m.splitChunks(ctx, collection, stream, storageSize)
 		return retryErr
 	})

drivers/mongodb/internal/config.go

@@ -116,7 +116,10 @@ func (c *Config) Validate() error {
 		c.MaxThreads = constants.DefaultThreadCount
 	}
 
-	if c.RetryCount <= 0 {
+	if c.RetryCount < 0 {
+		return fmt.Errorf("retry count is required")
+	}
+	if c.RetryCount == 0 {
 		c.RetryCount = constants.DefaultRetryCount
 	}
 

drivers/mongodb/internal/config_test.go

@@ -186,6 +186,18 @@ func TestConfig_Validate(t *testing.T) {
 			},
 			expectErr: true,
 		},
+		{
+			name: "negative retry count fails validation",
+			config: &Config{
+				Hosts:      []string{"mongo.example.com:27017"},
+				Database:   "testdb",
+				Username:   "user",
+				Password:   "pass",
+				AuthDB:     "admin",
+				RetryCount: -1,
+			},
+			expectErr: true,
+		},
 		{
 			name: "sets defaults",
 			config: &Config{

drivers/mongodb/internal/mon.go

@@ -171,8 +171,6 @@ func (m *Mongo) Setup(ctx context.Context) error {
 	m.client = conn
 	// no need to check from discover if it have cdc support or not
 	m.CDCSupport = true
-	// check for default backoff count
-	m.config.RetryCount = utils.Ternary(m.config.RetryCount == 0, 1, m.config.RetryCount+1).(int)
 	pingCtx, cancel := context.WithTimeout(ctx, 1*time.Minute)
 	defer cancel()
 

drivers/mssql/internal/config.go

@@ -55,7 +55,10 @@ func (c *Config) Validate() error {
 		c.MaxThreads = constants.DefaultThreadCount
 	}
 
-	if c.RetryCount <= 0 {
+	if c.RetryCount < 0 {
+		return fmt.Errorf("retry count is required")
+	}
+	if c.RetryCount == 0 {
 		c.RetryCount = constants.DefaultRetryCount
 	}
 

drivers/mssql/internal/config_test.go

@@ -136,6 +136,19 @@ func TestConfig_Validate(t *testing.T) {
 			},
 			expectErr: true,
 		},
+		{
+			// Rejects config with a negative retry count.
+			name: "invalid config - negative retry count",
+			config: &Config{
+				Host:       "mssql-host",
+				Port:       1433,
+				Database:   "testdb",
+				Username:   "sa",
+				Password:   "Password!123",
+				RetryCount: -1,
+			},
+			expectErr: true,
+		},
 		{
 			name: "valid config with primary_config",
 			config: &Config{

drivers/mssql/internal/mssql.go

@@ -74,7 +74,6 @@ func (m *MSSQL) Setup(ctx context.Context) error {
 		return fmt.Errorf("failed to connect to MSSQL: %s", err)
 	}
 
-	m.config.RetryCount = utils.Ternary(m.config.RetryCount <= 0, 1, m.config.RetryCount+1).(int)
 	// Enable CDC support if database-level CDC is enabled
 	cdcSupported, err := m.isDatabaseCDCEnabled(ctx)
 	if err != nil {

drivers/mysql/internal/config.go

@@ -107,9 +107,8 @@ func (c *Config) Validate() error {
 		return fmt.Errorf("password is required")
 	}
 
-	// Optional database name, default to 'mysql'
 	if c.Database == "" {
-		c.Database = "mysql"
+		return fmt.Errorf("database name is required")
 	}
 
 	// Set default number of threads if not provided
@@ -118,7 +117,10 @@ func (c *Config) Validate() error {
 	}
 
 	// Set default retry count if not provided
-	if c.RetryCount <= 0 {
+	if c.RetryCount < 0 {
+		return fmt.Errorf("retry count is required")
+	}
+	if c.RetryCount == 0 {
 		c.RetryCount = constants.DefaultRetryCount // Reasonable default for retries
 	}
 

drivers/mysql/internal/config_test.go

@@ -325,17 +325,30 @@ func TestConfig_Validate(t *testing.T) {
 			},
 			expectErr: true,
 		},
-		// Defaults database to mysql when empty.
+		// Rejects empty database.
 		{
-			name: "valid config - defaults database",
+			name: "invalid config - missing database",
 			config: &Config{
 				Host:     "localhost",
 				Port:     3306,
 				Username: "testuser",
 				Password: "testpass",
 				Database: "",
 			},
-			expectErr: false,
+			expectErr: true,
+		},
+		// Rejects negative retry count.
+		{
+			name: "invalid config - negative retry count",
+			config: &Config{
+				Host:       "localhost",
+				Port:       3306,
+				Username:   "testuser",
+				Password:   "testpass",
+				Database:   "testdb",
+				RetryCount: -1,
+			},
+			expectErr: true,
 		},
 	}
 
@@ -348,9 +361,6 @@ func TestConfig_Validate(t *testing.T) {
 			if !tt.expectErr && err != nil {
 				t.Errorf("Expected no error but got: %v", err)
 			}
-			if !tt.expectErr && err == nil && tt.config.Database == "" {
-				t.Errorf("Expected database to default to 'mysql'")
-			}
 		})
 	}
 }

drivers/mysql/internal/mysql.go

@@ -154,7 +154,6 @@ func (m *MySQL) Setup(ctx context.Context) error {
 		m.cdcConfig = *cdc
 	}
 	m.client = client
-	m.config.RetryCount = utils.Ternary(m.config.RetryCount <= 0, 1, m.config.RetryCount+1).(int)
 	// Enable CDC support if binlog is configured
 	cdcSupported, err := m.IsCDCSupported(ctx)
 	if err != nil {

drivers/oracle/internal/config.go

@@ -76,6 +76,13 @@ func (c *Config) Validate() error {
 		c.MaxThreads = constants.DefaultThreadCount
 	}
 
+	if c.RetryCount < 0 {
+		return fmt.Errorf("retry count is required")
+	}
+	if c.RetryCount == 0 {
+		c.RetryCount = constants.DefaultRetryCount
+	}
+
 	if c.SSLConfiguration == nil {
 		c.SSLConfiguration = &utils.SSLConfig{
 			Mode: "disable",

drivers/oracle/internal/oracle.go

@@ -85,7 +85,6 @@ func (o *Oracle) Setup(ctx context.Context) error {
 	}
 
 	o.client = client
-	o.config.RetryCount = utils.Ternary(o.config.RetryCount <= 0, 1, o.config.RetryCount+1).(int)
 	return nil
 }
 

drivers/postgres/internal/config.go

@@ -47,11 +47,23 @@ func (c *Config) Validate() error {
 		return fmt.Errorf("invalid port number: must be between 1 and 65535")
 	}
 
+	if c.Database == "" {
+		return fmt.Errorf("database name is required")
+	}
+
 	// default number of threads
 	if c.MaxThreads <= 0 {
 		c.MaxThreads = constants.DefaultThreadCount
 	}
 
+	// default backoff retry count
+	if c.RetryCount < 0 {
+		return fmt.Errorf("retry count is required")
+	}
+	if c.RetryCount == 0 {
+		c.RetryCount = constants.DefaultRetryCount
+	}
+
 	// Add the connection parameters to the url
 	parsed := &url.URL{
 		Scheme: "postgres",

drivers/postgres/internal/config_test.go

@@ -110,6 +110,29 @@ func TestConfig_Validate(t *testing.T) {
 			},
 			expectErr: true,
 		},
+		{
+			name: "invalid missing database",
+			config: &Config{
+				Host:     "localhost",
+				Port:     5432,
+				Username: "postgres",
+				Password: "secret",
+				Database: "",
+			},
+			expectErr: true,
+		},
+		{
+			name: "invalid negative retry count",
+			config: &Config{
+				Host:       "localhost",
+				Port:       5432,
+				Username:   "postgres",
+				Password:   "secret",
+				Database:   "postgres",
+				RetryCount: -1,
+			},
+			expectErr: true,
+		},
 	}
 
 	for _, tt := range tests {

drivers/postgres/internal/postgres.go

@@ -160,7 +160,6 @@ func (p *Postgres) Setup(ctx context.Context) error {
 		logger.Info("Standard Replication is selected")
 	}
 	p.client = pgClient
-	p.config.RetryCount = utils.Ternary(p.config.RetryCount <= 0, 1, p.config.RetryCount+1).(int)
 	return nil
 }
 

drivers/s3/internal/config.go

@@ -152,7 +152,10 @@ func (c *Config) Validate() error {
 	}
 
 	// Set default retry count
-	if c.RetryCount <= 0 {
+	if c.RetryCount < 0 {
+		return fmt.Errorf("retry count is required")
+	}
+	if c.RetryCount == 0 {
 		c.RetryCount = constants.DefaultRetryCount
 	}
 

utils/utils.go

@@ -473,6 +473,9 @@ func RetryWithSkip(ctx context.Context, maxRetries int, sleep time.Duration, sho
 
 // RetryOnBackoff retries the function f up to attempts times with a backoff sleep between attempts.
 func RetryOnBackoff(ctx context.Context, attempts int, sleep time.Duration, f func(ctx context.Context) error) (err error) {
+	// Add 1 because attempts represents the number of retries,
+	// while the function needs the total number of attempts.
+	attempts = attempts + 1
 	for cur := range attempts {
 		select {
 		case <-ctx.Done():