diff --git a/drivers/db2/internal/config.go b/drivers/db2/internal/config.go index d194c29f4..079bda3e6 100644 --- a/drivers/db2/internal/config.go +++ b/drivers/db2/internal/config.go @@ -79,10 +79,22 @@ func (c *Config) Validate() error { return fmt.Errorf("database name is required") } - if c.MaxThreads <= 0 { + if c.MaxThreads < 0 { + return fmt.Errorf("max threads is required") + } + + if c.MaxThreads == 0 { 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", diff --git a/drivers/db2/internal/config_test.go b/drivers/db2/internal/config_test.go index ae3d585d2..3929df929 100644 --- a/drivers/db2/internal/config_test.go +++ b/drivers/db2/internal/config_test.go @@ -110,6 +110,32 @@ func TestConfig_Validate(t *testing.T) { }, expectErr: true, }, + { + // Rejects config with a negative max threads. + name: "invalid config - negative max threads", + config: &Config{ + Host: "db2-host", + Port: 50000, + Database: "testdb", + Username: "db2inst1", + Password: "secret1234", + MaxThreads: -1, + }, + 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 { diff --git a/drivers/db2/internal/db2.go b/drivers/db2/internal/db2.go index 0f840a178..84407277a 100644 --- a/drivers/db2/internal/db2.go +++ b/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 } diff --git a/drivers/kafka/internal/config.go b/drivers/kafka/internal/config.go index ab0bc2826..9d5ca8763 100644 --- a/drivers/kafka/internal/config.go +++ b/drivers/kafka/internal/config.go @@ -65,11 +65,19 @@ func (c *Config) Validate() error { } } - if c.MaxThreads <= 0 { + if c.MaxThreads < 0 { + return fmt.Errorf("max threads is required") + } + + if c.MaxThreads == 0 { 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 } diff --git a/drivers/kafka/internal/kafka.go b/drivers/kafka/internal/kafka.go index 31d222d2f..abacd592b 100644 --- a/drivers/kafka/internal/kafka.go +++ b/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 } diff --git a/drivers/mongodb/internal/backfill.go b/drivers/mongodb/internal/backfill.go index 33c44f637..71fb6d29c 100644 --- a/drivers/mongodb/internal/backfill.go +++ b/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 }) diff --git a/drivers/mongodb/internal/config.go b/drivers/mongodb/internal/config.go index eac8e5497..85e1ea65a 100644 --- a/drivers/mongodb/internal/config.go +++ b/drivers/mongodb/internal/config.go @@ -112,11 +112,19 @@ func (c *Config) Validate() error { // MongoDB rejects at connect time if the mechanism actually needs a password. } - if c.MaxThreads <= 0 { + if c.MaxThreads < 0 { + return fmt.Errorf("max threads is required") + } + + if c.MaxThreads == 0 { 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 } diff --git a/drivers/mongodb/internal/config_test.go b/drivers/mongodb/internal/config_test.go index d5cef8a1d..96968dc49 100644 --- a/drivers/mongodb/internal/config_test.go +++ b/drivers/mongodb/internal/config_test.go @@ -186,6 +186,30 @@ func TestConfig_Validate(t *testing.T) { }, expectErr: true, }, + { + name: "negative max threads fails validation", + config: &Config{ + Hosts: []string{"mongo.example.com:27017"}, + Database: "testdb", + Username: "user", + Password: "pass", + AuthDB: "admin", + MaxThreads: -1, + }, + 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{ diff --git a/drivers/mongodb/internal/mon.go b/drivers/mongodb/internal/mon.go index 218df68e4..fbc9013c8 100644 --- a/drivers/mongodb/internal/mon.go +++ b/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() diff --git a/drivers/mssql/internal/config.go b/drivers/mssql/internal/config.go index 9bd4f5e98..866dbfde3 100644 --- a/drivers/mssql/internal/config.go +++ b/drivers/mssql/internal/config.go @@ -51,11 +51,19 @@ func (c *Config) Validate() error { return fmt.Errorf("database is required") } - if c.MaxThreads <= 0 { + if c.MaxThreads < 0 { + return fmt.Errorf("max threads is required") + } + + if c.MaxThreads == 0 { 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 } diff --git a/drivers/mssql/internal/config_test.go b/drivers/mssql/internal/config_test.go index 68a4d9b0e..2a53b9bf3 100644 --- a/drivers/mssql/internal/config_test.go +++ b/drivers/mssql/internal/config_test.go @@ -136,6 +136,32 @@ func TestConfig_Validate(t *testing.T) { }, expectErr: true, }, + { + // Rejects config with a negative max threads. + name: "invalid config - negative max threads", + config: &Config{ + Host: "mssql-host", + Port: 1433, + Database: "testdb", + Username: "sa", + Password: "Password!123", + MaxThreads: -1, + }, + 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{ diff --git a/drivers/mssql/internal/mssql.go b/drivers/mssql/internal/mssql.go index 42402b615..538d08dea 100644 --- a/drivers/mssql/internal/mssql.go +++ b/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: %w", 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 { diff --git a/drivers/mysql/internal/config.go b/drivers/mysql/internal/config.go index 8afcb466b..fbf3cdbdf 100644 --- a/drivers/mysql/internal/config.go +++ b/drivers/mysql/internal/config.go @@ -113,12 +113,20 @@ func (c *Config) Validate() error { } // Set default number of threads if not provided - if c.MaxThreads <= 0 { + if c.MaxThreads < 0 { + return fmt.Errorf("max threads is required") + } + + if c.MaxThreads == 0 { c.MaxThreads = constants.DefaultThreadCount // Aligned with PostgreSQL default } // 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 } diff --git a/drivers/mysql/internal/config_test.go b/drivers/mysql/internal/config_test.go index 364070269..6d7f9c710 100644 --- a/drivers/mysql/internal/config_test.go +++ b/drivers/mysql/internal/config_test.go @@ -337,6 +337,32 @@ func TestConfig_Validate(t *testing.T) { }, expectErr: false, }, + // Rejects negative max threads. + { + name: "invalid config - negative max threads", + config: &Config{ + Host: "localhost", + Port: 3306, + Username: "testuser", + Password: "testpass", + Database: "testdb", + MaxThreads: -1, + }, + 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, + }, } for _, tt := range tests { @@ -348,8 +374,8 @@ 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'") + if !tt.expectErr && err == nil && tt.name == "valid config - defaults database" && tt.config.Database != "mysql" { + t.Errorf("Expected database to default to 'mysql', got %q", tt.config.Database) } }) } diff --git a/drivers/mysql/internal/mysql.go b/drivers/mysql/internal/mysql.go index dc3558977..d67717cdb 100644 --- a/drivers/mysql/internal/mysql.go +++ b/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 { diff --git a/drivers/oracle/internal/config.go b/drivers/oracle/internal/config.go index 2d75d78f4..14863aca0 100644 --- a/drivers/oracle/internal/config.go +++ b/drivers/oracle/internal/config.go @@ -72,10 +72,22 @@ func (c *Config) Validate() error { } // Set default number of threads if not provided - if c.MaxThreads <= 0 { + if c.MaxThreads < 0 { + return fmt.Errorf("max threads is required") + } + + if c.MaxThreads == 0 { 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", diff --git a/drivers/oracle/internal/oracle.go b/drivers/oracle/internal/oracle.go index 811e839f6..fc1c9e937 100644 --- a/drivers/oracle/internal/oracle.go +++ b/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 } diff --git a/drivers/postgres/internal/config.go b/drivers/postgres/internal/config.go index f20d72f00..04d43cd48 100644 --- a/drivers/postgres/internal/config.go +++ b/drivers/postgres/internal/config.go @@ -47,11 +47,28 @@ 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 { + if c.MaxThreads < 0 { + return fmt.Errorf("max threads is required") + } + + 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", diff --git a/drivers/postgres/internal/config_test.go b/drivers/postgres/internal/config_test.go index b21bf0010..067e7711b 100644 --- a/drivers/postgres/internal/config_test.go +++ b/drivers/postgres/internal/config_test.go @@ -110,6 +110,41 @@ 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 max threads", + config: &Config{ + Host: "localhost", + Port: 5432, + Username: "postgres", + Password: "secret", + Database: "postgres", + MaxThreads: -1, + }, + 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 { diff --git a/drivers/postgres/internal/postgres.go b/drivers/postgres/internal/postgres.go index df876f5ca..bf903fa2a 100644 --- a/drivers/postgres/internal/postgres.go +++ b/drivers/postgres/internal/postgres.go @@ -165,7 +165,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 } diff --git a/drivers/s3/internal/config.go b/drivers/s3/internal/config.go index b7d6e4b44..9193a40b1 100644 --- a/drivers/s3/internal/config.go +++ b/drivers/s3/internal/config.go @@ -147,12 +147,20 @@ func (c *Config) Validate() error { c.XML = &parser.XMLConfig{} } // Set default thread count - if c.MaxThreads <= 0 { + if c.MaxThreads < 0 { + return fmt.Errorf("max threads is required") + } + + if c.MaxThreads == 0 { c.MaxThreads = constants.DefaultThreadCount } // 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 } diff --git a/drivers/s3/internal/edge_cases_test.go b/drivers/s3/internal/edge_cases_test.go index df08c0707..d6f872179 100644 --- a/drivers/s3/internal/edge_cases_test.go +++ b/drivers/s3/internal/edge_cases_test.go @@ -391,7 +391,7 @@ func TestThreadCountEdgeCases(t *testing.T) { {"default threads (0)", 0, true}, {"single thread", 1, true}, {"many threads", 100, true}, - {"negative threads", -1, true}, // Should be normalized to default + {"negative threads", -1, false}, } for _, tt := range tests { diff --git a/utils/utils.go b/utils/utils.go index 40def30c8..9b1653ef8 100644 --- a/utils/utils.go +++ b/utils/utils.go @@ -480,6 +480,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():