Delivery timeout

config/channels/in-memory-channel/resources/in-memory-channel.yaml

@@ -74,6 +74,7 @@ spec:
                     description: Retry is the minimum number of retries the sender should attempt when sending an event before moving it to the dead letter sink.
                     type: integer
                     format: int32
+                x-kubernetes-preserve-unknown-fields: true # This is necessary to enable the experimental feature delivery-timeout
               subscribers:
                 description: This is the list of subscriptions for this subscribable.
                 type: array

config/core/configmaps/features.yaml

@@ -25,3 +25,7 @@ data:
   # ALPHA feature: The kreference-group allows you to use the Group field in KReferences.
   # For more details: https://github.com/knative/eventing/issues/5086
   kreference-group: "disabled"
+
+  # ALPHA feature: The delivery-timeout allows you to use the Timeout field in DeliverySpec.
+  # For more details: https://github.com/knative/eventing/issues/5148
+  delivery-timeout: "disabled"

config/core/resources/broker.yaml

@@ -90,6 +90,7 @@ spec:
                     description: Retry is the minimum number of retries the sender should attempt when sending an event before moving it to the dead letter sink.
                     type: integer
                     format: int32
+                x-kubernetes-preserve-unknown-fields: true # This is necessary to enable the experimental feature delivery-timeout
           status:
             description: Status represents the current state of the Broker. This data may be out of date.
             type: object

config/core/resources/channel.yaml

@@ -102,6 +102,7 @@ spec:
                     description: Retry is the minimum number of retries the sender should attempt when sending an event before moving it to the dead letter sink.
                     type: integer
                     format: int32
+                x-kubernetes-preserve-unknown-fields: true # This is necessary to enable the experimental feature delivery-timeout
               subscribers:
                 description: This is the list of subscriptions for this subscribable.
                 type: array

config/core/resources/parallel.yaml

@@ -104,6 +104,7 @@ spec:
                               sink.
                           type: integer
                           format: int32
+                      x-kubernetes-preserve-unknown-fields: true # This is necessary to enable the experimental feature delivery-timeout
                     filter:
                       description: Filter is the expression guarding the branch
                       type: object

config/core/resources/sequence.yaml

@@ -115,6 +115,7 @@ spec:
                           description: Retry is the minimum number of retries the sender should attempt when sending an event before moving it to the dead letter sink.
                           type: integer
                           format: int32
+                      x-kubernetes-preserve-unknown-fields: true # This is necessary to enable the experimental feature delivery-timeout
                     ref:
                       description: Ref points to an Addressable.
                       type: object

config/core/resources/subscription.yaml

@@ -84,6 +84,7 @@ spec:
                     description: Retry is the minimum number of retries the sender should attempt when sending an event before moving it to the dead letter sink.
                     type: integer
                     format: int32
+                x-kubernetes-preserve-unknown-fields: true # This is necessary to enable the experimental feature delivery-timeout
               reply:
                 description: Reply specifies (optionally) how to handle events returned from the Subscriber target.
                 type: object

docs/eventing-api.md

@@ -325,6 +325,22 @@ sending an event before moving it to the dead letter sink.</p>
 </tr>
 <tr>
 <td>
+<code>timeout</code><br/>
+<em>
+string
+</em>
+</td>
+<td>
+<em>(Optional)</em>
+<p>Timeout is the timeout of each single request. The value must be greater than 0.
+More information on Duration format:
+- <a href="https://www.iso.org/iso-8601-date-and-time-format.html">https://www.iso.org/iso-8601-date-and-time-format.html</a>
+- <a href="https://en.wikipedia.org/wiki/ISO_8601">https://en.wikipedia.org/wiki/ISO_8601</a></p>
+<p>Note: This API is EXPERIMENTAL and might break anytime. For more details: <a href="https://github.com/knative/eventing/issues/5148">https://github.com/knative/eventing/issues/5148</a></p>
+</td>
+</tr>
+<tr>
+<td>
 <code>backoffPolicy</code><br/>
 <em>
 <a href="#duck.knative.dev/v1.BackoffPolicyType">
