split consumer and producer topic configuration (#50)

Signed-off-by: Amir Malka <[email protected]>
kubescape · Feb 6, 2024 · 7035e5a · 7035e5a
1 parent 12c9602
commit 7035e5a
Show file tree

Hide file tree

Showing 4 changed files with 8 additions and 5 deletions.
diff --git a/adapters/backend/v1/pulsar.go b/adapters/backend/v1/pulsar.go
@@ -45,7 +45,7 @@ func NewPulsarMessageReader(cfg config.Config, pulsarClient pulsarconnector.Clie
 		panic(err)
 	}
 	readerName := fmt.Sprintf("%s-%s", cfg.Backend.Subscription, hostname)
-	topic := pulsarconnector.BuildPersistentTopic(pulsarClient.GetConfig().Tenant, pulsarClient.GetConfig().Namespace, cfg.Backend.Topic)
+	topic := pulsarconnector.BuildPersistentTopic(pulsarClient.GetConfig().Tenant, pulsarClient.GetConfig().Namespace, cfg.Backend.ConsumerTopic)
 	logger.L().Debug("creating new pulsar reader",
 		helpers.String("readerName", readerName),
 		helpers.String("topic", topic))
@@ -333,7 +333,7 @@ type PulsarMessageProducer struct {
 }
 
 func NewPulsarMessageProducer(cfg config.Config, pulsarClient pulsarconnector.Client) (*PulsarMessageProducer, error) {
-	topic := cfg.Backend.Topic
+	topic := cfg.Backend.ProducerTopic
 	fullTopic := pulsarconnector.BuildPersistentTopic(pulsarClient.GetConfig().Tenant, pulsarClient.GetConfig().Namespace, topic)
 
 	options := pulsar.ProducerOptions{

diff --git a/config/config.go b/config/config.go
@@ -25,7 +25,8 @@ type Backend struct {
 	AuthenticationServer *AuthenticationServerConfig `mapstructure:"authenticationServer"`
 	Subscription         string                      `mapstructure:"subscription"`
 	PulsarConfig         *pulsarconfig.PulsarConfig  `mapstructure:"pulsarConfig"`
-	Topic                pulsarconnector.TopicName   `mapstructure:"topic"`
+	ProducerTopic        pulsarconnector.TopicName   `mapstructure:"producerTopic"`
+	ConsumerTopic        pulsarconnector.TopicName   `mapstructure:"consumerTopic"`
 	Prometheus           *PrometheusConfig           `mapstructure:"prometheusConfig"`
 	ReconciliationTask   *ReconciliationTaskConfig   `mapstructure:"reconciliationTaskConfig"`
 }

diff --git a/config/config_test.go b/config/config_test.go
@@ -93,7 +93,8 @@ func TestLoadConfig(t *testing.T) {
 						RedeliveryDelaySeconds: 5,
 						MaxDeliveryAttempts:    20,
 					},
-					Topic: "synchronizer",
+					ProducerTopic: "synchronizer",
+					ConsumerTopic: "synchronizer",
 				},
 			},
 		},

diff --git a/configuration/server/config.json b/configuration/server/config.json
@@ -1,7 +1,8 @@
 {
   "backend": {
     "subscription": "synchronizer-server",
-    "topic": "synchronizer",
+    "producerTopic": "synchronizer",
+    "consumerTopic": "synchronizer",
     "pulsarConfig": {
       "url": "pulsar://localhost:6650",
       "tenant": "armo",