Repository: knative-sandbox/eventing-kafka-broker (go)

Issue:
Webclient timeout is not set correctly
**Describe the bug**
Setting web client timeouts (idleTimeout, connectTimeout) in ConfigMap **config-kafka-source-data-plane** section **config-kafka-broker-webclient.properties** seems to have no effect.
Timeout 10 minutes is used anyway.

 **Expected behavior**
Timeouts are taken from ConfigMap **config-kafka-broker-webclient.properties**.

 **To Reproduce**

1. Install knative eventing, serving and Kafka Source. 
2. Specify web client timeout for Kafka Source in ConfigMap **config-kafka-source-data-plane** section **config-kafka-broker-webclient.properties**

    idleTimeout=0 (or 3600000 as 1 hour)
    connectTimeout=0 (or 3600000 as 1 hour)
3. Enqueue the task which takes more than 10 minutes
4. Get the error in kafka-source-dispatcher pod:
`{"@timestamp":"2023-01-10T06:12:25.834Z","@version":"1","message":"failed to send event to subscriber context={topics=[report-message-commands], consumerGroup='report-message-commands-group2', reference=uuid: \"da9e3045-019c-4347-902f-11ceb8d80cee\"\nnamespace: \"knative-test\"\nname: \"kafka-source2\"\n} target=http://job-service.knative-test.svc.cluster.local/report-message-commands","logger_name":"dev.knative.eventing.kafka.broker.dispatcher.impl.http.WebClientCloudEventSender","thread_name":"vert.x-worker-thread-16","level":"ERROR","level_value":40000,"stack_trace":"io.vertx.core.http.impl.NoStackTraceTimeoutException: The timeout period of 600000ms has been exceeded while executing POST /report-message-commands for server job-service.knative-test.svc.cluster.local:80\n","context":{},"target":"http://job-service.knative-test.svc.cluster.local/report-message-commands"}`

 **Knative release version**
1.8.3 with eventing-kafka-broker 1.8.5

 **Additional context**
Maybe I'm doing something wrong and should use another setting?


Pull request:
Added delivery to kafkasource spec
Part of #2875 

<!-- Please include the 'why' behind your changes if no issue exists -->

## Proposed Changes

- Add delivery spec to KafkaSource

<!--
If this change has user-visible impact, follow the instructions below.
Examples include:

- :gift: Add new feature
- :bug: Fix bug
- :broom: Update or clean up current behavior
- :wastebasket: Remove feature or internal logic

Otherwise delete the rest of this template.
-->

**Release Note**

<!--
:page_facing_up: If this change has user-visible impact, write a release note in the block
below. Include the string "action required" if additional action is required of
users switching to the new release, for example in case of a breaking change.

Write as if you are speaking to users, not other Knative contributors. If this
change has no user-visible impact, no release-note is needed.
-->

```release-note
KafkaSource resources now allow specifying delivery spec.
This means you can specify dead letter sink, number of retries, timeout and backoff options.
```

**Docs**

<!--
:book: If this change has user-visible impact, link to an issue or PR in
https://github.com/knative/docs.
-->
Docs may be updated to include this new feature of KafkaSource.

Patch:
diff --git a/.github/workflows/knative-go-build.yaml b/.github/workflows/knative-go-build.yaml
index b83849869..0ecbdbccb 100644
--- a/.github/workflows/knative-go-build.yaml
+++ b/.github/workflows/knative-go-build.yaml
@@ -11,4 +11,4 @@ on:
 
 jobs:
   build:
-    uses: knative/actions/.github/workflows/go-build.yaml@main
+    uses: knative/actions/.github/workflows/reusable-go-build.yaml@main
diff --git a/.github/workflows/knative-releasability.yaml b/.github/workflows/knative-releasability.yaml
index be5f765eb..7ca69caa1 100644
--- a/.github/workflows/knative-releasability.yaml
+++ b/.github/workflows/knative-releasability.yaml
@@ -23,7 +23,7 @@ on:
 
 jobs:
   releasability:
