feat: kafka multiple-broker support. Closes argoproj#892 (argoproj#894)

icecoffee531 · Oct 11, 2020 · 9d23043 · 9d23043
1 parent 73539ce
commit 9d23043
Show file tree

Hide file tree

Showing 7 changed files with 11 additions and 8 deletions.
diff --git a/api/event-source.html b/api/event-source.html
diff --git a/api/event-source.md b/api/event-source.md
diff --git a/api/openapi-spec/swagger.json b/api/openapi-spec/swagger.json
diff --git a/eventsources/sources/kafka/start.go b/eventsources/sources/kafka/start.go
@@ -20,6 +20,7 @@ import (
 	"context"
 	"encoding/json"
 	"strconv"
+	"strings"
 	"sync"
 	"time"
 
@@ -106,7 +107,8 @@ func (listener *EventListener) consumerGroupConsumer(ctx context.Context, log *z
 		kafkaEventSource: kafkaEventSource,
 	}
 
-	client, err := sarama.NewConsumerGroup([]string{kafkaEventSource.URL}, kafkaEventSource.ConsumerGroup.GroupName, config)
+	urls := strings.Split(kafkaEventSource.URL, ",")
+	client, err := sarama.NewConsumerGroup(urls, kafkaEventSource.ConsumerGroup.GroupName, config)
 	if err != nil {
 		log.Errorf("Error creating consumer group client: %v", err)
 		return err
@@ -163,7 +165,8 @@ func (el *EventListener) partitionConsumer(ctx context.Context, log *zap.Sugared
 			return err
 		}
 
-		consumer, err = sarama.NewConsumer([]string{kafkaEventSource.URL}, config)
+		urls := strings.Split(kafkaEventSource.URL, ",")
+		consumer, err = sarama.NewConsumer(urls, config)
 		if err != nil {
 			return err
 		}

diff --git a/pkg/apis/eventsource/v1alpha1/generated.proto b/pkg/apis/eventsource/v1alpha1/generated.proto
diff --git a/pkg/apis/eventsource/v1alpha1/openapi_generated.go b/pkg/apis/eventsource/v1alpha1/openapi_generated.go
diff --git a/pkg/apis/eventsource/v1alpha1/types.go b/pkg/apis/eventsource/v1alpha1/types.go
@@ -288,7 +288,7 @@ type AMQPEventSource struct {
 
 // KafkaEventSource refers to event-source for Kafka related events
 type KafkaEventSource struct {
-	// URL to kafka cluster
+	// URL to kafka cluster, multiple URLs separated by comma
 	URL string `json:"url" protobuf:"bytes,1,opt,name=url"`
 	// Partition name
 	Partition string `json:"partition" protobuf:"bytes,2,opt,name=partition"`