Repository: knative-sandbox/eventing-kafka-broker
Pull request #213 (2020-09-24 08:16:57)

Upload Codecov Java report
It's different from https://github.com/knative-sandbox/eventing-kafka-broker/pull/199 since in this one we upload reports in two different actions independently with two different flags.

## Proposed Changes

- Upload Codecov Java report

knative-prow-robot: [APPROVALNOTIFIER] This PR is **APPROVED**

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/213#" title="Author self-approved">pierDipi</a>*

The full list of commands accepted by this bot can be found [here](https://go.k8s.io/bot-commands?repo=knative-sandbox%2Feventing-kafka-broker).

The pull request process is described [here](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process)

<details >
Needs approval from an approver in each of these files:

- ~~[OWNERS](https://github.com/knative-sandbox/eventing-kafka-broker/blob/master/OWNERS)~~ [pierDipi]

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":[]} -->
codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/213?src=pr&el=h1) Report
> Merging [#213](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/213?src=pr&el=desc) into [master](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/commit/04d4dc504f724f5610f95c1c1d949a67122539a0?el=desc) will **decrease** coverage by `0.14%`.
> The diff coverage is `n/a`.

[![Impacted file tree graph](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/213/graphs/tree.svg?width=650&height=150&src=pr&token=xyPIgGOiwy)](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/213?src=pr&el=tree)

```diff
@@             Coverage Diff              @@
##             master     #213      +/-   ##
============================================
- Coverage     76.32%   76.18%   -0.15%     
  Complexity      164      164              
============================================
  Files            47       25      -22     
  Lines          1394      676     -718     
  Branches         45       45              
============================================
- Hits           1064      515     -549     
+ Misses          261      139     -122     
+ Partials         69       22      -47     
```

| Flag | Coverage Δ | Complexity Δ | |
|---|---|---|---|
| #unittests | `76.18% <ø> (-0.15%)` | `164.00 <ø> (ø)` | |

Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags#carryforward-flags-in-the-pull-request-comment) to find out more.

| [Impacted Files](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/213?src=pr&el=tree) | Coverage Δ | Complexity Δ | |
|---|---|---|---|
| [...ntrol-plane/pkg/reconciler/broker/broker\_config.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/213/diff?src=pr&el=tree#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9icm9rZXIvYnJva2VyX2NvbmZpZy5nbw==) | | | |
| [control-plane/pkg/receiver/path.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/213/diff?src=pr&el=tree#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjZWl2ZXIvcGF0aC5nbw==) | | | |
| [control-plane/pkg/config/env\_config.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/213/diff?src=pr&el=tree#diff-Y29udHJvbC1wbGFuZS9wa2cvY29uZmlnL2Vudl9jb25maWcuZ28=) | | | |
| [control-plane/pkg/reconciler/kafka/admin.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/213/diff?src=pr&el=tree#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9rYWZrYS9hZG1pbi5nbw==) | | | |
| [...kg/apis/eventing/v1alpha1/kafka\_sink\_conversion.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/213/diff?src=pr&el=tree#diff-Y29udHJvbC1wbGFuZS9wa2cvYXBpcy9ldmVudGluZy92MWFscGhhMS9rYWZrYV9zaW5rX2NvbnZlcnNpb24uZ28=) | | | |
| [control-plane/pkg/core/config/egress.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/213/diff?src=pr&el=tree#diff-Y29udHJvbC1wbGFuZS9wa2cvY29yZS9jb25maWcvZWdyZXNzLmdv) | | | |
| [control-plane/pkg/reconciler/trigger/trigger.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/213/diff?src=pr&el=tree#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci90cmlnZ2VyL3RyaWdnZXIuZ28=) | | | |
| [...ane/pkg/apis/eventing/v1alpha1/kafka\_sink\_types.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/213/diff?src=pr&el=tree#diff-Y29udHJvbC1wbGFuZS9wa2cvYXBpcy9ldmVudGluZy92MWFscGhhMS9rYWZrYV9zaW5rX3R5cGVzLmdv) | | | |
| [...pkg/apis/eventing/v1alpha1/kafka\_sink\_lifecycle.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/213/diff?src=pr&el=tree#diff-Y29udHJvbC1wbGFuZS9wa2cvYXBpcy9ldmVudGluZy92MWFscGhhMS9rYWZrYV9zaW5rX2xpZmVjeWNsZS5nbw==) | | | |
| [control-plane/pkg/reconciler/trigger/controller.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/213/diff?src=pr&el=tree#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci90cmlnZ2VyL2NvbnRyb2xsZXIuZ28=) | | | |
| ... and [12 more](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/213/diff?src=pr&el=tree-more) | |

------

[Continue to review full report at Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/213?src=pr&el=continue).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/213?src=pr&el=footer). Last update [04d4dc5...21b65f3](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/213?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).

pierDipi: It seems to be working: https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/213?src=pr&el=continue, also see the table of the report message above.
slinkydeveloper: Ah, two different flags does the trick, thanks!

/lgtm


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/.github/workflows/knative-java-test.yaml b/.github/workflows/knative-java-test.yaml
new file mode 100644
index 000000000..505c22163
--- /dev/null
+++ b/.github/workflows/knative-java-test.yaml
@@ -0,0 +1,63 @@
+# 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.
+
+# This file is automagically synced here from github.com/knative-sandbox/.github
+# repo by knobots: https://github.com/mattmoor/knobots and will be overwritten.
+
+name: Test
+
+on:
+
+  push:
+    branches: [ 'master' ]
+
+  pull_request:
+    branches: [ 'master', 'release-*' ]
+
+jobs:
+
+  test:
+    name: Java Unit Tests
+    strategy:
+      matrix:
+        java-version: [ 14 ]
+        platform: [ ubuntu-latest ]
+
+    runs-on: ${{ matrix.platform }}
+
+    steps:
+      - name: Check out code
+        uses: actions/checkout@v2
+
+      - name: Setup java
+        uses: actions/setup-java@v1
+        with:
+          java-version: ${{ matrix.java-version }}
+
+      - name: Check for .codecov.yaml
+        id: codecov-enabled
+        uses: andstor/file-existence-action@v1
+        with:
+          files: .codecov.yaml
+
+      - if: steps.codecov-enabled.outputs.files_exists == 'true'
+        name: Java Test
+        run: mvn verify --file data-plane/pom.xml --no-transfer-progress
+
+      - if: steps.codecov-enabled.outputs.files_exists == 'true'
+        name: Codecov
+        uses: codecov/codecov-action@v1
+        with:
+          file: ./data-plane/target/jacoco/jacoco.xml
+          flags: java-unittests
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