feat: Added dlq option to aws-sqs eventsource spec

api/jsonschema/schema.json

@@ -1968,6 +1968,10 @@
           "$ref": "#/definitions/io.k8s.api.core.v1.SecretKeySelector",
           "description": "AccessKey refers K8s secret containing aws access key"
         },
+        "dlq": {
+          "description": "DLQ specifies if a dead-letter queue is configured for messages that can't be processed successfully. If set to true, messages with invalid payload won't be acknowledged to allow to forward them farther to the dead-letter queue. The default value is false.",
+          "type": "boolean"
+        },
         "jsonBody": {
           "description": "JSONBody specifies that all event body payload coming from this source will be JSON",
           "type": "boolean"

eventsources/sources/awssqs/start.go

@@ -142,7 +142,10 @@ func (el *EventListener) processMessage(ctx context.Context, message *sqslib.Mes
 	if err != nil {
 		log.Errorw("failed to marshal event data, will process next message...", zap.Error(err))
 		el.Metrics.EventProcessingFailed(el.GetEventSourceName(), el.GetEventName())
-		ack()
+		// Don't ack if a DLQ is configured to allow to forward the message to the DLQ
+		if !el.SQSEventSource.DLQ {
+			ack()
+		}
 		return
 	}
 	if err = dispatch(eventBytes); err != nil {

pkg/apis/eventsource/v1alpha1/types.go

@@ -604,6 +604,11 @@ type SQSEventSource struct {
 	// Metadata holds the user defined metadata which will passed along the event payload.
 	// +optional
 	Metadata map[string]string `json:"metadata,omitempty" protobuf:"bytes,9,rep,name=metadata"`
+	// DLQ specifies if a dead-letter queue is configured for messages that can't be processed successfully.
+	// If set to true, messages with invalid payload won't be acknowledged to allow to forward them farther to the dead-letter queue.
+	// The default value is false.
+	// +optional
+	DLQ bool `json:"dlq,omitempty" protobuf:"varint,10,opt,name=dlq"`
 }
 
 // PubSubEventSource refers to event-source for GCP PubSub related events.