feat: ability to specify version in kafka trigger. Closes #990. (#991)

* feat: ability to specify version in kafka trigger. Closes #990. Signed-off-by: Derek Wang <[email protected]> * upgrade slack version Signed-off-by: Derek Wang <[email protected]> * omitempty Signed-off-by: Derek Wang <[email protected]>
argoproj · Dec 21, 2020 · 31d3e93 · 31d3e93
1 parent 97e4926
commit 31d3e93
Show file tree

Hide file tree

Showing 10 changed files with 323 additions and 230 deletions.
diff --git a/api/openapi-spec/swagger.json b/api/openapi-spec/swagger.json
diff --git a/api/sensor.html b/api/sensor.html
diff --git a/api/sensor.md b/api/sensor.md
diff --git a/go.mod b/go.mod
@@ -67,7 +67,7 @@ require (
 	github.com/robfig/cron v1.2.0
 	github.com/sergi/go-diff v1.1.0 // indirect
 	github.com/sirupsen/logrus v1.6.0 // indirect
-	github.com/slack-go/slack v0.6.5
+	github.com/slack-go/slack v0.7.4
 	github.com/smartystreets/assertions v0.0.0-20190401211740-f487f9de1cd3 // indirect
 	github.com/smartystreets/goconvey v1.6.4
 	github.com/spf13/viper v1.7.0

diff --git a/go.sum b/go.sum
@@ -757,8 +757,8 @@ github.com/sirupsen/logrus v1.4.1/go.mod h1:ni0Sbl8bgC9z8RoU9G6nDWqqs/fq4eDPysMB
 github.com/sirupsen/logrus v1.4.2/go.mod h1:tLMulIdttU9McNUspp0xgXVQah82FyeX6MwdIuYE2rE=
 github.com/sirupsen/logrus v1.6.0 h1:UBcNElsrwanuuMsnGSlYmtmgbb23qDR5dG+6X6Oo89I=
 github.com/sirupsen/logrus v1.6.0/go.mod h1:7uNnSEd1DgxDLC74fIahvMZmmYsHGZGEOFrfsX/uA88=
-github.com/slack-go/slack v0.6.5 h1:IkDKtJ2IROJNoe3d6mW870/NRKvq2fhLB/Q5XmzWk00=
-github.com/slack-go/slack v0.6.5/go.mod h1:FGqNzJBmxIsZURAxh2a8D21AnOVvvXZvGligs4npPUM=
+github.com/slack-go/slack v0.7.4 h1:Z+7CmUDV+ym4lYLA4NNLFIpr3+nDgViHrx8xsuXgrYs=
+github.com/slack-go/slack v0.7.4/go.mod h1:FGqNzJBmxIsZURAxh2a8D21AnOVvvXZvGligs4npPUM=
 github.com/smartystreets/assertions v0.0.0-20180927180507-b2de0cb4f26d/go.mod h1:OnSkiWE9lh6wB0YB77sQom3nweQdgAjqCqsofrRNTgc=
 github.com/smartystreets/assertions v0.0.0-20190116191733-b6c0e53d7304/go.mod h1:OnSkiWE9lh6wB0YB77sQom3nweQdgAjqCqsofrRNTgc=
 github.com/smartystreets/assertions v0.0.0-20190401211740-f487f9de1cd3 h1:hBSHahWMEgzwRyS6dRpxY0XyjZsHyQ61s084wo5PJe0=

diff --git a/pkg/apis/sensor/v1alpha1/generated.pb.go b/pkg/apis/sensor/v1alpha1/generated.pb.go
diff --git a/pkg/apis/sensor/v1alpha1/generated.proto b/pkg/apis/sensor/v1alpha1/generated.proto
diff --git a/pkg/apis/sensor/v1alpha1/openapi_generated.go b/pkg/apis/sensor/v1alpha1/openapi_generated.go
diff --git a/pkg/apis/sensor/v1alpha1/types.go b/pkg/apis/sensor/v1alpha1/types.go
@@ -439,6 +439,10 @@ type KafkaTrigger struct {
 	// Defaults to broker url.
 	// +optional.
 	PartitioningKey string `json:"partitioningKey,omitempty" protobuf:"bytes,10,opt,name=partitioningKey"`
+
+	// Specify what kafka version is being connected to enables certain features in sarama, defaults to 1.0.0
+	// +optional
+	Version string `json:"version,omitempty" protobuf:"bytes,11,opt,name=version"`
 }
 
 // NATSTrigger refers to the specification of the NATS trigger.

diff --git a/sensors/triggers/kafka/kafka.go b/sensors/triggers/kafka/kafka.go
@@ -49,6 +49,16 @@ func NewKafkaTrigger(sensor *v1alpha1.Sensor, trigger *v1alpha1.Trigger, kafkaPr
 		var err error
 		config := sarama.NewConfig()
 
+		if kafkatrigger.Version == "" {
+			config.Version = sarama.V1_0_0_0
+		} else {
+			version, err := sarama.ParseKafkaVersion(kafkatrigger.Version)
+			if err != nil {
+				return nil, errors.Wrap(err, "failed to parse Kafka version")
+			}
+			config.Version = version
+		}
+
 		if kafkatrigger.TLS != nil {
 			tlsConfig, err := common.GetTLSConfig(kafkatrigger.TLS)
 			if err != nil {