idea: need to sleep to avoid test ending before slot closed

PeerDB-io · Jan 4, 2024 · 51760c9 · 51760c9
1 parent 73dcd18
commit 51760c9
Show file tree

Hide file tree

Showing 2 changed files with 3 additions and 3 deletions.
diff --git a/flow/connectors/postgres/postgres.go b/flow/connectors/postgres/postgres.go
@@ -196,9 +196,7 @@ 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 func() {
-		req.RecordStream.Close()
-	}()
+	defer req.RecordStream.Close()
 
 	// Slotname would be the job name prefixed with "peerflow_slot_"
 	slotName := fmt.Sprintf("peerflow_slot_%s", req.FlowJobName)

diff --git a/flow/e2e/postgres/peer_flow_pg_test.go b/flow/e2e/postgres/peer_flow_pg_test.go
@@ -267,6 +267,7 @@ func (s PeerFlowE2ETestSuitePG) Test_Simple_Schema_Changes_PG() {
 	}()
 
 	env.ExecuteWorkflow(peerflow.CDCFlowWorkflowWithConfig, flowConnConfig, &limits, nil)
+	time.Sleep(10 * time.Second)
 	wg.Wait()
 	env.AssertExpectations(s.t)
 }
@@ -337,6 +338,7 @@ func (s PeerFlowE2ETestSuitePG) Test_Composite_PKey_PG() {
 	}()
 
 	env.ExecuteWorkflow(peerflow.CDCFlowWorkflowWithConfig, flowConnConfig, &limits, nil)
+	time.Sleep(10 * time.Second)
 	wg.Wait()
 	env.AssertExpectations(s.t)
 }