-    uses: knative/actions/.github/workflows/releasability.yaml@main
+    uses: knative/actions/.github/workflows/reusable-releasability.yaml@main
     with:
       releaseFamily: ${{ github.event.inputs.releaseFamily || 'v1.9' }}
       moduleReleaseFamily: ${{ github.event.inputs.moduleReleaseFamily || 'v0.36' }}
diff --git a/.github/workflows/knative-release-notes.yaml b/.github/workflows/knative-release-notes.yaml
index 45de1cc4f..1e63da78a 100644
--- a/.github/workflows/knative-release-notes.yaml
+++ b/.github/workflows/knative-release-notes.yaml
@@ -21,7 +21,7 @@ on:
 
 jobs:
   release-notes:
-    uses: knative/actions/.github/workflows/release-notes.yaml@main
+    uses: knative/actions/.github/workflows/reusable-release-notes.yaml@main
     with:
       branch: ${{ github.event.inputs.branch }}
       start-rev: ${{ github.event.inputs.start-rev }}
diff --git a/.github/workflows/knative-security.yaml b/.github/workflows/knative-security.yaml
index fbc343842..4a0848ca6 100644
--- a/.github/workflows/knative-security.yaml
+++ b/.github/workflows/knative-security.yaml
@@ -14,4 +14,4 @@ on:
 
 jobs:
   analyze:
-    uses: knative/actions/.github/workflows/security.yaml@main
+    uses: knative/actions/.github/workflows/reusable-security.yaml@main
diff --git a/.github/workflows/knative-stale.yaml b/.github/workflows/knative-stale.yaml
index bca559c0c..e46461c1e 100644
--- a/.github/workflows/knative-stale.yaml
+++ b/.github/workflows/knative-stale.yaml
@@ -11,4 +11,4 @@ on:
 jobs:
 
   stale:
-    uses: knative/actions/.github/workflows/stale.yaml@main
+    uses: knative/actions/.github/workflows/reusable-stale.yaml@main
diff --git a/.github/workflows/knative-style.yaml b/.github/workflows/knative-style.yaml
index ee7d49f8f..7ca9e6024 100644
--- a/.github/workflows/knative-style.yaml
+++ b/.github/workflows/knative-style.yaml
@@ -12,4 +12,4 @@ on:
 jobs:
 
   style:
-    uses: knative/actions/.github/workflows/style.yaml@main
+    uses: knative/actions/.github/workflows/reusable-style.yaml@main
diff --git a/.github/workflows/knative-verify.yaml b/.github/workflows/knative-verify.yaml
index 982a72711..28a4af54b 100644
--- a/.github/workflows/knative-verify.yaml
+++ b/.github/workflows/knative-verify.yaml
@@ -23,4 +23,4 @@ on:
 
 jobs:
   verify:
-    uses: knative/actions/.github/workflows/verify-codegen.yaml@main
+    uses: knative/actions/.github/workflows/reusable-verify-codegen.yaml@main
diff --git a/control-plane/pkg/apis/sources/v1beta1/kafka_defaults.go b/control-plane/pkg/apis/sources/v1beta1/kafka_defaults.go
index c44382f64..f6865f057 100644
--- a/control-plane/pkg/apis/sources/v1beta1/kafka_defaults.go
+++ b/control-plane/pkg/apis/sources/v1beta1/kafka_defaults.go
@@ -71,4 +71,5 @@ func (k *KafkaSource) SetDefaults(ctx context.Context) {
 	}
 
 	k.Spec.Sink.SetDefaults(ctx)
+	k.Spec.Delivery.SetDefaults(ctx)
 }
