Repository: knative-sandbox/eventing-kafka-broker
Pull request #1469 (2021-11-11 12:50:19)

Add KafkaSource condition for Sink resolved
> when KafkaSource cannot resolve the sink, it just reports status "Unknown"

## Proposed Changes

- Add KafkaSource condition for Sink resolved

<!--
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
None
```

**Docs**

<!--
:book: If this change has user-visible impact, link to an issue or PR in
https://github.com/knative/docs.
-->
```
None
```


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

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

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

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

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

- ~~[OWNERS](https://github.com/knative-sandbox/eventing-kafka-broker/blob/main/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-metrics-robot: The following is the coverage report on the affected files.
Say `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage` to re-run this coverage report

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[control-plane/pkg/reconciler/source/source.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1469/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1458779122963582976/artifacts/line-cov.html#file1) | 74.2% | 77.0% | 2.8

codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1469?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) Report
> Merging [#1469](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1469?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (895d352) into [main](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/commit/1b5771f35a34395650f28ce1b07a00886dbd7d08?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (1b5771f) will **decrease** coverage by `7.97%`.
> The diff coverage is `46.15%`.

[![Impacted file tree graph](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1469/graphs/tree.svg?width=650&height=150&src=pr&token=xyPIgGOiwy&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1469?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)

```diff
@@             Coverage Diff              @@
##               main    #1469      +/-   ##
============================================
- Coverage     73.41%   65.43%   -7.98%     
============================================
  Files           101       36      -65     
  Lines          3949     2086    -1863     
  Branches        164        0     -164     
