Repository: knative-sandbox/eventing-kafka
Pull request #518 (2021-04-08 10:31:33)

[release-0.22] [Consolidated] Rewrote the dispatcher reconciler
This is an automated cherry-pick of #509

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

This pull-request has been approved by:
To complete the [pull request process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process), please assign **aliok** after the PR has been reviewed.
You can assign the PR to them by writing `/assign @aliok` 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).

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

- **[OWNERS](https://github.com/knative-sandbox/eventing-kafka/blob/release-0.22/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":["aliok"]} -->
google-cla[bot]: All (the pull request submitter and all commit authors) CLAs are signed, **but** one or more commits were authored or co-authored by someone other than the pull request submitter.

We need to confirm that all authors are ok with their commits being contributed to this project.  Please have them confirm that by leaving a comment that contains only `@googlebot I consent.` in this pull request.

*Note to project maintainer:* There may be cases where the author cannot leave a comment, or the comment is not properly detected as consent.  In those cases, you can manually confirm consent of the commit author(s), and set the `cla` label to `yes` (if enabled on your project).

ℹ️ **Googlers: [Go here](https://goto.google.com/prinfo/https%3A%2F%2Fgithub.com%2Fknative-sandbox%2Feventing-kafka%2Fpull%2F518) for more info**.

<!-- need_author_consent -->
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/pull/518#pullrequestreview-631234533" title="LGTM">devguyio</a>*
To complete the [pull request process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process), please ask for approval from **slinkydeveloper** after the PR has been reviewed.

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

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

- **[OWNERS](https://github.com/knative-sandbox/eventing-kafka/blob/release-0.22/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":["slinkydeveloper"]} -->
slinkydeveloper: @googlebot I consent.
codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/518?src=pr&el=h1) Report
> :exclamation: No coverage uploaded for pull request base (`release-0.22@2e59cf5`). [Click here to learn what that means](https://docs.codecov.io/docs/error-reference#section-missing-base-commit).
> The diff coverage is `n/a`.

[![Impacted file tree graph](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/518/graphs/tree.svg?width=650&height=150&src=pr&token=W765CT93QC)](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/518?src=pr&el=tree)

```diff
@@               Coverage Diff               @@
##             release-0.22     #518   +/-   ##
===============================================
  Coverage                ?   73.47%           
===============================================
  Files                   ?      147           
  Lines                   ?     5753           
  Branches                ?        0           
===============================================
  Hits                    ?     4227           
  Misses                  ?     1282           
  Partials                ?      244           
```



------

[Continue to review full report at Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/518?src=pr&el=continue).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/518?src=pr&el=footer). Last update [2e59cf5...daefb64](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/518?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).

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

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka/pull/518#pullrequestreview-631234533" title="LGTM">devguyio</a>*, *<a href="https://github.com/knative-sandbox/eventing-kafka/pull/518#pullrequestreview-631246606" title="Approved">matzew</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).

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/blob/release-0.22/OWNERS)~~ [matzew]

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":[]} -->

Patch:
diff --git a/pkg/channel/consolidated/dispatcher/config.go b/pkg/channel/consolidated/dispatcher/config.go
index 1524b42a2..ce2a0800e 100644
--- a/pkg/channel/consolidated/dispatcher/config.go
+++ b/pkg/channel/consolidated/dispatcher/config.go
@@ -16,14 +16,17 @@ limitations under the License.
 
 package dispatcher
 
-type Config struct {
-	// The configuration of each channel in this handler.
-	ChannelConfigs []ChannelConfig
-}
-
 type ChannelConfig struct {
 	Namespace     string
 	Name          string
 	HostName      string
 	Subscriptions []Subscription
 }
+
+func (cc ChannelConfig) SubscriptionsUIDs() []string {
+	res := make([]string, 0, len(cc.Subscriptions))
+	for _, s := range cc.Subscriptions {
+		res = append(res, string(s.UID))
+	}
+	return res
+}
diff --git a/pkg/channel/consolidated/dispatcher/dispatcher.go b/pkg/channel/consolidated/dispatcher/dispatcher.go
index a3927949e..c073a74d7 100644
--- a/pkg/channel/consolidated/dispatcher/dispatcher.go
+++ b/pkg/channel/consolidated/dispatcher/dispatcher.go
@@ -17,11 +17,10 @@ package dispatcher
 
 import (
 	"context"
-	"errors"
 	"fmt"
 	nethttp "net/http"
+	"strings"
 	"sync"
-	"sync/atomic"
 
 	"github.com/Shopify/sarama"
 	protocolkafka "github.com/cloudevents/sdk-go/protocol/kafka_sarama/v2"
@@ -31,15 +30,17 @@ import (
 	"go.uber.org/zap"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/sets"
+	"knative.dev/pkg/logging"
+
+	eventingchannels "knative.dev/eventing/pkg/channel"
+	"knative.dev/eventing/pkg/kncloudevents"
+	"knative.dev/pkg/kmeta"
 
 	"knative.dev/eventing-kafka/pkg/channel/consolidated/utils"
 	"knative.dev/eventing-kafka/pkg/channel/distributed/common/env"
 	"knative.dev/eventing-kafka/pkg/common/client"
 	"knative.dev/eventing-kafka/pkg/common/consumer"
 	"knative.dev/eventing-kafka/pkg/common/tracing"
-	eventingchannels "knative.dev/eventing/pkg/channel"
-	"knative.dev/eventing/pkg/kncloudevents"
-	"knative.dev/pkg/kmeta"
 )
 
 const (
@@ -55,23 +56,23 @@ type KafkaDispatcherArgs struct {
 	KafkaAuthConfig          *client.KafkaAuthConfig
 	SaramaSettingsYamlString string
 	TopicFunc                TopicFunc
-	Logger                   *zap.SugaredLogger
 }
 
 type KafkaDispatcher struct {
-	hostToChannelMap atomic.Value
-	// hostToChannelMapLock is used to update hostToChannelMap
-	hostToChannelMapLock sync.Mutex
-
 	receiver   *eventingchannels.MessageReceiver
 	dispatcher *eventingchannels.MessageDispatcherImpl
 
-	kafkaSyncProducer    sarama.SyncProducer
-	channelSubscriptions map[eventingchannels.ChannelReference]*KafkaSubscription
+	// Receiver data structures
+	// map[string]eventingchannels.ChannelReference
+	hostToChannelMap  sync.Map
+	kafkaSyncProducer sarama.SyncProducer
+
+	// Dispatcher data structures
+	// consumerUpdateLock must be used to update all the below maps
+	consumerUpdateLock   sync.Mutex
+	channelSubscriptions map[types.NamespacedName]*KafkaSubscription
 	subsConsumerGroups   map[types.UID]sarama.ConsumerGroup
 	subscriptions        map[types.UID]Subscription
-	// consumerUpdateLock must be used to update kafkaConsumers
-	consumerUpdateLock   sync.Mutex
 	kafkaConsumerFactory consumer.KafkaConsumerGroupFactory
 
 	topicFunc TopicFunc
@@ -96,30 +97,30 @@ func NewDispatcher(ctx context.Context, args *KafkaDispatcherArgs) (*KafkaDispat
 	}
 
 	dispatcher := &KafkaDispatcher{
-		dispatcher:           eventingchannels.NewMessageDispatcher(args.Logger.Desugar()),
+		dispatcher:           eventingchannels.NewMessageDispatcher(logging.FromContext(ctx).Desugar()),
 		kafkaConsumerFactory: consumer.NewConsumerGroupFactory(args.Brokers, conf),
-		channelSubscriptions: make(map[eventingchannels.ChannelReference]*KafkaSubscription),
+		channelSubscriptions: make(map[types.NamespacedName]*KafkaSubscription),
 		subsConsumerGroups:   make(map[types.UID]sarama.ConsumerGroup),
 		subscriptions:        make(map[types.UID]Subscription),
 		kafkaSyncProducer:    producer,
-		logger:               args.Logger,
+		logger:               logging.FromContext(ctx),
 		topicFunc:            args.TopicFunc,
 	}
 
 	// initialize and start the subscription endpoint server
 	subscriptionEndpoint := &subscriptionEndpoint{
 		dispatcher: dispatcher,
-		logger:     args.Logger,
+		logger:     logging.FromContext(ctx),
 	}
 	go func() {
 		subscriptionEndpoint.start()
 	}()
 
-	podName, err := env.GetRequiredConfigValue(args.Logger.Desugar(), env.PodNameEnvVarKey)
+	podName, err := env.GetRequiredConfigValue(logging.FromContext(ctx).Desugar(), env.PodNameEnvVarKey)
 	if err != nil {
 		return nil, err
 	}
-	containerName, err := env.GetRequiredConfigValue(args.Logger.Desugar(), env.ContainerNameEnvVarKey)
+	containerName, err := env.GetRequiredConfigValue(logging.FromContext(ctx).Desugar(), env.ContainerNameEnvVarKey)
 	if err != nil {
 		return nil, err
 	}
@@ -148,7 +149,7 @@ func NewDispatcher(ctx context.Context, args *KafkaDispatcherArgs) (*KafkaDispat
 
 			return err
 		},
-		args.Logger.Desugar(),
+		logging.FromContext(ctx).Desugar(),
 		reporter,
 		eventingchannels.ResolveMessageChannelFromHostHeader(dispatcher.getChannelReferenceFromHost))
 	if err != nil {
@@ -156,7 +157,6 @@ func NewDispatcher(ctx context.Context, args *KafkaDispatcherArgs) (*KafkaDispat
 	}
 
 	dispatcher.receiver = receiverFunc
-	dispatcher.setHostToChannelMap(map[string]eventingchannels.ChannelReference{})
 	return dispatcher, nil
 }
 
@@ -169,101 +169,152 @@ func (d *KafkaDispatcher) Start(ctx context.Context) error {
 	return d.receiver.Start(ctx)
 }
 
-// UpdateKafkaConsumers will be called by new CRD based kafka channel dispatcher controller.
-func (d *KafkaDispatcher) UpdateKafkaConsumers(config *Config) (map[types.UID]error, error) {
-	if config == nil {
-		return nil, fmt.Errorf("nil config")
+// UpdateError is the error returned from the ReconcileConsumers method, with the details of which
+// subscriptions failed to subscribe to.
+type UpdateError map[types.UID]error
+
+func (k UpdateError) Error() string {
+	errs := make([]string, 0, len(k))
+	for uid, err := range k {
+		errs = append(errs, fmt.Sprintf("subscription %s: %v", uid, err))
 	}
+	return strings.Join(errs, ",")
+}
+
+// ReconcileConsumers will be called by new CRD based kafka channel dispatcher controller.
+func (d *KafkaDispatcher) ReconcileConsumers(config *ChannelConfig) error {
+	channelNamespacedName := types.NamespacedName{
+		Namespace: config.Namespace,
+		Name:      config.Name,
+	}
+
+	// Aux data structures to reconcile
+	toAddSubs := make(map[types.UID]Subscription)
+	toRemoveSubs := sets.NewString()
 
 	d.consumerUpdateLock.Lock()
 	defer d.consumerUpdateLock.Unlock()
 
-	var newSubs []types.UID
-	failedToSubscribe := make(map[types.UID]error)
-	for _, cc := range config.ChannelConfigs {
-		channelRef := eventingchannels.ChannelReference{
-			Name:      cc.Name,
-			Namespace: cc.Namespace,
-		}
-		for _, subSpec := range cc.Subscriptions {
-			newSubs = append(newSubs, subSpec.UID)
-
-			// Check if sub already exists
-			exists := false
-			if _, ok := d.channelSubscriptions[channelRef]; ok {
-				for _, s := range d.channelSubscriptions[channelRef].subs {
-					if s == subSpec.UID {
-						exists = true
-					}
-				}
-			} else { //ensure the pointer is populated or things go boom
-				d.channelSubscriptions[channelRef] = &KafkaSubscription{
-					logger:                    d.logger,
-					subs:                      []types.UID{},
-					channelReadySubscriptions: map[string]sets.Int32{},
-				}
-			}
+	// This loop takes care of filling toAddSubs and toRemoveSubs for new and existing channels
+	thisChannelKafkaSubscriptions := d.channelSubscriptions[channelNamespacedName]
 
-			if !exists {
-				// only subscribe when not exists in channel-subscriptions map
-				// do not need to resubscribe every time channel fanout config is updated
-				if err := d.subscribe(channelRef, subSpec); err != nil {
-					failedToSubscribe[subSpec.UID] = err
-				}
-			}
+	var existingSubsForThisChannel sets.String
+	if thisChannelKafkaSubscriptions != nil {
+		existingSubsForThisChannel = thisChannelKafkaSubscriptions.subs
+	} else {
+		existingSubsForThisChannel = sets.NewString()
+	}
+
+	newSubsForThisChannel := sets.NewString(config.SubscriptionsUIDs()...)
+
+	// toRemoveSubs += existing subs of this channel - new subs of this channel
+	thisChannelToRemoveSubs := existingSubsForThisChannel.Difference(newSubsForThisChannel).UnsortedList()
+	toRemoveSubs.Insert(
+		thisChannelToRemoveSubs...,
+	)
+
+	// toAddSubs += new subs of this channel - existing subs of this channel
+	thisChannelToAddSubs := newSubsForThisChannel.Difference(existingSubsForThisChannel)
+	for _, subSpec := range config.Subscriptions {
+		if thisChannelToAddSubs.Has(string(subSpec.UID)) {
+			toAddSubs[subSpec.UID] = subSpec
 		}
 	}
 
-	d.logger.Debug("Number of new subs", zap.Any("subs", len(newSubs)))
-	d.logger.Debug("Number of subs failed to subscribe", zap.Any("subs", len(failedToSubscribe)))
+	d.logger.Debug("Number of new subs", zap.Any("subs", len(toAddSubs)))
+	d.logger.Debug("Number of old subs", zap.Any("subs", len(toRemoveSubs)))
 
-	// Unsubscribe and close consumer for any deleted subscriptions
-	subsToRemove := make(map[eventingchannels.ChannelReference][]types.UID)
-	for channelRef, actualSubs := range d.channelSubscriptions {
-		subsToRemove[channelRef] = uidSetDifference(actualSubs.subs, newSubs)
+	failedToSubscribe := make(UpdateError)
+	for subUid, subSpec := range toAddSubs {
+		if err := d.subscribe(channelNamespacedName, subSpec); err != nil {
+			failedToSubscribe[subUid] = err
+		}
 	}
+	d.logger.Debug("Number of subs failed to subscribe", zap.Any("subs", len(failedToSubscribe)))
 
-	for channelRef, subs := range subsToRemove {
-		for _, s := range subs {
-			if err := d.unsubscribe(channelRef, d.subscriptions[s]); err != nil {
-				return nil, err
-			}
+	for _, subUid := range toRemoveSubs.UnsortedList() {
+		// We don't signal to the caller the unsubscribe invocation
+		if err := d.unsubscribe(channelNamespacedName, d.subscriptions[types.UID(subUid)]); err != nil {
+			d.logger.Warnw("Error while unsubscribing", zap.Error(err))
 		}
-		d.channelSubscriptions[channelRef].subs = newSubs
 	}
 
-	return failedToSubscribe, nil
+	if len(failedToSubscribe) == 0 {
+		return nil
+	}
+	return failedToSubscribe
 }
 
-// UpdateHostToChannelMap will be called by new CRD based kafka channel dispatcher controller.
-func (d *KafkaDispatcher) UpdateHostToChannelMap(config *Config) error {
-	if config == nil {
-		return errors.New("nil config")
+// RegisterChannelHost adds a new channel to the host-channel mapping.
+func (d *KafkaDispatcher) RegisterChannelHost(channelConfig *ChannelConfig) error {
+	old, ok := d.hostToChannelMap.LoadOrStore(channelConfig.HostName, eventingchannels.ChannelReference{
+		Name:      channelConfig.Name,
+		Namespace: channelConfig.Namespace,
+	})
+	if ok {
+		oldChannelRef := old.(eventingchannels.ChannelReference)
+		if !(oldChannelRef.Namespace == channelConfig.Namespace && oldChannelRef.Name == channelConfig.Name) {
+			// If something is already there, but it's not the same channel, then fail
+			return fmt.Errorf(
+				"duplicate hostName found. Each channel must have a unique host header. HostName:%s, channel:%s.%s, channel:%s.%s",
+				channelConfig.HostName,
+				old.(eventingchannels.ChannelReference).Namespace,
+				old.(eventingchannels.ChannelReference).Name,
+				channelConfig.Namespace,
+				channelConfig.Name,
+			)
+		}
 	}
+	return nil
+}
 
-	d.hostToChannelMapLock.Lock()
-	defer d.hostToChannelMapLock.Unlock()
+func (d *KafkaDispatcher) CleanupChannel(name, namespace, hostname string) error {
+	channelRef := types.NamespacedName{
+		Name:      name,
+		Namespace: namespace,
+	}
 
-	hcMap, err := createHostToChannelMap(config)
-	if err != nil {
-		return err
+	// Remove from the hostToChannel map the mapping with this channel
+	d.hostToChannelMap.Delete(hostname)
+
+	// Remove all subs
+	d.consumerUpdateLock.Lock()
+	defer d.consumerUpdateLock.Unlock()
+
+	if d.channelSubscriptions[channelRef] == nil {
+		// No subs to remove
+		return nil
+	}
+
+	for _, s := range d.channelSubscriptions[channelRef].subs.UnsortedList() {
+		if err := d.unsubscribe(channelRef, d.subscriptions[types.UID(s)]); err != nil {
+			return err
+		}
 	}
 
-	d.setHostToChannelMap(hcMap)
 	return nil
 }
 
 // subscribe reads kafkaConsumers which gets updated in UpdateConfig in a separate go-routine.
 // subscribe must be called under updateLock.
-func (d *KafkaDispatcher) subscribe(channelRef eventingchannels.ChannelReference, sub Subscription) error {
+func (d *KafkaDispatcher) subscribe(channelRef types.NamespacedName, sub Subscription) error {
 	d.logger.Infow("Subscribing to Kafka Channel", zap.Any("channelRef", channelRef), zap.Any("subscription", sub.UID))
+
 	topicName := d.topicFunc(utils.KafkaChannelSeparator, channelRef.Namespace, channelRef.Name)
 	groupID := fmt.Sprintf("kafka.%s.%s.%s", channelRef.Namespace, channelRef.Name, string(sub.UID))
+
+	// Get or create the channel kafka subscription
+	kafkaSubscription, ok := d.channelSubscriptions[channelRef]
+	if !ok {
+		kafkaSubscription = NewKafkaSubscription(d.logger)
+		d.channelSubscriptions[channelRef] = kafkaSubscription
+	}
+
 	handler := &consumerMessageHandler{
 		d.logger,
 		sub,
 		d.dispatcher,
-		d.channelSubscriptions[channelRef],
+		kafkaSubscription,
 		groupID,
 	}
 	d.logger.Debugw("Starting consumer group", zap.Any("channelRef", channelRef),
@@ -284,7 +335,8 @@ func (d *KafkaDispatcher) subscribe(channelRef eventingchannels.ChannelReference
 		}
 	}()
 
-	d.channelSubscriptions[channelRef].subs = append(d.channelSubscriptions[channelRef].subs, sub.UID)
+	// Update the data structures that holds the reconciliation data
+	kafkaSubscription.subs.Insert(string(sub.UID))
 	d.subscriptions[sub.UID] = sub
 	d.subsConsumerGroups[sub.UID] = consumerGroup
 
@@ -293,74 +345,37 @@ func (d *KafkaDispatcher) subscribe(channelRef eventingchannels.ChannelReference
 
 // unsubscribe reads kafkaConsumers which gets updated in UpdateConfig in a separate go-routine.
 // unsubscribe must be called under updateLock.
-func (d *KafkaDispatcher) unsubscribe(channel eventingchannels.ChannelReference, sub Subscription) error {
-	d.logger.Infow("Unsubscribing from channel", zap.Any("channel", channel), zap.Any("subscription", sub.UID))
+func (d *KafkaDispatcher) unsubscribe(channelRef types.NamespacedName, sub Subscription) error {
+	d.logger.Infow("Unsubscribing from channel", zap.Any("channel", channelRef), zap.Any("subscription", sub.UID))
+
+	// Remove the sub spec
 	delete(d.subscriptions, sub.UID)
-	if _, ok := d.channelSubscriptions[channel]; !ok {
+
+	// Remove the sub from the channel
+	kafkaSubscription, ok := d.channelSubscriptions[channelRef]
+	if !ok {
+		// If this happens, then there's a bug somewhere...
 		return nil
 	}
-	if subsSlice := d.channelSubscriptions[channel].subs; subsSlice != nil {
-		var newSlice []types.UID
-		for _, oldSub := range subsSlice {
-			if oldSub != sub.UID {
-				newSlice = append(newSlice, oldSub)
-			}
-		}
-		d.channelSubscriptions[channel].subs = newSlice
+	kafkaSubscription.subs.Delete(string(sub.UID))
+	if kafkaSubscription.subs.Len() == 0 {
+		// We can get rid of this
+		delete(d.channelSubscriptions, channelRef)
 	}
-	if consumer, ok := d.subsConsumerGroups[sub.UID]; ok {
+
+	// Delete the consumer group
+	if consumerGroup, ok := d.subsConsumerGroups[sub.UID]; ok {
 		delete(d.subsConsumerGroups, sub.UID)
-		d.logger.Debugw("Closing cached consumer group", zap.Any("consumer group", consumer))
-		return consumer.Close()
+		d.logger.Debugw("Closing cached consumerGroup group", zap.Any("consumer group", consumerGroup))
+		return consumerGroup.Close()
 	}
 	return nil
 }
 
-func (d *KafkaDispatcher) getHostToChannelMap() map[string]eventingchannels.ChannelReference {
-	return d.hostToChannelMap.Load().(map[string]eventingchannels.ChannelReference)
-}
-
-func (d *KafkaDispatcher) setHostToChannelMap(hcMap map[string]eventingchannels.ChannelReference) {
-	d.hostToChannelMap.Store(hcMap)
-}
-
 func (d *KafkaDispatcher) getChannelReferenceFromHost(host string) (eventingchannels.ChannelReference, error) {
-	chMap := d.getHostToChannelMap()
-	cr, ok := chMap[host]
+	cr, ok := d.hostToChannelMap.Load(host)
 	if !ok {
-		return cr, eventingchannels.UnknownHostError(host)
-	}
-	return cr, nil
-}
-
-func uidSetDifference(a, b []types.UID) (diff []types.UID) {
-	m := make(map[types.UID]bool)
-
-	for _, item := range b {
-		m[item] = true
-	}
-
-	for _, item := range a {
-		if _, ok := m[item]; !ok {
-			diff = append(diff, item)
-		}
-	}
-	return
-}
-
-func createHostToChannelMap(config *Config) (map[string]eventingchannels.ChannelReference, error) {
-	hcMap := make(map[string]eventingchannels.ChannelReference, len(config.ChannelConfigs))
-	for _, cConfig := range config.ChannelConfigs {
-		if cr, ok := hcMap[cConfig.HostName]; ok {
-			return nil, fmt.Errorf(
-				"duplicate hostName found. Each channel must have a unique host header. HostName:%s, channel:%s.%s, channel:%s.%s",
-				cConfig.HostName,
-				cConfig.Namespace,
-				cConfig.Name,
-				cr.Namespace,
-				cr.Name)
-		}
-		hcMap[cConfig.HostName] = eventingchannels.ChannelReference{Name: cConfig.Name, Namespace: cConfig.Namespace}
+		return eventingchannels.ChannelReference{}, eventingchannels.UnknownHostError(host)
 	}
-	return hcMap, nil
+	return cr.(eventingchannels.ChannelReference), nil
 }
diff --git a/pkg/channel/consolidated/dispatcher/kafka_subscription.go b/pkg/channel/consolidated/dispatcher/kafka_subscription.go
index 8f54c4a77..6cd088a96 100644
--- a/pkg/channel/consolidated/dispatcher/kafka_subscription.go
+++ b/pkg/channel/consolidated/dispatcher/kafka_subscription.go
@@ -27,12 +27,20 @@ import (
 
 type KafkaSubscription struct {
 	logger *zap.SugaredLogger
-	subs   []types.UID
+	subs   sets.String
 	// readySubscriptionsLock must be used to synchronize access to channelReadySubscriptions
 	readySubscriptionsLock    sync.RWMutex
 	channelReadySubscriptions map[string]sets.Int32
 }
 
+func NewKafkaSubscription(logger *zap.SugaredLogger) *KafkaSubscription {
+	return &KafkaSubscription{
+		logger:                    logger,
+		subs:                      sets.NewString(),
+		channelReadySubscriptions: map[string]sets.Int32{},
+	}
+}
+
 // SetReady will mark the subid in the KafkaSubscription and call any registered callbacks
 func (ks *KafkaSubscription) SetReady(subID types.UID, partition int32, ready bool) {
 	ks.logger.Debugw("Setting subscription readiness", zap.Any("subscription", subID), zap.Bool("ready", ready))
diff --git a/pkg/channel/consolidated/dispatcher/subscription_endpoint.go b/pkg/channel/consolidated/dispatcher/subscription_endpoint.go
index 3612e4464..942d9e26c 100644
--- a/pkg/channel/consolidated/dispatcher/subscription_endpoint.go
+++ b/pkg/channel/consolidated/dispatcher/subscription_endpoint.go
@@ -6,7 +6,7 @@ import (
 	"strings"
 
 	"go.uber.org/zap"
-	eventingchannels "knative.dev/eventing/pkg/channel"
+	"k8s.io/apimachinery/pkg/types"
 )
 
 // subscriptionEndpoint is serving the subscription status of the Kafka channel.
@@ -29,7 +29,7 @@ func (d *subscriptionEndpoint) ServeHTTP(w nethttp.ResponseWriter, r *nethttp.Re
 		return
 	}
 	channelRefNamespace, channelRefName := uriSplit[1], uriSplit[2]
-	channelRef := eventingchannels.ChannelReference{
+	channelRef := types.NamespacedName{
 		Name:      channelRefName,
 		Namespace: channelRefNamespace,
 	}
diff --git a/pkg/channel/consolidated/reconciler/dispatcher/kafkachannel.go b/pkg/channel/consolidated/reconciler/dispatcher/kafkachannel.go
index e55d10d56..119dbb419 100644
--- a/pkg/channel/consolidated/reconciler/dispatcher/kafkachannel.go
+++ b/pkg/channel/consolidated/reconciler/dispatcher/kafkachannel.go
@@ -21,9 +21,9 @@ import (
 	"fmt"
 
 	"go.uber.org/zap"
-	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/client-go/kubernetes/scheme"
 	"k8s.io/client-go/tools/cache"
+
 	"knative.dev/eventing-kafka/pkg/common/constants"
 
 	"knative.dev/eventing/pkg/apis/eventing"
@@ -64,7 +64,6 @@ type Reconciler struct {
 	impl                 *controller.Impl
 }
 
-// Check that our Reconciler implements controller.Reconciler.
 var _ kafkachannelreconciler.Interface = (*Reconciler)(nil)
 
 // NewController initializes the controller and is called by the generated code.
@@ -102,7 +101,6 @@ func NewController(ctx context.Context, cmw configmap.Watcher) *controller.Impl
 		KafkaAuthConfig:          kafkaAuthCfg,
 		SaramaSettingsYamlString: kafkaConfig.SaramaSettingsYamlString,
 		TopicFunc:                utils.TopicName,
-		Logger:                   logger,
 	}
 	kafkaDispatcher, err := dispatcher.NewDispatcher(ctx, args)
 	if err != nil {
@@ -127,7 +125,16 @@ func NewController(ctx context.Context, cmw configmap.Watcher) *controller.Impl
 	kafkaChannelInformer.Informer().AddEventHandler(
 		cache.FilteringResourceEventHandler{
 			FilterFunc: filterWithAnnotation(injection.HasNamespaceScope(ctx)),
-			Handler:    controller.HandleAll(r.impl.Enqueue),
+			Handler: cache.ResourceEventHandlerFuncs{
+				AddFunc:    r.impl.Enqueue,
+				UpdateFunc: controller.PassNew(r.impl.Enqueue),
+				DeleteFunc: func(obj interface{}) {
+					// TODO when finalize kind is fixed, we'll need to handle that error properly
+					if err := r.CleanupChannel(obj.(*v1beta1.KafkaChannel)); err != nil {
+						logger.Warnw("Unable to remove kafka channel", zap.Any("kafkachannel", obj), zap.Error(err))
+					}
+				},
+			},
 		})
 
 	logger.Info("Starting dispatcher.")
@@ -149,50 +156,43 @@ func filterWithAnnotation(namespaced bool) func(obj interface{}) bool {
 
 func (r *Reconciler) ReconcileKind(ctx context.Context, kc *v1beta1.KafkaChannel) pkgreconciler.Event {
 	logging.FromContext(ctx).Debugw("ReconcileKind for channel", zap.String("channel", kc.Name))
-	return r.syncDispatcher(ctx)
+	return r.syncChannel(ctx, kc)
 }
 
 func (r *Reconciler) ObserveKind(ctx context.Context, kc *v1beta1.KafkaChannel) pkgreconciler.Event {
 	logging.FromContext(ctx).Debugw("ObserveKind for channel", zap.String("channel", kc.Name))
-	return r.syncDispatcher(ctx)
+	return r.syncChannel(ctx, kc)
 }
 
-func (r *Reconciler) syncDispatcher(ctx context.Context) pkgreconciler.Event {
-	channels, err := r.kafkachannelLister.List(labels.Everything())
-	if err != nil {
-		logging.FromContext(ctx).Error("Error listing kafka channels")
-		return err
+func (r *Reconciler) syncChannel(ctx context.Context, kc *v1beta1.KafkaChannel) pkgreconciler.Event {
+	if !kc.Status.IsReady() {
+		logging.FromContext(ctx).Debugw("KafkaChannel still not ready, short-circuiting the reconciler", zap.String("channel", kc.Name))
+		return nil
 	}
 
-	// TODO: revisit this code. Instead of reading all channels and updating consumers and hostToChannel map for all
-	// why not just reconcile the current channel. With this the UpdateKafkaConsumers can now return SubscribableStatus
-	// for the subscriptions on the channel that is being reconciled.
-	kafkaChannels := make([]*v1beta1.KafkaChannel, 0)
-	for _, channel := range channels {
-		if channel.Status.IsReady() {
-			kafkaChannels = append(kafkaChannels, channel)
-		}
-	}
-	config := r.newConfigFromKafkaChannels(kafkaChannels)
-	if err := r.kafkaDispatcher.UpdateHostToChannelMap(config); err != nil {
+	config := r.newConfigFromKafkaChannel(kc)
+
+	// Update receiver side
+	if err := r.kafkaDispatcher.RegisterChannelHost(config); err != nil {
 		logging.FromContext(ctx).Error("Error updating host to channel map in dispatcher")
 		return err
 	}
 
-	failedSubscriptions, err := r.kafkaDispatcher.UpdateKafkaConsumers(config)
+	// Update dispatcher side
+	err := r.kafkaDispatcher.ReconcileConsumers(config)
 	if err != nil {
-		logging.FromContext(ctx).Error("Error updating kafka consumers in dispatcher")
-		return err
-	}
-	if len(failedSubscriptions) > 0 {
-		logging.FromContext(ctx).Error("Some kafka subscriptions failed to subscribe")
-		return fmt.Errorf("Some kafka subscriptions failed to subscribe")
+		logging.FromContext(ctx).Errorw("Some kafka subscriptions failed to subscribe", zap.Error(err))
+		return fmt.Errorf("some kafka subscriptions failed to subscribe: %v", err)
 	}
 	return nil
 }
 
-// newConfigFromKafkaChannels creates a new Config from the list of kafka channels.
-func (r *Reconciler) newChannelConfigFromKafkaChannel(c *v1beta1.KafkaChannel) *dispatcher.ChannelConfig {
+func (r *Reconciler) CleanupChannel(kc *v1beta1.KafkaChannel) pkgreconciler.Event {
+	return r.kafkaDispatcher.CleanupChannel(kc.Name, kc.Namespace, kc.Status.Address.URL.Host)
+}
+
+// newConfigFromKafkaChannel creates a new Config from the list of kafka channels.
+func (r *Reconciler) newConfigFromKafkaChannel(c *v1beta1.KafkaChannel) *dispatcher.ChannelConfig {
 	channelConfig := dispatcher.ChannelConfig{
 		Namespace: c.Namespace,
 		Name:      c.Name,
@@ -213,15 +213,3 @@ func (r *Reconciler) newChannelConfigFromKafkaChannel(c *v1beta1.KafkaChannel) *
 
 	return &channelConfig
 }
-
-// newConfigFromKafkaChannels creates a new Config from the list of kafka channels.
-func (r *Reconciler) newConfigFromKafkaChannels(channels []*v1beta1.KafkaChannel) *dispatcher.Config {
-	cc := make([]dispatcher.ChannelConfig, 0)
-	for _, c := range channels {
-		channelConfig := r.newChannelConfigFromKafkaChannel(c)
-		cc = append(cc, *channelConfig)
-	}
-	return &dispatcher.Config{
-		ChannelConfigs: cc,
-	}
-}


Test patch:
diff --git a/pkg/channel/consolidated/dispatcher/dispatcher_it_test.go b/pkg/channel/consolidated/dispatcher/dispatcher_it_test.go
index 8e89ed9e5..fc87e7f3e 100644
--- a/pkg/channel/consolidated/dispatcher/dispatcher_it_test.go
+++ b/pkg/channel/consolidated/dispatcher/dispatcher_it_test.go
@@ -27,6 +27,8 @@ import (
 	"testing"
 	"time"
 
+	"github.com/stretchr/testify/require"
+
 	"knative.dev/eventing-kafka/pkg/common/constants"
 
 	"github.com/cloudevents/sdk-go/v2/binding"
@@ -76,7 +78,6 @@ func TestDispatcher(t *testing.T) {
 		ClientID:  "testing",
 		Brokers:   []string{"localhost:9092"},
 		TopicFunc: utils.TopicName,
-		Logger:    logger.Sugar(),
 	}
 
 	// Create the dispatcher. At this point, if Kafka is not up, this thing fails
@@ -156,58 +157,46 @@ func TestDispatcher(t *testing.T) {
 	)
 
 	// send -> channela -> sub with transformationServer and reply to channelb -> channelb -> sub with receiver -> receiver
-	config := Config{
-		ChannelConfigs: []ChannelConfig{
+	channelAConfig := &ChannelConfig{
+		Namespace: "default",
+		Name:      "channela",
+		HostName:  "channela.svc",
+		Subscriptions: []Subscription{
 			{
-				Namespace: "default",
-				Name:      "channela",
-				HostName:  "channela.svc",
-				Subscriptions: []Subscription{
-					{
-						UID: "aaaa",
-						Subscription: fanout.Subscription{
-							Subscriber: mustParseUrl(t, transformationsServer.URL),
-							Reply:      mustParseUrl(t, channelBProxy.URL),
-						},
-					},
-					{
-						UID: "cccc",
-						Subscription: fanout.Subscription{
-							Subscriber: mustParseUrl(t, transformationsFailureServer.URL),
-							Reply:      mustParseUrl(t, channelBProxy.URL),
-							DeadLetter: mustParseUrl(t, deadLetterServer.URL),
-						},
-					},
+				UID: "aaaa",
+				Subscription: fanout.Subscription{
+					Subscriber: mustParseUrl(t, transformationsServer.URL),
+					Reply:      mustParseUrl(t, channelBProxy.URL),
 				},
 			},
 			{
-				Namespace: "default",
-				Name:      "channelb",
-				HostName:  "channelb.svc",
-				Subscriptions: []Subscription{
-					{
-						UID: "bbbb",
-						Subscription: fanout.Subscription{
-							Subscriber: mustParseUrl(t, receiverServer.URL),
-						},
-					},
+				UID: "cccc",
+				Subscription: fanout.Subscription{
+					Subscriber: mustParseUrl(t, transformationsFailureServer.URL),
+					Reply:      mustParseUrl(t, channelBProxy.URL),
+					DeadLetter: mustParseUrl(t, deadLetterServer.URL),
 				},
 			},
 		},
 	}
+	require.NoError(t, dispatcher.RegisterChannelHost(channelAConfig))
+	require.NoError(t, dispatcher.ReconcileConsumers(channelAConfig))
 
-	err = dispatcher.UpdateHostToChannelMap(&config)
-	if err != nil {
-		t.Fatal(err)
-	}
-
-	failed, err := dispatcher.UpdateKafkaConsumers(&config)
-	if err != nil {
-		t.Fatal(err)
-	}
-	if len(failed) != 0 {
-		t.Fatal(err)
+	channelBConfig := &ChannelConfig{
+		Namespace: "default",
+		Name:      "channelb",
+		HostName:  "channelb.svc",
+		Subscriptions: []Subscription{
+			{
+				UID: "bbbb",
+				Subscription: fanout.Subscription{
+					Subscriber: mustParseUrl(t, receiverServer.URL),
+				},
+			},
+		},
 	}
+	require.NoError(t, dispatcher.RegisterChannelHost(channelBConfig))
+	require.NoError(t, dispatcher.ReconcileConsumers(channelBConfig))
 
 	time.Sleep(5 * time.Second)
 
@@ -240,18 +229,8 @@ func TestDispatcher(t *testing.T) {
 	receiverWg.Wait()
 
 	// Try to close consumer groups
-	err = dispatcher.UpdateHostToChannelMap(&Config{})
-	if err != nil {
-		t.Fatal(err)
-	}
-
-	failed, err = dispatcher.UpdateKafkaConsumers(&Config{})
-	if err != nil {
-		t.Fatal(err)
-	}
-	if len(failed) != 0 {
-		t.Fatal(err)
-	}
+	require.NoError(t, dispatcher.CleanupChannel("channela", "default", "channela.svc"))
+	require.NoError(t, dispatcher.CleanupChannel("channelb", "default", "channelb.svc"))
 }
 
 func createReverseProxy(t *testing.T, host string) *httputil.ReverseProxy {
diff --git a/pkg/channel/consolidated/dispatcher/dispatcher_test.go b/pkg/channel/consolidated/dispatcher/dispatcher_test.go
index 39ce1a12f..a93ff2d81 100644
--- a/pkg/channel/consolidated/dispatcher/dispatcher_test.go
+++ b/pkg/channel/consolidated/dispatcher/dispatcher_test.go
@@ -20,11 +20,14 @@ import (
 	"context"
 	"errors"
 	"net/url"
+	"sync"
 	"testing"
 
 	"github.com/Shopify/sarama"
 	"github.com/google/go-cmp/cmp"
 	"github.com/google/go-cmp/cmp/cmpopts"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 	"go.uber.org/zap"
 	"go.uber.org/zap/zaptest"
 	"k8s.io/apimachinery/pkg/types"
@@ -73,30 +76,69 @@ var _ sarama.ConsumerGroup = (*mockConsumerGroup)(nil)
 
 // ----- Tests
 
-// test util for various config checks
-func (d *KafkaDispatcher) checkConfigAndUpdate(config *Config) error {
-	if config == nil {
-		return errors.New("nil config")
+func (d *KafkaDispatcher) getHostToChannelMap() map[string]eventingchannels.ChannelReference {
+	m := make(map[string]eventingchannels.ChannelReference)
+	d.hostToChannelMap.Range(func(key, value interface{}) bool {
+		m[key.(string)] = value.(eventingchannels.ChannelReference)
+		return true
+	})
+	return m
+}
+
+func TestKafkaDispatcher_RegisterChannelHost(t *testing.T) {
+	firstChannelConfig := &ChannelConfig{
+		Namespace: "default",
+		Name:      "test-channel-1",
+		HostName:  "a.b.c.d",
+	}
+	secondChannelConfig := &ChannelConfig{
+		Namespace: "default",
+		Name:      "test-channel-2",
+		HostName:  "a.b.c.d",
 	}
 
-	if _, err := d.UpdateKafkaConsumers(config); err != nil {
-		// failed to update dispatchers consumers
-		return err
+	d := &KafkaDispatcher{
+		kafkaConsumerFactory: &mockKafkaConsumerFactory{},
+		channelSubscriptions: make(map[types.NamespacedName]*KafkaSubscription),
+		subsConsumerGroups:   make(map[types.UID]sarama.ConsumerGroup),
+		subscriptions:        make(map[types.UID]Subscription),
+		topicFunc:            utils.TopicName,
+		logger:               zaptest.NewLogger(t).Sugar(),
 	}
-	if err := d.UpdateHostToChannelMap(config); err != nil {
-		return err
+
+	require.NoError(t, d.RegisterChannelHost(firstChannelConfig))
+	require.Error(t, d.RegisterChannelHost(secondChannelConfig))
+}
+
+func TestKafkaDispatcher_RegisterSameChannelTwiceShouldNotFail(t *testing.T) {
+	channelConfig := &ChannelConfig{
+		Namespace: "default",
+		Name:      "test-channel-1",
+		HostName:  "a.b.c.d",
 	}
 
-	return nil
+	d := &KafkaDispatcher{
+		kafkaConsumerFactory: &mockKafkaConsumerFactory{},
+		channelSubscriptions: make(map[types.NamespacedName]*KafkaSubscription),
+		subsConsumerGroups:   make(map[types.UID]sarama.ConsumerGroup),
+		subscriptions:        make(map[types.UID]Subscription),
+		topicFunc:            utils.TopicName,
+		logger:               zaptest.NewLogger(t).Sugar(),
+	}
+
+	require.NoError(t, d.RegisterChannelHost(channelConfig))
+	require.Contains(t, d.getHostToChannelMap(), "a.b.c.d")
+	require.NoError(t, d.RegisterChannelHost(channelConfig))
+	require.Contains(t, d.getHostToChannelMap(), "a.b.c.d")
 }
 
-func TestDispatcher_UpdateConfig(t *testing.T) {
+func TestDispatcher_UpdateConsumers(t *testing.T) {
 	subscriber, _ := url.Parse("http://test/subscriber")
 
 	testCases := []struct {
 		name             string
-		oldConfig        *Config
-		newConfig        *Config
+		oldConfig        *ChannelConfig
+		newConfig        *ChannelConfig
 		subscribes       []string
 		unsubscribes     []string
 		createErr        string
@@ -105,7 +147,7 @@ func TestDispatcher_UpdateConfig(t *testing.T) {
 	}{
 		{
 			name:             "nil config",
-			oldConfig:        &Config{},
+			oldConfig:        &ChannelConfig{},
 			newConfig:        nil,
 			createErr:        "nil config",
 			oldHostToChanMap: map[string]eventingchannels.ChannelReference{},
@@ -113,22 +155,18 @@ func TestDispatcher_UpdateConfig(t *testing.T) {
 		},
 		{
 			name:             "same config",
-			oldConfig:        &Config{},
-			newConfig:        &Config{},
+			oldConfig:        &ChannelConfig{},
+			newConfig:        &ChannelConfig{},
 			oldHostToChanMap: map[string]eventingchannels.ChannelReference{},
 			newHostToChanMap: map[string]eventingchannels.ChannelReference{},
 		},
 		{
 			name:      "config with no subscription",
-			oldConfig: &Config{},
-			newConfig: &Config{
-				ChannelConfigs: []ChannelConfig{
-					{
-						Namespace: "default",
-						Name:      "test-channel",
-						HostName:  "a.b.c.d",
-					},
-				},
+			oldConfig: &ChannelConfig{},
+			newConfig: &ChannelConfig{
+				Namespace: "default",
+				Name:      "test-channel",
+				HostName:  "a.b.c.d",
 			},
 			oldHostToChanMap: map[string]eventingchannels.ChannelReference{},
 			newHostToChanMap: map[string]eventingchannels.ChannelReference{
@@ -137,26 +175,22 @@ func TestDispatcher_UpdateConfig(t *testing.T) {
 		},
 		{
 			name:      "single channel w/ new subscriptions",
-			oldConfig: &Config{},
-			newConfig: &Config{
-				ChannelConfigs: []ChannelConfig{
+			oldConfig: &ChannelConfig{},
+			newConfig: &ChannelConfig{
+				Namespace: "default",
+				Name:      "test-channel",
+				HostName:  "a.b.c.d",
+				Subscriptions: []Subscription{
+					{
+						UID: "subscription-1",
+						Subscription: fanout.Subscription{
+							Subscriber: subscriber,
+						},
+					},
 					{
-						Namespace: "default",
-						Name:      "test-channel",
-						HostName:  "a.b.c.d",
-						Subscriptions: []Subscription{
-							{
-								UID: "subscription-1",
-								Subscription: fanout.Subscription{
-									Subscriber: subscriber,
-								},
-							},
-							{
-								UID: "subscription-2",
-								Subscription: fanout.Subscription{
-									Subscriber: subscriber,
-								},
-							},
+						UID: "subscription-2",
+						Subscription: fanout.Subscription{
+							Subscriber: subscriber,
 						},
 					},
 				},
@@ -169,173 +203,72 @@ func TestDispatcher_UpdateConfig(t *testing.T) {
 		},
 		{
 			name: "single channel w/ existing subscriptions",
-			oldConfig: &Config{
-				ChannelConfigs: []ChannelConfig{
-					{
-						Namespace: "default",
-						Name:      "test-channel",
-						HostName:  "a.b.c.d",
-						Subscriptions: []Subscription{
-							{
-								UID: "subscription-1",
-								Subscription: fanout.Subscription{
-									Subscriber: subscriber,
-								},
-							},
-							{
-								UID: "subscription-2",
-								Subscription: fanout.Subscription{
-									Subscriber: subscriber,
-								},
-							},
-						},
-					},
-				},
-			},
-			newConfig: &Config{
-				ChannelConfigs: []ChannelConfig{
+			oldConfig: &ChannelConfig{
+				Namespace: "default",
+				Name:      "test-channel",
+				HostName:  "a.b.c.d",
+				Subscriptions: []Subscription{
 					{
-						Namespace: "default",
-						Name:      "test-channel",
-						HostName:  "a.b.c.d",
-						Subscriptions: []Subscription{
-							{
-								UID: "subscription-2",
-								Subscription: fanout.Subscription{
-									Subscriber: subscriber,
-								},
-							},
-							{
-								UID: "subscription-3",
-								Subscription: fanout.Subscription{
-									Subscriber: subscriber,
-								},
-							},
+						UID: "subscription-1",
+						Subscription: fanout.Subscription{
+							Subscriber: subscriber,
 						},
 					},
-				},
-			},
-			subscribes:   []string{"subscription-2", "subscription-3"},
-			unsubscribes: []string{"subscription-1"},
-			oldHostToChanMap: map[string]eventingchannels.ChannelReference{
-				"a.b.c.d": {Name: "test-channel", Namespace: "default"},
-			},
-			newHostToChanMap: map[string]eventingchannels.ChannelReference{
-				"a.b.c.d": {Name: "test-channel", Namespace: "default"},
-			},
-		},
-		{
-			name: "multi channel w/old and new subscriptions",
-			oldConfig: &Config{
-				ChannelConfigs: []ChannelConfig{
 					{
-						Namespace: "default",
-						Name:      "test-channel-1",
-						HostName:  "a.b.c.d",
-						Subscriptions: []Subscription{
-							{
-								UID: "subscription-1",
-								Subscription: fanout.Subscription{
-									Subscriber: subscriber,
-								},
-							},
-							{
-								UID: "subscription-2",
-								Subscription: fanout.Subscription{
-									Subscriber: subscriber,
-								},
-							},
+						UID: "subscription-2",
+						Subscription: fanout.Subscription{
+							Subscriber: subscriber,
 						},
 					},
 				},
 			},
-			newConfig: &Config{
-				ChannelConfigs: []ChannelConfig{
+			newConfig: &ChannelConfig{
+				Namespace: "default",
+				Name:      "test-channel",
+				HostName:  "a.b.c.d",
+				Subscriptions: []Subscription{
 					{
-						Namespace: "default",
-						Name:      "test-channel-1",
-						HostName:  "a.b.c.d",
-						Subscriptions: []Subscription{
-							{
-								UID: "subscription-1",
-								Subscription: fanout.Subscription{
-									Subscriber: subscriber,
-								},
-							},
+						UID: "subscription-2",
+						Subscription: fanout.Subscription{
+							Subscriber: subscriber,
 						},
 					},
 					{
-						Namespace: "default",
-						Name:      "test-channel-2",
-						HostName:  "e.f.g.h",
-						Subscriptions: []Subscription{
-							{
-								UID: "subscription-3",
-								Subscription: fanout.Subscription{
-									Subscriber: subscriber,
-								},
-							},
-							{
-								UID: "subscription-4",
-								Subscription: fanout.Subscription{
-									Subscriber: subscriber,
-								},
-							},
+						UID: "subscription-3",
+						Subscription: fanout.Subscription{
+							Subscriber: subscriber,
 						},
 					},
 				},
 			},
-			subscribes:   []string{"subscription-1", "subscription-3", "subscription-4"},
-			unsubscribes: []string{"subscription-2"},
+			subscribes:   []string{"subscription-2", "subscription-3"},
+			unsubscribes: []string{"subscription-1"},
 			oldHostToChanMap: map[string]eventingchannels.ChannelReference{
-				"a.b.c.d": {Name: "test-channel-1", Namespace: "default"},
+				"a.b.c.d": {Name: "test-channel", Namespace: "default"},
 			},
 			newHostToChanMap: map[string]eventingchannels.ChannelReference{
-				"a.b.c.d": {Name: "test-channel-1", Namespace: "default"},
-				"e.f.g.h": {Name: "test-channel-2", Namespace: "default"},
+				"a.b.c.d": {Name: "test-channel", Namespace: "default"},
 			},
 		},
-		{
-			name:      "Duplicate hostnames",
-			oldConfig: &Config{},
-			newConfig: &Config{
-				ChannelConfigs: []ChannelConfig{
-					{
-						Namespace: "default",
-						Name:      "test-channel-1",
-						HostName:  "a.b.c.d",
-					},
-					{
-						Namespace: "default",
-						Name:      "test-channel-2",
-						HostName:  "a.b.c.d",
-					},
-				},
-			},
-			createErr:        "duplicate hostName found. Each channel must have a unique host header. HostName:a.b.c.d, channel:default.test-channel-2, channel:default.test-channel-1",
-			oldHostToChanMap: map[string]eventingchannels.ChannelReference{},
-		},
 	}
 
 	for _, tc := range testCases {
 		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
 			t.Logf("Running %s", t.Name())
 			d := &KafkaDispatcher{
 				kafkaConsumerFactory: &mockKafkaConsumerFactory{},
-				channelSubscriptions: make(map[eventingchannels.ChannelReference]*KafkaSubscription),
+				channelSubscriptions: make(map[types.NamespacedName]*KafkaSubscription),
 				subsConsumerGroups:   make(map[types.UID]sarama.ConsumerGroup),
 				subscriptions:        make(map[types.UID]Subscription),
 				topicFunc:            utils.TopicName,
 				logger:               zaptest.NewLogger(t).Sugar(),
 			}
-			d.setHostToChannelMap(map[string]eventingchannels.ChannelReference{})
 
 			// Initialize using oldConfig
-			err := d.checkConfigAndUpdate(tc.oldConfig)
-			if err != nil {
+			require.NoError(t, d.RegisterChannelHost(tc.oldConfig))
+			require.NoError(t, d.ReconcileConsumers(tc.oldConfig))
 
-				t.Errorf("unexpected error: %v", err)
-			}
 			oldSubscribers := sets.NewString()
 			for _, sub := range d.subscriptions {
 				oldSubscribers.Insert(string(sub.UID))
@@ -348,7 +281,7 @@ func TestDispatcher_UpdateConfig(t *testing.T) {
 			}
 
 			// Update with new config
-			err = d.checkConfigAndUpdate(tc.newConfig)
+			err := d.ReconcileConsumers(tc.newConfig)
 			if tc.createErr != "" {
 				if err == nil {
 					t.Errorf("Expected UpdateConfig error: '%v'. Actual nil", tc.createErr)
@@ -376,15 +309,157 @@ func TestDispatcher_UpdateConfig(t *testing.T) {
 	}
 }
 
+func TestDispatcher_MultipleChannelsInParallel(t *testing.T) {
+	subscriber, _ := url.Parse("http://test/subscriber")
+
+	configs := []*ChannelConfig{
+		{
+			Namespace: "default",
+			Name:      "test-channel",
+			HostName:  "a.b.c.d",
+		},
+		{
+			Namespace: "default",
+			Name:      "test-channel-1",
+			HostName:  "x.y.w.z",
+			Subscriptions: []Subscription{
+				{
+					UID: "subscription-1",
+					Subscription: fanout.Subscription{
+						Subscriber: subscriber,
+					},
+				},
+			},
+		},
+		{
+			Namespace: "default",
+			Name:      "test-channel-2",
+			HostName:  "e.f.g.h",
+			Subscriptions: []Subscription{
+				{
+					UID: "subscription-3",
+					Subscription: fanout.Subscription{
+						Subscriber: subscriber,
+					},
+				},
+				{
+					UID: "subscription-4",
+					Subscription: fanout.Subscription{
+						Subscriber: subscriber,
+					},
+				},
+			},
+		},
+	}
+
+	d := &KafkaDispatcher{
+		kafkaConsumerFactory: &mockKafkaConsumerFactory{},
+		channelSubscriptions: make(map[types.NamespacedName]*KafkaSubscription),
+		subsConsumerGroups:   make(map[types.UID]sarama.ConsumerGroup),
+		subscriptions:        make(map[types.UID]Subscription),
+		topicFunc:            utils.TopicName,
+		logger:               zaptest.NewLogger(t).Sugar(),
+	}
+
+	// Let's register channel configs first
+	for _, c := range configs {
+		require.NoError(t, d.RegisterChannelHost(c))
+	}
+
+	var wg sync.WaitGroup
+	for i := 0; i < 10; i++ { // Let's reiterate several times to check everything is fine
+		for _, c := range configs {
+			wg.Add(1)
+			go func(c *ChannelConfig) {
+				defer wg.Done()
+				assert.NoError(t, d.ReconcileConsumers(c))
+			}(c)
+		}
+	}
+	wg.Wait()
+
+	// Assert the state is the final one we want
+	require.Contains(t, d.getHostToChannelMap(), "a.b.c.d")
+	require.Contains(t, d.getHostToChannelMap(), "x.y.w.z")
+	require.Contains(t, d.getHostToChannelMap(), "e.f.g.h")
+
+	require.Contains(t, d.subscriptions, types.UID("subscription-1"))
+	require.Contains(t, d.subscriptions, types.UID("subscription-3"))
+	require.Contains(t, d.subscriptions, types.UID("subscription-4"))
+
+	// Now let's remove all of them
+	wg = sync.WaitGroup{}
+	for _, c := range configs {
+		wg.Add(1)
+		go func(c *ChannelConfig) {
+			defer wg.Done()
+			assert.NoError(t, d.CleanupChannel(c.Name, c.Namespace, c.HostName))
+		}(c)
+	}
+	wg.Wait()
+
+	require.Empty(t, d.getHostToChannelMap())
+	require.Empty(t, d.subscriptions)
+	require.Empty(t, d.channelSubscriptions)
+	require.Empty(t, d.subsConsumerGroups)
+}
+
+func TestKafkaDispatcher_CleanupChannel(t *testing.T) {
+	subscriber, _ := url.Parse("http://test/subscriber")
+
+	d := &KafkaDispatcher{
+		kafkaConsumerFactory: &mockKafkaConsumerFactory{},
+		channelSubscriptions: make(map[types.NamespacedName]*KafkaSubscription),
+		subsConsumerGroups:   make(map[types.UID]sarama.ConsumerGroup),
+		subscriptions:        make(map[types.UID]Subscription),
+		topicFunc:            utils.TopicName,
+		logger:               zaptest.NewLogger(t).Sugar(),
+	}
+
+	channelConfig := &ChannelConfig{
+		Namespace: "default",
+		Name:      "test-channel",
+		HostName:  "a.b.c.d",
+		Subscriptions: []Subscription{
+			{
+				UID: "subscription-1",
+				Subscription: fanout.Subscription{
+					Subscriber: subscriber,
+				},
+			},
+			{
+				UID: "subscription-2",
+				Subscription: fanout.Subscription{
+					Subscriber: subscriber,
+				},
+			},
+		},
+	}
+	require.NoError(t, d.RegisterChannelHost(channelConfig))
+	require.NoError(t, d.ReconcileConsumers(channelConfig))
+
+	require.NoError(t, d.CleanupChannel(channelConfig.Name, channelConfig.Namespace, channelConfig.HostName))
+	require.NotContains(t, d.subscriptions, "subscription-1")
+	require.NotContains(t, d.subscriptions, "subscription-2")
+	require.NotContains(t, d.channelSubscriptions, eventingchannels.ChannelReference{
+		Namespace: "default",
+		Name:      "test-channel",
+	})
+	require.NotContains(t, d.subsConsumerGroups, "subscription-1")
+	require.NotContains(t, d.subsConsumerGroups, "subscription-2")
+}
+
 func TestSubscribeError(t *testing.T) {
 	cf := &mockKafkaConsumerFactory{createErr: true}
 	d := &KafkaDispatcher{
 		kafkaConsumerFactory: cf,
 		logger:               zap.NewNop().Sugar(),
 		topicFunc:            utils.TopicName,
+		subscriptions:        map[types.UID]Subscription{},
+		channelSubscriptions: map[types.NamespacedName]*KafkaSubscription{},
 	}
 
-	channelRef := eventingchannels.ChannelReference{
+	channelRef := types.NamespacedName{
 		Name:      "test-channel",
 		Namespace: "test-ns",
 	}
@@ -406,7 +481,7 @@ func TestUnsubscribeUnknownSub(t *testing.T) {
 		logger:               zap.NewNop().Sugar(),
 	}
 
-	channelRef := eventingchannels.ChannelReference{
+	channelRef := types.NamespacedName{
 		Name:      "test-channel",
 		Namespace: "test-ns",
 	}
@@ -433,7 +508,6 @@ func TestNewDispatcher(t *testing.T) {
 		ClientID:  "kafka-ch-dispatcher",
 		Brokers:   []string{"localhost:10000"},
 		TopicFunc: utils.TopicName,
-		Logger:    nil,
 	}
 	_, err := NewDispatcher(context.TODO(), args)
 	if err == nil {
@@ -460,7 +534,7 @@ func TestSetReady(t *testing.T) {
 				channelReadySubscriptions: map[string]sets.Int32{"bar": sets.NewInt32(0)},
 			},
 			desiredKafkaSub: &KafkaSubscription{
-				subs: []types.UID{},
+				subs: sets.NewString(),
 				channelReadySubscriptions: map[string]sets.Int32{
 					"bar": sets.NewInt32(0),
 					"foo": sets.NewInt32(0),
@@ -479,7 +553,7 @@ func TestSetReady(t *testing.T) {
 				},
 			},
 			desiredKafkaSub: &KafkaSubscription{
-				subs: []types.UID{},
+				subs: sets.NewString(),
 				channelReadySubscriptions: map[string]sets.Int32{
 					"bar": sets.NewInt32(0),
 					"foo": sets.NewInt32(0, 1),
diff --git a/pkg/channel/consolidated/dispatcher/subscription_endpoint_test.go b/pkg/channel/consolidated/dispatcher/subscription_endpoint_test.go
index 219cf1dc7..63d7b0a6f 100644
--- a/pkg/channel/consolidated/dispatcher/subscription_endpoint_test.go
+++ b/pkg/channel/consolidated/dispatcher/subscription_endpoint_test.go
@@ -10,7 +10,6 @@ import (
 	"github.com/google/go-cmp/cmp"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/sets"
-	eventingchannels "knative.dev/eventing/pkg/channel"
 	klogtesting "knative.dev/pkg/logging/testing"
 )
 
@@ -22,7 +21,7 @@ func TestServeHTTP(t *testing.T) {
 		name               string
 		responseReturnCode int
 		desiredJson        []byte
-		channelSubs        map[eventingchannels.ChannelReference]*KafkaSubscription
+		channelSubs        map[types.NamespacedName]*KafkaSubscription
 		requestURI         string
 		httpMethod         string
 	}{
@@ -43,9 +42,9 @@ func TestServeHTTP(t *testing.T) {
 			httpMethod:         httpGet,
 			responseReturnCode: http.StatusOK,
 			desiredJson:        []byte(`{}`),
-			channelSubs: map[eventingchannels.ChannelReference]*KafkaSubscription{
+			channelSubs: map[types.NamespacedName]*KafkaSubscription{
 				{Name: "foo", Namespace: "bar"}: {
-					subs:                      []types.UID{},
+					subs:                      sets.NewString(),
 					channelReadySubscriptions: map[string]sets.Int32{},
 				},
 			},
@@ -55,9 +54,9 @@ func TestServeHTTP(t *testing.T) {
 			httpMethod:         httpGet,
 			desiredJson:        []byte{},
 			responseReturnCode: http.StatusNotFound,
-			channelSubs: map[eventingchannels.ChannelReference]*KafkaSubscription{
+			channelSubs: map[types.NamespacedName]*KafkaSubscription{
 				{Name: "foo", Namespace: "baz"}: {
-					subs: []types.UID{"a", "b"},
+					subs: sets.NewString("a", "b"),
 					channelReadySubscriptions: map[string]sets.Int32{
 						"a": sets.NewInt32(0),
 						"b": sets.NewInt32(0),
@@ -70,9 +69,9 @@ func TestServeHTTP(t *testing.T) {
 			httpMethod:         httpGet,
 			desiredJson:        []byte(`{"a":[0],"b":[0,2,5]}`),
 			responseReturnCode: http.StatusOK,
-			channelSubs: map[eventingchannels.ChannelReference]*KafkaSubscription{
+			channelSubs: map[types.NamespacedName]*KafkaSubscription{
 				{Name: "foo", Namespace: "bar"}: {
-					subs: []types.UID{"a", "b"},
+					subs: sets.NewString("a", "b"),
 					channelReadySubscriptions: map[string]sets.Int32{
 						"a": sets.NewInt32(0),
 						"b": sets.NewInt32(0, 2, 5),
@@ -85,15 +84,15 @@ func TestServeHTTP(t *testing.T) {
 			httpMethod:         httpGet,
 			desiredJson:        []byte(`{"a":[0],"b":[0,2,5]}`),
 			responseReturnCode: http.StatusOK,
-			channelSubs: map[eventingchannels.ChannelReference]*KafkaSubscription{
+			channelSubs: map[types.NamespacedName]*KafkaSubscription{
 				{Name: "table", Namespace: "flip"}: {
-					subs: []types.UID{"c", "d"},
+					subs: sets.NewString("c", "d"),
 					channelReadySubscriptions: map[string]sets.Int32{
 						"c": sets.NewInt32(0),
 						"d": sets.NewInt32(0),
 					}},
 				{Name: "foo", Namespace: "bar"}: {
-					subs: []types.UID{"a", "b"},
+					subs: sets.NewString("a", "b"),
 					channelReadySubscriptions: map[string]sets.Int32{
 						"a": sets.NewInt32(0),
 						"b": sets.NewInt32(0, 2, 5),
@@ -116,7 +115,7 @@ func TestServeHTTP(t *testing.T) {
 	}
 	logger := klogtesting.TestLogger(t)
 	d := &KafkaDispatcher{
-		channelSubscriptions: make(map[eventingchannels.ChannelReference]*KafkaSubscription),
+		channelSubscriptions: make(map[types.NamespacedName]*KafkaSubscription),
 		logger:               logger,
 	}
 	subscriptionEndpoint := &subscriptionEndpoint{