Method Open

broker.go:189–337

Open tries to connect to the Broker if it is not already connected or connecting, but does not block waiting for the connection to complete. This means that any subsequent operations on the broker will block waiting for the connection to succeed or fail. To get the effect of a fully synchronous Open

(conf *Config)

// follow it by a call to Connected(). The only errors Open will return directly are ConfigurationError or
// AlreadyConnected. If conf is nil, the result of NewConfig() is used.
func (b *Broker) Open(conf *Config) error {
	if !b.opened.CompareAndSwap(false, true) {
		return ErrAlreadyConnected
	}

	if conf == nil {
		conf = NewConfig()
	}

	err := conf.Validate()
	if err != nil {
		return err
	}

	b.lock.Lock()

	if b.metricRegistry == nil {
		b.metricRegistry = newCleanupRegistry(conf.MetricRegistry)
	}

	go withRecover(func() {
		defer b.lock.Unlock()

		dialer := conf.getDialer()
		b.conn, b.connErr = dialer.Dial("tcp", b.addr)
		if b.connErr != nil {
			Logger.Printf("Failed to connect to broker %s: %s\n", b.addr, b.connErr)
			b.conn = nil
			b.opened.Store(false)
			return
		}
		if conf.Net.TLS.Enable {
			b.conn = tls.Client(b.conn, validServerNameTLS(b.addr, conf.Net.TLS.Config))
		}

		b.conn = newBufConn(b.conn)
		b.conf = conf

		// Create or reuse the global metrics shared between brokers
		b.incomingByteRate = metrics.GetOrRegisterMeter("incoming-byte-rate", b.metricRegistry)
		b.requestRate = metrics.GetOrRegisterMeter("request-rate", b.metricRegistry)
		b.fetchRate = metrics.GetOrRegisterMeter("consumer-fetch-rate", b.metricRegistry)
		b.requestSize = getOrRegisterHistogram("request-size", b.metricRegistry)
		b.requestLatency = getOrRegisterHistogram("request-latency-in-ms", b.metricRegistry)
		b.outgoingByteRate = metrics.GetOrRegisterMeter("outgoing-byte-rate", b.metricRegistry)
		b.responseRate = metrics.GetOrRegisterMeter("response-rate", b.metricRegistry)
		b.responseSize = getOrRegisterHistogram("response-size", b.metricRegistry)
		b.requestsInFlight = metrics.GetOrRegisterCounter("requests-in-flight", b.metricRegistry)
		b.protocolRequestsRate = map[int16]metrics.Meter{}
		// Do not gather metrics for seeded broker (only used during bootstrap) because they share
		// the same id (-1) and are already exposed through the global metrics above
		if b.id >= 0 && !metrics.UseNilMetrics {
			b.registerMetrics()
		}

		// Send an ApiVersionsRequest to identify the client (KIP-511).
		// Store the response in the brokerAPIVersions map.
		// It will be used to determine the supported API versions for each request.