Repository: knative-sandbox/eventing-kafka-broker
Pull request #2279 (2022-06-07 10:20:40)

[release-1.4] Avoid waiting for `Consumer.close` when removing resources
This is an automated cherry-pick of #2277

```release-note

```

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/2279#" title="Author self-approved">knative-prow-robot</a>*, *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/2279#pullrequestreview-997894796" title="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/release-1.4/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":[]} -->
knative-prow[bot]: @knative-prow-robot: 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_release-1.4 | 2eb1c41ebfc0cad50dbe7fafafd075eb793a90c7 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2279/unit-tests_eventing-kafka-broker_release-1.4/1534118142941335552) | true | `/test unit-tests_eventing-kafka-broker_release-1.4`

[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 -->
pierDipi: /retest
knative-prow[bot]: @knative-prow-robot: 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_release-1.4 | 2eb1c41ebfc0cad50dbe7fafafd075eb793a90c7 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2279/unit-tests_eventing-kafka-broker_release-1.4/1534119995003375616) | true | `/test unit-tests_eventing-kafka-broker_release-1.4`

[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 -->
codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2279?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) Report
> Merging [#2279](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2279?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (2eb1c41) into [release-1.4](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/commit/1cf5da62a6798f1670b7458c3fdb668bfd4dcd37?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (1cf5da6) will **increase** coverage by `21.86%`.
> The diff coverage is `100.00%`.

```diff
@@                Coverage Diff                 @@
##             release-1.4    #2279       +/-   ##
==================================================
+ Coverage          60.50%   82.36%   +21.86%     
- Complexity             0      696      +696     
==================================================
  Files                 71       72        +1     
  Lines               6937     2359     -4578     
  Branches               0      202      +202     
==================================================
- Hits                4197     1943     -2254     
+ Misses              2440      306     -2134     
+ Partials             300      110      -190     
```

| Flag | Coverage Δ | |
|---|---|---|
| java-unittests | `82.36% <100.00%> (?)` | |

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/2279?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | Coverage Δ | |
|---|---|---|
| [...dispatcher/impl/consumer/BaseConsumerVerticle.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2279/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9jb25zdW1lci9CYXNlQ29uc3VtZXJWZXJ0aWNsZS5qYXZh) | `93.75% <100.00%> (ø)` | |
| [control-plane/pkg/prober/async\_prober.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2279/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcHJvYmVyL2FzeW5jX3Byb2Jlci5nbw==) | | |
| [control-plane/pkg/security/secret.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2279/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvc2VjdXJpdHkvc2VjcmV0Lmdv) | | |
| [...rnals/kafka/eventing/v1alpha1/consumer\_defaults.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2279/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvYXBpcy9pbnRlcm5hbHMva2Fma2EvZXZlbnRpbmcvdjFhbHBoYTEvY29uc3VtZXJfZGVmYXVsdHMuZ28=) | | |
| [control-plane/pkg/reconciler/broker/controller.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2279/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9icm9rZXIvY29udHJvbGxlci5nbw==) | | |
| [...ol-plane/pkg/reconciler/channel/v2/controllerv2.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2279/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9jaGFubmVsL3YyL2NvbnRyb2xsZXJ2Mi5nbw==) | | |
| [control-plane/pkg/reconciler/base/reconciler.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2279/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9iYXNlL3JlY29uY2lsZXIuZ28=) | | |
| [control-plane/pkg/kafka/broker\_class.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2279/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cva2Fma2EvYnJva2VyX2NsYXNzLmdv) | | |
| [...ntrol-plane/pkg/reconciler/consumergroup/filter.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2279/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9jb25zdW1lcmdyb3VwL2ZpbHRlci5nbw==) | | |
| [...rol-plane/pkg/reconciler/source/v2/controllerv2.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2279/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9zb3VyY2UvdjIvY29udHJvbGxlcnYyLmdv) | | |
| ... and [134 more](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2279/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/2279?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/2279?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Last update [1cf5da6...2eb1c41](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2279?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).

knative-prow[bot]: @knative-prow-robot: 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
--- | --- | --- | --- | ---
integration-tests_eventing-kafka-broker_release-1.4 | 2eb1c41ebfc0cad50dbe7fafafd075eb793a90c7 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2279/integration-tests_eventing-kafka-broker_release-1.4/1534118142991667200) | true | `/test integration-tests_eventing-kafka-broker_release-1.4`

[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: /retest
matzew: good old `TestKafkaSourceUpdate` flake :cry: 

matzew: /retest
/retest
/retest
/retest
/retest

knative-prow[bot]: @knative-prow-robot: 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
--- | --- | --- | --- | ---
integration-tests_eventing-kafka-broker_release-1.4 | 2eb1c41ebfc0cad50dbe7fafafd075eb793a90c7 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2279/integration-tests_eventing-kafka-broker_release-1.4/1534154231752167424) | true | `/test integration-tests_eventing-kafka-broker_release-1.4`

[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: /retest
knative-prow[bot]: @knative-prow-robot: 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
--- | --- | --- | --- | ---
integration-tests_eventing-kafka-broker_release-1.4 | 2eb1c41ebfc0cad50dbe7fafafd075eb793a90c7 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2279/integration-tests_eventing-kafka-broker_release-1.4/1534184655845068800) | true | `/test integration-tests_eventing-kafka-broker_release-1.4`

[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 integration-tests_eventing-kafka-broker_release-1.4


knative-prow[bot]: @knative-prow-robot: 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
--- | --- | --- | --- | ---
integration-tests_eventing-kafka-broker_release-1.4 | 2eb1c41ebfc0cad50dbe7fafafd075eb793a90c7 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2279/integration-tests_eventing-kafka-broker_release-1.4/1534263618999160832) | true | `/test integration-tests_eventing-kafka-broker_release-1.4`

[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 -->

Patch:
diff --git a/control-plane/pkg/prober/async_prober.go b/control-plane/pkg/prober/async_prober.go
index c4f211c09..3caa1c6cd 100644
--- a/control-plane/pkg/prober/async_prober.go
+++ b/control-plane/pkg/prober/async_prober.go
@@ -26,7 +26,7 @@ import (
 	"knative.dev/pkg/logging"
 )
 
-type IPsLister func() ([]string, error)
+type IPsLister func(addressable Addressable) ([]string, error)
 
 type asyncProber struct {
 	client    httpClient
@@ -44,8 +44,8 @@ func NewAsync(ctx context.Context, client httpClient, port string, IPsLister IPs
 	logger := logging.FromContext(ctx).Desugar().
 		With(zap.String("scope", "prober"))
 
-	if len(port) == 0 {
-		logger.Fatal("Port is required")
+	if len(port) > 0 && port[0] != ':' {
+		port = ":" + port
 	}
 	return &asyncProber{
 		client:    client,
@@ -59,7 +59,7 @@ func NewAsync(ctx context.Context, client httpClient, port string, IPsLister IPs
 
 func (a *asyncProber) Probe(ctx context.Context, addressable Addressable, expected Status) Status {
 	address := addressable.Address
-	IPs, err := a.IPsLister()
+	IPs, err := a.IPsLister(addressable)
 	if err != nil {
 		a.logger.Error("Failed to list IPs", zap.Error(err))
 		return StatusUnknown
@@ -88,7 +88,7 @@ func (a *asyncProber) Probe(ctx context.Context, addressable Addressable, expect
 
 	for _, IP := range IPs {
 		podUrl := *address
-		podUrl.Host = IP + ":" + a.port
+		podUrl.Host = IP + a.port
 		address := podUrl.String()
 
 		logger := a.logger.
diff --git a/control-plane/pkg/prober/prober.go b/control-plane/pkg/prober/prober.go
index b0cf2f7d8..fa0f514d1 100644
--- a/control-plane/pkg/prober/prober.go
+++ b/control-plane/pkg/prober/prober.go
@@ -92,7 +92,13 @@ func probe(ctx context.Context, client httpClient, logger *zap.Logger, address s
 }
 
 func IPsListerFromService(svc types.NamespacedName) IPsLister {
-	return func() ([]string, error) {
+	return func(addressable Addressable) ([]string, error) {
 		return []string{fmt.Sprintf("%s.%s.svc", svc.Name, svc.Namespace)}, nil
 	}
 }
+
+func IdentityIPsLister() IPsLister {
+	return func(addressable Addressable) ([]string, error) {
+		return []string{addressable.Address.Host}, nil
+	}
+}
diff --git a/control-plane/pkg/receiver/path.go b/control-plane/pkg/receiver/routing.go
similarity index 74%
rename from control-plane/pkg/receiver/path.go
rename to control-plane/pkg/receiver/routing.go
index 3139b1c26..20f0473f3 100644
--- a/control-plane/pkg/receiver/path.go
+++ b/control-plane/pkg/receiver/routing.go
@@ -20,6 +20,8 @@ import (
 	"fmt"
 
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"knative.dev/eventing-kafka/pkg/channel/consolidated/reconciler/controller/resources"
+	"knative.dev/pkg/network"
 )
 
 // PathFromObject returns an HTTP request path given a generic object.
@@ -31,3 +33,8 @@ func PathFromObject(obj metav1.Object) string {
 func Path(namespace, name string) string {
 	return fmt.Sprintf("/%s/%s", namespace, name)
 }
+
+// Host returns an HTTP host given namespace and name of an object.
+func Host(namespace, name string) string {
+	return fmt.Sprintf("%s.%s.svc.%s", resources.MakeChannelServiceName(name), namespace, network.GetClusterDomainName())
+}
diff --git a/control-plane/pkg/receiver/path_test.go b/control-plane/pkg/receiver/routing_test.go
similarity index 100%
rename from control-plane/pkg/receiver/path_test.go
rename to control-plane/pkg/receiver/routing_test.go
diff --git a/control-plane/pkg/reconciler/channel/channel.go b/control-plane/pkg/reconciler/channel/channel.go
index 33104e4d4..ff8431866 100644
--- a/control-plane/pkg/reconciler/channel/channel.go
+++ b/control-plane/pkg/reconciler/channel/channel.go
@@ -19,12 +19,14 @@ package channel
 import (
 	"context"
 	"fmt"
+	"net/url"
 	"strconv"
 	"time"
 
 	"k8s.io/apimachinery/pkg/api/equality"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"knative.dev/pkg/network"
 
 	"github.com/Shopify/sarama"
 
@@ -62,7 +64,7 @@ const (
 	// TopicPrefix is the Kafka Channel topic prefix - (topic name: knative-messaging-kafka.<channel-namespace>.<channel-name>).
 	TopicPrefix                  = "knative-messaging-kafka"
 	DefaultDeliveryOrder         = contract.DeliveryOrder_ORDERED
-	NewChannelIngressServiceName = "kafka-channel-dispatcher"
+	NewChannelIngressServiceName = "kafka-channel-ingress"
 )
 
 type Reconciler struct {
@@ -278,7 +280,13 @@ func (r *Reconciler) reconcileKind(ctx context.Context, channel *messagingv1beta
 		return subscriptionError
 	}
 
-	address := receiver.Address(r.IngressHost, channel)
+	channelService, err := r.reconcileChannelService(ctx, channel)
+	if err != nil {
+		logger.Error("Error reconciling the backwards compatibility channel service.", zap.Error(err))
+		return err
+	}
+
+	address, _ := url.Parse(fmt.Sprintf("http://%s.%s.svc.%s", channelService.Name, channelService.Namespace, network.GetClusterDomainName()))
 	proberAddressable := prober.Addressable{
 		Address: address,
 		ResourceKey: types.NamespacedName{
@@ -295,12 +303,6 @@ func (r *Reconciler) reconcileKind(ctx context.Context, channel *messagingv1beta
 	}
 	statusConditionManager.Addressable(address)
 
-	err = r.reconcileChannelService(ctx, channel)
-	if err != nil {
-		logger.Error("Error reconciling the backwards compatibility channel service.", zap.Error(err))
-		return err
-	}
-
 	return nil
 }
 
@@ -580,7 +582,7 @@ func (r *Reconciler) getChannelContractResource(ctx context.Context, topic strin
 		Uid:    string(channel.UID),
 		Topics: []string{topic},
 		Ingress: &contract.Ingress{
-			Path: receiver.PathFromObject(channel),
+			Host: receiver.Host(channel.GetNamespace(), channel.GetName()),
 		},
 		BootstrapServers: config.GetBootstrapServers(),
 		Reference: &contract.Reference{
@@ -605,10 +607,10 @@ func (r *Reconciler) getChannelContractResource(ctx context.Context, topic strin
 	return resource, nil
 }
 
-func (r *Reconciler) reconcileChannelService(ctx context.Context, channel *messagingv1beta1.KafkaChannel) error {
+func (r *Reconciler) reconcileChannelService(ctx context.Context, channel *messagingv1beta1.KafkaChannel) (*corev1.Service, error) {
 	expected, err := resources.MakeK8sService(channel, resources.ExternalService(system.Namespace(), NewChannelIngressServiceName))
 	if err != nil {
-		return fmt.Errorf("failed to create the channel service object: %w", err)
+		return expected, fmt.Errorf("failed to create the channel service object: %w", err)
 	}
 
 	svc, err := r.ServiceLister.Services(channel.Namespace).Get(resources.MakeChannelServiceName(channel.Name))
@@ -616,25 +618,25 @@ func (r *Reconciler) reconcileChannelService(ctx context.Context, channel *messa
 		if apierrors.IsNotFound(err) {
 			_, err = r.KubeClient.CoreV1().Services(channel.Namespace).Create(ctx, expected, metav1.CreateOptions{})
 			if err != nil {
-				return fmt.Errorf("failed to create the channel service object: %w", err)
+				return expected, fmt.Errorf("failed to create the channel service object: %w", err)
 			}
-			return nil
+			return expected, nil
 		}
-		return err
+		return expected, err
 	} else if !equality.Semantic.DeepEqual(svc.Spec, expected.Spec) {
 		svc = svc.DeepCopy()
 		svc.Spec = expected.Spec
 
 		_, err = r.KubeClient.CoreV1().Services(channel.Namespace).Update(ctx, svc, metav1.UpdateOptions{})
 		if err != nil {
-			return fmt.Errorf("failed to update the channel service: %w", err)
+			return expected, fmt.Errorf("failed to update the channel service: %w", err)
 		}
 	}
 	// Check to make sure that the KafkaChannel owns this service and if not, complain.
 	if !metav1.IsControlledBy(svc, channel) {
-		return fmt.Errorf("kafkachannel: %s/%s does not own Service: %q", channel.Namespace, channel.Name, svc.Name)
+		return expected, fmt.Errorf("kafkachannel: %s/%s does not own Service: %q", channel.Namespace, channel.Name, svc.Name)
 	}
-	return nil
+	return expected, nil
 }
 
 // consumerGroup returns a consumerGroup name for the given channel and subscription
diff --git a/control-plane/pkg/reconciler/channel/controller.go b/control-plane/pkg/reconciler/channel/controller.go
index c9831742d..c45d597a1 100644
--- a/control-plane/pkg/reconciler/channel/controller.go
+++ b/control-plane/pkg/reconciler/channel/controller.go
@@ -23,7 +23,6 @@ import (
 	"github.com/Shopify/sarama"
 	"go.uber.org/zap"
 	corev1 "k8s.io/api/core/v1"
-	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/tools/cache"
 
 	messagingv1beta "knative.dev/eventing-kafka/pkg/apis/messaging/v1beta1"
@@ -85,8 +84,8 @@ func NewController(ctx context.Context, configs *config.Env) *controller.Impl {
 	}
 
 	impl := kafkachannelreconciler.NewImpl(ctx, reconciler)
-	IPsLister := prober.IPsListerFromService(types.NamespacedName{Namespace: configs.SystemNamespace, Name: configs.IngressName})
-	reconciler.Prober = prober.NewAsync(ctx, http.DefaultClient, configs.IngressPodPort, IPsLister, impl.EnqueueKey)
+	IPsLister := prober.IdentityIPsLister()
+	reconciler.Prober = prober.NewAsync(ctx, http.DefaultClient, "", IPsLister, impl.EnqueueKey)
 	reconciler.IngressHost = network.GetServiceHostname(configs.IngressName, configs.SystemNamespace)
 
 	channelInformer := kafkachannelinformer.Get(ctx)
diff --git a/control-plane/pkg/reconciler/channel/v2/channelv2.go b/control-plane/pkg/reconciler/channel/v2/channelv2.go
index f78b7fe79..b07024c9f 100644
--- a/control-plane/pkg/reconciler/channel/v2/channelv2.go
+++ b/control-plane/pkg/reconciler/channel/v2/channelv2.go
@@ -20,6 +20,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"net/url"
 	"strconv"
 	"strings"
 	"time"
@@ -34,6 +35,8 @@ import (
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/types"
 	corelisters "k8s.io/client-go/listers/core/v1"
+	"knative.dev/eventing-kafka/pkg/channel/consolidated/reconciler/controller/resources"
+	"knative.dev/pkg/network"
 	"knative.dev/pkg/resolver"
 
 	v1 "knative.dev/eventing/pkg/apis/duck/v1"
@@ -53,6 +56,7 @@ import (
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/prober"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/receiver"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/base"
+	channelreconciler "knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/channel"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/security"
 
 	internals "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing"
@@ -96,6 +100,7 @@ type Reconciler struct {
 	NewKafkaClient kafka.NewClientFunc
 
 	ConfigMapLister corelisters.ConfigMapLister
+	ServiceLister   corelisters.ServiceLister
 
 	Prober prober.Prober
 
@@ -277,7 +282,13 @@ func (r *Reconciler) ReconcileKind(ctx context.Context, channel *messagingv1beta
 		logger.Debug("Updated dispatcher pod annotation")
 	}
 
-	address := receiver.Address(r.IngressHost, channel)
+	channelService, err := r.reconcileChannelService(ctx, channel)
+	if err != nil {
+		logger.Error("Error reconciling the backwards compatibility channel service.", zap.Error(err))
+		return err
+	}
+
+	address, _ := url.Parse(fmt.Sprintf("http://%s.%s.svc.%s", channelService.Name, channelService.Namespace, network.GetClusterDomainName()))
 	proberAddressable := prober.Addressable{
 		Address: address,
 		ResourceKey: types.NamespacedName{
@@ -286,7 +297,9 @@ func (r *Reconciler) ReconcileKind(ctx context.Context, channel *messagingv1beta
 		},
 	}
 
+	logger.Debug("Going to probe address to check ingress readiness for the channel.", zap.Any("proberAddressable", proberAddressable))
 	if status := r.Prober.Probe(ctx, proberAddressable, prober.StatusReady); status != prober.StatusReady {
+		logger.Debug("Ingress is not ready for the channel. Going to requeue.", zap.Any("proberAddressable", proberAddressable))
 		statusConditionManager.ProbesStatusNotReady(status)
 		return nil // Object will get re-queued once probe status changes.
 	}
@@ -295,6 +308,38 @@ func (r *Reconciler) ReconcileKind(ctx context.Context, channel *messagingv1beta
 	return nil
 }
 
+func (r *Reconciler) reconcileChannelService(ctx context.Context, channel *messagingv1beta1.KafkaChannel) (*corev1.Service, error) {
+	expected, err := resources.MakeK8sService(channel, resources.ExternalService(r.Env.SystemNamespace, channelreconciler.NewChannelIngressServiceName))
+	if err != nil {
+		return expected, fmt.Errorf("failed to create the channel service object: %w", err)
+	}
+
+	svc, err := r.ServiceLister.Services(channel.Namespace).Get(resources.MakeChannelServiceName(channel.Name))
+	if err != nil {
+		if apierrors.IsNotFound(err) {
+			_, err = r.KubeClient.CoreV1().Services(channel.Namespace).Create(ctx, expected, metav1.CreateOptions{})
+			if err != nil {
+				return expected, fmt.Errorf("failed to create the channel service object: %w", err)
+			}
+			return expected, nil
+		}
+		return expected, err
+	} else if !equality.Semantic.DeepEqual(svc.Spec, expected.Spec) {
+		svc = svc.DeepCopy()
+		svc.Spec = expected.Spec
+
+		_, err = r.KubeClient.CoreV1().Services(channel.Namespace).Update(ctx, svc, metav1.UpdateOptions{})
+		if err != nil {
+			return expected, fmt.Errorf("failed to update the channel service: %w", err)
+		}
+	}
+	// Check to make sure that the KafkaChannel owns this service and if not, complain.
+	if !metav1.IsControlledBy(svc, channel) {
+		return expected, fmt.Errorf("kafkachannel: %s/%s does not own Service: %q", channel.Namespace, channel.Name, svc.Name)
+	}
+	return expected, nil
+}
+
 func (r *Reconciler) FinalizeKind(ctx context.Context, channel *messagingv1beta1.KafkaChannel) reconciler.Event {
 	logger := kafkalogging.CreateFinalizeMethodLogger(ctx, channel)
 
@@ -571,7 +616,7 @@ func (r *Reconciler) getChannelContractResource(ctx context.Context, topic strin
 		Uid:    string(channel.UID),
 		Topics: []string{topic},
 		Ingress: &contract.Ingress{
-			Path: receiver.PathFromObject(channel),
+			Host: receiver.Host(channel.GetNamespace(), channel.GetName()),
 		},
 		BootstrapServers: config.GetBootstrapServers(),
 		Reference: &contract.Reference{
diff --git a/control-plane/pkg/reconciler/channel/v2/controllerv2.go b/control-plane/pkg/reconciler/channel/v2/controllerv2.go
index 2454e5e8f..671dffa3c 100644
--- a/control-plane/pkg/reconciler/channel/v2/controllerv2.go
+++ b/control-plane/pkg/reconciler/channel/v2/controllerv2.go
@@ -22,8 +22,8 @@ import (
 
 	"github.com/Shopify/sarama"
 	corev1 "k8s.io/api/core/v1"
-	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/tools/cache"
+	serviceinformer "knative.dev/pkg/client/injection/kube/informers/core/v1/service"
 	"knative.dev/pkg/resolver"
 
 	messagingv1beta "knative.dev/eventing-kafka/pkg/apis/messaging/v1beta1"
@@ -69,13 +69,14 @@ func NewController(ctx context.Context, configs *config.Env) *controller.Impl {
 		NewKafkaClusterAdminClient: sarama.NewClusterAdmin,
 		Env:                        configs,
 		ConfigMapLister:            configmapInformer.Lister(),
+		ServiceLister:              serviceinformer.Get(ctx).Lister(),
 		ConsumerGroupLister:        consumerGroupInformer.Lister(),
 		InternalsClient:            consumergroupclient.Get(ctx),
 	}
 
 	impl := kafkachannelreconciler.NewImpl(ctx, reconciler)
-	IPsLister := prober.IPsListerFromService(types.NamespacedName{Namespace: configs.SystemNamespace, Name: configs.IngressName})
-	reconciler.Prober = prober.NewAsync(ctx, http.DefaultClient, configs.IngressPodPort, IPsLister, impl.EnqueueKey)
+	IPsLister := prober.IdentityIPsLister()
+	reconciler.Prober = prober.NewAsync(ctx, http.DefaultClient, "", IPsLister, impl.EnqueueKey)
 	reconciler.IngressHost = network.GetServiceHostname(configs.IngressName, configs.SystemNamespace)
 	reconciler.Resolver = resolver.NewURIResolverFromTracker(ctx, impl.Tracker)
 
diff --git a/data-plane/config/broker/100-config-kafka-broker-data-plane.yaml b/data-plane/config/broker/100-config-kafka-broker-data-plane.yaml
index 0cee845f3..5d8189997 100644
--- a/data-plane/config/broker/100-config-kafka-broker-data-plane.yaml
+++ b/data-plane/config/broker/100-config-kafka-broker-data-plane.yaml
@@ -38,7 +38,7 @@ data:
     max.request.size=1048576
     partitioner.class=org.apache.kafka.clients.producer.internals.DefaultPartitioner
     receive.buffer.bytes=-1
-    request.timeout.ms=30000
+    request.timeout.ms=2000
     enable.idempotence=false
     max.in.flight.requests.per.connection=5
     metadata.max.age.ms=300000
@@ -67,7 +67,7 @@ data:
     auto.offset.reset=earliest
     client.dns.lookup=use_all_dns_ips
     connections.max.idle.ms=540000
-    default.api.timeout.ms=60000
+    default.api.timeout.ms=2000
     enable.auto.commit=false
     exclude.internal.topics=true
     fetch.max.bytes=52428800
@@ -76,7 +76,7 @@ data:
     max.poll.records=500
     # partition.assignment.strategy=org.apache.kafka.clients.consumer.CooperativeStickyAssignor
     receive.buffer.bytes=65536
-    request.timeout.ms=30000
+    request.timeout.ms=2000
     # sasl.client.callback.handler.class=
     # sasl.jaas.config=
     # sasl.kerberos.service.name=
diff --git a/data-plane/config/channel/100-config-kafka-channel-data-plane.yaml b/data-plane/config/channel/100-config-kafka-channel-data-plane.yaml
index f691d02fb..083832fd8 100644
--- a/data-plane/config/channel/100-config-kafka-channel-data-plane.yaml
+++ b/data-plane/config/channel/100-config-kafka-channel-data-plane.yaml
@@ -38,7 +38,7 @@ data:
     max.request.size=1048576
     partitioner.class=org.apache.kafka.clients.producer.internals.DefaultPartitioner
     receive.buffer.bytes=-1
-    request.timeout.ms=30000
+    request.timeout.ms=2000
     enable.idempotence=false
     max.in.flight.requests.per.connection=5
     metadata.max.age.ms=300000
@@ -66,7 +66,7 @@ data:
     allow.auto.create.topics=true
     client.dns.lookup=use_all_dns_ips
     connections.max.idle.ms=540000
-    default.api.timeout.ms=60000
+    default.api.timeout.ms=2000
     enable.auto.commit=false
     exclude.internal.topics=true
     fetch.max.bytes=52428800
@@ -75,7 +75,7 @@ data:
     max.poll.records=500
     # partition.assignment.strategy=org.apache.kafka.clients.consumer.CooperativeStickyAssignor
     receive.buffer.bytes=65536
-    request.timeout.ms=30000
+    request.timeout.ms=2000
     # sasl.client.callback.handler.class=
     # sasl.jaas.config=
     # sasl.kerberos.service.name=
diff --git a/data-plane/config/sink/100-config-kafka-sink-data-plane.yaml b/data-plane/config/sink/100-config-kafka-sink-data-plane.yaml
index b5f42bd3c..0869932aa 100644
--- a/data-plane/config/sink/100-config-kafka-sink-data-plane.yaml
+++ b/data-plane/config/sink/100-config-kafka-sink-data-plane.yaml
@@ -38,7 +38,7 @@ data:
     max.request.size=1048576
     partitioner.class=org.apache.kafka.clients.producer.internals.DefaultPartitioner
     receive.buffer.bytes=-1
-    request.timeout.ms=30000
+    request.timeout.ms=2000
     enable.idempotence=false
     max.in.flight.requests.per.connection=5
     metadata.max.age.ms=300000
diff --git a/data-plane/config/source/100-config-kafka-source-data-plane.yaml b/data-plane/config/source/100-config-kafka-source-data-plane.yaml
index 1670b34b5..c3ddd9ad1 100644
--- a/data-plane/config/source/100-config-kafka-source-data-plane.yaml
+++ b/data-plane/config/source/100-config-kafka-source-data-plane.yaml
@@ -38,7 +38,7 @@ data:
     max.request.size=1048576
     partitioner.class=org.apache.kafka.clients.producer.internals.DefaultPartitioner
     receive.buffer.bytes=-1
-    request.timeout.ms=30000
+    request.timeout.ms=2000
     enable.idempotence=false
     max.in.flight.requests.per.connection=5
     metadata.max.age.ms=300000
@@ -69,7 +69,7 @@ data:
     auto.offset.reset=earliest
     client.dns.lookup=use_all_dns_ips
     connections.max.idle.ms=540000
-    default.api.timeout.ms=60000
+    default.api.timeout.ms=2000
     enable.auto.commit=false
     exclude.internal.topics=true
     fetch.max.bytes=52428800
@@ -78,7 +78,7 @@ data:
     max.poll.records=500
     # partition.assignment.strategy=org.apache.kafka.clients.consumer.CooperativeStickyAssignor
     receive.buffer.bytes=65536
-    request.timeout.ms=30000
+    request.timeout.ms=2000
     # sasl.client.callback.handler.class=
     # sasl.jaas.config=
     # sasl.kerberos.service.name=
diff --git a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/BaseConsumerVerticle.java b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/BaseConsumerVerticle.java
index 27668d56c..6041f3def 100644
--- a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/BaseConsumerVerticle.java
+++ b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/BaseConsumerVerticle.java
@@ -74,8 +74,13 @@ public abstract class BaseConsumerVerticle extends AbstractVerticle {
     logger.info("Stopping consumer");
 
     AsyncCloseable
-      .compose(this.recordDispatcher, this.closeable, this.consumer::close)
+      .compose(this.recordDispatcher, this.closeable)
       .close(stopPromise);
+
+    stopPromise.future()
+      .onComplete(v -> this.consumer.close()
+        .onFailure(cause -> logger.error("Failed to close consumer {}", keyValue("topics", topics), cause))
+      );
   }
 
   public void setConsumer(KafkaConsumer<Object, CloudEvent> consumer) {
diff --git a/data-plane/receiver/src/main/java/dev/knative/eventing/kafka/broker/receiver/impl/IngressProducerReconcilableStore.java b/data-plane/receiver/src/main/java/dev/knative/eventing/kafka/broker/receiver/impl/IngressProducerReconcilableStore.java
index 5e5146c2b..25ce687ea 100644
--- a/data-plane/receiver/src/main/java/dev/knative/eventing/kafka/broker/receiver/impl/IngressProducerReconcilableStore.java
+++ b/data-plane/receiver/src/main/java/dev/knative/eventing/kafka/broker/receiver/impl/IngressProducerReconcilableStore.java
@@ -89,9 +89,27 @@ public class IngressProducerReconcilableStore implements IngressReconcilerListen
     // That means, we support these modes:
     // - Request coming to "/path" --> path is used for matching
     // - Request coming to "/" --> hostname is used for matching
-    if (pathMapper.get(path) != null) {
-      return pathMapper.get(path);
+    final var p = pathMapper.get(path);
+    if (p != null) {
+      return p;
     }
+
+    // Host based routing is not as simple as path based routing, since the host header
+    // is variable.
+    // For example an external name service might be hit in multiple ways producing
+    // different host headers:
+    // - my-kafka-channel-kn-channel.default
+    // - my-kafka-channel-kn-channel.default:80
+    // - my-kafka-channel-kn-channel.default.svc
+    // - my-kafka-channel-kn-channel.default.svc:80
+    // - my-kafka-channel-kn-channel.default.svc.cluster.local [1]
+    // - my-kafka-channel-kn-channel.default.svc.cluster.local:80
+    //
+    // Current implementations expects only the long form that contains the cluster
+    // domain without the port [1], however, this is clearly a non-robust implementation,
+    // or it might be wrong with some clients, for now, we keep the same behavior as
+    // the old implementations.
+
     return hostMapper.get(host);
   }
 
@@ -137,14 +155,14 @@ public class IngressProducerReconcilableStore implements IngressReconcilerListen
         producerProps
       );
 
-      if(isRootPath(ingress.getPath()) && Strings.isNullOrEmpty(ingress.getHost())){
+      if (isRootPath(ingress.getPath()) && Strings.isNullOrEmpty(ingress.getHost())) {
         throw new IllegalArgumentException("Ingress path and host is blank. One of them should be defined. Resource UID: " + resource.getUid());
       }
 
-      if(!isRootPath(ingress.getPath())){
+      if (!isRootPath(ingress.getPath())) {
         this.pathMapper.put(ingress.getPath(), ingressInfo);
       }
-      if (!Strings.isNullOrEmpty(ingress.getHost())){
+      if (!Strings.isNullOrEmpty(ingress.getHost())) {
         this.hostMapper.put(ingress.getHost(), ingressInfo);
       }
       this.ingressInfos.put(resource.getUid(), ingressInfo);
@@ -183,20 +201,20 @@ public class IngressProducerReconcilableStore implements IngressReconcilerListen
       return rc.getValue().close()
         .onSuccess(r -> {
           // Remove ingress info from the maps
-          if(!isRootPath(ingressInfo.getPath())){
+          if (!isRootPath(ingressInfo.getPath())) {
             this.pathMapper.remove(ingressInfo.getPath());
           }
-          if(!Strings.isNullOrEmpty(ingressInfo.getHost())){
+          if (!Strings.isNullOrEmpty(ingressInfo.getHost())) {
             this.hostMapper.remove(ingressInfo.getHost());
           }
           this.ingressInfos.remove(resource.getUid());
         });
     }
     // Remove ingress info from the maps
-    if(!isRootPath(ingressInfo.getPath())){
+    if (!isRootPath(ingressInfo.getPath())) {
       this.pathMapper.remove(ingressInfo.getPath());
     }
-    if(!Strings.isNullOrEmpty(ingressInfo.getHost())){
+    if (!Strings.isNullOrEmpty(ingressInfo.getHost())) {
       this.hostMapper.remove(ingressInfo.getHost());
     }
     this.ingressInfos.remove(resource.getUid());
diff --git a/data-plane/receiver/src/main/java/dev/knative/eventing/kafka/broker/receiver/impl/ReceiverVerticle.java b/data-plane/receiver/src/main/java/dev/knative/eventing/kafka/broker/receiver/impl/ReceiverVerticle.java
index a8a056e07..f817b775d 100644
--- a/data-plane/receiver/src/main/java/dev/knative/eventing/kafka/broker/receiver/impl/ReceiverVerticle.java
+++ b/data-plane/receiver/src/main/java/dev/knative/eventing/kafka/broker/receiver/impl/ReceiverVerticle.java
@@ -108,9 +108,9 @@ public class ReceiverVerticle extends AbstractVerticle implements Handler<HttpSe
   @Override
   public void stop(Promise<Void> stopPromise) {
     CompositeFuture.all(
-      server.close().mapEmpty(),
-      messageConsumer.unregister()
-    )
+        server.close().mapEmpty(),
+        messageConsumer.unregister()
+      )
       .<Void>mapEmpty()
       .onComplete(stopPromise);
   }
@@ -124,7 +124,11 @@ public class ReceiverVerticle extends AbstractVerticle implements Handler<HttpSe
     IngressProducer producer = this.ingressProducerStore.resolve(request.host(), request.path());
     if (producer == null) {
       request.response().setStatusCode(NOT_FOUND.code()).end();
-      logger.warn("Resource not found {}", keyValue("path", request.path()));
+      logger.warn("Resource not found {} {} {}",
+        keyValue("path", request.path()),
+        keyValue("host", request.host()),
+        keyValue("hostHeader", request.getHeader("Host"))
+      );
       return;
     }
 
diff --git a/go.mod b/go.mod
index 732e0e4f0..a5009940e 100644
--- a/go.mod
+++ b/go.mod
@@ -25,7 +25,7 @@ require (
 	k8s.io/client-go v0.23.5
 	k8s.io/utils v0.0.0-20220210201930-3a6ce19ff2f9
 	knative.dev/eventing v0.31.0
-	knative.dev/eventing-kafka v0.31.0
+	knative.dev/eventing-kafka v0.31.1
 	knative.dev/hack v0.0.0-20220411131823-6ffd8417de7c
 	knative.dev/pkg v0.0.0-20220412134708-e325df66cb51
 	knative.dev/reconciler-test v0.0.0-20220412165608-994f0c3fab62
diff --git a/go.sum b/go.sum
index 76d2e176a..44a3bc29b 100644
--- a/go.sum
+++ b/go.sum
@@ -1395,8 +1395,8 @@ k8s.io/utils v0.0.0-20220210201930-3a6ce19ff2f9/go.mod h1:jPW/WVKK9YHAvNhRxK0md/
 knative.dev/control-protocol v0.0.0-20220412172608-013cfc57d96d/go.mod h1:1lv5PHz2hL0DiqXrlEjTRUhdHAA8F5I7mIL/EsQ1W9A=
 knative.dev/eventing v0.31.0 h1:Bu1cBSwxAT1BsaexQ6PJPWc1mxx2LF/DeszvYEIytJ0=
 knative.dev/eventing v0.31.0/go.mod h1:XgJY27IxyBjmu/mz53cVlz+oMPPzzRaVXlPmWKCqEd8=
-knative.dev/eventing-kafka v0.31.0 h1:1zB6/uGyhvCjmVd0z7Wg2vxQNnhTDd9ItL+QUzr708s=
-knative.dev/eventing-kafka v0.31.0/go.mod h1:RZ/b/xO0sEPL0it4Xm0kSbrhATdsND1WVQq5AcHi7ro=
+knative.dev/eventing-kafka v0.31.1 h1:qEb1zP7hFhg63o7UYiH3Klc+HRUHhDXJxE83aq5fLz0=
+knative.dev/eventing-kafka v0.31.1/go.mod h1:RZ/b/xO0sEPL0it4Xm0kSbrhATdsND1WVQq5AcHi7ro=
 knative.dev/hack v0.0.0-20220411131823-6ffd8417de7c h1:aXsFXeky/GccNQxwf72CS4NR3EoqTqsCVNKQnblfwr0=
 knative.dev/hack v0.0.0-20220411131823-6ffd8417de7c/go.mod h1:PHt8x8yX5Z9pPquBEfIj0X66f8iWkWfR0S/sarACJrI=
 knative.dev/hack/schema v0.0.0-20220411131823-6ffd8417de7c/go.mod h1:ffjwmdcrH5vN3mPhO8RrF2KfNnbHeCE2C60A+2cv3U0=
diff --git a/vendor/knative.dev/eventing-kafka/pkg/channel/consolidated/reconciler/controller/resources/service.go b/vendor/knative.dev/eventing-kafka/pkg/channel/consolidated/reconciler/controller/resources/service.go
index 6dc5f148b..4b9d753bd 100644
--- a/vendor/knative.dev/eventing-kafka/pkg/channel/consolidated/reconciler/controller/resources/service.go
+++ b/vendor/knative.dev/eventing-kafka/pkg/channel/consolidated/reconciler/controller/resources/service.go
@@ -21,9 +21,10 @@ import (
 
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"knative.dev/eventing-kafka/pkg/apis/messaging/v1beta1"
 	"knative.dev/pkg/kmeta"
 	"knative.dev/pkg/network"
+
+	"knative.dev/eventing-kafka/pkg/apis/messaging/v1beta1"
 )
 
 const (
@@ -31,13 +32,21 @@ const (
 	portNumber         = 80
 	MessagingRoleLabel = "messaging.knative.dev/role"
 	MessagingRole      = "kafka-channel"
+
+	ChannelSuffix = "-kn-channel"
+	// MaxResourceNameLength is the maximum number of characters for a Kubernetes resource name.
+	// See vendor/k8s.io/apiserver/pkg/storage/names/generate.go
+	MaxResourceNameLength = 63
 )
 
 // ServiceOption can be used to optionally modify the K8s service in MakeK8sService.
 type ServiceOption func(*corev1.Service) error
 
 func MakeChannelServiceName(name string) string {
-	return fmt.Sprintf("%s-kn-channel", name)
+	if len(name)+len(ChannelSuffix) > MaxResourceNameLength {
+		return name
+	}
+	return fmt.Sprintf("%s%s", name, ChannelSuffix)
 }
 
 // ExternalService is a functional option for MakeK8sService to create a K8s service of type ExternalName
diff --git a/vendor/modules.txt b/vendor/modules.txt
index 5a140aab8..8f09b17cf 100644
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -1299,7 +1299,7 @@ knative.dev/eventing/test/upgrade/prober/wathola/fetcher
 knative.dev/eventing/test/upgrade/prober/wathola/forwarder
 knative.dev/eventing/test/upgrade/prober/wathola/receiver
 knative.dev/eventing/test/upgrade/prober/wathola/sender
-# knative.dev/eventing-kafka v0.31.0
+# knative.dev/eventing-kafka v0.31.1
 ## explicit
 knative.dev/eventing-kafka/pkg/apis/bindings
 knative.dev/eventing-kafka/pkg/apis/bindings/v1beta1


Test patch:
diff --git a/control-plane/pkg/prober/async_prober_test.go b/control-plane/pkg/prober/async_prober_test.go
index 893a3b6c8..7a749f72b 100644
--- a/control-plane/pkg/prober/async_prober_test.go
+++ b/control-plane/pkg/prober/async_prober_test.go
@@ -132,7 +132,7 @@ func TestAsyncProber(t *testing.T) {
 			u, _ := url.Parse(s.URL)
 
 			wantRequeueCountMin := atomic.NewInt64(int64(tc.wantRequeueCountMin))
-			var IPsLister IPsLister = func() ([]string, error) {
+			var IPsLister IPsLister = func(addressable Addressable) ([]string, error) {
 				pods, err := podinformer.Get(ctx).Lister().List(tc.podsLabelsSelector)
 				if err != nil {
 					return nil, err
diff --git a/control-plane/pkg/prober/prober_test.go b/control-plane/pkg/prober/prober_test.go
index 596c8985d..cafb496ae 100644
--- a/control-plane/pkg/prober/prober_test.go
+++ b/control-plane/pkg/prober/prober_test.go
@@ -60,7 +60,7 @@ func TestIPsListerFromService(t *testing.T) {
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			got, err := IPsListerFromService(tt.svc)()
+			got, err := IPsListerFromService(tt.svc)(Addressable{})
 			if tt.wantErr != (err != nil) {
 				t.Errorf("Got err %v, wantErr %v", err, tt.wantErr)
 			}
diff --git a/control-plane/pkg/reconciler/channel/channel_test.go b/control-plane/pkg/reconciler/channel/channel_test.go
index 62cd532f0..1414e6dbc 100644
--- a/control-plane/pkg/reconciler/channel/channel_test.go
+++ b/control-plane/pkg/reconciler/channel/channel_test.go
@@ -74,7 +74,7 @@ var DefaultEnv = &config.Env{
 	DataPlaneConfigMapNamespace: "knative-eventing",
 	DataPlaneConfigMapName:      "kafka-channel-channels-subscriptions",
 	GeneralConfigMapName:        "kafka-channel-config",
-	IngressName:                 "kafka-channel-dispatcher",
+	IngressName:                 "kafka-channel-ingress",
 	SystemNamespace:             "knative-eventing",
 	DataPlaneConfigFormat:       base.Json,
 }
@@ -164,7 +164,7 @@ func TestReconcileKind(t *testing.T) {
 							BootstrapServers: ChannelBootstrapServers,
 							Reference:        ChannelReference(),
 							Ingress: &contract.Ingress{
-								Path: receiver.Path(ChannelNamespace, ChannelName),
+								Host: receiver.Host(ChannelNamespace, ChannelName),
 							},
 						},
 					},
@@ -237,7 +237,7 @@ func TestReconcileKind(t *testing.T) {
 							BootstrapServers: ChannelBootstrapServers,
 							Reference:        ChannelReference(),
 							Ingress: &contract.Ingress{
-								Path: receiver.Path(ChannelNamespace, ChannelName),
+								Host: receiver.Host(ChannelNamespace, ChannelName),
 							},
 							EgressConfig: &contract.EgressConfig{
 								DeadLetter: ServiceURL,
@@ -312,7 +312,7 @@ func TestReconcileKind(t *testing.T) {
 							BootstrapServers: ChannelBootstrapServers,
 							Reference:        ChannelReference(),
 							Ingress: &contract.Ingress{
-								Path: receiver.Path(ChannelNamespace, ChannelName),
+								Host: receiver.Host(ChannelNamespace, ChannelName),
 							},
 						},
 					},
@@ -329,6 +329,7 @@ func TestReconcileKind(t *testing.T) {
 			SkipNamespaceValidation: true, // WantCreates compare the channel namespace with configmap namespace, so skip it
 			WantCreates: []runtime.Object{
 				NewConfigMapWithBinaryData(&env, nil),
+				NewPerChannelService(&env),
 			},
 			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
 				{
@@ -380,7 +381,7 @@ func TestReconcileKind(t *testing.T) {
 							BootstrapServers: ChannelBootstrapServers,
 							Reference:        ChannelReference(),
 							Ingress: &contract.Ingress{
-								Path: receiver.Path(ChannelNamespace, ChannelName),
+								Host: receiver.Host(ChannelNamespace, ChannelName),
 							},
 						},
 					},
@@ -397,6 +398,7 @@ func TestReconcileKind(t *testing.T) {
 			SkipNamespaceValidation: true, // WantCreates compare the channel namespace with configmap namespace, so skip it
 			WantCreates: []runtime.Object{
 				NewConfigMapWithBinaryData(&env, nil),
+				NewPerChannelService(&env),
 			},
 			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
 				{
@@ -449,7 +451,7 @@ func TestReconcileKind(t *testing.T) {
 							BootstrapServers: ChannelBootstrapServers,
 							Reference:        ChannelReference(),
 							Ingress: &contract.Ingress{
-								Path: receiver.Path(ChannelNamespace, ChannelName),
+								Host: receiver.Host(ChannelNamespace, ChannelName),
 							},
 							Egresses: []*contract.Egress{},
 						},
@@ -520,7 +522,7 @@ func TestReconcileKind(t *testing.T) {
 							BootstrapServers: ChannelBootstrapServers,
 							Reference:        ChannelReference(),
 							Ingress: &contract.Ingress{
-								Path: receiver.Path(ChannelNamespace, ChannelName),
+								Host: receiver.Host(ChannelNamespace, ChannelName),
 							},
 							Egresses: []*contract.Egress{{
 								ConsumerGroup: "kafka." + ChannelNamespace + "." + ChannelName + "." + Subscription1UUID,
@@ -593,7 +595,7 @@ func TestReconcileKind(t *testing.T) {
 							BootstrapServers: ChannelBootstrapServers,
 							Reference:        ChannelReference(),
 							Ingress: &contract.Ingress{
-								Path: receiver.Path(ChannelNamespace, ChannelName),
+								Host: receiver.Host(ChannelNamespace, ChannelName),
 							},
 							Egresses: []*contract.Egress{{
 								ConsumerGroup: "kafka." + ChannelNamespace + "." + ChannelName + "." + Subscription1UUID,
@@ -666,7 +668,7 @@ func TestReconcileKind(t *testing.T) {
 							BootstrapServers: ChannelBootstrapServers,
 							Reference:        ChannelReference(),
 							Ingress: &contract.Ingress{
-								Path: receiver.Path(ChannelNamespace, ChannelName),
+								Host: receiver.Host(ChannelNamespace, ChannelName),
 							},
 							Egresses: []*contract.Egress{{
 								ConsumerGroup: "kafka." + ChannelNamespace + "." + ChannelName + "." + Subscription1UUID,
@@ -742,7 +744,7 @@ func TestReconcileKind(t *testing.T) {
 							BootstrapServers: ChannelBootstrapServers,
 							Reference:        ChannelReference(),
 							Ingress: &contract.Ingress{
-								Path: receiver.Path(ChannelNamespace, ChannelName),
+								Host: receiver.Host(ChannelNamespace, ChannelName),
 							},
 							Egresses: []*contract.Egress{{
 								ConsumerGroup: "kafka." + ChannelNamespace + "." + ChannelName + "." + Subscription1UUID,
@@ -1020,7 +1022,7 @@ func TestReconcileKind(t *testing.T) {
 							BootstrapServers: ChannelBootstrapServers,
 							Reference:        ChannelReference(),
 							Ingress: &contract.Ingress{
-								Path: receiver.Path(ChannelNamespace, ChannelName),
+								Host: receiver.Host(ChannelNamespace, ChannelName),
 							},
 						},
 					},
@@ -1112,7 +1114,7 @@ func TestReconcileKind(t *testing.T) {
 								},
 							},
 							Ingress: &contract.Ingress{
-								Path: receiver.Path(ChannelNamespace, ChannelName),
+								Host: receiver.Host(ChannelNamespace, ChannelName),
 							},
 							Egresses: []*contract.Egress{{
 								ConsumerGroup: "kafka." + ChannelNamespace + "." + ChannelName + "." + Subscription1UUID,
@@ -1209,7 +1211,7 @@ func TestReconcileKind(t *testing.T) {
 								},
 							},
 							Ingress: &contract.Ingress{
-								Path: receiver.Path(ChannelNamespace, ChannelName),
+								Host: receiver.Host(ChannelNamespace, ChannelName),
 							},
 							Egresses: []*contract.Egress{{
 								ConsumerGroup: "kafka." + ChannelNamespace + "." + ChannelName + "." + Subscription1UUID,
@@ -1303,7 +1305,7 @@ func TestReconcileKind(t *testing.T) {
 								},
 							},
 							Ingress: &contract.Ingress{
-								Path: receiver.Path(ChannelNamespace, ChannelName),
+								Host: receiver.Host(ChannelNamespace, ChannelName),
 							},
 							Egresses: []*contract.Egress{{
 								ConsumerGroup: "kafka." + ChannelNamespace + "." + ChannelName + "." + Subscription1UUID,
@@ -1375,7 +1377,7 @@ func TestReconcileKind(t *testing.T) {
 							BootstrapServers: ChannelBootstrapServers,
 							Reference:        ChannelReference(),
 							Ingress: &contract.Ingress{
-								Path: receiver.Path(ChannelNamespace, ChannelName),
+								Host: receiver.Host(ChannelNamespace, ChannelName),
 							},
 						},
 					},
@@ -1443,7 +1445,7 @@ func TestReconcileKind(t *testing.T) {
 							BootstrapServers: ChannelBootstrapServers,
 							Reference:        ChannelReference(),
 							Ingress: &contract.Ingress{
-								Path: receiver.Path(ChannelNamespace, ChannelName),
+								Host: receiver.Host(ChannelNamespace, ChannelName),
 							},
 						},
 					},
@@ -1553,7 +1555,7 @@ func TestFinalizeKind(t *testing.T) {
 							BootstrapServers: ChannelBootstrapServers,
 							Reference:        ChannelReference(),
 							Ingress: &contract.Ingress{
-								Path: receiver.Path(ChannelNamespace, ChannelName),
+								Host: receiver.Host(ChannelNamespace, ChannelName),
 							},
 						},
 					},
diff --git a/control-plane/pkg/reconciler/channel/v2/channelv2_test.go b/control-plane/pkg/reconciler/channel/v2/channelv2_test.go
index e043e11b6..8c738f201 100644
--- a/control-plane/pkg/reconciler/channel/v2/channelv2_test.go
+++ b/control-plane/pkg/reconciler/channel/v2/channelv2_test.go
@@ -155,7 +155,7 @@ func TestReconcileKind(t *testing.T) {
 							BootstrapServers: ChannelBootstrapServers,
 							Reference:        ChannelReference(),
 							Ingress: &contract.Ingress{
-								Path: receiver.Path(ChannelNamespace, ChannelName),
+								Host: receiver.Host(ChannelNamespace, ChannelName),
 							},
 						},
 					},
@@ -197,7 +197,7 @@ func TestReconcileKind(t *testing.T) {
 							BootstrapServers: ChannelBootstrapServers,
 							Reference:        ChannelReference(),
 							Ingress: &contract.Ingress{
-								Path: receiver.Path(ChannelNamespace, ChannelName),
+								Host: receiver.Host(ChannelNamespace, ChannelName),
 							},
 						},
 					},
@@ -214,6 +214,7 @@ func TestReconcileKind(t *testing.T) {
 			SkipNamespaceValidation: true, // WantCreates compare the channel namespace with configmap namespace, so skip it
 			WantCreates: []runtime.Object{
 				NewConfigMapWithBinaryData(&env, nil),
+				NewPerChannelService(&env),
 			},
 			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
 				{
@@ -267,7 +268,7 @@ func TestReconcileKind(t *testing.T) {
 							BootstrapServers: ChannelBootstrapServers,
 							Reference:        ChannelReference(),
 							Ingress: &contract.Ingress{
-								Path: receiver.Path(ChannelNamespace, ChannelName),
+								Host: receiver.Host(ChannelNamespace, ChannelName),
 							},
 							EgressConfig: &contract.EgressConfig{
 								DeadLetter: ServiceURL,
@@ -288,6 +289,7 @@ func TestReconcileKind(t *testing.T) {
 			SkipNamespaceValidation: true, // WantCreates compare the channel namespace with configmap namespace, so skip it
 			WantCreates: []runtime.Object{
 				NewConfigMapWithBinaryData(&env, nil),
+				NewPerChannelService(&env),
 			},
 			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
 				{
@@ -333,7 +335,7 @@ func TestReconcileKind(t *testing.T) {
 							BootstrapServers: ChannelBootstrapServers,
 							Reference:        ChannelReference(),
 							Ingress: &contract.Ingress{
-								Path: receiver.Path(ChannelNamespace, ChannelName),
+								Host: receiver.Host(ChannelNamespace, ChannelName),
 							},
 						},
 					},
@@ -342,6 +344,7 @@ func TestReconcileKind(t *testing.T) {
 			SkipNamespaceValidation: true, // WantCreates compare the channel namespace with configmap namespace, so skip it
 			WantCreates: []runtime.Object{
 				NewConfigMapWithBinaryData(&env, nil),
+				NewPerChannelService(&env),
 			},
 			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
 				{
@@ -384,7 +387,7 @@ func TestReconcileKind(t *testing.T) {
 							BootstrapServers: ChannelBootstrapServers,
 							Reference:        ChannelReference(),
 							Ingress: &contract.Ingress{
-								Path: receiver.Path(ChannelNamespace, ChannelName),
+								Host: receiver.Host(ChannelNamespace, ChannelName),
 							},
 						},
 					},
@@ -393,6 +396,7 @@ func TestReconcileKind(t *testing.T) {
 			SkipNamespaceValidation: true, // WantCreates compare the channel namespace with configmap namespace, so skip it
 			WantCreates: []runtime.Object{
 				NewConfigMapWithBinaryData(&env, nil),
+				NewPerChannelService(&env),
 			},
 			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
 				{
@@ -427,6 +431,7 @@ func TestReconcileKind(t *testing.T) {
 			},
 			Key: testKey,
 			WantCreates: []runtime.Object{
+				NewPerChannelService(&env),
 				NewConsumerGroup(
 					WithConsumerGroupName(Subscription1UUID),
 					WithConsumerGroupNamespace(ChannelNamespace),
@@ -454,7 +459,7 @@ func TestReconcileKind(t *testing.T) {
 							BootstrapServers: ChannelBootstrapServers,
 							Reference:        ChannelReference(),
 							Ingress: &contract.Ingress{
-								Path: receiver.Path(ChannelNamespace, ChannelName),
+								Host: receiver.Host(ChannelNamespace, ChannelName),
 							},
 						},
 					},
@@ -493,6 +498,7 @@ func TestReconcileKind(t *testing.T) {
 			},
 			Key: testKey,
 			WantCreates: []runtime.Object{
+				NewPerChannelService(&env),
 				NewConsumerGroup(
 					WithConsumerGroupName(Subscription1UUID),
 					WithConsumerGroupNamespace(ChannelNamespace),
@@ -520,7 +526,7 @@ func TestReconcileKind(t *testing.T) {
 							BootstrapServers: ChannelBootstrapServers,
 							Reference:        ChannelReference(),
 							Ingress: &contract.Ingress{
-								Path: receiver.Path(ChannelNamespace, ChannelName),
+								Host: receiver.Host(ChannelNamespace, ChannelName),
 							},
 						},
 					},
@@ -559,6 +565,7 @@ func TestReconcileKind(t *testing.T) {
 			},
 			Key: testKey,
 			WantCreates: []runtime.Object{
+				NewPerChannelService(&env),
 				NewConsumerGroup(
 					WithConsumerGroupName(Subscription1UUID),
 					WithConsumerGroupNamespace(ChannelNamespace),
@@ -586,7 +593,7 @@ func TestReconcileKind(t *testing.T) {
 							BootstrapServers: ChannelBootstrapServers,
 							Reference:        ChannelReference(),
 							Ingress: &contract.Ingress{
-								Path: receiver.Path(ChannelNamespace, ChannelName),
+								Host: receiver.Host(ChannelNamespace, ChannelName),
 							},
 						},
 					},
@@ -631,6 +638,9 @@ func TestReconcileKind(t *testing.T) {
 				),
 			},
 			Key: testKey,
+			WantCreates: []runtime.Object{
+				NewPerChannelService(&env),
+			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
 				ConfigMapUpdate(&env, &contract.Contract{
 					Generation: 1,
@@ -641,7 +651,7 @@ func TestReconcileKind(t *testing.T) {
 							BootstrapServers: ChannelBootstrapServers,
 							Reference:        ChannelReference(),
 							Ingress: &contract.Ingress{
-								Path: receiver.Path(ChannelNamespace, ChannelName),
+								Host: receiver.Host(ChannelNamespace, ChannelName),
 							},
 						},
 					},
@@ -714,6 +724,9 @@ func TestReconcileKind(t *testing.T) {
 				),
 			},
 			Key: testKey,
+			WantCreates: []runtime.Object{
+				NewPerChannelService(&env),
+			},
 			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
 				{
 					Object: NewChannel(
@@ -738,7 +751,7 @@ func TestReconcileKind(t *testing.T) {
 							BootstrapServers: ChannelBootstrapServers,
 							Reference:        ChannelReference(),
 							Ingress: &contract.Ingress{
-								Path: receiver.Path(ChannelNamespace, ChannelName),
+								Host: receiver.Host(ChannelNamespace, ChannelName),
 							},
 						},
 					},
@@ -765,6 +778,7 @@ func TestReconcileKind(t *testing.T) {
 			},
 			Key: testKey,
 			WantCreates: []runtime.Object{
+				NewPerChannelService(&env),
 				NewConsumerGroup(
 					WithConsumerGroupName(Subscription1UUID),
 					WithConsumerGroupNamespace(ChannelNamespace),
@@ -808,7 +822,7 @@ func TestReconcileKind(t *testing.T) {
 							BootstrapServers: ChannelBootstrapServers,
 							Reference:        ChannelReference(),
 							Ingress: &contract.Ingress{
-								Path: receiver.Path(ChannelNamespace, ChannelName),
+								Host: receiver.Host(ChannelNamespace, ChannelName),
 							},
 						},
 					},
@@ -865,6 +879,7 @@ func TestReconcileKind(t *testing.T) {
 			},
 			Key: testKey,
 			WantCreates: []runtime.Object{
+				NewPerChannelService(&env),
 				NewConsumerGroup(
 					WithConsumerGroupName(Subscription1UUID),
 					WithConsumerGroupNamespace(ChannelNamespace),
@@ -892,7 +907,7 @@ func TestReconcileKind(t *testing.T) {
 							BootstrapServers: ChannelBootstrapServers,
 							Reference:        ChannelReference(),
 							Ingress: &contract.Ingress{
-								Path: receiver.Path(ChannelNamespace, ChannelName),
+								Host: receiver.Host(ChannelNamespace, ChannelName),
 							},
 						},
 					},
@@ -1049,7 +1064,7 @@ func TestReconcileKind(t *testing.T) {
 							BootstrapServers: ChannelBootstrapServers,
 							Reference:        ChannelReference(),
 							Ingress: &contract.Ingress{
-								Path: receiver.Path(ChannelNamespace, ChannelName),
+								Host: receiver.Host(ChannelNamespace, ChannelName),
 							},
 						},
 					},
@@ -1058,6 +1073,7 @@ func TestReconcileKind(t *testing.T) {
 			SkipNamespaceValidation: true, // WantCreates compare the channel namespace with configmap namespace, so skip it
 			WantCreates: []runtime.Object{
 				NewConfigMapWithBinaryData(&env, nil),
+				NewPerChannelService(&env),
 			},
 			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
 				{
@@ -1112,6 +1128,9 @@ func TestReconcileKind(t *testing.T) {
 				),
 			},
 			Key: testKey,
+			WantCreates: []runtime.Object{
+				NewPerChannelService(&env),
+			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
 				ConfigMapUpdate(&env, &contract.Contract{
 					Generation: 1,
@@ -1130,7 +1149,7 @@ func TestReconcileKind(t *testing.T) {
 								},
 							},
 							Ingress: &contract.Ingress{
-								Path: receiver.Path(ChannelNamespace, ChannelName),
+								Host: receiver.Host(ChannelNamespace, ChannelName),
 							},
 						},
 					},
@@ -1177,7 +1196,7 @@ func TestReconcileKind(t *testing.T) {
 							BootstrapServers: ChannelBootstrapServers,
 							Reference:        ChannelReference(),
 							Ingress: &contract.Ingress{
-								Path: receiver.Path(ChannelNamespace, ChannelName),
+								Host: receiver.Host(ChannelNamespace, ChannelName),
 							},
 						},
 					},
@@ -1186,6 +1205,7 @@ func TestReconcileKind(t *testing.T) {
 			SkipNamespaceValidation: true, // WantCreates compare the channel namespace with configmap namespace, so skip it
 			WantCreates: []runtime.Object{
 				NewConfigMapWithBinaryData(&env, nil),
+				NewPerChannelService(&env),
 			},
 			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
 				{
@@ -1217,6 +1237,9 @@ func TestReconcileKind(t *testing.T) {
 				NewConfigMapWithBinaryData(&env, nil),
 			},
 			Key: testKey,
+			WantCreates: []runtime.Object{
+				NewPerChannelService(&env),
+			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
 				ConfigMapUpdate(&env, &contract.Contract{
 					Generation: 1,
@@ -1227,7 +1250,7 @@ func TestReconcileKind(t *testing.T) {
 							BootstrapServers: ChannelBootstrapServers,
 							Reference:        ChannelReference(),
 							Ingress: &contract.Ingress{
-								Path: receiver.Path(ChannelNamespace, ChannelName),
+								Host: receiver.Host(ChannelNamespace, ChannelName),
 							},
 						},
 					},
@@ -1336,6 +1359,7 @@ func TestReconcileKind(t *testing.T) {
 				}, nil
 			},
 			ConfigMapLister:     listers.GetConfigMapLister(),
+			ServiceLister:       listers.GetServiceLister(),
 			ConsumerGroupLister: listers.GetConsumerGroupLister(),
 			InternalsClient:     fakeconsumergroupinformer.Get(ctx),
 			Prober:              proberMock,
diff --git a/control-plane/pkg/reconciler/channel/v2/controllerv2_test.go b/control-plane/pkg/reconciler/channel/v2/controllerv2_test.go
index 469317c78..34459582f 100644
--- a/control-plane/pkg/reconciler/channel/v2/controllerv2_test.go
+++ b/control-plane/pkg/reconciler/channel/v2/controllerv2_test.go
@@ -24,16 +24,18 @@ import (
 	"k8s.io/apimachinery/pkg/runtime"
 	fakekubeclientset "k8s.io/client-go/kubernetes/fake"
 
-	_ "knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/injection/informers/eventing/v1alpha1/consumergroup/fake"
 	_ "knative.dev/eventing-kafka/pkg/client/injection/informers/messaging/v1beta1/kafkachannel/fake"
 	_ "knative.dev/pkg/client/injection/ducks/duck/v1/addressable/fake"
 	fakekubeclient "knative.dev/pkg/client/injection/kube/client/fake"
 	_ "knative.dev/pkg/client/injection/kube/informers/core/v1/configmap/fake"
 	_ "knative.dev/pkg/client/injection/kube/informers/core/v1/pod/fake"
 	_ "knative.dev/pkg/client/injection/kube/informers/core/v1/secret/fake"
+	_ "knative.dev/pkg/client/injection/kube/informers/core/v1/service/fake"
 	dynamicclient "knative.dev/pkg/injection/clients/dynamicclient/fake"
 	reconcilertesting "knative.dev/pkg/reconciler/testing"
 
+	_ "knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/injection/informers/eventing/v1alpha1/consumergroup/fake"
+
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
 )
 
diff --git a/control-plane/pkg/reconciler/testing/objects_channel.go b/control-plane/pkg/reconciler/testing/objects_channel.go
index f52359054..2982313c9 100644
--- a/control-plane/pkg/reconciler/testing/objects_channel.go
+++ b/control-plane/pkg/reconciler/testing/objects_channel.go
@@ -193,8 +193,7 @@ func ChannelAddressable(env *config.Env) func(obj duckv1.KRShaped) {
 
 		channel.Status.Address.URL = &apis.URL{
 			Scheme: "http",
-			Host:   network.GetServiceHostname(env.IngressName, env.SystemNamespace),
-			Path:   fmt.Sprintf("/%s/%s", channel.Namespace, channel.Name),
+			Host:   fmt.Sprintf("%s.%s.svc.%s", resources.MakeChannelServiceName(channel.Name), channel.Namespace, network.GetClusterDomainName()),
 		}
 
 		channel.GetConditionSet().Manage(&channel.Status).MarkTrue(base.ConditionAddressable)
diff --git a/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/AbstractConsumerVerticleTest.java b/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/AbstractConsumerVerticleTest.java
index 89c546472..cb700ce9e 100644
--- a/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/AbstractConsumerVerticleTest.java
+++ b/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/AbstractConsumerVerticleTest.java
@@ -48,6 +48,7 @@ import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.stream.Collectors;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.awaitility.Awaitility.await;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.times;
@@ -138,11 +139,11 @@ public abstract class AbstractConsumerVerticleTest {
       .onFailure(context::failNow);
 
     undeployPromise.future()
-      .onSuccess(ignored -> {
-        assertThat(consumer.closed()).isTrue();
-        context.completeNow();
-      })
+      .onSuccess(ignored -> {})
       .onFailure(context::failNow);
+
+    await().untilAsserted(() -> assertThat(consumer.closed()).isTrue());
+    context.completeNow();
   }
 
   @Test
@@ -151,6 +152,8 @@ public abstract class AbstractConsumerVerticleTest {
     final var topics = new String[]{"a"};
     final KafkaConsumer<Object, CloudEvent> consumer = mock(KafkaConsumer.class);
 
+    final var checkpoints = context.checkpoint(2);
+
     when(consumer.close()).thenReturn(Future.succeededFuture());
     when(consumer.subscribe((Set<String>) any(), any())).then(answer -> {
       final Handler<AsyncResult<Void>> callback = answer.getArgument(1);
@@ -226,11 +229,13 @@ public abstract class AbstractConsumerVerticleTest {
           assertThat(consumerRecordSenderClosed.get()).isTrue();
           assertThat(sinkClosed.get()).isTrue();
           assertThat(dlsSenderClosed.get()).isTrue();
-          verify(consumer, times(1)).close();
-
-          context.completeNow();
+          checkpoints.flag();
         }))
       );
+
+
+    await().untilAsserted(() -> verify(consumer, times(1)).close());
+    checkpoints.flag();
   }
 
   abstract BaseConsumerVerticle createConsumerVerticle(BaseConsumerVerticle.Initializer initializer,
diff --git a/test/upgrade/continual.go b/test/upgrade/continual.go
index 6f1c62e82..0eda11517 100644
--- a/test/upgrade/continual.go
+++ b/test/upgrade/continual.go
@@ -20,13 +20,11 @@ import (
 	pkgupgrade "knative.dev/pkg/test/upgrade"
 
 	"knative.dev/eventing-kafka-broker/test/upgrade/continual"
-	eventingkafkacontinual "knative.dev/eventing-kafka/test/upgrade/continual"
 )
 
 // ContinualTests returns all continual tests.
 func ContinualTests() []pkgupgrade.BackgroundOperation {
 	c := BrokerContinualTests()
-	c = append(c, ChannelContinualTests()...)
 	c = append(c, SinkContinualTests()...)
 	return c
 }
@@ -40,16 +38,6 @@ func BrokerContinualTests() []pkgupgrade.BackgroundOperation {
 	}
 }
 
-// ChannelContinualTests returns background operations to test KafkaChannel
-// functionality in continual manner during the whole upgrade and downgrade
-// process asserting that all events are propagated.
-func ChannelContinualTests() []pkgupgrade.BackgroundOperation {
-	return []pkgupgrade.BackgroundOperation{
-		eventingkafkacontinual.ChannelTest(eventingkafkacontinual.ChannelTestOptions{}),
-		eventingkafkacontinual.BrokerBackedByChannelTest(eventingkafkacontinual.ChannelTestOptions{}),
-	}
-}
-
 // SinkContinualTests returns background operations to test KafkaSink
 // functionality in continual manner during the whole upgrade and downgrade
 // process asserting that all events are propagated.
diff --git a/vendor/knative.dev/eventing-kafka/test/upgrade/continual/channel.go b/vendor/knative.dev/eventing-kafka/test/upgrade/continual/channel.go
index c082763a2..559ea835c 100644
--- a/vendor/knative.dev/eventing-kafka/test/upgrade/continual/channel.go
+++ b/vendor/knative.dev/eventing-kafka/test/upgrade/continual/channel.go
@@ -250,7 +250,7 @@ func (b *brokerBackedByKafkaChannelSut) setKafkaAsDefaultForBroker(ctx sut.Conte
 			Name:      "config-br-default-channel",
 		},
 		Data: map[string]string{
-			"channelTemplateSpec": fmt.Sprintf(`apiVersion: %s
+			"channel-template-spec": fmt.Sprintf(`apiVersion: %s
 kind: %s
 spec:
   numPartitions: %d