remove the env

.github/workflows/flow.yml

@@ -121,4 +121,3 @@ jobs:
           PEERDB_CATALOG_PASSWORD: postgres
           PEERDB_CATALOG_DATABASE: postgres
           PEERDB_CDC_IDLE_TIMEOUT_SECONDS: 3
-          PEERDB_ENVIRONMENT: test

flow/connectors/postgres/cdc.go

@@ -244,7 +244,7 @@ func (p *PostgresCDCSource) consumeStream(
 		var ctx context.Context
 		var cancel context.CancelFunc
 
-		if len(localRecords) == 0 {
+		if len(localRecords) == 0 && !receivedPKM {
 			// if length of localRecords is 0, then we are waiting for the first record
 			// indefinitely. So we should not timeout.
 			ctx, cancel = context.WithCancel(p.ctx)
@@ -256,8 +256,8 @@ func (p *PostgresCDCSource) consumeStream(
 		cancel()
 		if err != nil && !p.commitLock {
 			if pgconn.Timeout(err) {
-				log.Infof("Idle timeout reached")
-				if utils.IsTestEnv() || receivedPKM || len(localRecords) > 0 {
+				log.Infof("Idle timeout of %d seconds reached", standbyMessageTimeout/time.Second)
+				if receivedPKM || len(localRecords) > 0 {
 					log.Infof("Returning currently accumulated records - %d", len(localRecords))
 					return nil
 				} else {

flow/connectors/utils/env.go

@@ -45,23 +45,3 @@ func GetEnvInt(name string, defaultValue int) int {
 
 	return i
 }
-
-// GetEnvString returns the value of the environment variable with the given
-// name or defaultValue if the environment variable is not set.
-func GetEnvString(name string, defaultValue string) string {
-	val, ok := GetEnv(name)
-	if !ok {
-		return defaultValue
-	}
-
-	return val
-}
-
-func IsTestEnv() bool {
-	envVal := GetEnvString("PEERDB_ENVIRONMENT", "")
-	if envVal == "test" {
-		return true
-	} else {
-		return false
-	}
-}