Make replication factor for topics configurable

cloudhut · Dec 8, 2020 · 2d22b69 · 2d22b69
1 parent 2b3d74e
commit 2d22b69
Show file tree

Hide file tree

Showing 5 changed files with 8 additions and 6 deletions.
diff --git a/pkg/kafka/config.go b/pkg/kafka/config.go
@@ -10,8 +10,9 @@ type Config struct {
 	Brokers  []string `yaml:"brokers"`
 	ClientID string   `yaml:"clientId"`
 
-	TLS  TLSConfig  `yaml:"tls"`
-	SASL SASLConfig `yaml:"sasl"`
+	TLS                    TLSConfig  `yaml:"tls"`
+	SASL                   SASLConfig `yaml:"sasl"`
+	TopicReplicationFactor int16      `yaml:"topicReplicationFactor"`
 }
 
 // RegisterFlags for all sensitive Kafka SASL configs.
@@ -36,6 +37,7 @@ func (c *Config) Validate() error {
 // SetDefaults for Kafka config
 func (c *Config) SetDefaults() {
 	c.ClientID = "owl-shop"
+	c.TopicReplicationFactor = 3
 
 	c.SASL.SetDefaults()
 }
diff --git a/pkg/shop/address_service.go b/pkg/shop/address_service.go
@@ -194,7 +194,7 @@ func (svc *AddressService) createKafkaTopic(ctx context.Context) error {
 			{
 				Topic:             svc.topicName,
 				NumPartitions:     6,
-				ReplicationFactor: 3,
+				ReplicationFactor: svc.cfg.Kafka.TopicReplicationFactor,
 				Configs: []kmsg.CreateTopicsRequestTopicConfig{
 					{"cleanup.policy", &cleanupPolicy},
 				},

diff --git a/pkg/shop/customer_service.go b/pkg/shop/customer_service.go
@@ -216,7 +216,7 @@ func (svc *CustomerService) createKafkaTopic(ctx context.Context) error {
 			{
 				Topic:             svc.topicName,
 				NumPartitions:     6,
-				ReplicationFactor: 3,
+				ReplicationFactor: svc.cfg.Kafka.TopicReplicationFactor,
 				Configs: []kmsg.CreateTopicsRequestTopicConfig{
 					{"cleanup.policy", &cleanupPolicy},
 				},

diff --git a/pkg/shop/frontend_service.go b/pkg/shop/frontend_service.go
@@ -120,7 +120,7 @@ func (svc *FrontendService) createKafkaTopic(ctx context.Context) error {
 			{
 				Topic:             svc.topicName,
 				NumPartitions:     6,
-				ReplicationFactor: 3,
+				ReplicationFactor: svc.cfg.Kafka.TopicReplicationFactor,
 				Configs: []kmsg.CreateTopicsRequestTopicConfig{
 					{"cleanup.policy", &cleanupPolicy},
 					{"retention.bytes", &retentionBytes},

diff --git a/pkg/shop/order_service.go b/pkg/shop/order_service.go
@@ -197,7 +197,7 @@ func (svc *OrderService) createKafkaTopic(ctx context.Context) error {
 			{
 				Topic:             svc.topicName,
 				NumPartitions:     6,
-				ReplicationFactor: 3,
+				ReplicationFactor: svc.cfg.Kafka.TopicReplicationFactor,
 				Configs: []kmsg.CreateTopicsRequestTopicConfig{
 					{"cleanup.policy", &cleanupPolicy},
 				},