code: func New(config *localconfig.TopLevel, metricsProvider metrics.Provider, healthChecker healthChecker, migCtrl migration.Controller) (consensus.Consenter, *Metrics) {
  	if config.Kafka.Verbose {
  		logging.SetLevel(logging.DEBUG, "orderer.consensus.kafka.sarama")
  	}
  
  	brokerConfig := newBrokerConfig(
  		config.Kafka.TLS,
  		config.Kafka.SASLPlain,
  		config.Kafka.Retry,
  		config.Kafka.Version,
  		defaultPartition)
  
  	bootFile := ""
  	if config.General.GenesisMethod == "file" {
  		bootFile = config.General.GenesisFile
  	}
  
  	return &consenterImpl{
  		brokerConfigVal: brokerConfig,
  		tlsConfigVal:    config.Kafka.TLS,
  		retryOptionsVal: config.Kafka.Retry,
  		kafkaVersionVal: config.Kafka.Version,
  		topicDetailVal: &sarama.TopicDetail{
  			NumPartitions:     1,
  			ReplicationFactor: config.Kafka.Topic.ReplicationFactor,
  		},
  		healthChecker:     healthChecker,
  		migController:     migCtrl,
  		bootstrapFileName: bootFile,
  	}, NewMetrics(metricsProvider, brokerConfig.MetricRegistry)
  }
docs: // New creates a Kafka-based consenter. Called by orderer's main.go.