func (gc *T) rewireMuxAsync(topic string, wg *sync.WaitGroup, mux *multiplexer.T, tc *topiccsm.T, assigned []int32) {
	actor.Spawn(gc.actDesc.NewChild("rewire", topic), wg, func() {
		if tc == nil {
			mux.WireUp(nil, nil)
			return
		}
		mux.WireUp(tc, assigned)
	})
}
rewireMuxAsync calls muxInputs in another goroutine.
package groupcsm
import (
	"fmt"
	"sort"
	"sync"
	"time"
	"github.com/Shopify/sarama"
	"github.com/mailgun/kafka-pixy/actor"
	"github.com/mailgun/kafka-pixy/config"
	"github.com/mailgun/kafka-pixy/consumer"
	"github.com/mailgun/kafka-pixy/consumer/dispatcher"
	"github.com/mailgun/kafka-pixy/consumer/msgfetcher"
	"github.com/mailgun/kafka-pixy/consumer/multiplexer"
	"github.com/mailgun/kafka-pixy/consumer/partitioncsm"
	"github.com/mailgun/kafka-pixy/consumer/subscriber"
	"github.com/mailgun/kafka-pixy/consumer/topiccsm"
	"github.com/mailgun/kafka-pixy/offsetmgr"
	"github.com/mailgun/kafka-pixy/prettyfmt"
	"github.com/pkg/errors"
	"github.com/samuel/go-zookeeper/zk"
)
type T struct {
	actDesc     *actor.Descriptor
	cfg         *config.Proxy
	group       string
	dispatcher  *dispatcher.T
	kafkaClt    sarama.Client
	zkConn      *zk.Conn
	msgFetcherF msgfetcher.Factory
	offsetMgrF  offsetmgr.Factory
	subscriber  *subscriber.T
	topicCsmCh  chan *topiccsm.T
	wg          sync.WaitGroup
	multiplexersMu sync.Mutex
	multiplexers   map[string]*multiplexer.T
}
func Spawn(parentActDesc *actor.Descriptor, childSpec dispatcher.ChildSpec,
	cfg *config.Proxy, kafkaClt sarama.Client, zkConn *zk.Conn, offsetMgrF offsetmgr.Factory,
) *T {
	group := string(childSpec.Key())
	actDesc := parentActDesc.NewChild(fmt.Sprintf("%s", group))
	actDesc.AddLogField("kafka.group", group)
	gc := &T{
		actDesc:      actDesc,
		cfg:          cfg,
		group:        group,
		kafkaClt:     kafkaClt,
		zkConn:       zkConn,
		offsetMgrF:   offsetMgrF,
		multiplexers: make(map[string]*multiplexer.T),
		topicCsmCh:   make(chan *topiccsm.T, cfg.Consumer.ChannelBufferSize),
	}
	gc.subscriber = subscriber.Spawn(gc.actDesc, gc.group, gc.cfg, gc.zkConn)
	gc.msgFetcherF = msgfetcher.SpawnFactory(gc.actDesc, gc.cfg, gc.kafkaClt)
	actor.Spawn(gc.actDesc, &gc.wg, gc.run)
	gc.dispatcher = dispatcher.Spawn(gc.actDesc, gc, cfg,
		dispatcher.WithChildSpec(childSpec),
		dispatcher.WithFinalizer(gc.finalizer))
	return gc
}
func (gc *T) KeyOf(req consumer.Request) dispatcher.Key {
	return dispatcher.Key(req.Topic)
}
func (gc *T) SpawnChild(childSpec dispatcher.ChildSpec) {
	topic := string(childSpec.Key())
	topiccsm.Spawn(gc.actDesc, gc.group, childSpec, gc.cfg, gc.topicCsmCh,
		func() bool { return gc.isSafe2Stop(topic) })
}
func (gc *T) String() string {
	return gc.actDesc.String()
}
func (gc *T) finalizer() {
	gc.subscriber.Stop()
	gc.wg.Wait()
	gc.msgFetcherF.Stop()
	gc.subscriber.DeleteGroupIfEmpty()
}
func (gc *T) isSafe2Stop(topic string) bool {
	gc.multiplexersMu.Lock()
	mux := gc.multiplexers[topic]
	gc.multiplexersMu.Unlock()
	if mux == nil {
		return true
	}
	return mux.IsSafe2Stop()
}
func (gc *T) run() {
	var (
		topicConsumers          = make(map[string]*topiccsm.T)
		topics                  []string
		subscriptions           map[string][]string
		ok                      = true
		nilOrRetryCh            <-chan time.Time
		nilOrSubscriberTopicsCh chan<- []string
		rebalanceRequired       = false
		rebalancePending        = false
		rebalanceScheduled      = false
		stopped                 = false
		rebalanceResultCh       = make(chan error, 1)
	)
	for {
		select {
		case tc := <-gc.topicCsmCh:
			if topicConsumers[tc.Topic()] == tc {
				delete(topicConsumers, tc.Topic())
			} else {
				topicConsumers[tc.Topic()] = tc
			}
			topics = listTopics(topicConsumers)
			gc.actDesc.Log().Infof("Topics updated: %s", topics)
			nilOrSubscriberTopicsCh = gc.subscriber.Topics()
			continue
		case nilOrSubscriberTopicsCh <- topics:
			nilOrSubscriberTopicsCh = nil
			continue
		case subscriptions, ok = <-gc.subscriber.Subscriptions():
			nilOrRetryCh = nil
			if !ok {
				if !rebalancePending {
					goto done
				}
				stopped = true
				continue
			}
			rebalanceRequired = true
		case err := <-rebalanceResultCh:
			rebalancePending = false
			if err != nil {
				gc.actDesc.Log().WithError(err).Error("rebalancing failed")
				if stopped {
					goto done
				}
				nilOrRetryCh = time.After(gc.cfg.Consumer.RetryBackoff)
				rebalanceScheduled = true
			}
			if stopped {
				goto done
			}
		case <-nilOrRetryCh:
			rebalanceScheduled = false
		}
		if rebalanceRequired && !rebalancePending && !rebalanceScheduled {
			rebalanceActDesc := gc.actDesc.NewChild("rebalance")
			topicConsumersCopy := make(map[string]*topiccsm.T, len(topicConsumers))
			for topic, tc := range topicConsumers {
				topicConsumersCopy[topic] = tc
			}
			subscriptions := subscriptions
			actor.Spawn(rebalanceActDesc, nil, func() {
				gc.rebalance(rebalanceActDesc, topicConsumersCopy, subscriptions, rebalanceResultCh)
			})
			rebalancePending = true
			rebalanceRequired = false
		}
	}
done:
	var wg sync.WaitGroup
	gc.multiplexersMu.Lock()
	for _, mux := range gc.multiplexers {
		wg.Add(1)
		go func(mux *multiplexer.T) {
			defer wg.Done()
			mux.Stop()
		}(mux)
	}
	gc.multiplexersMu.Unlock()
	wg.Wait()
}
func (gc *T) rebalance(actDesc *actor.Descriptor, topicConsumers map[string]*topiccsm.T,
	subscriptions map[string][]string, rebalanceResultCh chan<- error,
) {
	assignedPartitions, err := gc.resolvePartitions(subscriptions, gc.kafkaClt.Partitions)
	if err != nil {
		rebalanceResultCh <- err
		return
	}
	actDesc.Log().Infof("assigned partitions: %s", prettyfmt.Val(assignedPartitions))
	var wg sync.WaitGroup
	gc.multiplexersMu.Lock()
	defer gc.multiplexersMu.Unlock()
	for topic, mux := range gc.multiplexers {
		gc.rewireMuxAsync(topic, &wg, mux, topicConsumers[topic], assignedPartitions[topic])
	}
	for topic, assignedTopicPartitions := range assignedPartitions {
		tc := topicConsumers[topic]
		mux := gc.multiplexers[topic]
		if tc == nil || mux != nil {
			continue
		}
		topic := topic
		spawnInFn := func(partition int32) multiplexer.In {
			return partitioncsm.Spawn(gc.actDesc, gc.group, topic, partition,
				gc.cfg, gc.subscriber, gc.msgFetcherF, gc.offsetMgrF)
		}
		mux = multiplexer.New(gc.actDesc, spawnInFn)
		gc.rewireMuxAsync(topic, &wg, mux, tc, assignedTopicPartitions)
		gc.multiplexers[topic] = mux
	}
	wg.Wait()
	for topic, mux := range gc.multiplexers {
		if !mux.IsRunning() {
			delete(gc.multiplexers, topic)
		}
	}
	rebalanceResultCh <- nil
	return
}