Take the list of kafka brokers from the config

cs3org · Mar 6, 2024 · 4ebcec4 · 4ebcec4
1 parent 10fa589
commit 4ebcec4
Show file tree

Hide file tree

Showing 3 changed files with 6 additions and 5 deletions.
diff --git a/pkg/storage/fs/posix/options/options.go b/pkg/storage/fs/posix/options/options.go
@@ -27,8 +27,9 @@ import (
 type Options struct {
 	decomposedoptions.Options
 
-	WatchType string `mapstructure:"watch_type"`
-	WatchPath string `mapstructure:"watch_path"`
+	WatchType               string `mapstructure:"watch_type"`
+	WatchPath               string `mapstructure:"watch_path"`
+	WatchFolderKafkaBrokers string `mapstructure:"watch_folder_kafka_brokers"`
 }
 
 // New returns a new Options instance for the given configuration

diff --git a/pkg/storage/fs/posix/tree/gpfswatchfolderwatcher.go b/pkg/storage/fs/posix/tree/gpfswatchfolderwatcher.go
@@ -18,9 +18,9 @@ type GpfsWatchFolderWatcher struct {
 	c    *kafka.Consumer
 }
 
-func NewGpfsWatchFolderWatcher(tree *Tree, kafkaservers []string) (*GpfsWatchFolderWatcher, error) {
+func NewGpfsWatchFolderWatcher(tree *Tree, kafkaBrokers []string) (*GpfsWatchFolderWatcher, error) {
 	c, err := kafka.NewConsumer(&kafka.ConfigMap{
-		"bootstrap.servers": strings.Join(kafkaservers, ","),
+		"bootstrap.servers": strings.Join(kafkaBrokers, ","),
 		"group.id":          "ocis-posixfs",
 		"auto.offset.reset": "earliest",
 	})

diff --git a/pkg/storage/fs/posix/tree/tree.go b/pkg/storage/fs/posix/tree/tree.go
@@ -112,7 +112,7 @@ func New(lu node.PathLookup, bs Blobstore, o *options.Options, cache store.Store
 	var err error
 	switch o.WatchType {
 	case "gpfswatchfolder":
-		t.watcher, err = NewGpfsWatchFolderWatcher(t, []string{"192.168.3.99:29092"})
+		t.watcher, err = NewGpfsWatchFolderWatcher(t, strings.Split(o.WatchFolderKafkaBrokers, ","))
 		if err != nil {
 			return nil, err
 		}