Method connect

dialer.go:265–301

connect opens a socket connection to the broker, wraps it to create a kafka connection, and performs SASL authentication if configured to do so.

(ctx context.Context, network, address string, connCfg ConnConfig)

// connect opens a socket connection to the broker, wraps it to create a
// kafka connection, and performs SASL authentication if configured to do so.
func (d *Dialer) connect(ctx context.Context, network, address string, connCfg ConnConfig) (*Conn, error) {
	if d.Timeout != 0 {
		var cancel context.CancelFunc
		ctx, cancel = context.WithTimeout(ctx, d.Timeout)
		defer cancel()
	}

	if !d.Deadline.IsZero() {
		var cancel context.CancelFunc
		ctx, cancel = context.WithDeadline(ctx, d.Deadline)
		defer cancel()
	}

	c, err := d.dialContext(ctx, network, address)
	if err != nil {
		return nil, fmt.Errorf("failed to dial: %w", err)
	}

	conn := NewConnWith(c, connCfg)

	if d.SASLMechanism != nil {
		host, port, err := splitHostPortNumber(address)
		if err != nil {
			return nil, fmt.Errorf("could not determine host/port for SASL authentication: %w", err)
		}
		metadata := &sasl.Metadata{
			Host: host,
			Port: port,
		}
		if err := d.authenticateSASL(sasl.WithMetadata(ctx, metadata), conn); err != nil {
			_ = conn.Close()
			return nil, fmt.Errorf("could not successfully authenticate to %s:%d with SASL: %w", host, port, err)
		}
	}

	return conn, nil
}

// authenticateSASL performs all of the required requests to authenticate this
// connection.  If any step fails, this function returns with an error.  A nil