fix(sync): Pass DeterministicCQID option to scheduler (#12424)

erezrokah · web-flow · commit eaac2e684bd6 · 2023-07-20T11:57:07.000Z
#### Summary Fixes #12421 Did a sweep on all sources to add this option. Will look into adding a test for it in the SDK <!--
diff --git a/plugins/source/aws/resources/plugin/client.go b/plugins/source/aws/resources/plugin/client.go
@@ -99,5 +99,5 @@ func (c *Client) Sync(ctx context.Context, options plugin.SyncOptions, res chan<
 	// for each sync we want to create a copy of the client so they won't share state
 	awsClient = awsClient.Duplicate()
 	awsClient.Backend = stateClient
-	return c.scheduler.Sync(ctx, awsClient, tt, res)
+	return c.scheduler.Sync(ctx, awsClient, tt, res, scheduler.WithSyncDeterministicCQID(options.DeterministicCQID))
 }
diff --git a/plugins/source/azure/resources/plugin/client.go b/plugins/source/azure/resources/plugin/client.go
@@ -60,5 +60,5 @@ func (c *Client) Sync(ctx context.Context, options plugin.SyncOptions, res chan<
 	if err != nil {
 		return err
 	}
-	return c.scheduler.Sync(ctx, c.syncClient.Duplicate(), tables, res)
+	return c.scheduler.Sync(ctx, c.syncClient.Duplicate(), tables, res, scheduler.WithSyncDeterministicCQID(options.DeterministicCQID))
 }
diff --git a/plugins/source/cloudflare/resources/plugin/plugin.go b/plugins/source/cloudflare/resources/plugin/plugin.go
@@ -75,7 +75,7 @@ func (c *Client) Sync(ctx context.Context, options plugin.SyncOptions, res chan<
 	if err != nil {
 		return err
 	}
-	return c.scheduler.Sync(ctx, c.syncClient, tables, res)
+	return c.scheduler.Sync(ctx, c.syncClient, tables, res, scheduler.WithSyncDeterministicCQID(options.DeterministicCQID))
 }
 
 func Plugin() *plugin.Plugin {
diff --git a/plugins/source/facebookmarketing/resources/plugin/plugin.go b/plugins/source/facebookmarketing/resources/plugin/plugin.go
@@ -103,7 +103,7 @@ func (c *Client) Sync(ctx context.Context, options plugin.SyncOptions, res chan<
 	if err != nil {
 		return err
 	}
-	return c.scheduler.Sync(ctx, c.syncClient, tables, res)
+	return c.scheduler.Sync(ctx, c.syncClient, tables, res, scheduler.WithSyncDeterministicCQID(options.DeterministicCQID))
 }
 
 func Plugin() *plugin.Plugin {
diff --git a/plugins/source/fastly/resources/plugin/plugin.go b/plugins/source/fastly/resources/plugin/plugin.go
@@ -63,7 +63,7 @@ func (c *Client) Sync(ctx context.Context, options plugin.SyncOptions, res chan<
 	if err != nil {
 		return err
 	}
-	return c.scheduler.Sync(ctx, c.syncClient, tables, res)
+	return c.scheduler.Sync(ctx, c.syncClient, tables, res, scheduler.WithSyncDeterministicCQID(options.DeterministicCQID))
 }
 
 func Plugin() *plugin.Plugin {
diff --git a/plugins/source/gcp/resources/plugin/client.go b/plugins/source/gcp/resources/plugin/client.go
@@ -83,5 +83,5 @@ func (c *Client) Sync(ctx context.Context, options plugin.SyncOptions, res chan<
 		}
 		syncClient = c.syncClient.WithBackend(stateClient)
 	}
-	return c.scheduler.Sync(ctx, syncClient, tables, res)
+	return c.scheduler.Sync(ctx, syncClient, tables, res, scheduler.WithSyncDeterministicCQID(options.DeterministicCQID))
 }
diff --git a/plugins/source/gitlab/resources/plugin/plugin.go b/plugins/source/gitlab/resources/plugin/plugin.go
@@ -73,7 +73,7 @@ func (c *Client) Sync(ctx context.Context, options plugin.SyncOptions, res chan<
 	if err != nil {
 		return err
 	}
-	return c.scheduler.Sync(ctx, c.syncClient, tables, res)
+	return c.scheduler.Sync(ctx, c.syncClient, tables, res, scheduler.WithSyncDeterministicCQID(options.DeterministicCQID))
 }
 
 func Plugin() *plugin.Plugin {
diff --git a/plugins/source/hubspot/resources/plugin/plugin.go b/plugins/source/hubspot/resources/plugin/plugin.go
@@ -86,7 +86,7 @@ func (c *Client) Sync(ctx context.Context, options plugin.SyncOptions, res chan<
 	if err != nil {
 		return err
 	}
-	return c.scheduler.Sync(ctx, c.syncClient, tables, res)
+	return c.scheduler.Sync(ctx, c.syncClient, tables, res, scheduler.WithSyncDeterministicCQID(options.DeterministicCQID))
 }
 
 func getTables() schema.Tables {
diff --git a/plugins/source/k8s/resources/plugin/plugin.go b/plugins/source/k8s/resources/plugin/plugin.go
@@ -105,7 +105,7 @@ func (c *Client) Sync(ctx context.Context, options plugin.SyncOptions, res chan<
 	if err != nil {
 		return err
 	}
-	return c.scheduler.Sync(ctx, c.syncClient, tables, res)
+	return c.scheduler.Sync(ctx, c.syncClient, tables, res, scheduler.WithSyncDeterministicCQID(options.DeterministicCQID))
 }
 
 func getTables() schema.Tables {

Original file line number	Diff line number	Diff line change
`@@ -99,5 +99,5 @@ func (c *Client) Sync(ctx context.Context, options plugin.SyncOptions, res chan<`
`99`	`99`	`// for each sync we want to create a copy of the client so they won't share state`
`100`	`100`	`awsClient = awsClient.Duplicate()`
`101`	`101`	`awsClient.Backend = stateClient`
`102`		`- return c.scheduler.Sync(ctx, awsClient, tt, res)`
	`102`	`+ return c.scheduler.Sync(ctx, awsClient, tt, res, scheduler.WithSyncDeterministicCQID(options.DeterministicCQID))`
`103`	`103`	`}`
Original file line number	Diff line number	Diff line change
`@@ -60,5 +60,5 @@ func (c *Client) Sync(ctx context.Context, options plugin.SyncOptions, res chan<`
`60`	`60`	`if err != nil {`
`61`	`61`	`return err`
`62`	`62`	`}`
`63`		`- return c.scheduler.Sync(ctx, c.syncClient.Duplicate(), tables, res)`
	`63`	`+ return c.scheduler.Sync(ctx, c.syncClient.Duplicate(), tables, res, scheduler.WithSyncDeterministicCQID(options.DeterministicCQID))`
`64`	`64`	`}`
Original file line number	Diff line number	Diff line change
`@@ -75,7 +75,7 @@ func (c *Client) Sync(ctx context.Context, options plugin.SyncOptions, res chan<`
`75`	`75`	`if err != nil {`
`76`	`76`	`return err`
`77`	`77`	`}`
`78`		`- return c.scheduler.Sync(ctx, c.syncClient, tables, res)`
	`78`	`+ return c.scheduler.Sync(ctx, c.syncClient, tables, res, scheduler.WithSyncDeterministicCQID(options.DeterministicCQID))`
`79`	`79`	`}`
`80`	`80`
`81`	`81`	`func Plugin() *plugin.Plugin {`
Original file line number	Diff line number	Diff line change
`@@ -103,7 +103,7 @@ func (c *Client) Sync(ctx context.Context, options plugin.SyncOptions, res chan<`
`103`	`103`	`if err != nil {`
`104`	`104`	`return err`
`105`	`105`	`}`
`106`		`- return c.scheduler.Sync(ctx, c.syncClient, tables, res)`
	`106`	`+ return c.scheduler.Sync(ctx, c.syncClient, tables, res, scheduler.WithSyncDeterministicCQID(options.DeterministicCQID))`
`107`	`107`	`}`
`108`	`108`
`109`	`109`	`func Plugin() *plugin.Plugin {`
Original file line number	Diff line number	Diff line change
`@@ -63,7 +63,7 @@ func (c *Client) Sync(ctx context.Context, options plugin.SyncOptions, res chan<`
`63`	`63`	`if err != nil {`
`64`	`64`	`return err`
`65`	`65`	`}`
`66`		`- return c.scheduler.Sync(ctx, c.syncClient, tables, res)`
	`66`	`+ return c.scheduler.Sync(ctx, c.syncClient, tables, res, scheduler.WithSyncDeterministicCQID(options.DeterministicCQID))`
`67`	`67`	`}`
`68`	`68`
`69`	`69`	`func Plugin() *plugin.Plugin {`
Original file line number	Diff line number	Diff line change
`@@ -83,5 +83,5 @@ func (c *Client) Sync(ctx context.Context, options plugin.SyncOptions, res chan<`
`83`	`83`	`}`
`84`	`84`	`syncClient = c.syncClient.WithBackend(stateClient)`
`85`	`85`	`}`
`86`		`- return c.scheduler.Sync(ctx, syncClient, tables, res)`
	`86`	`+ return c.scheduler.Sync(ctx, syncClient, tables, res, scheduler.WithSyncDeterministicCQID(options.DeterministicCQID))`
`87`	`87`	`}`
Original file line number	Diff line number	Diff line change
`@@ -73,7 +73,7 @@ func (c *Client) Sync(ctx context.Context, options plugin.SyncOptions, res chan<`
`73`	`73`	`if err != nil {`
`74`	`74`	`return err`
`75`	`75`	`}`
`76`		`- return c.scheduler.Sync(ctx, c.syncClient, tables, res)`
	`76`	`+ return c.scheduler.Sync(ctx, c.syncClient, tables, res, scheduler.WithSyncDeterministicCQID(options.DeterministicCQID))`
`77`	`77`	`}`
`78`	`78`
`79`	`79`	`func Plugin() *plugin.Plugin {`
Original file line number	Diff line number	Diff line change
`@@ -86,7 +86,7 @@ func (c *Client) Sync(ctx context.Context, options plugin.SyncOptions, res chan<`
`86`	`86`	`if err != nil {`
`87`	`87`	`return err`
`88`	`88`	`}`
`89`		`- return c.scheduler.Sync(ctx, c.syncClient, tables, res)`
	`89`	`+ return c.scheduler.Sync(ctx, c.syncClient, tables, res, scheduler.WithSyncDeterministicCQID(options.DeterministicCQID))`
`90`	`90`	`}`
`91`	`91`
`92`	`92`	`func getTables() schema.Tables {`
Original file line number	Diff line number	Diff line change
`@@ -105,7 +105,7 @@ func (c *Client) Sync(ctx context.Context, options plugin.SyncOptions, res chan<`
`105`	`105`	`if err != nil {`
`106`	`106`	`return err`
`107`	`107`	`}`
`108`		`- return c.scheduler.Sync(ctx, c.syncClient, tables, res)`
	`108`	`+ return c.scheduler.Sync(ctx, c.syncClient, tables, res, scheduler.WithSyncDeterministicCQID(options.DeterministicCQID))`
`109`	`109`	`}`
`110`	`110`
`111`	`111`	`func getTables() schema.Tables {`