code: func (p *Plugin) AfterInit() error {
  	if p.disabled {
  		p.Log.Debugf("kafka plugin disabled, skipping AfterInit")
  		return nil
  	}
  
  	if p.mux != nil {
  		err := p.mux.Start()
  		if err != nil {
  			return err
  		}
  	}
  
  	// Register for providing status reports (polling mode)
  	if p.StatusCheck != nil {
  		p.StatusCheck.Register(p.PluginName, func() (statuscheck.PluginState, error) {
  			if p.hsClient == nil || p.hsClient.Closed() {
  				return statuscheck.Error, fmt.Errorf("kafka client/consumer not available")
  			}
  			// Method 'RefreshMetadata()' returns error if kafka server is unavailable
  			err := p.hsClient.RefreshMetadata(topic)
  			if err == nil {
  				return statuscheck.OK, nil
  			}
  			p.Log.Errorf("Kafka server unavailable")
  			return statuscheck.Error, err
  		})
  	} else {
  		p.Log.Warnf("Unable to start status check for kafka")
  	}
  
  	return nil
  }
docs: // AfterInit is called in the second phase of the initialization. The kafka multiplexerNewWatcher
  // is started, all consumers have to be subscribed until this phase.