Why Go
- Case Studies
  
  Common problems companies solve with Go
- Use Cases
  
  Stories about how and why companies use Go
- Security
  
  How Go can help keep you secure by default
Learn
Docs
- Effective Go
  
  Tips for writing clear, performant, and idiomatic Go code
- Go User Manual
  
  A complete introduction to building software with Go
- Standard library
  
  Reference documentation for Go's standard library
- Release Notes
  
  Learn what's new in each Go release
Packages
Community
- Recorded Talks
  
  Videos from prior events
- Meetups
  
  Meet other local Go developers
- Conferences
  
  Learn and network with Go developers from around the world
- Go blog
  
  The Go project's official blog.
- Go project
  
  Get help and stay informed from Go
- Get connected

broker

package

Go to main page

Versions in this module

v1

v1.5.0

Jan 30, 2026

Changes in this version

+ func ContextWithConnInfo(ctx context.Context, info *ConnContext) context.Context

+ type ConnContext struct

+ Principal string

+ ProxyAddr string

+ RemoteAddr string

+ func ConnInfoFromContext(ctx context.Context) *ConnContext

+ type ConnContextFunc func(conn net.Conn) (net.Conn, *ConnContext, error)

+ type ProxyInfo struct

+ DestAddr string

+ DestIP string

+ DestPort int

+ Local bool

+ SourceAddr string

+ SourceIP string

+ SourcePort int

+ func ReadProxyProtocol(conn net.Conn) (net.Conn, *ProxyInfo, error)

type Server

+ ConnContextFunc ConnContextFunc

v1.4.2

Jan 19, 2026

Changes in this version

+ type CoordinatorConfig struct

+ CleanupInterval time.Duration

+ type GroupCoordinator struct

+ func NewGroupCoordinator(store metadata.Store, broker protocol.MetadataBroker, cfg *CoordinatorConfig) *GroupCoordinator

+ func (c *GroupCoordinator) DeleteGroups(ctx context.Context, req *protocol.DeleteGroupsRequest, correlationID int32) (*protocol.DeleteGroupsResponse, error)

+ func (c *GroupCoordinator) DescribeGroups(ctx context.Context, req *protocol.DescribeGroupsRequest, correlationID int32) (*protocol.DescribeGroupsResponse, error)

+ func (c *GroupCoordinator) FindCoordinatorResponse(correlationID int32, errorCode int16) *protocol.FindCoordinatorResponse

+ func (c *GroupCoordinator) Heartbeat(ctx context.Context, req *protocol.HeartbeatRequest, correlationID int32) *protocol.HeartbeatResponse

+ func (c *GroupCoordinator) JoinGroup(ctx context.Context, req *protocol.JoinGroupRequest, correlationID int32) (*protocol.JoinGroupResponse, error)

+ func (c *GroupCoordinator) LeaveGroup(ctx context.Context, req *protocol.LeaveGroupRequest, correlationID int32) *protocol.LeaveGroupResponse

+ func (c *GroupCoordinator) ListGroups(ctx context.Context, req *protocol.ListGroupsRequest, correlationID int32) (*protocol.ListGroupsResponse, error)

+ func (c *GroupCoordinator) OffsetCommit(ctx context.Context, req *protocol.OffsetCommitRequest, correlationID int32) (*protocol.OffsetCommitResponse, error)

+ func (c *GroupCoordinator) OffsetFetch(ctx context.Context, req *protocol.OffsetFetchRequest, correlationID int32) (*protocol.OffsetFetchResponse, error)

+ func (c *GroupCoordinator) Stop()

+ func (c *GroupCoordinator) SyncGroup(ctx context.Context, req *protocol.SyncGroupRequest, correlationID int32) (*protocol.SyncGroupResponse, error)

+ type Handler interface

+ Handle func(ctx context.Context, header *protocol.RequestHeader, req protocol.Request) ([]byte, error)

+ type S3HealthConfig struct

+ ErrorCrit float64

+ ErrorWarn float64

+ LatencyCrit time.Duration

+ LatencyWarn time.Duration

+ MaxSamples int

+ Window time.Duration

+ type S3HealthMonitor struct

+ func NewS3HealthMonitor(cfg S3HealthConfig) *S3HealthMonitor

+ func (m *S3HealthMonitor) RecordOperation(op string, latency time.Duration, err error)

+ func (m *S3HealthMonitor) RecordUpload(latency time.Duration, err error)

+ func (m *S3HealthMonitor) Snapshot() S3HealthSnapshot

+ func (m *S3HealthMonitor) State() S3HealthState

+ type S3HealthSnapshot struct

+ AvgLatency time.Duration

+ ErrorRate float64

+ Since time.Time

+ State S3HealthState

+ type S3HealthState string

+ const S3StateDegraded

+ const S3StateHealthy

+ const S3StateUnavailable

+ type Server struct

+ Addr string

+ Handler Handler

+ func (s *Server) ListenAddress() string

+ func (s *Server) ListenAndServe(ctx context.Context) error

+ func (s *Server) Wait()