Repository: knative-sandbox/eventing-kafka-broker
Pull request #1883 (2022-02-09 12:42:27)

[release-1.1] Do not delete finalizer until the probe returns status not ready
This is an automated cherry-pick of #1805

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

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/1883#" title="Author self-approved">knative-prow-robot</a>*
To complete the [pull request process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process), please assign **nak3** after the PR has been reviewed.
You can assign the PR to them by writing `/assign @nak3` in a comment when ready.

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).

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

- **[OWNERS](https://github.com/knative-sandbox/eventing-kafka-broker/blob/release-1.1/OWNERS)**

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":["nak3"]} -->
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/channel/channel.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1883/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1491392051365810176/artifacts/line-cov.html#file1) | 78.8% | 78.6% | -0.2

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/1883#" title="Author self-approved">knative-prow-robot</a>*, *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/1883#pullrequestreview-877406574" 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.1/OWNERS)~~ [pierDipi]

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":[]} -->
codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1883?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) Report
> :exclamation: No coverage uploaded for pull request base (`release-1.1@a7d6071`). [Click here to learn what that means](https://docs.codecov.io/docs/error-reference?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#section-missing-base-commit).
> The diff coverage is `n/a`.

[![Impacted file tree graph](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1883/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/1883?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)

```diff
@@              Coverage Diff               @@
##             release-1.1    #1883   +/-   ##
==============================================
  Coverage               ?   66.81%           
==============================================
  Files                  ?       52           
  Lines                  ?     2480           
  Branches               ?        0           
==============================================
  Hits                   ?     1657           
  Misses                 ?      649           
  Partials               ?      174           
```



------

[Continue to review full report at Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1883?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/1883?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Last update [a7d6071...6e3c639](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1883?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).


Patch:
diff --git a/control-plane/pkg/reconciler/broker/broker.go b/control-plane/pkg/reconciler/broker/broker.go
index 280dbd7ab..47175d86d 100644
--- a/control-plane/pkg/reconciler/broker/broker.go
+++ b/control-plane/pkg/reconciler/broker/broker.go
@@ -20,6 +20,7 @@ import (
 	"context"
 	"fmt"
 	"strings"
+	"time"
 
 	"go.uber.org/zap"
 	corev1 "k8s.io/api/core/v1"
@@ -299,7 +300,9 @@ func (r *Reconciler) finalizeKind(ctx context.Context, broker *eventing.Broker)
 		},
 	}
 	if status := r.Prober.Probe(ctx, proberAddressable); status != prober.StatusNotReady {
-		return nil // Object will get re-queued once probe status changes.
+		// Return a requeueKeyError that doesn't generate an event and it re-queues the object
+		// for a new reconciliation.
+		return controller.NewRequeueAfter(5 * time.Second)
 	}
 
 	topicConfig, brokerConfig, err := r.topicConfig(ctx, logger, broker)
diff --git a/control-plane/pkg/reconciler/channel/channel.go b/control-plane/pkg/reconciler/channel/channel.go
index 07b334a96..45f78be55 100644
--- a/control-plane/pkg/reconciler/channel/channel.go
+++ b/control-plane/pkg/reconciler/channel/channel.go
@@ -20,6 +20,7 @@ import (
 	"context"
 	"fmt"
 	"strconv"
+	"time"
 
 	"github.com/Shopify/sarama"
 
@@ -352,7 +353,9 @@ func (r *Reconciler) finalizeKind(ctx context.Context, channel *messagingv1beta1
 		},
 	}
 	if status := r.Prober.Probe(ctx, proberAddressable); status != prober.StatusNotReady {
-		return nil // Object will get re-queued once probe status changes.
+		// Return a requeueKeyError that doesn't generate an event and it re-queues the object
+		// for a new reconciliation.
+		return controller.NewRequeueAfter(5 * time.Second)
 	}
 
 	// get the channel configmap
diff --git a/control-plane/pkg/reconciler/sink/kafka_sink.go b/control-plane/pkg/reconciler/sink/kafka_sink.go
index 622f9413c..64e14fe58 100644
--- a/control-plane/pkg/reconciler/sink/kafka_sink.go
+++ b/control-plane/pkg/reconciler/sink/kafka_sink.go
@@ -19,6 +19,7 @@ package sink
 import (
 	"context"
 	"fmt"
+	"time"
 
 	"github.com/Shopify/sarama"
 	"go.uber.org/zap"
@@ -298,7 +299,9 @@ func (r *Reconciler) finalizeKind(ctx context.Context, ks *eventing.KafkaSink) e
 		},
 	}
 	if status := r.Prober.Probe(ctx, proberAddressable); status != prober.StatusNotReady {
-		return nil // Object will get re-queued once probe status changes.
+		// Return a requeueKeyError that doesn't generate an event and it re-queues the object
+		// for a new reconciliation.
+		return controller.NewRequeueAfter(5 * time.Second)
 	}
 
 	if ks.GetStatus().Annotations[base.TopicOwnerAnnotation] == ControllerTopicOwner {


Test patch:
diff --git a/control-plane/pkg/reconciler/broker/broker_test.go b/control-plane/pkg/reconciler/broker/broker_test.go
index b1ae05ea2..dd84a7b62 100644
--- a/control-plane/pkg/reconciler/broker/broker_test.go
+++ b/control-plane/pkg/reconciler/broker/broker_test.go
@@ -1839,6 +1839,36 @@ func brokerFinalization(t *testing.T, format string, env config.Env) {
 				testProber: probertesting.MockProber(prober.StatusNotReady),
 			},
 		},
+		{
+			Name: "Reconciled failed - probe not ready",
+			Objects: []runtime.Object{
+				NewDeletedBroker(),
+				BrokerConfig(bootstrapServers, 20, 5,
+					BrokerConfigFinalizer(ConfigMapFinalizerName),
+				),
+				NewConfigMapFromContract(&contract.Contract{
+					Resources: []*contract.Resource{
+						{
+							Uid:     BrokerUUID,
+							Topics:  []string{BrokerTopic()},
+							Ingress: &contract.Ingress{IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
+						},
+					},
+					Generation: 1,
+				}, &env),
+			},
+			Key: testKey,
+			WantUpdates: []clientgotesting.UpdateActionImpl{
+				ConfigMapUpdate(&env, &contract.Contract{
+					Resources:  []*contract.Resource{},
+					Generation: 2,
+				}),
+			},
+			WantErr: true,
+			OtherTestData: map[string]interface{}{
+				testProber: probertesting.MockProber(prober.StatusReady),
+			},
+		},
 		{
 			Name: "Reconciled normal - with DLS",
 			Objects: []runtime.Object{
diff --git a/control-plane/pkg/reconciler/channel/channel_test.go b/control-plane/pkg/reconciler/channel/channel_test.go
index 438e1d885..0065ab296 100644
--- a/control-plane/pkg/reconciler/channel/channel_test.go
+++ b/control-plane/pkg/reconciler/channel/channel_test.go
@@ -109,7 +109,7 @@ func TestReconcileKind(t *testing.T) {
 			Key:  testKey,
 		},
 		{
-			Name: "Channel is being deleted",
+			Name: "Channel is being deleted, probe not ready",
 			Key:  testKey,
 			Objects: []runtime.Object{
 				NewChannel(
@@ -120,6 +120,26 @@ func TestReconcileKind(t *testing.T) {
 				}),
 				NewConfigMapWithBinaryData(&env, nil),
 			},
+			OtherTestData: map[string]interface{}{
+				testProber: probertesting.MockProber(prober.StatusNotReady),
+			},
+		},
+		{
+			Name: "Channel is being deleted, probe ready",
+			Key:  testKey,
+			Objects: []runtime.Object{
+				NewChannel(
+					WithInitKafkaChannelConditions,
+					WithDeletedTimeStamp),
+				NewConfigMapWithTextData(system.Namespace(), DefaultEnv.GeneralConfigMapName, map[string]string{
+					kafka.BootstrapServersConfigMapKey: ChannelBootstrapServers,
+				}),
+				NewConfigMapWithBinaryData(&env, nil),
+			},
+			WantErr: true,
+			OtherTestData: map[string]interface{}{
+				testProber: probertesting.MockProber(prober.StatusReady),
+			},
 		},
 		{
 			Name: "Reconciled normal - no subscription - no auth",
diff --git a/control-plane/pkg/reconciler/sink/kafka_sink_test.go b/control-plane/pkg/reconciler/sink/kafka_sink_test.go
index 6fb4ebd7f..a184d2966 100644
--- a/control-plane/pkg/reconciler/sink/kafka_sink_test.go
+++ b/control-plane/pkg/reconciler/sink/kafka_sink_test.go
@@ -1021,6 +1021,9 @@ func sinkFinalization(t *testing.T, format string, env config.Env) {
 					Generation: 2,
 				}),
 			},
+			OtherTestData: map[string]interface{}{
+				testProber: probertesting.MockProber(prober.StatusNotReady),
+			},
 		},
 		{
 			Name: "Reconciled normal - topic controlled by us",
@@ -1075,6 +1078,124 @@ func sinkFinalization(t *testing.T, format string, env config.Env) {
 					Generation: 2,
 				}),
 			},
+			OtherTestData: map[string]interface{}{
+				testProber: probertesting.MockProber(prober.StatusNotReady),
+			},
+		},
+		{
+			Name: "Reconciled normal, probe not ready",
+			Objects: []runtime.Object{
+				NewDeletedSink(
+					StatusControllerOwnsTopic(reconciler.ControllerTopicOwner),
+				),
+				NewConfigMapFromContract(&contract.Contract{
+					Resources: []*contract.Resource{
+						{
+							Uid:    SinkUUID + "a",
+							Topics: []string{"topic"},
+							Ingress: &contract.Ingress{
+								IngressType: &contract.Ingress_Path{
+									Path: "path",
+								},
+								ContentMode: contract.ContentMode_STRUCTURED,
+							},
+							BootstrapServers: bootstrapServers,
+						},
+						{
+							Uid:    SinkUUID,
+							Topics: []string{"topic-2"},
+							Ingress: &contract.Ingress{
+								IngressType: &contract.Ingress_Path{
+									Path: "path",
+								},
+								ContentMode: contract.ContentMode_STRUCTURED,
+							},
+							BootstrapServers: bootstrapServers,
+						},
+					},
+					Generation: 1,
+				}, &env),
+			},
+			Key: testKey,
+			WantUpdates: []clientgotesting.UpdateActionImpl{
+				ConfigMapUpdate(&env, &contract.Contract{
+					Resources: []*contract.Resource{
+						{
+							Uid:    SinkUUID + "a",
+							Topics: []string{"topic"},
+							Ingress: &contract.Ingress{
+								IngressType: &contract.Ingress_Path{
+									Path: "path",
+								},
+								ContentMode: contract.ContentMode_STRUCTURED,
+							},
+							BootstrapServers: bootstrapServers,
+						},
+					},
+					Generation: 2,
+				}),
+			},
+			OtherTestData: map[string]interface{}{
+				testProber: probertesting.MockProber(prober.StatusNotReady),
+			},
+		},
+		{
+			Name: "Reconciled failed, probe ready",
+			Objects: []runtime.Object{
+				NewDeletedSink(
+					StatusControllerOwnsTopic(reconciler.ControllerTopicOwner),
+				),
+				NewConfigMapFromContract(&contract.Contract{
+					Resources: []*contract.Resource{
+						{
+							Uid:    SinkUUID + "a",
+							Topics: []string{"topic"},
+							Ingress: &contract.Ingress{
+								IngressType: &contract.Ingress_Path{
+									Path: "path",
+								},
+								ContentMode: contract.ContentMode_STRUCTURED,
+							},
+							BootstrapServers: bootstrapServers,
+						},
+						{
+							Uid:    SinkUUID,
+							Topics: []string{"topic-2"},
+							Ingress: &contract.Ingress{
+								IngressType: &contract.Ingress_Path{
+									Path: "path",
+								},
+								ContentMode: contract.ContentMode_STRUCTURED,
+							},
+							BootstrapServers: bootstrapServers,
+						},
+					},
+					Generation: 1,
+				}, &env),
+			},
+			Key: testKey,
+			WantUpdates: []clientgotesting.UpdateActionImpl{
+				ConfigMapUpdate(&env, &contract.Contract{
+					Resources: []*contract.Resource{
+						{
+							Uid:    SinkUUID + "a",
+							Topics: []string{"topic"},
+							Ingress: &contract.Ingress{
+								IngressType: &contract.Ingress_Path{
+									Path: "path",
+								},
+								ContentMode: contract.ContentMode_STRUCTURED,
+							},
+							BootstrapServers: bootstrapServers,
+						},
+					},
+					Generation: 2,
+				}),
+			},
+			WantErr: true,
+			OtherTestData: map[string]interface{}{
+				testProber: probertesting.MockProber(prober.StatusReady),
+			},
 		},
 		{
 			Name: "Reconciled normal - topic controlled by us - error deleting topic",
diff --git a/test/upgrade-tests.sh b/test/upgrade-tests.sh
index 5973232ae..419b20a4d 100755
--- a/test/upgrade-tests.sh
+++ b/test/upgrade-tests.sh
@@ -28,7 +28,7 @@ fi
 
 set -Eeuo pipefail
 
-TIMEOUT=${TIMEOUT:-60m}
+TIMEOUT=${TIMEOUT:-100m}
 GO_TEST_VERBOSITY="${GO_TEST_VERBOSITY:-standard-verbose}"
 
 go_test_e2e -v \