Wait for at least one record or PKM before returning even after timeout

.gitignore

@@ -7,3 +7,7 @@ tmp/
 private/
 nexus/server/tests/assets/*.json
 nexus/server/tests/results/actual/
+
+# go workspace
+go.work
+go.work.sum

flow/connectors/postgres/cdc.go

@@ -205,6 +205,7 @@ func (p *PostgresCDCSource) consumeStream(
 	}()
 
 	tablePKeyLastSeen := make(map[model.TableWithPkey]int)
+	receivedPKM := false
 
 	addRecord := func(rec model.Record) {
 		records.AddRecord(rec)
@@ -245,8 +246,15 @@ func (p *PostgresCDCSource) consumeStream(
 		cancel()
 		if err != nil && !p.commitLock {
 			if pgconn.Timeout(err) {
-				log.Infof("Idle timeout reached, returning currently accumulated records - %d", len(localRecords))
-				return nil
+				log.Infof("Idle timeout reached")
+				if receivedPKM || len(localRecords) > 0 {
+					log.Infof("Returning currently accumulated records - %d", len(localRecords))
+					return nil
+				} else {
+					log.Infof("No records or pkms received, reseting timemout and continuing to wait...")
+					nextStandbyMessageDeadline = time.Now().Add(standbyMessageTimeout)
+					continue
+				}
 			} else {
 				return fmt.Errorf("ReceiveMessage failed: %w", err)
 			}
@@ -285,8 +293,13 @@ func (p *PostgresCDCSource) consumeStream(
 				return fmt.Errorf("ParseXLogData failed: %w", err)
 			}
 
-			log.Debugf("XLogData => WALStart %s ServerWALEnd %s ServerTime %s\n",
-				xld.WALStart, xld.ServerWALEnd, xld.ServerTime)
+			if !receivedPKM {
+				receivedPKM = true
+				log.WithField("flowName", req.FlowJobName).Infof(
+					"Received first PKM. XLogData => WALStart %s ServerWALEnd %s ServerTime %s\n",
+					xld.WALStart, xld.ServerWALEnd, xld.ServerTime)
+			}
+
 			rec, err := p.processMessage(records, xld)
 
 			if err != nil {