diff --git a/control-plane/pkg/apis/sources/v1beta1/kafka_types.go b/control-plane/pkg/apis/sources/v1beta1/kafka_types.go
index 0670f0a73..0cdd5f911 100644
--- a/control-plane/pkg/apis/sources/v1beta1/kafka_types.go
+++ b/control-plane/pkg/apis/sources/v1beta1/kafka_types.go
@@ -25,6 +25,7 @@ import (
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	bindingsv1beta1 "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/bindings/v1beta1"
+	eventingduckv1 "knative.dev/eventing/pkg/apis/duck/v1"
 	"knative.dev/pkg/apis"
 	duckv1 "knative.dev/pkg/apis/duck/v1"
 	"knative.dev/pkg/kmeta"
@@ -77,6 +78,10 @@ type KafkaSourceSpec struct {
 	// +optional
 	InitialOffset Offset `json:"initialOffset,omitempty"`
 
+	// Delivery contains the delivery spec for this source
+	// +optional
+	Delivery *eventingduckv1.DeliverySpec `json:"delivery,omitempty"`
+
 	// inherits duck/v1 SourceSpec, which currently provides:
 	// * Sink - a reference to an object that will resolve to a domain name or
 	//   a URI directly to use as the sink.
diff --git a/control-plane/pkg/apis/sources/v1beta1/zz_generated.deepcopy.go b/control-plane/pkg/apis/sources/v1beta1/zz_generated.deepcopy.go
index 4b58012a3..e7d0e7e2a 100644
--- a/control-plane/pkg/apis/sources/v1beta1/zz_generated.deepcopy.go
+++ b/control-plane/pkg/apis/sources/v1beta1/zz_generated.deepcopy.go
@@ -23,6 +23,7 @@ package v1beta1
 
 import (
 	runtime "k8s.io/apimachinery/pkg/runtime"
+	v1 "knative.dev/eventing/pkg/apis/duck/v1"
 )
 
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
@@ -100,6 +101,11 @@ func (in *KafkaSourceSpec) DeepCopyInto(out *KafkaSourceSpec) {
 		*out = make([]string, len(*in))
 		copy(*out, *in)
 	}
+	if in.Delivery != nil {
+		in, out := &in.Delivery, &out.Delivery
+		*out = new(v1.DeliverySpec)
+		(*in).DeepCopyInto(*out)
+	}
 	in.SourceSpec.DeepCopyInto(&out.SourceSpec)
 	return
 }
diff --git a/control-plane/pkg/contract/subscriptionsapi.go b/control-plane/pkg/contract/subscriptionsapi.go
index aa39475fe..703dfd76d 100644
--- a/control-plane/pkg/contract/subscriptionsapi.go
+++ b/control-plane/pkg/contract/subscriptionsapi.go
@@ -96,7 +96,7 @@ func newAllFilter(filters []v1.SubscriptionsAPIFilter) *DialectedFilter {
 // newAnyFilter converts the SubscriptionsAPIFilter into the any dialect of the
 // DialectedFilter as defined in CloudEvents Subscriptions API.
 //
-// All filter evaluates to true when all nested filters evaluate to true.
+// Any filter evaluates to true when at least one nested filters evaluate to true.
 //
 // See "CNCF CloudEvents Subscriptions API" > "3.2.4.1 Filters Dialects"
 // https://github.com/cloudevents/spec/blob/main/subscriptions/spec.md#any-filter-dialect
@@ -162,7 +162,7 @@ func FromSubscriptionFilter(f v1.SubscriptionsAPIFilter) *DialectedFilter {
 		filters = append(filters, newAllFilter(f.All))
 	}
 	if len(f.Any) > 0 {
-		filters = append(filters, newAnyFilter(f.All))
+		filters = append(filters, newAnyFilter(f.Any))
 	}
 	if f.Not != nil {
 		filters = append(filters, newNotFilter(*f.Not))
diff --git a/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/reconciler/impl/ResourcesReconcilerMessageHandler.java b/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/reconciler/impl/ResourcesReconcilerMessageHandler.java
index d12176bb5..1fb25bd48 100644
--- a/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/reconciler/impl/ResourcesReconcilerMessageHandler.java
+++ b/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/reconciler/impl/ResourcesReconcilerMessageHandler.java
@@ -37,6 +37,7 @@ public class ResourcesReconcilerMessageHandler implements Handler<Message<Object
 
   public final static String ADDRESS = "resourcesreconciler.core";
   public static final int RECONCILE_TIMEOUT = 10000;
+  public static final int RECONCILE_FAILED_RETRY_DELAY = 5000;
 
   private final Vertx vertx;
   private final ResourcesReconciler resourcesReconciler;
@@ -109,6 +110,13 @@ public class ResourcesReconcilerMessageHandler implements Handler<Message<Object
           // If that's the case, reconcile it.
           if (last.get().getGeneration() != contract.getGeneration()) {
             reconcileLast(null);
+          } else if (r.failed()) {
+            // Retry failed reconciliations after a few seconds.
+            // We add a delay to avoid looping too much on errors.
+            vertx.setTimer(RECONCILE_FAILED_RETRY_DELAY, v -> {
+              logger.info("Retrying previously failed reconciliation");
+              reconcileLast(null);
+            });
           }
         });
     }


