Resolve deadLetterSink uri and set it in the KafkaChannel status (#1126)

The subscription reconciler requires that the resolved
deadLetterSink, if specified, is set in the status.

Signed-off-by: Pierangelo Di Pilato <pdipilat@redhat.com>

Co-authored-by: Pierangelo Di Pilato <pdipilat@redhat.com>

pkg/channel/consolidated/reconciler/controller/controller.go

@@ -41,6 +41,7 @@ import (
 	"knative.dev/pkg/controller"
 	"knative.dev/pkg/logging"
 	knativeReconciler "knative.dev/pkg/reconciler"
+	"knative.dev/pkg/resolver"
 	"knative.dev/pkg/system"
 
 	kafkaChannelClient "knative.dev/eventing-kafka/pkg/client/injection/client"
@@ -104,6 +105,7 @@ func NewController(
 	r.controllerRef = *ownerRef
 
 	impl := kafkaChannelReconciler.NewImpl(ctx, r)
+	r.resolver = resolver.NewURIResolverFromTracker(ctx, impl.Tracker)
 
 	// Call GlobalResync on kafkachannels.
 	grCh := func(interface{}) {

pkg/channel/consolidated/reconciler/controller/kafkachannel.go

@@ -46,6 +46,7 @@ import (
 	"knative.dev/pkg/logging"
 	"knative.dev/pkg/network"
 	pkgreconciler "knative.dev/pkg/reconciler"
+	"knative.dev/pkg/resolver"
 
 	"knative.dev/eventing-kafka/pkg/apis/messaging/v1beta1"
 	"knative.dev/eventing-kafka/pkg/channel/consolidated/reconciler/controller/resources"
@@ -140,6 +141,7 @@ type Reconciler struct {
 	serviceAccountLister corev1listers.ServiceAccountLister
 	roleBindingLister    rbacv1listers.RoleBindingLister
 	controllerRef        metav1.OwnerReference
+	resolver             *resolver.URIResolver
 }
 
 type envConfig struct {
@@ -258,6 +260,10 @@ func (r *Reconciler) ReconcileKind(ctx context.Context, kc *v1beta1.KafkaChannel
 		return fmt.Errorf("error reconciling subscribers %v", err)
 	}
 
+	if err := r.reconcileDeadLetterSink(ctx, kc); err != nil {
+		return fmt.Errorf("failed to reconcile deadLetterSink: %w", err)
+	}
+
 	// Ok, so now the Dispatcher Deployment & Service have been created, we're golden since the
 	// dispatcher watches the Channel and where it needs to dispatch events to.
 	return newReconciledNormal(kc.Namespace, kc.Name)
@@ -690,6 +696,18 @@ func (r *Reconciler) FinalizeKind(ctx context.Context, kc *v1beta1.KafkaChannel)
 	return newReconciledNormal(kc.Namespace, kc.Name) //ok to remove finalizer
 }
 
+func (r *Reconciler) reconcileDeadLetterSink(ctx context.Context, kc *v1beta1.KafkaChannel) error {
+	if kc.Spec.Delivery == nil || kc.Spec.Delivery.DeadLetterSink == nil {
+		return nil
+	}
+	dls, err := r.resolver.URIFromDestinationV1(ctx, *kc.Spec.Delivery.DeadLetterSink, kc)
+	if err != nil {
+		return fmt.Errorf("failed to resolve spec.delivery.deadLetterSink: %w", err)
+	}
+	kc.Status.DeadLetterSinkURI = dls
+	return nil
+}
+
 func findSubscriptionStatus(kc *v1beta1.KafkaChannel, subUID types.UID) *v1.SubscriberStatus {
 	for _, subStatus := range kc.Status.Subscribers {
 		if subStatus.UID == subUID {