Make idle timeout configurable

`PEERDB_CDC_IDLE_TIMEOUT_SECONDS` is the env var

flow/activities/flowable.go

@@ -193,14 +193,16 @@ func (a *FlowableActivity) StartFlow(ctx context.Context,
 		tblNameMapping[v.SourceTableIdentifier] = v.DestinationTableIdentifier
 	}
 
+	idleTimeout := utils.GetEnvInt("PEERDB_CDC_IDLE_TIMEOUT_SECONDS", 10)
+
 	startTime := time.Now()
 	recordsWithTableSchemaDelta, err := srcConn.PullRecords(&model.PullRecordsRequest{
 		FlowJobName:                 input.FlowConnectionConfigs.FlowJobName,
 		SrcTableIDNameMapping:       input.FlowConnectionConfigs.SrcTableIdNameMapping,
 		TableNameMapping:            tblNameMapping,
 		LastSyncState:               input.LastSyncState,
 		MaxBatchSize:                uint32(input.SyncFlowOptions.BatchSize),
-		IdleTimeout:                 10 * time.Second,
+		IdleTimeout:                 time.Duration(idleTimeout) * time.Second,
 		TableNameSchemaMapping:      input.FlowConnectionConfigs.TableNameSchemaMapping,
 		OverridePublicationName:     input.FlowConnectionConfigs.PublicationName,
 		OverrideReplicationSlotName: input.FlowConnectionConfigs.ReplicationSlotName,

flow/connectors/utils/env.go

@@ -28,3 +28,20 @@ func GetEnvBool(name string, defaultValue bool) bool {
 
 	return b
 }
+
+// GetEnvInt returns the value of the environment variable with the given name
+// or defaultValue if the environment variable is not set or is not a valid
+// integer value.
+func GetEnvInt(name string, defaultValue int) int {
+	val, ok := GetEnv(name)
+	if !ok {
+		return defaultValue
+	}
+
+	i, err := strconv.Atoi(val)
+	if err != nil {
+		return defaultValue
+	}
+
+	return i
+}