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

Issue:
Don't reconcile Triggers we don't own
**Describe the bug**

We reconcile every trigger in the cluster.

 **Expected behavior**

We reconciler only Trigger that are associated with a Broker that has the `Kafka` broker class.

 **To Reproduce**

Create an MTChannelBasedBroker.

 **Knative release version**
`<= devel`

/assign
/area control-plane
/kind bug


Pull request:
Don't reconcile Triggers we don't own
This PR adds a check to verify that the Trigger we're reconciling
is associated with a Broker with the annotation
`eventing.knative.dev/broker.class` set to `Kafka`.

Signed-off-by: Pierangelo Di Pilato <pierangelodipilato@gmail.com>

Fixes #211 

<!-- Please include the 'why' behind your changes if no issue exists -->

## Proposed Changes

- Don't reconcile Triggers we don't own

Patch:
diff --git a/control-plane/pkg/reconciler/trigger/trigger.go b/control-plane/pkg/reconciler/trigger/trigger.go
index b322f3ea9..26b2b729b 100644
--- a/control-plane/pkg/reconciler/trigger/trigger.go
+++ b/control-plane/pkg/reconciler/trigger/trigger.go
@@ -22,6 +22,7 @@ import (
 	"math"
 
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/contract"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/kafka"
 
 	"go.uber.org/zap"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
@@ -206,6 +207,12 @@ func (r *Reconciler) reconcileKind(ctx context.Context, trigger *eventing.Trigge
 		}
 	}
 