@@ -977,6 +993,20 @@ sending an event before moving it to the dead letter sink.</p>
 </tr>
 <tr>
 <td>
+<code>timeout</code><br/>
+<em>
+string
+</em>
+</td>
+<td>
+<p>Timeout is the timeout of each single request.
+More information on Duration format:
+- <a href="https://www.iso.org/iso-8601-date-and-time-format.html">https://www.iso.org/iso-8601-date-and-time-format.html</a>
+- <a href="https://en.wikipedia.org/wiki/ISO_8601">https://en.wikipedia.org/wiki/ISO_8601</a></p>
+</td>
+</tr>
+<tr>
+<td>
 <code>backoffPolicy</code><br/>
 <em>
 <a href="#duck.knative.dev/v1beta1.BackoffPolicyType">

pkg/apis/duck/v1/delivery_types.go

@@ -20,6 +20,7 @@ import (
 	"context"
 
 	"github.com/rickb777/date/period"
+	"knative.dev/eventing/pkg/apis/feature"
 	"knative.dev/pkg/apis"
 	duckv1 "knative.dev/pkg/apis/duck/v1"
 )
@@ -37,6 +38,15 @@ type DeliverySpec struct {
 	// +optional
 	Retry *int32 `json:"retry,omitempty"`
 
+	// Timeout is the timeout of each single request. The value must be greater than 0.
+	// More information on Duration format:
+	//  - https://www.iso.org/iso-8601-date-and-time-format.html
+	//  - https://en.wikipedia.org/wiki/ISO_8601
+	//
+	// Note: This API is EXPERIMENTAL and might break anytime. For more details: https://github.com/knative/eventing/issues/5148
+	// +optional
+	Timeout *string `json:"timeout,omitempty"`
+
 	// BackoffPolicy is the retry backoff policy (linear, exponential).
 	// +optional
 	BackoffPolicy *BackoffPolicyType `json:"backoffPolicy,omitempty"`
@@ -65,6 +75,17 @@ func (ds *DeliverySpec) Validate(ctx context.Context) *apis.FieldError {
 		errs = errs.Also(apis.ErrInvalidValue(*ds.Retry, "retry"))
 	}
 
+	if ds.Timeout != nil {
+		if feature.FromContext(ctx).IsEnabled(feature.DeliveryTimeout) {
+			t, te := period.Parse(*ds.Timeout)
+			if te != nil || t.IsZero() {
+				errs = errs.Also(apis.ErrInvalidValue(*ds.Timeout, "timeout"))
+			}
+		} else {
+			errs = errs.Also(apis.ErrDisallowedFields("timeout"))
+		}
+	}
+
 	if ds.BackoffPolicy != nil {
 		switch *ds.BackoffPolicy {
 		case BackoffPolicyExponential, BackoffPolicyLinear:

pkg/apis/duck/v1/delivery_types_test.go

@@ -21,18 +21,24 @@ import (
 
 	"github.com/google/go-cmp/cmp"
 	"k8s.io/utils/pointer"
+	"knative.dev/eventing/pkg/apis/feature"
 	"knative.dev/pkg/apis"
 	duckv1 "knative.dev/pkg/apis/duck/v1"
 )
 
 func TestDeliverySpecValidation(t *testing.T) {
+	deliveryTimeoutEnabledCtx := feature.ToContext(context.TODO(), feature.Flags{
+		feature.DeliveryTimeout: feature.Enabled,
+	})
+
 	invalidString := "invalid time"
 	bop := BackoffPolicyExponential
-	validBackoffDelay := "PT2S"
-	invalidBackoffDelay := "1985-04-12T23:20:50.52Z"
+	validDuration := "PT2S"
+	invalidDuration := "1985-04-12T23:20:50.52Z"
 	tests := []struct {
 		name string
 		spec *DeliverySpec
+		ctx  context.Context
 		want *apis.FieldError
 	}{{
 		name: "nil is valid",
@@ -50,25 +56,48 @@ func TestDeliverySpecValidation(t *testing.T) {
 		want: func() *apis.FieldError {
 			return apis.ErrGeneric("expected at least one, got none", "ref", "uri").ViaField("deadLetterSink")
 		}(),
+	}, {
+		name: "valid timeout",
+		spec: &DeliverySpec{Timeout: &validDuration},
+		ctx:  deliveryTimeoutEnabledCtx,
+		want: nil,
+	}, {
+		name: "invalid timeout",
+		spec: &DeliverySpec{Timeout: &invalidDuration},
+		ctx:  deliveryTimeoutEnabledCtx,
+		want: func() *apis.FieldError {
+			return apis.ErrInvalidValue(invalidDuration, "timeout")
+		}(),
+	}, {
+		name: "zero timeout",
+		spec: &DeliverySpec{Timeout: pointer.StringPtr("PT0S")},
+		ctx:  deliveryTimeoutEnabledCtx,
+		want: func() *apis.FieldError {
+			return apis.ErrInvalidValue("PT0S", "timeout")
+		}(),
+	}, {
+		name: "disabled timeout",
+		spec: &DeliverySpec{Timeout: &validDuration},
+		want: apis.ErrDisallowedFields("timeout"),
 	}, {
 		name: "valid backoffPolicy",
 		spec: &DeliverySpec{BackoffPolicy: &bop},
 		want: nil,
 	}, {
 		name: "valid backoffDelay",
-		spec: &DeliverySpec{BackoffDelay: &validBackoffDelay},
+		spec: &DeliverySpec{BackoffDelay: &validDuration},
 		want: nil,
 	}, {
 		name: "invalid backoffDelay",
-		spec: &DeliverySpec{BackoffDelay: &invalidBackoffDelay},
+		spec: &DeliverySpec{BackoffDelay: &invalidDuration},
 		want: func() *apis.FieldError {
-			return apis.ErrGeneric("invalid value: "+invalidBackoffDelay, "backoffDelay")
+			return apis.ErrInvalidValue(invalidDuration, "backoffDelay")
 		}(),
 	}, {
 		name: "negative retry",
 		spec: &DeliverySpec{Retry: pointer.Int32Ptr(-1)},
 		want: func() *apis.FieldError {
-			return apis.ErrGeneric("invalid value: -1", "retry")
+			return apis.ErrInvalidValue("-1", "retry")
 		}(),
 	}, {
 		name: "valid retry 0",
@@ -80,7 +109,11 @@ func TestDeliverySpecValidation(t *testing.T) {
 
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
-			got := test.spec.Validate(context.TODO())
+			ctx := test.ctx
+			if ctx == nil {
+				ctx = context.TODO()
+			}
+			got := test.spec.Validate(ctx)
 			if diff := cmp.Diff(test.want.Error(), got.Error()); diff != "" {
 				t.Error("DeliverySpec.Validate (-want, +got) =", diff)
 			}

pkg/apis/duck/v1beta1/delivery_conversion.go

@@ -31,6 +31,7 @@ func (source *DeliverySpec) ConvertTo(ctx context.Context, to apis.Convertible)
 	case *eventingduckv1.DeliverySpec:
 		sink.Retry = source.Retry
 		sink.BackoffDelay = source.BackoffDelay
+		sink.Timeout = source.Timeout
 		if source.BackoffPolicy != nil {
 			if *source.BackoffPolicy == BackoffPolicyLinear {
 				linear := eventingduckv1.BackoffPolicyLinear
@@ -55,6 +56,7 @@ func (sink *DeliverySpec) ConvertFrom(ctx context.Context, from apis.Convertible
 	case *eventingduckv1.DeliverySpec:
 		sink.Retry = source.Retry
 		sink.BackoffDelay = source.BackoffDelay
+		sink.Timeout = source.Timeout
 		if source.BackoffPolicy != nil {
 			if *source.BackoffPolicy == eventingduckv1.BackoffPolicyLinear {
 				linear := BackoffPolicyLinear

pkg/apis/duck/v1beta1/delivery_types.go

@@ -38,6 +38,13 @@ type DeliverySpec struct {
 	// +optional
 	Retry *int32 `json:"retry,omitempty"`
 
+	// Timeout is the timeout of each single request.
+	// More information on Duration format:
+	//  - https://www.iso.org/iso-8601-date-and-time-format.html
+	//  - https://en.wikipedia.org/wiki/ISO_8601
+	//
+	Timeout *string `json:"timeout,omitempty"`
+
 	// BackoffPolicy is the retry backoff policy (linear, exponential).
 	// +optional
 	BackoffPolicy *BackoffPolicyType `json:"backoffPolicy,omitempty"`
@@ -66,6 +73,13 @@ func (ds *DeliverySpec) Validate(ctx context.Context) *apis.FieldError {
 		errs = errs.Also(apis.ErrInvalidValue(*ds.Retry, "retry"))
 	}
 
+	if ds.Timeout != nil {
+		_, te := period.Parse(*ds.Timeout)
+		if te != nil {
+			errs = errs.Also(apis.ErrInvalidValue(*ds.Timeout, "timeout"))
+		}
+	}
+
 	if ds.BackoffPolicy != nil {
 		switch *ds.BackoffPolicy {
 		case BackoffPolicyExponential, BackoffPolicyLinear:

pkg/apis/duck/v1beta1/delivery_types_test.go

@@ -28,8 +28,8 @@ import (
 func TestDeliverySpecValidation(t *testing.T) {
 	invalidString := "invalid time"
 	bop := BackoffPolicyExponential
-	validBackoffDelay := "PT2S"
-	invalidBackoffDelay := "1985-04-12T23:20:50.52Z"
+	validDuration := "PT2S"
+	invalidDuration := "1985-04-12T23:20:50.52Z"
 	tests := []struct {
 		name string
 		spec *DeliverySpec
@@ -50,25 +50,35 @@ func TestDeliverySpecValidation(t *testing.T) {
 		want: func() *apis.FieldError {
 			return apis.ErrGeneric("expected at least one, got none", "ref", "uri").ViaField("deadLetterSink")
 		}(),
+	}, {
+		name: "valid timeout",
+		spec: &DeliverySpec{Timeout: &validDuration},
+		want: nil,
+	}, {
+		name: "invalid timeout",
+		spec: &DeliverySpec{Timeout: &invalidDuration},
+		want: func() *apis.FieldError {
+			return apis.ErrInvalidValue(invalidDuration, "timeout")
+		}(),
 	}, {
 		name: "valid backoffPolicy",
 		spec: &DeliverySpec{BackoffPolicy: &bop},
 		want: nil,
 	}, {
 		name: "valid backoffDelay",
-		spec: &DeliverySpec{BackoffDelay: &validBackoffDelay},
+		spec: &DeliverySpec{BackoffDelay: &validDuration},
 		want: nil,
 	}, {
 		name: "invalid backoffDelay",
-		spec: &DeliverySpec{BackoffDelay: &invalidBackoffDelay},
+		spec: &DeliverySpec{BackoffDelay: &invalidDuration},
 		want: func() *apis.FieldError {
-			return apis.ErrGeneric("invalid value: "+invalidBackoffDelay, "backoffDelay")
+			return apis.ErrInvalidValue(invalidDuration, "backoffDelay")
 		}(),
 	}, {
 		name: "negative retry",
 		spec: &DeliverySpec{Retry: pointer.Int32Ptr(-1)},
 		want: func() *apis.FieldError {
-			return apis.ErrGeneric("invalid value: -1", "retry")
+			return apis.ErrInvalidValue("-1", "retry")
 		}(),
 	}, {
 		name: "valid retry 0",

pkg/apis/feature/flag_names.go

@@ -18,4 +18,5 @@ package feature
 
 const (
 	KReferenceGroup = "kreference-group"
+	DeliveryTimeout = "delivery-timeout"
 )