============================================
- Hits           2899     1365    -1534     
+ Misses          810      569     -241     
+ Partials        240      152      -88     
```

| Flag | Coverage Δ | |
|---|---|---|
| java-unittests | `?` | |

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/1469?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | Coverage Δ | |
|---|---|---|
| [...lane/pkg/reconciler/base/receiver\_condition\_set.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1469/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9iYXNlL3JlY2VpdmVyX2NvbmRpdGlvbl9zZXQuZ28=) | `0.00% <0.00%> (ø)` | |
| [control-plane/pkg/reconciler/source/source.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1469/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9zb3VyY2Uvc291cmNlLmdv) | `61.65% <100.00%> (+3.96%)` | :arrow_up: |
| [control-plane/pkg/reconciler/channel/channel.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1469/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9jaGFubmVsL2NoYW5uZWwuZ28=) | `29.96% <0.00%> (-0.12%)` | :arrow_down: |
| [...tive/eventing/kafka/broker/receiver/main/Main.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1469/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9yZWNlaXZlci9zcmMvbWFpbi9qYXZhL2Rldi9rbmF0aXZlL2V2ZW50aW5nL2thZmthL2Jyb2tlci9yZWNlaXZlci9tYWluL01haW4uamF2YQ==) | | |
| [...spatcher/impl/consumer/CloudEventDeserializer.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1469/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9jb25zdW1lci9DbG91ZEV2ZW50RGVzZXJpYWxpemVyLmphdmE=) | | |
| [...ka/broker/core/reconciler/ResourcesReconciler.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1469/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9jb3JlL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2NvcmUvcmVjb25jaWxlci9SZXNvdXJjZXNSZWNvbmNpbGVyLmphdmE=) | | |
| [...er/dispatcher/impl/consumer/InvalidCloudEvent.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1469/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9jb25zdW1lci9JbnZhbGlkQ2xvdWRFdmVudC5qYXZh) | | |
| [...ka/broker/dispatcher/impl/filter/SuffixFilter.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1469/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9maWx0ZXIvU3VmZml4RmlsdGVyLmphdmE=) | | |
| [...cher/impl/consumer/CloudEventOverridesMutator.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1469/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9jb25zdW1lci9DbG91ZEV2ZW50T3ZlcnJpZGVzTXV0YXRvci5qYXZh) | | |
| [...ve/eventing/kafka/broker/core/metrics/Metrics.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1469/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9jb3JlL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2NvcmUvbWV0cmljcy9NZXRyaWNzLmphdmE=) | | |
| ... and [59 more](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1469/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/1469?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/1469?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Last update [1b5771f...895d352](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1469?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-metrics-robot: The following is the coverage report on the affected files.
Say `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage` to re-run this coverage report

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[control-plane/pkg/reconciler/source/source.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1469/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1458780221892202496/artifacts/line-cov.html#file1) | 74.2% | 77.0% | 2.8

knative-prow-robot: @pierDipi: 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
--- | --- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-broker-integration-tests | c3f6e990281d4d8ecd902add4c8057f498b07e83 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1469/pull-knative-sandbox-eventing-kafka-broker-integration-tests/1458780221867036672) | true | `/test pull-knative-sandbox-eventing-kafka-broker-integration-tests`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka-broker/1469). [Your PR dashboard](https://gubernator.knative.dev/pr/pierDipi).

<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/reconciler/base/receiver_condition_set.go b/control-plane/pkg/reconciler/base/receiver_condition_set.go
index 0def56f54..e904d8710 100644
--- a/control-plane/pkg/reconciler/base/receiver_condition_set.go
+++ b/control-plane/pkg/reconciler/base/receiver_condition_set.go
@@ -28,6 +28,8 @@ import (
 	"knative.dev/pkg/network"
 	"knative.dev/pkg/reconciler"
 
+	sources "knative.dev/eventing-kafka/pkg/apis/sources/v1beta1"
+
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
 )
 
@@ -53,6 +55,7 @@ var EgressConditionSet = apis.NewLivingConditionSet(
 	ConditionTopicReady,
 	ConditionConfigMapUpdated,
 	ConditionInitialOffsetsCommitted,
+	sources.KafkaConditionSinkProvided,
 )
 
 const (
@@ -266,3 +269,16 @@ func (manager *StatusConditionManager) InitialOffsetNotCommitted(err error) erro
 func (manager *StatusConditionManager) InitialOffsetsCommitted() {
 	manager.Object.GetConditionSet().Manage(manager.Object.GetStatus()).MarkTrue(ConditionInitialOffsetsCommitted)
 }
+
+func (manager *StatusConditionManager) SinkResolved() {
+	manager.Object.GetConditionSet().Manage(manager.Object.GetStatus()).MarkTrue(sources.KafkaConditionSinkProvided)
+}
+
+func (manager *StatusConditionManager) FailedToResolveSink(err error) error {
+	manager.Object.GetConditionSet().Manage(manager.Object.GetStatus()).MarkFalse(
+		sources.KafkaConditionSinkProvided,
+		"FailedToResolveSink",
+		err.Error(),
+	)
+	return fmt.Errorf("failed to resolve sink: %w", err)
+}
diff --git a/control-plane/pkg/reconciler/source/source.go b/control-plane/pkg/reconciler/source/source.go
index d27d0238a..9e406c49c 100644
--- a/control-plane/pkg/reconciler/source/source.go
+++ b/control-plane/pkg/reconciler/source/source.go
@@ -145,6 +145,12 @@ func (r *Reconciler) reconcileKind(ctx context.Context, ks *sources.KafkaSource)
 		}
 	}
 	statusConditionManager.InitialOffsetsCommitted()
+	// Get resource configuration.
+	resource, err := r.reconcileKafkaSourceResource(ctx, ks, authContext.MultiSecretReference)
+	if err != nil {
+		return statusConditionManager.FailedToResolveSink(err)
+	}
+	statusConditionManager.SinkResolved()
 
 	// Get contract config map.
 	contractConfigMap, err := r.GetOrCreateDataPlaneConfigMap(ctx)
@@ -162,12 +168,6 @@ func (r *Reconciler) reconcileKind(ctx context.Context, ks *sources.KafkaSource)
 
 	logger.Debug("Got contract data from config map", zap.Any(base.ContractLogKey, ct))
 
-	// Get resource configuration.
-	resource, err := r.reconcileKafkaSourceResource(ctx, ks, authContext.MultiSecretReference)
-	if err != nil {
-		return statusConditionManager.FailedToGetConfig(err)
-	}
-
 	sourceIndex := coreconfig.FindResource(ct, ks.GetUID())
 	// Update contract data with the new contract configuration
 	changed := coreconfig.AddOrUpdateResourceConfig(ct, resource, sourceIndex, logger)
@@ -261,8 +261,10 @@ func (r *Reconciler) reconcileKafkaSourceResource(ctx context.Context, ks *sourc
 	}
 	destination, err := r.Resolver.URIFromDestinationV1(ctx, destinationSpec, ks)
 	if err != nil {
+		ks.Status.SinkURI = nil
 		return nil, fmt.Errorf("failed to resolve destination: %w", err)
 	}
+	ks.Status.SinkURI = destination
 
 	egressConfig := proto.Clone(&DefaultEgressConfig).(*contract.EgressConfig)
 
diff --git a/go.mod b/go.mod
index 36e640843..380f3a440 100644
--- a/go.mod
+++ b/go.mod
@@ -24,9 +24,9 @@ require (
 	k8s.io/apiserver v0.21.4
 	k8s.io/client-go v0.21.4
 	k8s.io/utils v0.0.0-20210111153108-fddb29f9d009
-	knative.dev/eventing v0.27.1-0.20211109172546-31e4e2eb26d1
-	knative.dev/eventing-kafka v0.27.1-0.20211109165044-86d36b670174
+	knative.dev/eventing v0.27.1-0.20211111093638-536f2b97d020
+	knative.dev/eventing-kafka v0.27.1-0.20211110051645-e696b902011b
 	knative.dev/hack v0.0.0-20211108170701-96aac1c30be3
-	knative.dev/pkg v0.0.0-20211109100843-91d1932616a7
+	knative.dev/pkg v0.0.0-20211111114938-0b0c3390a475
 	knative.dev/reconciler-test v0.0.0-20211101213739-c5f938aa9974
 )
diff --git a/go.sum b/go.sum
index 1f3079d5e..cc5aeb4aa 100644
--- a/go.sum
+++ b/go.sum
@@ -1328,10 +1328,10 @@ k8s.io/utils v0.0.0-20210111153108-fddb29f9d009 h1:0T5IaWHO3sJTEmCP6mUlBvMukxPKU
 k8s.io/utils v0.0.0-20210111153108-fddb29f9d009/go.mod h1:jPW/WVKK9YHAvNhRxK0md/EJ228hCsBRufyofKtW8HA=
 knative.dev/control-protocol v0.0.0-20211108155701-88c4c1e549c4/go.mod h1:5hOr3rdQDEBE/3LihWgyfkfowQn8V0Vur/qg+g42MXY=
 knative.dev/eventing v0.27.1-0.20211108221643-a72cfc91c421/go.mod h1:lM+1rTzjcNmrPjTGaEk+QrmQGuM3/O+e3aN96GH4V0Y=
-knative.dev/eventing v0.27.1-0.20211109172546-31e4e2eb26d1 h1:vC/EpcYF1lsvaapxNGmz6vGNZfq94/1Fd770SQTZ9Es=
-knative.dev/eventing v0.27.1-0.20211109172546-31e4e2eb26d1/go.mod h1:zHDfFKGqgq9bZvUd3QC6iHRAaj2AbHercYhlxdtc9fA=
-knative.dev/eventing-kafka v0.27.1-0.20211109165044-86d36b670174 h1:mMZM9bJesUECCHg2kO5VvdOYK9cEIfAg8NNSNPgcPhY=
-knative.dev/eventing-kafka v0.27.1-0.20211109165044-86d36b670174/go.mod h1:6RjAzpeBOW7/fG+2E4d0siS7Zfy0YfxJIC2Qal+3K6M=
+knative.dev/eventing v0.27.1-0.20211111093638-536f2b97d020 h1:G4qzDjgXDiS6U/yYbNmzDyEGBG7/8cG08JQIJ7QL80o=
+knative.dev/eventing v0.27.1-0.20211111093638-536f2b97d020/go.mod h1:3pTt6KXnOuVLB8NPnD8qb2xdcRTJREOHdIDd2sd7AjM=
+knative.dev/eventing-kafka v0.27.1-0.20211110051645-e696b902011b h1:osp3npuQlECXMsCjMtCURHWTNKlAGSOY3t9kf1ca6QE=
+knative.dev/eventing-kafka v0.27.1-0.20211110051645-e696b902011b/go.mod h1:6RjAzpeBOW7/fG+2E4d0siS7Zfy0YfxJIC2Qal+3K6M=
 knative.dev/hack v0.0.0-20211101195839-11d193bf617b/go.mod h1:PHt8x8yX5Z9pPquBEfIj0X66f8iWkWfR0S/sarACJrI=
 knative.dev/hack v0.0.0-20211105231158-29f86c2653b5/go.mod h1:PHt8x8yX5Z9pPquBEfIj0X66f8iWkWfR0S/sarACJrI=
 knative.dev/hack v0.0.0-20211108170701-96aac1c30be3 h1:oSvRgnKoU308k7aXbPV3iL5Zh5kBGM2Ptar4hyeda+A=
@@ -1340,8 +1340,8 @@ knative.dev/hack/schema v0.0.0-20211105231158-29f86c2653b5/go.mod h1:ffjwmdcrH5v
 knative.dev/hack/schema v0.0.0-20211108170701-96aac1c30be3/go.mod h1:ffjwmdcrH5vN3mPhO8RrF2KfNnbHeCE2C60A+2cv3U0=
 knative.dev/pkg v0.0.0-20211101212339-96c0204a70dc/go.mod h1:SkfDk9bWIiNZD7XtILGkG7AKVyF/M6M0bGxLgl0SYL8=
 knative.dev/pkg v0.0.0-20211108064904-3cc697a3cb09/go.mod h1:YiSfFDMdYHZTuO5XS8UzxSeOEmocy+Kb2YOGik9C8lw=
-knative.dev/pkg v0.0.0-20211109100843-91d1932616a7 h1:uqo0nLez3NJf03gE9mdYpFxbqZ8iWdVElQa+s8W7VNQ=
-knative.dev/pkg v0.0.0-20211109100843-91d1932616a7/go.mod h1:plYIuKk6WNC9LyiG/9xBnOBYuxO0Ro6xN0z7iXWVUtk=
+knative.dev/pkg v0.0.0-20211111114938-0b0c3390a475 h1:BoZ88UHlgwRiGkkz7MQtoSlQyBRQxfu1ROOpUNDG3bU=
+knative.dev/pkg v0.0.0-20211111114938-0b0c3390a475/go.mod h1:plYIuKk6WNC9LyiG/9xBnOBYuxO0Ro6xN0z7iXWVUtk=
 knative.dev/reconciler-test v0.0.0-20211101213739-c5f938aa9974 h1:4eo67cEiDvEl1TgITT+3n1ivAcLFCQ6DBJnX855727E=
 knative.dev/reconciler-test v0.0.0-20211101213739-c5f938aa9974/go.mod h1:gTsbLk496j/M9xqMpx/liyCQ0X3bwDpRtcs2Zzws364=
 modernc.org/cc v1.0.0/go.mod h1:1Sk4//wdnYJiUIxnW8ddKpaOJCF37yAdqYnkxUpaYxw=
diff --git a/vendor/modules.txt b/vendor/modules.txt
index f7d6da2d4..27d75f606 100644
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -1104,7 +1104,7 @@ k8s.io/utils/buffer
 k8s.io/utils/integer
 k8s.io/utils/pointer
 k8s.io/utils/trace
-# knative.dev/eventing v0.27.1-0.20211109172546-31e4e2eb26d1
+# knative.dev/eventing v0.27.1-0.20211111093638-536f2b97d020
 ## explicit
 knative.dev/eventing/pkg/apis/config
 knative.dev/eventing/pkg/apis/duck
@@ -1213,7 +1213,7 @@ knative.dev/eventing/test/test_images/performance
 knative.dev/eventing/test/test_images/print
 knative.dev/eventing/test/test_images/recordevents
 knative.dev/eventing/test/test_images/request-sender
-# knative.dev/eventing-kafka v0.27.1-0.20211109165044-86d36b670174
+# knative.dev/eventing-kafka v0.27.1-0.20211110051645-e696b902011b
 ## explicit
 knative.dev/eventing-kafka/pkg/apis/bindings
 knative.dev/eventing-kafka/pkg/apis/bindings/v1beta1
@@ -1270,7 +1270,7 @@ knative.dev/eventing-kafka/test/lib/resources
 # knative.dev/hack v0.0.0-20211108170701-96aac1c30be3
 ## explicit
 knative.dev/hack
-# knative.dev/pkg v0.0.0-20211109100843-91d1932616a7
+# knative.dev/pkg v0.0.0-20211111114938-0b0c3390a475
 ## explicit
 knative.dev/pkg/apis
 knative.dev/pkg/apis/duck


Test patch:
diff --git a/control-plane/pkg/reconciler/source/source_test.go b/control-plane/pkg/reconciler/source/source_test.go
index 094312360..d9b86eca8 100644
--- a/control-plane/pkg/reconciler/source/source_test.go
+++ b/control-plane/pkg/reconciler/source/source_test.go
@@ -28,13 +28,14 @@ import (
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/types"
 	clientgotesting "k8s.io/client-go/testing"
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
 	sources "knative.dev/eventing-kafka/pkg/apis/sources/v1beta1"
 	duckv1 "knative.dev/pkg/apis/duck/v1"
 	kubeclient "knative.dev/pkg/client/injection/kube/client/fake"
 	"knative.dev/pkg/resolver"
 	"knative.dev/pkg/tracker"
 
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
+
 	"knative.dev/pkg/controller"
 	"knative.dev/pkg/logging"
 	. "knative.dev/pkg/reconciler/testing"
@@ -127,6 +128,7 @@ func TestReconcileKind(t *testing.T) {
 						StatusTopicReadyWithName(strings.Join(SourceTopics, ", ")),
 						StatusDataPlaneAvailable,
 						StatusInitialOffsetsCommitted,
+						StatusSourceSinkResolved(ServiceURL),
 					),
 				},
 			},
@@ -194,6 +196,7 @@ func TestReconcileKind(t *testing.T) {
 						StatusTopicReadyWithName(strings.Join(SourceTopics, ", ")),
 						StatusDataPlaneAvailable,
 						StatusInitialOffsetsCommitted,
+						StatusSourceSinkResolved(ServiceURL),
 					),
 				},
 			},
@@ -255,6 +258,7 @@ func TestReconcileKind(t *testing.T) {
 						StatusTopicReadyWithName(strings.Join(SourceTopics, ", ")),
 						StatusDataPlaneAvailable,
 						StatusInitialOffsetsCommitted,
+						StatusSourceSinkResolved(ServiceURL),
 					),
 				},
 			},
@@ -316,6 +320,7 @@ func TestReconcileKind(t *testing.T) {
 						StatusTopicReadyWithName(strings.Join(SourceTopics, ", ")),
 						StatusDataPlaneAvailable,
 						StatusInitialOffsetsCommitted,
+						StatusSourceSinkResolved(ServiceURL),
 					),
 				},
 			},
@@ -377,6 +382,7 @@ func TestReconcileKind(t *testing.T) {
 						StatusTopicReadyWithName(strings.Join(SourceTopics, ", ")),
 						StatusDataPlaneAvailable,
 						StatusInitialOffsetsCommitted,
+						StatusSourceSinkResolved(ServiceURL),
 					),
 				},
 			},
@@ -438,6 +444,35 @@ func TestReconcileKind(t *testing.T) {
 						StatusTopicReadyWithName(strings.Join(SourceTopics, ", ")),
 						StatusDataPlaneAvailable,
 						StatusInitialOffsetsCommitted,
+						StatusSourceSinkResolved(ServiceURL),
+					),
+				},
+			},
+			WantPatches: []clientgotesting.PatchActionImpl{
+				patchFinalizers(),
+			},
+			WantEvents: []string{
+				finalizerUpdatedEvent,
+			},
+		},
+		{
+			Name: "Reconciled failed - no sink",
+			Objects: []runtime.Object{
+				NewSource(),
+				SourceDispatcherPod(env.SystemNamespace, map[string]string{
+					"annotation_to_preserve": "value_to_preserve",
+				}),
+			},
+			Key:                     testKey,
+			SkipNamespaceValidation: true, // WantCreates compare the broker namespace with configmap namespace, so skip it
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: NewSource(
+						InitSourceConditions,
+						StatusTopicReadyWithName(strings.Join(SourceTopics, ", ")),
+						StatusDataPlaneAvailable,
+						StatusInitialOffsetsCommitted,
+						StatusSourceSinkNotResolved("failed to resolve destination: services \"test-service\" not found"),
 					),
 				},
 			},
@@ -446,7 +481,9 @@ func TestReconcileKind(t *testing.T) {
 			},
 			WantEvents: []string{
 				finalizerUpdatedEvent,
+				"Warning InternalError failed to resolve sink: failed to resolve destination: services \"test-service\" not found",
 			},
+			WantErr: true,
 		},
 	}
 
diff --git a/control-plane/pkg/reconciler/testing/objects_source.go b/control-plane/pkg/reconciler/testing/objects_source.go
index 18add224d..17feb9077 100644
--- a/control-plane/pkg/reconciler/testing/objects_source.go
+++ b/control-plane/pkg/reconciler/testing/objects_source.go
@@ -22,6 +22,7 @@ import (
 	"k8s.io/apimachinery/pkg/runtime"
 	"knative.dev/eventing-kafka/pkg/apis/bindings/v1beta1"
 	sources "knative.dev/eventing-kafka/pkg/apis/sources/v1beta1"
+	"knative.dev/pkg/apis"
 	duckv1 "knative.dev/pkg/apis/duck/v1"
 
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/base"
@@ -125,3 +126,28 @@ func SourceDispatcherPod(namespace string, annotations map[string]string) runtim
 		},
 	}
 }
+
+func InitSourceConditions(obj duckv1.KRShaped) {
+	sink := obj.(*sources.KafkaSource)
+	sink.Status.InitializeConditions()
+}
+
+func StatusSourceSinkResolved(uri string) KRShapedOption {
+	return func(obj duckv1.KRShaped) {
+		ks := obj.(*sources.KafkaSource)
+		ks.Status.SinkURI, _ = apis.ParseURL(uri)
+		ks.GetConditionSet().Manage(ks.GetStatus()).MarkTrue(sources.KafkaConditionSinkProvided)
+	}
+}
+
+func StatusSourceSinkNotResolved(err string) KRShapedOption {
+	return func(obj duckv1.KRShaped) {
+		ks := obj.(*sources.KafkaSource)
+		ks.Status.SinkURI = nil
+		ks.GetConditionSet().Manage(ks.GetStatus()).MarkFalse(
+			sources.KafkaConditionSinkProvided,
+			"FailedToResolveSink",
+			err,
+		)
+	}
+}
diff --git a/vendor/knative.dev/pkg/test/request.go b/vendor/knative.dev/pkg/test/request.go
index fd495c121..a22542e29 100644
--- a/vendor/knative.dev/pkg/test/request.go
+++ b/vendor/knative.dev/pkg/test/request.go
@@ -105,7 +105,7 @@ func WaitForEndpointStateWithTimeout(
 	timeout time.Duration,
 	opts ...interface{}) (*spoof.Response, error) {
 
-	client, rOpts, err := makeSpoofClient(ctx, kubeClient, logf, url, resolvable, timeout /* true, */, opts...)
+	client, rOpts, err := makeSpoofClient(ctx, kubeClient, logf, url, resolvable, timeout, opts...)
 	if err != nil {
 		return nil, err
 	}
@@ -152,7 +152,7 @@ func CheckEndpointState(
 	resolvable bool,
 	opts ...interface{},
 ) (*spoof.Response, error) {
-	client, rOpts, err := makeSpoofClient(ctx, kubeClient, logf, url, resolvable, Flags.SpoofRequestTimeout /* false, */, opts...)
+	client, rOpts, err := makeSpoofClient(ctx, kubeClient, logf, url, resolvable, Flags.SpoofRequestTimeout, opts...)
 	if err != nil {
 		return nil, err
 	}