func (b *Broker) Produce(request *ProduceRequest) (*ProduceResponse, error) {
	var (
		response *ProduceResponse
		err      error
	)
	if request.RequiredAcks == NoResponse {
		err = b.sendAndReceive(request, nil)
	} else {
		response = new(ProduceResponse)
		err = b.sendAndReceive(request, response)
	}
	if err != nil {
		return nil, err
	}
	return response, nil
}
Produce returns a produce response or error
package sarama
import (
	"crypto/tls"
	"encoding/binary"
	"fmt"
	"io"
	"net"
	"sort"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"time"
	"github.com/rcrowley/go-metrics"
)
type Broker struct {
	conf *Config
	rack *string
	id            int32
	addr          string
	correlationID int32
	conn          net.Conn
	connErr       error
	lock          sync.Mutex
	opened        int32
	responses     chan responsePromise
	done          chan bool
	registeredMetrics []string
	incomingByteRate       metrics.Meter
	requestRate            metrics.Meter
	requestSize            metrics.Histogram
	requestLatency         metrics.Histogram
	outgoingByteRate       metrics.Meter
	responseRate           metrics.Meter
	responseSize           metrics.Histogram
	requestsInFlight       metrics.Counter
	brokerIncomingByteRate metrics.Meter
	brokerRequestRate      metrics.Meter
	brokerRequestSize      metrics.Histogram
	brokerRequestLatency   metrics.Histogram
	brokerOutgoingByteRate metrics.Meter
	brokerResponseRate     metrics.Meter
	brokerResponseSize     metrics.Histogram
	brokerRequestsInFlight metrics.Counter
	kerberosAuthenticator GSSAPIKerberosAuth
}
type SASLMechanism string
const (
	SASLTypeOAuth = "OAUTHBEARER"
	SASLTypePlaintext = "PLAIN"
	SASLTypeSCRAMSHA256 = "SCRAM-SHA-256"
	SASLTypeSCRAMSHA512 = "SCRAM-SHA-512"
	SASLTypeGSSAPI      = "GSSAPI"
	SASLHandshakeV0 = int16(0)
	SASLHandshakeV1 = int16(1)
	SASLExtKeyAuth = "auth"
)
type AccessToken struct {
	Token string
	Extensions map[string]string
}
type AccessTokenProvider interface {
	Token() (*AccessToken, error)
}
type SCRAMClient interface {
	Begin(userName, password, authzID string) error
	Step(challenge string) (response string, err error)
	Done() bool
}
type responsePromise struct {
	requestTime   time.Time
	correlationID int32
	headerVersion int16
	packets       chan []byte
	errors        chan error
}
func NewBroker(addr string) *Broker {
	return &Broker{id: -1, addr: addr}
}
func (b *Broker) Open(conf *Config) error {
	if !atomic.CompareAndSwapInt32(&b.opened, 0, 1) {
		return ErrAlreadyConnected
	}
	if conf == nil {
		conf = NewConfig()
	}
	err := conf.Validate()
	if err != nil {
		return err
	}
	b.lock.Lock()
	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
			atomic.StoreInt32(&b.opened, 0)
			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
		b.incomingByteRate = metrics.GetOrRegisterMeter("incoming-byte-rate", conf.MetricRegistry)
		b.requestRate = metrics.GetOrRegisterMeter("request-rate", conf.MetricRegistry)
		b.requestSize = getOrRegisterHistogram("request-size", conf.MetricRegistry)
		b.requestLatency = getOrRegisterHistogram("request-latency-in-ms", conf.MetricRegistry)
		b.outgoingByteRate = metrics.GetOrRegisterMeter("outgoing-byte-rate", conf.MetricRegistry)
		b.responseRate = metrics.GetOrRegisterMeter("response-rate", conf.MetricRegistry)
		b.responseSize = getOrRegisterHistogram("response-size", conf.MetricRegistry)
		b.requestsInFlight = metrics.GetOrRegisterCounter("requests-in-flight", conf.MetricRegistry)
		if b.id >= 0 {
			b.registerMetrics()
		}
		if conf.Net.SASL.Enable {
			b.connErr = b.authenticateViaSASL()
			if b.connErr != nil {
				err = b.conn.Close()
				if err == nil {
					Logger.Printf("Closed connection to broker %s\n", b.addr)
				} else {
					Logger.Printf("Error while closing connection to broker %s: %s\n", b.addr, err)
				}
				b.conn = nil
				atomic.StoreInt32(&b.opened, 0)
				return
			}
		}
		b.done = make(chan bool)
		b.responses = make(chan responsePromise, b.conf.Net.MaxOpenRequests-1)
		if b.id >= 0 {
			Logger.Printf("Connected to broker at %s (registered as #%d)\n", b.addr, b.id)
		} else {
			Logger.Printf("Connected to broker at %s (unregistered)\n", b.addr)
		}
		go withRecover(b.responseReceiver)
	})
	return nil
}
func (b *Broker) Connected() (bool, error) {
	b.lock.Lock()
	defer b.lock.Unlock()
	return b.conn != nil, b.connErr
}
func (b *Broker) Close() error {
	b.lock.Lock()
	defer b.lock.Unlock()
	if b.conn == nil {
		return ErrNotConnected
	}
	close(b.responses)
	<-b.done
	err := b.conn.Close()
	b.conn = nil
	b.connErr = nil
	b.done = nil
	b.responses = nil
	b.unregisterMetrics()
	if err == nil {
		Logger.Printf("Closed connection to broker %s\n", b.addr)
	} else {
		Logger.Printf("Error while closing connection to broker %s: %s\n", b.addr, err)
	}
	atomic.StoreInt32(&b.opened, 0)
	return err
}
func (b *Broker) ID() int32 {
	return b.id
}
func (b *Broker) Addr() string {
	return b.addr
}
func (b *Broker) Rack() string {
	if b.rack == nil {
		return ""
	}
	return *b.rack
}
func (b *Broker) GetMetadata(request *MetadataRequest) (*MetadataResponse, error) {
	response := new(MetadataResponse)
	err := b.sendAndReceive(request, response)
	if err != nil {
		return nil, err
	}
	return response, nil
}
func (b *Broker) GetConsumerMetadata(request *ConsumerMetadataRequest) (*ConsumerMetadataResponse, error) {
	response := new(ConsumerMetadataResponse)
	err := b.sendAndReceive(request, response)
	if err != nil {
		return nil, err
	}
	return response, nil
}
func (b *Broker) FindCoordinator(request *FindCoordinatorRequest) (*FindCoordinatorResponse, error) {
	response := new(FindCoordinatorResponse)
	err := b.sendAndReceive(request, response)
	if err != nil {
		return nil, err
	}
	return response, nil
}
func (b *Broker) GetAvailableOffsets(request *OffsetRequest) (*OffsetResponse, error) {
	response := new(OffsetResponse)
	err := b.sendAndReceive(request, response)
	if err != nil {
		return nil, err
	}
	return response, nil
}