Set consumer and consumergroups finalizers when creating them (knativ…

…e-extensions#823)

It is possible that a delete consumer or consumergroup might
be reconciled and never finalized when it is deleted before
the finalizer is set.
This happens because the Knative generated reconciler uses
patch (as opposed to using update) for setting the finalizer
and patch doesn't have any optimistic concurrency controls.

Signed-off-by: Pierangelo Di Pilato <pierdipi@redhat.com>
Co-authored-by: Pierangelo Di Pilato <pierdipi@redhat.com>

control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_group_types.go

@@ -156,12 +156,13 @@ func (c *ConsumerGroup) GetStatus() *duckv1.Status {
 func (cg *ConsumerGroup) ConsumerFromTemplate(options ...ConsumerOption) *Consumer {
 	// TODO figure out naming strategy, is generateName enough?
 	c := &Consumer{
-		ObjectMeta: cg.Spec.Template.ObjectMeta,
-		Spec:       cg.Spec.Template.Spec,
+		ObjectMeta: *cg.Spec.Template.ObjectMeta.DeepCopy(),
+		Spec:       *cg.Spec.Template.Spec.DeepCopy(),
 	}
 
 	ownerRef := metav1.NewControllerRef(cg, ConsumerGroupGroupVersionKind)
 	c.OwnerReferences = append(c.OwnerReferences, *ownerRef)
+	c.Finalizers = []string{"consumers.internal.kafka.eventing.knative.dev"}
 
 	for _, opt := range options {
 		opt(c)

control-plane/pkg/reconciler/source/source.go

@@ -138,6 +138,9 @@ func (r Reconciler) reconcileConsumerGroup(ctx context.Context, ks *sources.Kafk
 			Labels: map[string]string{
 				internalscg.UserFacingResourceLabelSelector: strings.ToLower(ks.GetGroupVersionKind().Kind),
 			},
+			Finalizers: []string{
+				"consumergroups.internal.kafka.eventing.knative.dev",
+			},
 		},
 		Spec: internalscg.ConsumerGroupSpec{
 			Replicas: ks.Spec.Consumers,