fix: Fix hang in test destination plugin. (#19970)

marianogappa · web-flow · commit 864dd658fae0 · 2024-12-16T13:35:41.000Z
The StreamBatchWriter has a slightly different interface for its client, and the mocked method implementations didn't consume their input channels. As a result, in tests, the syncs got stuck, because sends on those channels would never be consumed.
diff --git a/plugins/destination/test/client/client.go b/plugins/destination/test/client/client.go
@@ -204,29 +204,37 @@ func newClientForStreamingBatchWriter(c *Client) streamingbatchwriter.Client {
 }
 
 func (c *ClientForStreamingBatchWriter) DeleteRecords(ctx context.Context, msgs <-chan *message.WriteDeleteRecord) error {
+	for range msgs {
+	}
 	return nil
 }
-func (c *ClientForStreamingBatchWriter) MigrateTable(context.Context, <-chan *message.WriteMigrateTable) error {
+func (c *ClientForStreamingBatchWriter) MigrateTable(ctx context.Context, msgs <-chan *message.WriteMigrateTable) error {
 	if c.spec.ErrorOnMigrate {
 		return ErrOnMigrate
 	}
 	if c.spec.ExitOnMigrate {
 		os.Exit(1)
 	}
+	for range msgs {
+	}
 	return nil
 }
 
-func (c *ClientForStreamingBatchWriter) DeleteStale(context.Context, <-chan *message.WriteDeleteStale) error {
+func (c *ClientForStreamingBatchWriter) DeleteStale(ctx context.Context, msgs <-chan *message.WriteDeleteStale) error {
+	for range msgs {
+	}
 	return nil
 }
 
-func (c *ClientForStreamingBatchWriter) WriteTable(context.Context, <-chan *message.WriteInsert) error {
+func (c *ClientForStreamingBatchWriter) WriteTable(ctx context.Context, msgs <-chan *message.WriteInsert) error {
 	if c.spec.ErrorOnInsert {
 		return ErrOnInsert
 	}
 	if c.spec.ExitOnInsert {
 		os.Exit(1)
 	}
+	for range msgs {
+	}
 	return nil
 }
 

Original file line number	Diff line number	Diff line change
`@@ -204,29 +204,37 @@ func newClientForStreamingBatchWriter(c *Client) streamingbatchwriter.Client {`
`204`	`204`	`}`
`205`	`205`
`206`	`206`	`func (c ClientForStreamingBatchWriter) DeleteRecords(ctx context.Context, msgs <-chan message.WriteDeleteRecord) error {`
	`207`	`+ for range msgs {`
	`208`	`+ }`
`207`	`209`	`return nil`
`208`	`210`	`}`
`209`		`-func (c ClientForStreamingBatchWriter) MigrateTable(context.Context, <-chan message.WriteMigrateTable) error {`
	`211`	`+func (c ClientForStreamingBatchWriter) MigrateTable(ctx context.Context, msgs <-chan message.WriteMigrateTable) error {`
`210`	`212`	`if c.spec.ErrorOnMigrate {`
`211`	`213`	`return ErrOnMigrate`
`212`	`214`	`}`
`213`	`215`	`if c.spec.ExitOnMigrate {`
`214`	`216`	`os.Exit(1)`
`215`	`217`	`}`
	`218`	`+ for range msgs {`
	`219`	`+ }`
`216`	`220`	`return nil`
`217`	`221`	`}`
`218`	`222`
`219`		`-func (c ClientForStreamingBatchWriter) DeleteStale(context.Context, <-chan message.WriteDeleteStale) error {`
	`223`	`+func (c ClientForStreamingBatchWriter) DeleteStale(ctx context.Context, msgs <-chan message.WriteDeleteStale) error {`
	`224`	`+ for range msgs {`
	`225`	`+ }`
`220`	`226`	`return nil`
`221`	`227`	`}`
`222`	`228`
`223`		`-func (c ClientForStreamingBatchWriter) WriteTable(context.Context, <-chan message.WriteInsert) error {`
	`229`	`+func (c ClientForStreamingBatchWriter) WriteTable(ctx context.Context, msgs <-chan message.WriteInsert) error {`
`224`	`230`	`if c.spec.ErrorOnInsert {`
`225`	`231`	`return ErrOnInsert`
`226`	`232`	`}`
`227`	`233`	`if c.spec.ExitOnInsert {`
`228`	`234`	`os.Exit(1)`
`229`	`235`	`}`
	`236`	`+ for range msgs {`
	`237`	`+ }`
`230`	`238`	`return nil`
`231`	`239`	`}`
`232`	`240`