Backport #1575 (#1602)

Signed-off-by: Francesco Guardiani <francescoguard@gmail.com>

kafka/channel/pkg/apis/messaging/v1alpha1/kafka_channel_conversion.go

@@ -64,13 +64,6 @@ func (source *KafkaChannel) ConvertTo(ctx context.Context, obj apis.Convertible)
 		if source.Spec.Subscribable != nil && len(source.Spec.Subscribable.Subscribers) > 0 {
 			subscribableSpec.Subscribers = make([]eventingduckv1.SubscriberSpec, len(source.Spec.Subscribable.Subscribers))
 			for i, ss := range source.Spec.Subscribable.Subscribers {
-				delivery := &eventingduckv1.DeliverySpec{}
-				if ss.Delivery != nil {
-					if err := ss.Delivery.ConvertTo(ctx, delivery); err != nil {
-						return err
-					}
-				}
-
 				var subscriberURI *apis.URL
 				if ss.SubscriberURI != nil {
 					subscriberURI = ss.SubscriberURI.DeepCopy()
@@ -81,7 +74,13 @@ func (source *KafkaChannel) ConvertTo(ctx context.Context, obj apis.Convertible)
 					Generation:    ss.Generation,
 					SubscriberURI: subscriberURI,
 					ReplyURI:      ss.ReplyURI,
-					Delivery:      delivery,
+				}
+				if ss.Delivery != nil {
+					delivery := &eventingduckv1.DeliverySpec{}
+					if err := ss.Delivery.ConvertTo(ctx, delivery); err != nil {
+						return err
+					}
+					subscribableSpec.Subscribers[i].Delivery = delivery
 				}
 			}
 		}
@@ -147,13 +146,6 @@ func (sink *KafkaChannel) ConvertFrom(ctx context.Context, obj apis.Convertible)
 		if len(source.Spec.SubscribableSpec.Subscribers) > 0 {
 			subscribableSpec.Subscribers = make([]eventingduckv1alpha1.SubscriberSpec, len(source.Spec.SubscribableSpec.Subscribers))
 			for i, ss := range source.Spec.SubscribableSpec.Subscribers {
-				delivery := &eventingduckv1beta1.DeliverySpec{}
-				if ss.Delivery != nil {
-					if err := delivery.ConvertFrom(ctx, ss.Delivery); err != nil {
-						return err
-					}
-				}
-
 				var deadLetterSinkURI *apis.URL
 				if ss.Delivery != nil && ss.Delivery.DeadLetterSink != nil && ss.Delivery.DeadLetterSink.URI != nil {
 					deadLetterSinkURI = ss.Delivery.DeadLetterSink.URI.DeepCopy()
@@ -165,7 +157,14 @@ func (sink *KafkaChannel) ConvertFrom(ctx context.Context, obj apis.Convertible)
 					SubscriberURI:     ss.SubscriberURI.DeepCopy(),
 					ReplyURI:          ss.ReplyURI,
 					DeadLetterSinkURI: deadLetterSinkURI,
-					Delivery:          delivery,
+				}
+
+				if ss.Delivery != nil {
+					delivery := &eventingduckv1beta1.DeliverySpec{}
+					if err := delivery.ConvertFrom(ctx, ss.Delivery); err != nil {
+						return err
+					}
+					subscribableSpec.Subscribers[i].Delivery = delivery
 				}
 			}
 		}