Repository: knative-sandbox/eventing-kafka-broker
Pull request #2391 (2022-07-07 15:56:37)

KafkaSink is binary now
Fixes #

<!-- Please include the 'why' behind your changes if no issue exists -->

## Proposed Changes

- align the `KafkaSink` with the rest of the components and default to CE binary mode

<!--
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
The KafkaSink now uses CloudEvent binary content mode as default
```

**Docs**

<!--
:book: If this change has user-visible impact, link to an issue or PR in
https://github.com/knative/docs.
-->

* PR for docs: https://github.com/knative/docs/pull/5084

codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2391?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) Report
> Merging [#2391](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2391?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (09ef270) into [main](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/commit/744cdc3bd8c1cb08e318a211239365d84204e87e?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (744cdc3) will **increase** coverage by `22.46%`.
> The diff coverage is `n/a`.

```diff
@@              Coverage Diff              @@
##               main    #2391       +/-   ##
=============================================
+ Coverage     59.97%   82.43%   +22.46%     
- Complexity        0      705      +705     
=============================================
  Files            69       72        +3     
  Lines          6963     2397     -4566     
  Branches          0      210      +210     
=============================================
- Hits           4176     1976     -2200     
+ Misses         2491      305     -2186     
+ Partials        296      116      -180     
```

| Flag | Coverage Δ | |
|---|---|---|
| java-unittests | `82.43% <ø> (?)` | |

Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#carryforward-flags-in-the-pull-request-comment) to find out more.

| [Impacted Files](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2391?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | Coverage Δ | |
|---|---|---|
| [.../pkg/apis/eventing/v1alpha1/kafka\_sink\_defaults.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2391/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvYXBpcy9ldmVudGluZy92MWFscGhhMS9rYWZrYV9zaW5rX2RlZmF1bHRzLmdv) | | |
| [control-plane/pkg/reconciler/sink/kafka\_sink.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2391/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9zaW5rL2thZmthX3NpbmsuZ28=) | | |
| [control-plane/pkg/reconciler/trigger/controller.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2391/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci90cmlnZ2VyL2NvbnRyb2xsZXIuZ28=) | | |
| [...ane/pkg/apis/eventing/v1alpha1/kafka\_sink\_types.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2391/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvYXBpcy9ldmVudGluZy92MWFscGhhMS9rYWZrYV9zaW5rX3R5cGVzLmdv) | | |
| [...trol-plane/pkg/reconciler/consumergroup/evictor.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2391/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9jb25zdW1lcmdyb3VwL2V2aWN0b3IuZ28=) | | |
| [control-plane/pkg/reconciler/sink/auth\_provider.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2391/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9zaW5rL2F1dGhfcHJvdmlkZXIuZ28=) | | |
| [...trol-plane/pkg/prober/probertesting/mock\_prober.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2391/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcHJvYmVyL3Byb2JlcnRlc3RpbmcvbW9ja19wcm9iZXIuZ28=) | | |
| [control-plane/pkg/prober/async\_prober.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2391/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcHJvYmVyL2FzeW5jX3Byb2Jlci5nbw==) | | |
| [...ntrol-plane/pkg/reconciler/consumergroup/filter.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2391/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9jb25zdW1lcmdyb3VwL2ZpbHRlci5nbw==) | | |
| [control-plane/pkg/core/config/resource.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2391/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvY29yZS9jb25maWcvcmVzb3VyY2UuZ28=) | | |
| ... and [131 more](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2391/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | |

------

[Continue to review full report at Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2391?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2391?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Last update [744cdc3...09ef270](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2391?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox).

matzew: /test unit-tests_eventing-kafka-broker_main
knative-prow[bot]: @matzew: The following test **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
unit-tests_eventing-kafka-broker_main | 7baeda77fdf3b2644f3314595d1f3e4ff4c99cb6 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2391/unit-tests_eventing-kafka-broker_main/1545116226085195776) | true | `/test unit-tests_eventing-kafka-broker_main`

[Your PR dashboard](https://prow.knative.dev/pr).

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
matzew: 

/test unit-tests_eventing-kafka-broker_main
pierDipi: /kind api-change
knative-prow[bot]: [APPROVALNOTIFIER] This PR is **APPROVED**

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/2391#" title="Author self-approved">matzew</a>*, *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/2391#pullrequestreview-1032579509" title="LGTM">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/main/OWNERS)~~ [matzew,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":[]} -->

Patch:
diff --git a/control-plane/config/eventing-kafka-broker/100-sink/100-kafka-sink.yaml b/control-plane/config/eventing-kafka-broker/100-sink/100-kafka-sink.yaml
index 14c9820b4..ce2aad482 100644
--- a/control-plane/config/eventing-kafka-broker/100-sink/100-kafka-sink.yaml
+++ b/control-plane/config/eventing-kafka-broker/100-sink/100-kafka-sink.yaml
@@ -73,15 +73,15 @@ spec:
                 contentMode:
                   description: |
                     CloudEvent content mode of Kafka messages sent to the topic.
-                    Possible values: [structured, binary] (default: structured)
+                    Possible values: [binary, structured] (default: binary)
                     - https://github.com/cloudevents/spec/blob/v1.0/spec.md#message
-                      - https://github.com/cloudevents/spec/blob/v1.0/kafka-protocol-binding.md#33-structured-content-mode
                       - https://github.com/cloudevents/spec/blob/v1.0/kafka-protocol-binding.md#32-binary-content-mode
+                      - https://github.com/cloudevents/spec/blob/v1.0/kafka-protocol-binding.md#33-structured-content-mode
                   type: string
                   enum:
-                    - structured
                     - binary
-                  default: structured
+                    - structured
+                  default: binary
                 auth:
                   description: 'Auth configurations'
                   type: object
diff --git a/control-plane/pkg/apis/eventing/v1alpha1/kafka_sink_defaults.go b/control-plane/pkg/apis/eventing/v1alpha1/kafka_sink_defaults.go
index 0e9edd5a5..ec1f1862b 100644
--- a/control-plane/pkg/apis/eventing/v1alpha1/kafka_sink_defaults.go
+++ b/control-plane/pkg/apis/eventing/v1alpha1/kafka_sink_defaults.go
@@ -25,7 +25,7 @@ func (ks *KafkaSink) SetDefaults(ctx context.Context) {
 
 // SetDefaults sets KafkaSinkSpec defaults.
 func (kss *KafkaSinkSpec) SetDefaults(ctx context.Context) {
-	defaultMode := ModeStructured
+	defaultMode := ModeBinary
 
 	if kss.ContentMode == nil || *kss.ContentMode == "" {
 		kss.ContentMode = &defaultMode
diff --git a/control-plane/pkg/reconciler/sink/kafka_sink.go b/control-plane/pkg/reconciler/sink/kafka_sink.go
index 2d68595ab..2235a2b4f 100644
--- a/control-plane/pkg/reconciler/sink/kafka_sink.go
+++ b/control-plane/pkg/reconciler/sink/kafka_sink.go
@@ -172,8 +172,7 @@ func (r *Reconciler) reconcileKind(ctx context.Context, ks *eventing.KafkaSink)
 		Uid:    string(ks.UID),
 		Topics: []string{ks.Spec.Topic},
 		Ingress: &contract.Ingress{
-			ContentMode: coreconfig.ContentModeFromString(*ks.Spec.ContentMode),
-			Path:        receiver.PathFromObject(ks),
+			Path: receiver.PathFromObject(ks),
 		},
 		BootstrapServers: kafka.BootstrapServersCommaSeparated(ks.Spec.BootstrapServers),
 		Reference: &contract.Reference{


Test patch:
diff --git a/control-plane/pkg/apis/eventing/v1alpha1/kafka_sink_defaults_test.go b/control-plane/pkg/apis/eventing/v1alpha1/kafka_sink_defaults_test.go
index 2a331a678..87898e306 100644
--- a/control-plane/pkg/apis/eventing/v1alpha1/kafka_sink_defaults_test.go
+++ b/control-plane/pkg/apis/eventing/v1alpha1/kafka_sink_defaults_test.go
@@ -25,7 +25,7 @@ import (
 
 func TestKafkaSinkSpecSetDefaults(t *testing.T) {
 
-	defaultMode := ModeStructured
+	defaultMode := ModeBinary
 
 	tests := []struct {
 		name string
@@ -33,7 +33,7 @@ func TestKafkaSinkSpecSetDefaults(t *testing.T) {
 		want *KafkaSinkSpec
 	}{
 		{
-			name: "structured content mode by default",
+			name: "binary content mode by default",
 			spec: &KafkaSinkSpec{},
 			want: &KafkaSinkSpec{
 				ContentMode: &defaultMode,
diff --git a/control-plane/pkg/reconciler/sink/kafka_sink_test.go b/control-plane/pkg/reconciler/sink/kafka_sink_test.go
index 6dd65342c..e51b5edf6 100644
--- a/control-plane/pkg/reconciler/sink/kafka_sink_test.go
+++ b/control-plane/pkg/reconciler/sink/kafka_sink_test.go
@@ -131,7 +131,7 @@ func sinkReconciliation(t *testing.T, format string, env config.Env) {
 						{
 							Uid:              SinkUUID,
 							Topics:           []string{SinkTopic()},
-							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_STRUCTURED, Path: receiver.Path(SinkNamespace, SinkName)},
+							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_BINARY, Path: receiver.Path(SinkNamespace, SinkName)},
 							BootstrapServers: bootstrapServers,
 							Reference:        SinkReference(),
 						},
@@ -188,7 +188,7 @@ func sinkReconciliation(t *testing.T, format string, env config.Env) {
 							Topics:           []string{SinkTopic()},
 							BootstrapServers: bootstrapServers,
 							Reference:        SinkReference(),
-							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_STRUCTURED, Path: receiver.Path(SinkNamespace, SinkName)},
+							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_BINARY, Path: receiver.Path(SinkNamespace, SinkName)},
 							Auth: &contract.Resource_AuthSecret{
 								AuthSecret: &contract.Reference{
 									Uuid:      SecretUUID,
@@ -256,7 +256,7 @@ func sinkReconciliation(t *testing.T, format string, env config.Env) {
 							BootstrapServers: bootstrapServers,
 							Reference:        SinkReference(),
 							Ingress: &contract.Ingress{
-								ContentMode: contract.ContentMode_STRUCTURED,
+								ContentMode: contract.ContentMode_BINARY,
 								Path:        receiver.Path(SinkNamespace, SinkName),
 							},
 						},
@@ -369,7 +369,7 @@ func sinkReconciliation(t *testing.T, format string, env config.Env) {
 						{
 							Uid:              SinkUUID,
 							Topics:           []string{"my-topic-1"},
-							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_STRUCTURED, Path: receiver.Path(SinkNamespace, SinkName)},
+							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_BINARY, Path: receiver.Path(SinkNamespace, SinkName)},
 							BootstrapServers: "kafka-broker:10000",
 							Reference:        SinkReference(),
 						},
@@ -476,7 +476,7 @@ func sinkReconciliation(t *testing.T, format string, env config.Env) {
 						{
 							Uid:              SinkUUID,
 							Topics:           []string{SinkTopic()},
-							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_STRUCTURED, Path: receiver.Path(SinkNamespace, SinkName)},
+							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_BINARY, Path: receiver.Path(SinkNamespace, SinkName)},
 							BootstrapServers: bootstrapServers,
 							Reference:        SinkReference(),
 						},
@@ -526,7 +526,7 @@ func sinkReconciliation(t *testing.T, format string, env config.Env) {
 						{
 							Uid:              SinkUUID,
 							Topics:           []string{SinkTopic()},
-							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_STRUCTURED, Path: receiver.Path(SinkNamespace, SinkName)},
+							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_BINARY, Path: receiver.Path(SinkNamespace, SinkName)},
 							BootstrapServers: bootstrapServers,
 							Reference:        SinkReference(),
 						},
@@ -599,7 +599,7 @@ func sinkReconciliation(t *testing.T, format string, env config.Env) {
 						{
 							Uid:              SinkUUID,
 							Topics:           []string{SinkTopic()},
-							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_STRUCTURED, Path: receiver.Path(SinkNamespace, SinkName)},
+							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_BINARY, Path: receiver.Path(SinkNamespace, SinkName)},
 							BootstrapServers: bootstrapServers,
 							Reference:        SinkReference(),
 						},
@@ -648,7 +648,7 @@ func sinkReconciliation(t *testing.T, format string, env config.Env) {
 							Topics:           []string{SinkTopic()},
 							BootstrapServers: bootstrapServers,
 							Reference:        SinkReference(),
-							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_STRUCTURED},
+							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_BINARY},
 						},
 					},
 				}, &env),
@@ -671,7 +671,7 @@ func sinkReconciliation(t *testing.T, format string, env config.Env) {
 						{
 							Uid:              SinkUUID,
 							Topics:           []string{SinkTopic()},
-							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_STRUCTURED, Path: receiver.Path(SinkNamespace, SinkName)},
+							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_BINARY, Path: receiver.Path(SinkNamespace, SinkName)},
 							BootstrapServers: bootstrapServers,
 							Reference:        SinkReference(),
 						},
@@ -749,7 +749,7 @@ func sinkReconciliation(t *testing.T, format string, env config.Env) {
 						{
 							Uid:              SinkUUID,
 							Topics:           []string{SinkTopic()},
-							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_STRUCTURED, Path: receiver.Path(SinkNamespace, SinkName)},
+							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_BINARY, Path: receiver.Path(SinkNamespace, SinkName)},
 							BootstrapServers: bootstrapServers,
 							Reference:        SinkReference(),
 						},
@@ -804,7 +804,7 @@ func sinkReconciliation(t *testing.T, format string, env config.Env) {
 						{
 							Uid:              SinkUUID,
 							Topics:           []string{SinkTopic()},
-							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_STRUCTURED, Path: receiver.Path(SinkNamespace, SinkName)},
+							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_BINARY, Path: receiver.Path(SinkNamespace, SinkName)},
 							BootstrapServers: bootstrapServers,
 							Reference:        SinkReference(),
 						},
@@ -864,7 +864,7 @@ func sinkReconciliation(t *testing.T, format string, env config.Env) {
 						{
 							Uid:              SinkUUID,
 							Topics:           []string{SinkTopic()},
-							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_STRUCTURED, Path: receiver.Path(SinkNamespace, SinkName)},
+							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_BINARY, Path: receiver.Path(SinkNamespace, SinkName)},
 							BootstrapServers: bootstrapServers,
 							Reference:        SinkReference(),
 						},
@@ -918,7 +918,7 @@ func sinkReconciliation(t *testing.T, format string, env config.Env) {
 						{
 							Uid:              SinkUUID,
 							Topics:           []string{SinkTopic()},
-							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_STRUCTURED, Path: receiver.Path(SinkNamespace, SinkName)},
+							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_BINARY, Path: receiver.Path(SinkNamespace, SinkName)},
 							BootstrapServers: bootstrapServers,
 							Reference:        SinkReference(),
 						},
@@ -991,7 +991,7 @@ func sinkFinalization(t *testing.T, format string, env config.Env) {
 							Topics: []string{"topic"},
 							Ingress: &contract.Ingress{
 								Path:        "path",
-								ContentMode: contract.ContentMode_STRUCTURED,
+								ContentMode: contract.ContentMode_BINARY,
 							},
 							BootstrapServers: bootstrapServers,
 						},
@@ -1000,7 +1000,7 @@ func sinkFinalization(t *testing.T, format string, env config.Env) {
 							Topics: []string{"topic"},
 							Ingress: &contract.Ingress{
 								Path:        "path",
-								ContentMode: contract.ContentMode_STRUCTURED,
+								ContentMode: contract.ContentMode_BINARY,
 							},
 							BootstrapServers: bootstrapServers,
 						},
@@ -1017,7 +1017,7 @@ func sinkFinalization(t *testing.T, format string, env config.Env) {
 							Topics: []string{"topic"},
 							Ingress: &contract.Ingress{
 								Path:        "path",
-								ContentMode: contract.ContentMode_STRUCTURED,
+								ContentMode: contract.ContentMode_BINARY,
 							},
 							BootstrapServers: bootstrapServers,
 						},
@@ -1042,7 +1042,7 @@ func sinkFinalization(t *testing.T, format string, env config.Env) {
 							Topics: []string{"topic"},
 							Ingress: &contract.Ingress{
 								Path:        "path",
-								ContentMode: contract.ContentMode_STRUCTURED,
+								ContentMode: contract.ContentMode_BINARY,
 							},
 							BootstrapServers: bootstrapServers,
 						},
@@ -1051,7 +1051,7 @@ func sinkFinalization(t *testing.T, format string, env config.Env) {
 							Topics: []string{"topic-2"},
 							Ingress: &contract.Ingress{
 								Path:        "path",
-								ContentMode: contract.ContentMode_STRUCTURED,
+								ContentMode: contract.ContentMode_BINARY,
 							},
 							BootstrapServers: bootstrapServers,
 						},
@@ -1068,7 +1068,7 @@ func sinkFinalization(t *testing.T, format string, env config.Env) {
 							Topics: []string{"topic"},
 							Ingress: &contract.Ingress{
 								Path:        "path",
-								ContentMode: contract.ContentMode_STRUCTURED,
+								ContentMode: contract.ContentMode_BINARY,
 							},
 							BootstrapServers: bootstrapServers,
 						},
@@ -1093,7 +1093,7 @@ func sinkFinalization(t *testing.T, format string, env config.Env) {
 							Topics: []string{"topic"},
 							Ingress: &contract.Ingress{
 								Path:        "path",
-								ContentMode: contract.ContentMode_STRUCTURED,
+								ContentMode: contract.ContentMode_BINARY,
 							},
 							BootstrapServers: bootstrapServers,
 						},
@@ -1102,7 +1102,7 @@ func sinkFinalization(t *testing.T, format string, env config.Env) {
 							Topics: []string{"topic-2"},
 							Ingress: &contract.Ingress{
 								Path:        "path",
-								ContentMode: contract.ContentMode_STRUCTURED,
+								ContentMode: contract.ContentMode_BINARY,
 							},
 							BootstrapServers: bootstrapServers,
 						},
@@ -1119,7 +1119,7 @@ func sinkFinalization(t *testing.T, format string, env config.Env) {
 							Topics: []string{"topic"},
 							Ingress: &contract.Ingress{
 								Path:        "path",
-								ContentMode: contract.ContentMode_STRUCTURED,
+								ContentMode: contract.ContentMode_BINARY,
 							},
 							BootstrapServers: bootstrapServers,
 						},
@@ -1144,7 +1144,7 @@ func sinkFinalization(t *testing.T, format string, env config.Env) {
 							Topics: []string{"topic"},
 							Ingress: &contract.Ingress{
 								Path:        "path",
-								ContentMode: contract.ContentMode_STRUCTURED,
+								ContentMode: contract.ContentMode_BINARY,
 							},
 							BootstrapServers: bootstrapServers,
 						},
@@ -1153,7 +1153,7 @@ func sinkFinalization(t *testing.T, format string, env config.Env) {
 							Topics: []string{"topic-2"},
 							Ingress: &contract.Ingress{
 								Path:        "path",
-								ContentMode: contract.ContentMode_STRUCTURED,
+								ContentMode: contract.ContentMode_BINARY,
 							},
 							BootstrapServers: bootstrapServers,
 						},
@@ -1170,7 +1170,7 @@ func sinkFinalization(t *testing.T, format string, env config.Env) {
 							Topics: []string{"topic"},
 							Ingress: &contract.Ingress{
 								Path:        "path",
-								ContentMode: contract.ContentMode_STRUCTURED,
+								ContentMode: contract.ContentMode_BINARY,
 							},
 							BootstrapServers: bootstrapServers,
 						},
@@ -1200,7 +1200,7 @@ func sinkFinalization(t *testing.T, format string, env config.Env) {
 							Topics: []string{"topic"},
 							Ingress: &contract.Ingress{
 								Path:        "path",
-								ContentMode: contract.ContentMode_STRUCTURED,
+								ContentMode: contract.ContentMode_BINARY,
 							},
 							BootstrapServers: bootstrapServers,
 						},
@@ -1209,7 +1209,7 @@ func sinkFinalization(t *testing.T, format string, env config.Env) {
 							Topics: []string{"topic-2"},
 							Ingress: &contract.Ingress{
 								Path:        "path",
-								ContentMode: contract.ContentMode_STRUCTURED,
+								ContentMode: contract.ContentMode_BINARY,
 							},
 							BootstrapServers: bootstrapServers,
 						},
@@ -1235,7 +1235,7 @@ func sinkFinalization(t *testing.T, format string, env config.Env) {
 							Topics: []string{"topic"},
 							Ingress: &contract.Ingress{
 								Path:        "path",
-								ContentMode: contract.ContentMode_STRUCTURED,
+								ContentMode: contract.ContentMode_BINARY,
 							},
 							BootstrapServers: bootstrapServers,
 						},
diff --git a/test/e2e_new/resources/kafkasink/kafkasink_test.go b/test/e2e_new/resources/kafkasink/kafkasink_test.go
index 2cf75fabf..33d372589 100644
--- a/test/e2e_new/resources/kafkasink/kafkasink_test.go
+++ b/test/e2e_new/resources/kafkasink/kafkasink_test.go
@@ -67,7 +67,7 @@ func Example_full() {
 		"topic":            "my-topic",
 		"bootstrapServers": []string{"my-bootstrap-server:8082"},
 	}
-	kafkasink.WithContentMode(eventingv1alpha1.ModeBinary)(cfg)
+	kafkasink.WithContentMode(eventingv1alpha1.ModeStructured)(cfg)
 	kafkasink.WithReplicationFactor(3)(cfg)
 	kafkasink.WithNumPartitions(10)(cfg)
 	kafkasink.WithAuthSecretName("abc")(cfg)
@@ -88,7 +88,7 @@ func Example_full() {
 	//   topic: "my-topic"
 	//   bootstrapServers:
 	//     - "my-bootstrap-server:8082"
-	//   contentMode: binary
+	//   contentMode: structured
 	//   numPartitions: 10
 	//   replicationFactor: 3
 	//   auth: