fix_import_conn

Signed-off-by: Song Gao <disxiaofei@163.com>
lf-edge · Oct 18, 2024 · d396180 · d396180
1 parent 343936b
commit d396180
Show file tree

Hide file tree

Showing 3 changed files with 16 additions and 0 deletions.
diff --git a/internal/meta/yamlConfigMeta.go b/internal/meta/yamlConfigMeta.go
@@ -25,6 +25,7 @@ import (
 	"github.com/lf-edge/ekuiper/v2/internal/pkg/store"
 	"github.com/lf-edge/ekuiper/v2/internal/pkg/util"
 	"github.com/lf-edge/ekuiper/v2/pkg/cast"
+	"github.com/lf-edge/ekuiper/v2/pkg/connection"
 	"github.com/lf-edge/ekuiper/v2/pkg/errorx"
 	"github.com/lf-edge/ekuiper/v2/pkg/kv"
 	"github.com/lf-edge/ekuiper/v2/pkg/replace"
@@ -768,6 +769,9 @@ func LoadConfigurations(configSets YamlConfigurationSet) YamlConfigurationSet {
 			continue
 		}
 	}
+	if err := connection.ReloadNamedConnection(); err != nil {
+		conf.Log.Errorf("reload connection config error: %s", err.Error())
+	}
 	return configResponse
 }
 
@@ -823,5 +827,8 @@ func LoadConfigurationsPartial(configSets YamlConfigurationSet) YamlConfiguratio
 			continue
 		}
 	}
+	if err := connection.ReloadNamedConnection(); err != nil {
+		conf.Log.Errorf("reload connection config error: %s", err.Error())
+	}
 	return configResponse
 }
diff --git a/internal/server/yaml_import_export.go b/internal/server/yaml_import_export.go
@@ -36,6 +36,7 @@ import (
 	"github.com/lf-edge/ekuiper/v2/internal/service"
 	"github.com/lf-edge/ekuiper/v2/pkg/ast"
 	"github.com/lf-edge/ekuiper/v2/pkg/cast"
+	"github.com/lf-edge/ekuiper/v2/pkg/connection"
 	"github.com/lf-edge/ekuiper/v2/pkg/replace"
 )
 
@@ -522,6 +523,9 @@ func importConfigurations(m *MetaConfiguration) error {
 			return err
 		}
 	}
+	if err := connection.ReloadNamedConnection(); err != nil {
+		conf.Log.Errorf("reload connection config error: %s", err.Error())
+	}
 	return nil
 }
 

diff --git a/pkg/connection/pool.go b/pkg/connection/pool.go
@@ -115,6 +115,8 @@ func FetchConnection(ctx api.StreamContext, refId, typ string, props map[string]
 
 // ReloadNamedConnection is called when server starts. It initializes all stored named connections
 func ReloadNamedConnection() error {
+	globalConnectionManager.Lock()
+	defer globalConnectionManager.Unlock()
 	cfgs, err := conf.GetCfgFromKVStorage("connections", "", "")
 	if err != nil {
 		return err
@@ -126,6 +128,9 @@ func ReloadNamedConnection() error {
 		}
 		typ := names[1]
 		id := names[2]
+		if _, ok := globalConnectionManager.connectionPool[id]; ok {
+			continue
+		}
 		meta := &Meta{
 			ID:    id,
 			Typ:   typ,