Implement multiple partions usage in Kafka trigger (#2360)

Signed-off-by: Thomas Ribeiro De Araujo <thomas10191@hotmail.com> Co-authored-by: Thomas Ribeiro De Araujo <thomas.ribeiro@ifood.com.br>
argoproj · Jan 3, 2023 · 1428cae · 1428cae
1 parent 3cbcd72
commit 1428cae
Show file tree

Hide file tree

Showing 11 changed files with 342 additions and 336 deletions.
diff --git a/api/jsonschema/schema.json b/api/jsonschema/schema.json
@@ -3582,12 +3582,12 @@
           "type": "array"
         },
         "partition": {
-          "description": "Partition to write data to.",
+          "description": "DEPRECATED",
           "format": "int32",
           "type": "integer"
         },
         "partitioningKey": {
-          "description": "The partitioning key for the messages put on the Kafka topic. Defaults to broker url.",
+          "description": "The partitioning key for the messages put on the Kafka topic.",
           "type": "string"
         },
         "payload": {
@@ -3626,7 +3626,6 @@
       "required": [
         "url",
         "topic",
-        "partition",
         "payload"
       ],
       "type": "object"

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/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
@@ -568,7 +568,8 @@ type KafkaTrigger struct {
 	// Name of the topic.
 	// More info at https://kafka.apache.org/documentation/#intro_topics
 	Topic string `json:"topic" protobuf:"bytes,2,opt,name=topic"`
-	// Partition to write data to.
+	// +optional
+	// DEPRECATED
 	Partition int32 `json:"partition" protobuf:"varint,3,opt,name=partition"`
 	// Parameters is the list of parameters that is applied to resolved Kafka trigger object.
 	Parameters []TriggerParameter `json:"parameters,omitempty" protobuf:"bytes,4,rep,name=parameters"`
@@ -591,9 +592,8 @@ type KafkaTrigger struct {
 	// Payload is the list of key-value extracted from an event payload to construct the request payload.
 	Payload []TriggerParameter `json:"payload" protobuf:"bytes,9,rep,name=payload"`
 	// The partitioning key for the messages put on the Kafka topic.
-	// Defaults to broker url.
 	// +optional.
-	PartitioningKey string `json:"partitioningKey,omitempty" protobuf:"bytes,10,opt,name=partitioningKey"`
+	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"`

diff --git a/pkg/apis/sensor/v1alpha1/zz_generated.deepcopy.go b/pkg/apis/sensor/v1alpha1/zz_generated.deepcopy.go
diff --git a/sensors/triggers/kafka/kafka.go b/sensors/triggers/kafka/kafka.go
@@ -189,20 +189,19 @@ func (t *KafkaTrigger) Execute(ctx context.Context, events map[string]*v1alpha1.
 		return nil, err
 	}
 
-	pk := trigger.PartitioningKey
-	if pk == "" {
-		pk = trigger.URL
-	}
-
-	t.Producer.Input() <- &sarama.ProducerMessage{
+	msg := &sarama.ProducerMessage{
 		Topic:     trigger.Topic,
-		Key:       sarama.StringEncoder(pk),
 		Value:     sarama.ByteEncoder(payload),
-		Partition: trigger.Partition,
 		Timestamp: time.Now().UTC(),
 	}
 
-	t.Logger.Infow("successfully produced a message", zap.Any("topic", trigger.Topic), zap.Any("partition", trigger.Partition))
+	if trigger.PartitioningKey != nil {
+		msg.Key = sarama.StringEncoder(*trigger.PartitioningKey)
+	}
+
+	t.Producer.Input() <- msg
+
+	t.Logger.Infow("successfully produced a message", zap.Any("topic", trigger.Topic))
 
 	return nil, nil
 }

diff --git a/sensors/triggers/kafka/kafka_test.go b/sensors/triggers/kafka/kafka_test.go
@@ -43,7 +43,6 @@ var sensorObj = &v1alpha1.Sensor{
 					Kafka: &v1alpha1.KafkaTrigger{
 						URL:            "fake-kafka-url",
 						Topic:          "fake-topic",
-						Partition:      0,
 						Parameters:     nil,
 						RequiredAcks:   1,
 						Compress:       false,
@@ -53,7 +52,7 @@ var sensorObj = &v1alpha1.Sensor{
 						},
 						TLS:             nil,
 						Payload:         nil,
-						PartitioningKey: "",
+						PartitioningKey: nil,
 					},
 				},
 			},