+	// Ignore Triggers that are associated with a Broker we don't own.
+	if isOur, brokerClass := isOurBroker(broker); !isOur {
+		logger.Debug("Ignoring Trigger", zap.String(eventing.BrokerClassAnnotationKey, brokerClass))
+		return nil
+	}
+
 	if !broker.GetDeletionTimestamp().IsZero() {
 
 		logger.Debug("broker deleted", zap.String("finalizeDuringReconcile", "deleted"))
@@ -286,6 +293,11 @@ func (r *Reconciler) reconcileKind(ctx context.Context, trigger *eventing.Trigge
 	return statusConditionManager.reconciled()
 }
 
+func isOurBroker(broker *eventing.Broker) (bool, string) {
+	brokerClass := broker.GetAnnotations()[eventing.BrokerClassAnnotationKey]
+	return brokerClass == kafka.BrokerClass, brokerClass
+}
+
 func incrementGeneration(generation uint64) uint64 {
 	return (generation + 1) % (math.MaxUint64 - 1)
 }
diff --git a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/http/HttpSinkResponseHandler.java b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/http/HttpSinkResponseHandler.java
index 25d11566e..f3968670c 100644
--- a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/http/HttpSinkResponseHandler.java
+++ b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/http/HttpSinkResponseHandler.java
@@ -62,6 +62,14 @@ public final class HttpSinkResponseHandler implements SinkResponseHandler<HttpRe
   public Future<Void> handle(final HttpResponse<Buffer> response) {
     MessageReader messageReader = VertxMessageFactory.createReader(response);
     if (messageReader.getEncoding() == Encoding.UNKNOWN) {
+
+      // When the sink returns a malformed event we return a failed future to avoid committing the message to Kafka.
+      if (response.body().length() > 0) {
+        return Future.failedFuture(
+          new IllegalResponseException("Unable to decode response: unknown encoding and non empty response")
+        );
+      }
+
       // Response is non-event, discard it
       return Future.succeededFuture();
     }
diff --git a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/http/IllegalResponseException.java b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/http/IllegalResponseException.java
new file mode 100644
index 000000000..d7a7ffdce
--- /dev/null
+++ b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/http/IllegalResponseException.java
@@ -0,0 +1,24 @@
+/*
+ * Copyright 2020 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package dev.knative.eventing.kafka.broker.dispatcher.http;
+
+public class IllegalResponseException extends Exception {
+
+  public IllegalResponseException(String s) {
+    super(s);
+  }
+}


Test patch:
diff --git a/control-plane/pkg/reconciler/testing/objects_broker.go b/control-plane/pkg/reconciler/testing/objects_broker.go
index f921518ff..620250e50 100644
--- a/control-plane/pkg/reconciler/testing/objects_broker.go
+++ b/control-plane/pkg/reconciler/testing/objects_broker.go
@@ -54,11 +54,13 @@ func NewBroker(options ...reconcilertesting.BrokerOption) runtime.Object {
 		BrokerName,
 		BrokerNamespace,
 		append(
-			options,
-			reconcilertesting.WithBrokerClass(kafka.BrokerClass),
-			func(broker *eventing.Broker) {
-				broker.UID = BrokerUUID
+			[]reconcilertesting.BrokerOption{
+				reconcilertesting.WithBrokerClass(kafka.BrokerClass),
+				func(broker *eventing.Broker) {
+					broker.UID = BrokerUUID
+				},
 			},
+			options...,
 		)...,
 	)
 }
diff --git a/control-plane/pkg/reconciler/trigger/trigger_test.go b/control-plane/pkg/reconciler/trigger/trigger_test.go
index ef1413652..288b6c5b2 100644
--- a/control-plane/pkg/reconciler/trigger/trigger_test.go
+++ b/control-plane/pkg/reconciler/trigger/trigger_test.go
@@ -1090,6 +1090,31 @@ func triggerReconciliation(t *testing.T, format string, configs broker.Configs)
 				},
 			},
 		},
+		{
+			Name: "Don't reconcile trigger associated with a broker with a different broker class",
+			Objects: []runtime.Object{
+				newTrigger(),
+				NewBroker(func(b *eventing.Broker) {
+					b.Annotations = map[string]string{
+						eventing.BrokerClassAnnotationKey: "MTChannelBasedBroker",
+					}
+				}),
+			},
+			Key: testKey,
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: newTrigger(
+						reconcilertesting.WithInitTriggerConditions,
+					),
+				},
+			},
+			WantEvents: []string{
+				finalizerUpdatedEvent,
+			},
+			WantPatches: []clientgotesting.PatchActionImpl{
+				patchFinalizers(),
+			},
+		},
 	}
 
 	for i := range table {
diff --git a/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/http/HttpSinkResponseHandlerTest.java b/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/http/HttpSinkResponseHandlerTest.java
index a862b6343..80910b700 100644
--- a/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/http/HttpSinkResponseHandlerTest.java
+++ b/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/http/HttpSinkResponseHandlerTest.java
@@ -48,7 +48,7 @@ public class HttpSinkResponseHandlerTest {
   private static final String TOPIC = "t1";
 
   @Test
-  public void shouldAlwaysSucceed(final Vertx vertx, final VertxTestContext context) {
+  public void shouldSucceedOnUnknownEncodingAndEmptyResponse(final Vertx vertx, final VertxTestContext context) {
     final var producer = new MockProducer<>(
       true,
       new StringSerializer(),
@@ -67,7 +67,32 @@ public class HttpSinkResponseHandlerTest {
 
     context
       .assertComplete(handler.handle(response))
-      .onComplete(v -> context.completeNow());
+      .onSuccess(v -> context.completeNow());
+  }
+
+  @Test
+  public void shouldFailOnUnknownEncodingAndNonEmptyResponse(final Vertx vertx, final VertxTestContext context) {
+
+    final var producer = new MockProducer<>(
+      true,
+      new StringSerializer(),
+      new CloudEventSerializer()
+    );
+    final var handler = new HttpSinkResponseHandler(
+      TOPIC,
+      KafkaProducer.create(vertx, producer)
+    );
+
+    // Empty response
+    final HttpResponse<Buffer> response = mock(HttpResponse.class);
+    when(response.statusCode()).thenReturn(202);
+    when(response.body()).thenReturn(Buffer.buffer(new byte[]{'a'}));
+    when(response.headers()).thenReturn(MultiMap.caseInsensitiveMultiMap());
+
+    context
+      .assertFailure(handler.handle(response))
+      .onSuccess(s -> context.failNow(new Exception("expected failed future")))
+      .onFailure(v -> context.completeNow());
   }
 
   @Test