Merge branch 'main' into wait-for

flow/connectors/bigquery/merge_statement_generator.go

@@ -204,7 +204,7 @@ func (m *mergeStmtGenerator) generateUpdateStatements(
 	updateStmts := make([]string, 0, len(unchangedToastCols))
 
 	for _, cols := range unchangedToastCols {
-		unchangedColsArray := strings.Split(cols, ", ")
+		unchangedColsArray := strings.Split(cols, ",")
 		otherCols := utils.ArrayMinus(allCols, unchangedColsArray)
 		tmpArray := make([]string, 0, len(otherCols))
 		for _, colName := range otherCols {

flow/connectors/bigquery/merge_stmt_generator_test.go

@@ -17,7 +17,7 @@ func TestGenerateUpdateStatement_WithUnchangedToastCols(t *testing.T) {
 		},
 	}
 	allCols := []string{"col1", "col2", "col3"}
-	unchangedToastCols := []string{"", "col2, col3", "col2", "col3"}
+	unchangedToastCols := []string{"", "col2,col3", "col2", "col3"}
 
 	expected := []string{
 		"WHEN MATCHED AND _rt!=2 AND _ut=''" +

flow/connectors/postgres/postgres.go

@@ -196,7 +196,9 @@ func (c *PostgresConnector) SetLastOffset(jobName string, lastOffset int64) erro
 
 // PullRecords pulls records from the source.
 func (c *PostgresConnector) PullRecords(catalogPool *pgxpool.Pool, req *model.PullRecordsRequest) error {
-	defer req.RecordStream.Close()
+	defer func() {
+		req.RecordStream.Close()
+	}()
 
 	// Slotname would be the job name prefixed with "peerflow_slot_"
 	slotName := fmt.Sprintf("peerflow_slot_%s", req.FlowJobName)

flow/connectors/utils/heartbeat.go

@@ -13,22 +13,24 @@ func HeartbeatRoutine(
 	ctx context.Context,
 	interval time.Duration,
 	message func() string,
-) context.CancelFunc {
-	ctx, cancel := context.WithCancel(ctx)
+) func() {
+	shutdown := make(chan struct{})
 	go func() {
 		counter := 0
 		for {
 			counter += 1
 			msg := fmt.Sprintf("heartbeat #%d: %s", counter, message())
 			RecordHeartbeatWithRecover(ctx, msg)
 			select {
+			case <-shutdown:
+				return
 			case <-ctx.Done():
 				return
 			case <-time.After(interval):
 			}
 		}
 	}()
-	return cancel
+	return func() { shutdown <- struct{}{} }
 }
 
 // if the functions are being called outside the context of a Temporal workflow,

flow/e2e/test_utils.go

@@ -133,7 +133,6 @@ func EnvWaitForEqualTables(
 	cols string,
 ) {
 	suite.T().Helper()
-
 	EnvWaitForEqualTablesWithNames(env, suite, reason, table, table, cols)
 }