code: func (p *Plugin) Init() (err error) {
  	// Prepare topic and  subscription for status check client
  	p.subscription = make(chan *client.ConsumerMessage)
  
  	// Get muxCfg data (contains kafka brokers ip addresses)
  	muxCfg := &mux.Config{}
  	found, err := p.Cfg.LoadValue(muxCfg)
  	if !found {
  		p.Log.Info("kafka config not found ", p.Cfg.GetConfigName(), " - skip loading this plugin")
  		p.disabled = true
  		return nil //skip loading the plugin
  	}
  	if err != nil {
  		return err
  	}
  	// retrieve clientCfg
  	clientCfg, err := p.getClientConfig(muxCfg, p.Log, topic)
  	if err != nil {
  		return err
  	}
  
  	// init 'hash' sarama client
  	p.hsClient, err = client.NewClient(clientCfg, client.Hash)
  	if err != nil {
  		return err
  	}
  
  	// init 'manual' sarama client
  	p.manClient, err = client.NewClient(clientCfg, client.Manual)
  	if err != nil {
  		return err
  	}
  
  	// Initialize both multiplexers to allow both, dynamic and manual mode
  	if p.mux == nil {
  		name := clientCfg.GroupID
  		p.Log.Infof("Group ID is set to %v", name)
  		p.mux, err = mux.InitMultiplexerWithConfig(clientCfg, p.hsClient, p.manClient, name, p.Log)
  		if err != nil {
  			return err
  		}
  		p.Log.Debug("Default multiplexer initialized")
  	}
  
  	return err
  }
docs: // Init is called at plugin initialization.