Test patch:
diff --git a/.github/workflows/knative-go-test.yaml b/.github/workflows/knative-go-test.yaml
index bb016dcda..52a2d60ed 100644
--- a/.github/workflows/knative-go-test.yaml
+++ b/.github/workflows/knative-go-test.yaml
@@ -14,4 +14,4 @@ on:
 
 jobs:
   test:
-    uses: knative/actions/.github/workflows/go-test.yaml@main
+    uses: knative/actions/.github/workflows/reusable-go-test.yaml@main
diff --git a/control-plane/pkg/contract/subscriptionsapi_test.go b/control-plane/pkg/contract/subscriptionsapi_test.go
index aa4b6c54b..2155368cb 100644
--- a/control-plane/pkg/contract/subscriptionsapi_test.go
+++ b/control-plane/pkg/contract/subscriptionsapi_test.go
@@ -49,6 +49,13 @@ func TestFromSubscriptionFilter(t *testing.T) {
 					{
 						Suffix: map[string]string{"subject": "source"},
 					},
+					{
+						Any: []v1.SubscriptionsAPIFilter{
+							{
+								Exact: map[string]string{"type": "dev.knative.kafkasrc.kafkarecord"},
+							},
+						},
+					},
 					{
 						Not: &v1.SubscriptionsAPIFilter{
 							Prefix: map[string]string{"source": "dev.knative"},
@@ -84,6 +91,21 @@ func TestFromSubscriptionFilter(t *testing.T) {
 									},
 								},
 							},
+							{
+								Filter: &DialectedFilter_Any{
+									Any: &Any{
+										Filters: []*DialectedFilter{
+											{
+												Filter: &DialectedFilter_Exact{
+													Exact: &Exact{
+														Attributes: map[string]string{"type": "dev.knative.kafkasrc.kafkarecord"},
+													},
+												},
+											},
+										},
+									},
+								},
+							},
 							{
 								Filter: &DialectedFilter_Not{
 									Not: &Not{
@@ -115,7 +137,7 @@ func TestFromSubscriptionFilter(t *testing.T) {
 		t.Run(tc.name, func(t *testing.T) {
 			got := FromSubscriptionFilter(tc.apiFilter)
 			if cmp.Diff(got, tc.want, cmpopts.IgnoreUnexported(DialectedFilter{},
-				All{}, Prefix{}, Suffix{}, Exact{}, Not{}, CESQL{})) != "" {
+				All{}, Any{}, Prefix{}, Suffix{}, Exact{}, Not{}, CESQL{})) != "" {
 				t.Errorf("FromSubscriptionFilter() = %v, want %v", got, tc.want)
 			}
 		})