do not unsubscribe every time channel fanout config is updated for existing consumers
@@ -112,18 +112,12 @@ func (d *KafkaDispatcher) UpdateConfig(config *multichannelfanout.Config) error
}
for _, subSpec := range cc.FanoutConfig.Subscriptions {
sub := newSubscription(subSpec)
- if _, ok := d.kafkaConsumers[channelRef][sub]; ok {
- // subscribe can be called multiple times for the same subscription,
- // unsubscribe before we resubscribe.
- // TODO The behavior to unsubscribe and re-subscribe is retained from the older kafka bus
- // implementation. It is not clear as to why this is needed instead of just re-using the same
- // consumer.
- err := d.unsubscribe(channelRef, sub)
- if err != nil {
- return err
- }
- }
+ if _, ok := d.kafkaConsumers[channelRef][sub]; !ok {
+ // only subscribe when not exists in channel-subscriptions map
+ // do not need to resubscribe every time channel fanout config is updated
d.subscribe(channelRef, sub)
+ }
+
newSubs[sub] = true
}
}
