kafka

package module

Go to main page

Versions in this module

v0

v0.4.19

Jun 1, 2021

v0.4.18

Jun 1, 2021

v0.4.17

Jun 1, 2021

v0.4.16

Apr 30, 2021

v0.4.15

Apr 29, 2021

Changes in this version

type Reader

+ Assignments map[string][]PartitionAssignment

v0.4.14

Apr 9, 2021

Changes in this version

+ const SeekDontCheck

+ var ErrGenerationEnded = errors.New("consumer group generation has ended")

+ var ErrGroupClosed = errors.New("consumer group is closed")

+ func Marshal(v interface{}) ([]byte, error)

+ func ReadAll(b Bytes) ([]byte, error)

+ func TCP(address ...string) net.Addr

+ func Unmarshal(b []byte, v interface{}) error

+ type AlterConfigRequestConfig struct

+ Name string

+ Value string

+ type AlterConfigRequestResource struct

+ Configs []AlterConfigRequestConfig

+ ResourceName string

+ ResourceType ResourceType

+ type AlterConfigsRequest struct

+ Addr net.Addr

+ Resources []AlterConfigRequestResource

+ ValidateOnly bool

+ type AlterConfigsResponse struct

+ Errors map[AlterConfigsResponseResource]error

+ Throttle time.Duration

+ type AlterConfigsResponseResource struct

+ Name string

+ Type int8

+ type AlterPartitionReassignmentsRequest struct

+ Addr net.Addr

+ Assignments []AlterPartitionReassignmentsRequestAssignment

+ Timeout time.Duration

+ Topic string

+ type AlterPartitionReassignmentsRequestAssignment struct

+ BrokerIDs []int

+ PartitionID int

+ type AlterPartitionReassignmentsResponse struct

+ Error error

+ PartitionResults []AlterPartitionReassignmentsResponsePartitionResult

+ type AlterPartitionReassignmentsResponsePartitionResult struct

+ Error error

+ PartitionID int

type ApiVersion

+ func (v ApiVersion) Format(w fmt.State, r rune)

+ type ApiVersionsRequest struct

+ Addr net.Addr

+ type ApiVersionsResponse struct

+ ApiKeys []ApiVersionsResponseApiKey

+ Error error

+ type ApiVersionsResponseApiKey struct

+ ApiKey int

+ ApiName string

+ MaxVersion int

+ MinVersion int

type Batch

+ func (batch *Batch) Err() error

+ func (batch *Batch) Partition() int

+ type BrokerResolver interface

+ LookupBrokerIPAddr func(ctx context.Context, broker Broker) ([]net.IPAddr, error)

+ func NewBrokerResolver(r *net.Resolver) BrokerResolver

+ type Bytes = protocol.Bytes

+ func NewBytes(b []byte) Bytes

+ type CRC32Balancer struct

+ Consistent bool

+ func (b CRC32Balancer) Balance(msg Message, partitions ...int) (partition int)

+ type Client struct

+ Addr net.Addr

+ Timeout time.Duration

+ Transport RoundTripper

+ func (c *Client) AlterConfigs(ctx context.Context, req *AlterConfigsRequest) (*AlterConfigsResponse, error)

+ func (c *Client) AlterPartitionReassignments(ctx context.Context, req *AlterPartitionReassignmentsRequest) (*AlterPartitionReassignmentsResponse, error)

+ func (c *Client) ApiVersions(ctx context.Context, req *ApiVersionsRequest) (*ApiVersionsResponse, error)

+ func (c *Client) ConsumerOffsets(ctx context.Context, tg TopicAndGroup) (map[int]int64, error)

+ func (c *Client) CreatePartitions(ctx context.Context, req *CreatePartitionsRequest) (*CreatePartitionsResponse, error)

+ func (c *Client) CreateTopics(ctx context.Context, req *CreateTopicsRequest) (*CreateTopicsResponse, error)

+ func (c *Client) DeleteTopics(ctx context.Context, req *DeleteTopicsRequest) (*DeleteTopicsResponse, error)

+ func (c *Client) DescribeConfigs(ctx context.Context, req *DescribeConfigsRequest) (*DescribeConfigsResponse, error)

+ func (c *Client) DescribeGroups(ctx context.Context, req *DescribeGroupsRequest) (*DescribeGroupsResponse, error)

+ func (c *Client) ElectLeaders(ctx context.Context, req *ElectLeadersRequest) (*ElectLeadersResponse, error)

+ func (c *Client) Fetch(ctx context.Context, req *FetchRequest) (*FetchResponse, error)

+ func (c *Client) FindCoordinator(ctx context.Context, req *FindCoordinatorRequest) (*FindCoordinatorResponse, error)

+ func (c *Client) IncrementalAlterConfigs(ctx context.Context, req *IncrementalAlterConfigsRequest) (*IncrementalAlterConfigsResponse, error)

+ func (c *Client) InitProducerID(ctx context.Context, req *InitProducerIDRequest) (*InitProducerIDResponse, error)

+ func (c *Client) ListGroups(ctx context.Context, req *ListGroupsRequest) (*ListGroupsResponse, error)

+ func (c *Client) ListOffsets(ctx context.Context, req *ListOffsetsRequest) (*ListOffsetsResponse, error)

+ func (c *Client) Metadata(ctx context.Context, req *MetadataRequest) (*MetadataResponse, error)

+ func (c *Client) OffsetFetch(ctx context.Context, req *OffsetFetchRequest) (*OffsetFetchResponse, error)

+ func (c *Client) Produce(ctx context.Context, req *ProduceRequest) (*ProduceResponse, error)

+ type Compression = compress.Compression

+ const Gzip

+ const Lz4

+ const Snappy

+ const Zstd

+ type ConfigOperation int8

+ const ConfigOperationAppend

+ const ConfigOperationDelete

+ const ConfigOperationSet

+ const ConfigOperationSubtract

+ type ConsumerGroup struct

+ func NewConsumerGroup(config ConsumerGroupConfig) (*ConsumerGroup, error)

+ func (cg *ConsumerGroup) Close() error

+ func (cg *ConsumerGroup) Next(ctx context.Context) (*Generation, error)

+ type ConsumerGroupConfig struct

+ Brokers []string

+ Dialer *Dialer

+ ErrorLogger Logger

+ GroupBalancers []GroupBalancer

+ HeartbeatInterval time.Duration

+ ID string

+ JoinGroupBackoff time.Duration

+ Logger Logger

+ PartitionWatchInterval time.Duration

+ RebalanceTimeout time.Duration

+ RetentionTime time.Duration

+ SessionTimeout time.Duration

+ StartOffset int64

+ Timeout time.Duration

+ Topics []string

+ WatchPartitionChanges bool

+ func (config *ConsumerGroupConfig) Validate() error

+ type CoordinatorKeyType int8

+ const CoordinatorKeyTypeConsumer

+ const CoordinatorKeyTypeTransaction

+ type CreatePartitionsRequest struct

+ Addr net.Addr

+ Topics []TopicPartitionsConfig

+ ValidateOnly bool

+ type CreatePartitionsResponse struct

+ Errors map[string]error

+ Throttle time.Duration

+ type CreateTopicsRequest struct

+ Addr net.Addr

+ Topics []TopicConfig

+ ValidateOnly bool

+ type CreateTopicsResponse struct

+ Errors map[string]error

+ Throttle time.Duration

+ type DeleteTopicsRequest struct

+ Addr net.Addr

+ Topics []string

+ type DeleteTopicsResponse struct

+ Errors map[string]error

+ Throttle time.Duration

+ type DescribeConfigRequestResource struct

+ ConfigNames []string

+ ResourceName string

+ ResourceType ResourceType

+ type DescribeConfigResponseConfigEntry struct

+ ConfigDocumentation string

+ ConfigName string

+ ConfigSource int8

+ ConfigSynonyms []DescribeConfigResponseConfigSynonym

+ ConfigType int8

+ ConfigValue string

+ IsDefault bool

+ IsSensitive bool

+ ReadOnly bool

+ type DescribeConfigResponseConfigSynonym struct

+ ConfigName string

+ ConfigSource int8

+ ConfigValue string

+ type DescribeConfigResponseResource struct

+ ConfigEntries []DescribeConfigResponseConfigEntry

+ Error error

+ ResourceName string

+ ResourceType int8

+ type DescribeConfigsRequest struct

+ Addr net.Addr

+ IncludeDocumentation bool

+ IncludeSynonyms bool

+ Resources []DescribeConfigRequestResource

+ type DescribeConfigsResponse struct

+ Resources []DescribeConfigResponseResource

+ Throttle time.Duration

+ type DescribeGroupsRequest struct

+ Addr net.Addr

+ GroupIDs []string

+ type DescribeGroupsResponse struct

+ Groups []DescribeGroupsResponseGroup

+ type DescribeGroupsResponseAssignments struct

+ Topics []GroupMemberTopic

+ UserData []byte

+ Version int

+ type DescribeGroupsResponseGroup struct

+ Error error

+ GroupID string

+ GroupState string

+ Members []DescribeGroupsResponseMember

+ type DescribeGroupsResponseMember struct

+ ClientHost string

+ ClientID string

+ MemberAssignments DescribeGroupsResponseAssignments

+ MemberID string

+ MemberMetadata DescribeGroupsResponseMemberMetadata

+ type DescribeGroupsResponseMemberMetadata struct

+ Topics []string

+ UserData []byte

+ Version int

type Dialer

+ DialFunc func(ctx context.Context, network string, address string) (net.Conn, error)

+ type ElectLeadersRequest struct

+ Addr net.Addr

+ Partitions []int

+ Timeout time.Duration

+ Topic string

+ type ElectLeadersResponse struct

+ Error error

+ PartitionResults []ElectLeadersResponsePartitionResult

+ type ElectLeadersResponsePartitionResult struct

+ Error error

+ Partition int

type Error

+ const ElectionNotNeeded

+ const EligibleLeadersNotAvailable

+ const FencedInstanceID

+ const GroupMaxSizeReached

+ const GroupSubscribedToTopic

+ const InvalidRecord

+ const MemberIDRequired

+ const NetworkException

+ const NoReassignmentInProgress

+ const OffsetNotAvailable

+ const PreferredLeaderNotAvailable

+ const StaleBrokerEpoch

+ const UnstableOffsetCommit

+ type FetchRequest struct

+ Addr net.Addr

+ IsolationLevel IsolationLevel

+ MaxBytes int64

+ MaxWait time.Duration

+ MinBytes int64

+ Offset int64

+ Partition int

+ Topic string

+ type FetchResponse struct

+ Error error

+ HighWatermark int64

+ LastStableOffset int64

+ LogStartOffset int64

+ Partition int

+ Records RecordReader

+ Throttle time.Duration

+ Topic string

+ type FindCoordinatorRequest struct

+ Addr net.Addr

+ Key string

+ KeyType CoordinatorKeyType

+ type FindCoordinatorResponse struct

+ Coordinator *FindCoordinatorResponseCoordinator

+ Error error

+ Throttle time.Duration

+ type FindCoordinatorResponseCoordinator struct

+ Host string

+ NodeID int

+ Port int

+ type Generation struct

+ Assignments map[string][]PartitionAssignment

+ GroupID string

+ ID int32

+ MemberID string

+ func (g *Generation) CommitOffsets(offsets map[string]map[int]int64) error

+ func (g *Generation) Start(fn func(ctx context.Context))

+ type GroupMemberTopic struct

+ Partitions []int

+ Topic string

+ type IncrementalAlterConfigsRequest struct

+ Addr net.Addr

+ Resources []IncrementalAlterConfigsRequestResource

+ ValidateOnly bool

+ type IncrementalAlterConfigsRequestConfig struct

+ ConfigOperation ConfigOperation

+ Name string

+ Value string

+ type IncrementalAlterConfigsRequestResource struct

+ Configs []IncrementalAlterConfigsRequestConfig

+ ResourceName string

+ ResourceType ResourceType

+ type IncrementalAlterConfigsResponse struct

+ Resources []IncrementalAlterConfigsResponseResource

+ type IncrementalAlterConfigsResponseResource struct

+ Error error

+ ResourceName string

+ ResourceType ResourceType

+ type InitProducerIDRequest struct

+ Addr net.Addr

+ TransactionTimeoutMs int

+ TransactionalID string

+ type InitProducerIDResponse struct

+ Error error

+ Producer *ProducerSession

+ Throttle time.Duration

+ type ListGroupsRequest struct

+ Addr net.Addr

+ type ListGroupsResponse struct

+ Error error

+ Groups []ListGroupsResponseGroup

+ type ListGroupsResponseGroup struct

+ Coordinator int

+ GroupID string

+ type ListOffsetsRequest struct

+ Addr net.Addr

+ IsolationLevel IsolationLevel

+ Topics map[string][]OffsetRequest

+ type ListOffsetsResponse struct

+ Throttle time.Duration

+ Topics map[string][]PartitionOffsets

+ type Logger interface

+ Printf func(string, ...interface{})

+ type LoggerFunc func(string, ...interface{})

+ func (f LoggerFunc) Printf(msg string, args ...interface{})

type Message

+ HighWaterMark int64

+ type MessageTooLargeError struct

+ Message Message

+ Remaining []Message

+ func (e MessageTooLargeError) Error() string

+ type MetadataRequest struct

+ Addr net.Addr

+ Topics []string

+ type MetadataResponse struct

+ Brokers []Broker

+ ClusterID string

+ Controller Broker

+ Throttle time.Duration

+ Topics []Topic

+ type Murmur2Balancer struct

+ Consistent bool

+ func (b Murmur2Balancer) Balance(msg Message, partitions ...int) (partition int)

+ type OffsetFetchPartition struct

+ CommittedOffset int64

+ Error error

+ Metadata string

+ Partition int

+ type OffsetFetchRequest struct

+ Addr net.Addr

+ GroupID string

+ Topics map[string][]int

+ type OffsetFetchResponse struct

+ Error error

+ Throttle time.Duration

+ Topics map[string][]OffsetFetchPartition

+ type OffsetRequest struct

+ Partition int

+ Timestamp int64

+ func FirstOffsetOf(partition int) OffsetRequest

+ func LastOffsetOf(partition int) OffsetRequest

+ func TimeOffsetOf(partition int, at time.Time) OffsetRequest

type Partition

+ Error error

+ type PartitionAssignment struct

+ ID int

+ Offset int64

+ type PartitionOffsets struct

+ Error error

+ FirstOffset int64

+ LastOffset int64

+ Offsets map[int64]time.Time

+ Partition int

+ type ProduceRequest struct

+ Addr net.Addr

+ Compression Compression

+ MessageVersion int

+ Partition int

+ Records RecordReader

+ RequiredAcks RequiredAcks

+ Topic string

+ TransactionalID string

+ type ProduceResponse struct

+ BaseOffset int64

+ Error error

+ LogAppendTime time.Time

+ LogStartOffset int64

+ RecordErrors map[int]error

+ Throttle time.Duration

+ type ProducerSession struct

+ ProducerEpoch int

+ ProducerID int

+ type RackAffinityGroupBalancer struct

+ Rack string

+ func (r RackAffinityGroupBalancer) AssignGroups(members []GroupMember, partitions []Partition) GroupMemberAssignments

+ func (r RackAffinityGroupBalancer) ProtocolName() string

+ func (r RackAffinityGroupBalancer) UserData() ([]byte, error)

type ReadBatchConfig

+ MaxWait time.Duration

type ReaderConfig

+ GroupTopics []string

+ JoinGroupBackoff time.Duration

+ ReadBackoffMax time.Duration

+ ReadBackoffMin time.Duration

+ StartOffset int64

+ type Record = protocol.Record

+ type RecordReader = protocol.RecordReader

+ func NewRecordReader(records ...Record) RecordReader

+ type Request = protocol.Message

+ type RequiredAcks int

+ const RequireAll

+ const RequireNone

+ const RequireOne

+ func (acks RequiredAcks) String() string

+ type ResourceType int8

+ const ResourceTypeBroker

+ const ResourceTypeTopic

+ const ResourceTypeUnknown

+ type Response = protocol.Message

+ type RoundTripper interface

+ RoundTrip func(context.Context, net.Addr, Request) (Response, error)

+ var DefaultTransport RoundTripper = &Transport{ ... }

+ type Topic struct

+ Error error

+ Internal bool

+ Name string

+ Partitions []Partition

+ type TopicAndGroup struct

+ GroupId string

+ Topic string

+ type TopicPartitionAssignment struct

+ BrokerIDs []int32

+ type TopicPartitionsConfig struct

+ Count int32

+ Name string

+ TopicPartitionAssignments []TopicPartitionAssignment

+ type Transport struct

+ ClientID string

+ Context context.Context

+ Dial func(context.Context, string, string) (net.Conn, error)

+ DialTimeout time.Duration

+ IdleTimeout time.Duration

+ MetadataTTL time.Duration

+ Resolver BrokerResolver

+ SASL sasl.Mechanism

+ TLS *tls.Config

+ func (t *Transport) CloseIdleConnections()

+ func (t *Transport) RoundTrip(ctx context.Context, addr net.Addr, req Request) (Response, error)

+ type Version int16

+ func (n Version) Marshal(v interface{}) ([]byte, error)

+ func (n Version) Unmarshal(b []byte, v interface{}) error

+ type WriteErrors []error

+ func (err WriteErrors) Count() int

+ func (err WriteErrors) Error() string

type Writer

+ Addr net.Addr

+ Async bool

+ Balancer Balancer

+ BatchBytes int64

+ BatchSize int

+ BatchTimeout time.Duration

+ Completion func(messages []Message, err error)

+ Compression Compression

+ ErrorLogger Logger

+ Logger Logger

+ MaxAttempts int

+ ReadTimeout time.Duration

+ RequiredAcks RequiredAcks

+ Topic string

+ Transport RoundTripper

+ WriteTimeout time.Duration

type WriterConfig

+ BatchBytes int

+ IdleConnTimeout time.Duration

type WriterStats

+ BatchBytes SummaryStats

+ BatchTime DurationStats

v0.2.3

May 3, 2019

Changes in this version

+ type ApiVersion struct

+ ApiKey int16

+ MaxVersion int16

+ MinVersion int16

type Conn

+ func (c *Conn) ApiVersions() ([]ApiVersion, error)

+ func (c *Conn) Brokers() ([]Broker, error)

+ func (c *Conn) Controller() (broker Broker, err error)

+ func (c *Conn) ReadBatchWith(cfg ReadBatchConfig) *Batch

+ func (c *Conn) WriteCompressedMessagesAt(codec CompressionCodec, msgs ...Message) (nbytes int, partition int32, offset int64, appendTime time.Time, err error)

type ConnConfig

+ TransactionalID string

type Dialer

+ SASLMechanism sasl.Mechanism

+ TransactionalID string

type Error

+ const DelegationTokenAuthDisabled

+ const DelegationTokenAuthorizationFailed

+ const DelegationTokenExpired

+ const DelegationTokenNotFound

+ const DelegationTokenOwnerMismatch

+ const DelegationTokenRequestNotAllowed

+ const FencedLeaderEpoch

+ const FetchSessionIDNotFound

+ const GroupIdNotFound

+ const InvalidFetchSessionEpoch

+ const InvalidPrincipalType

+ const KafkaStorageError

+ const ListenerNotFound

+ const LogDirNotFound

+ const NonEmptyGroup

+ const ReassignmentInProgress

+ const SASLAuthenticationFailed

+ const TopicDeletionDisabled

+ const UnknownLeaderEpoch

+ const UnknownProducerId

+ const UnsupportedCompressionType

+ type Header struct

+ Key string

+ Value []byte

+ type IsolationLevel int8

+ const ReadCommitted

+ const ReadUncommitted

type Message

+ Headers []Header

+ type ReadBatchConfig struct

+ IsolationLevel IsolationLevel

+ MaxBytes int

+ MinBytes int

type Reader

+ func (r *Reader) SetOffsetAt(ctx context.Context, t time.Time) error

type ReaderConfig

+ IsolationLevel IsolationLevel

+ MaxAttempts int

+ PartitionWatchInterval time.Duration

+ WatchPartitionChanges bool

+ func (config *ReaderConfig) Validate() error

type WriterConfig

+ func (config *WriterConfig) Validate() error

v0.2.2

Nov 27, 2018

Changes in this version

type ReaderStats

+ DeprecatedFetchesWithTypo int64

v0.2.1

Nov 15, 2018

Changes in this version

type Conn

+ func DialPartition(ctx context.Context, network string, address string, partition Partition) (*Conn, error)

+ func (c *Conn) WriteCompressedMessages(codec CompressionCodec, msgs ...Message) (int, error)

type Dialer

+ func (d *Dialer) DialPartition(ctx context.Context, network string, address string, partition Partition) (*Conn, error)

+ func (d *Dialer) LookupPartition(ctx context.Context, network string, address string, topic string, ...) (Partition, error)

type Partition

+ func LookupPartition(ctx context.Context, network string, address string, topic string, ...) (Partition, error)

+ func LookupPartitions(ctx context.Context, network string, address string, topic string) ([]Partition, error)

type WriterConfig

+ ErrorLogger *log.Logger

+ Logger *log.Logger

v0.2.0

Oct 25, 2018

Changes in this version

+ const FirstOffset

+ const LastOffset

type Broker

+ Rack string

+ type GroupBalancer interface

+ AssignGroups func(members []GroupMember, partitions []Partition) GroupMemberAssignments

+ ProtocolName func() string

+ UserData func() ([]byte, error)

+ type GroupMember struct

+ ID string

+ Topics []string

+ UserData []byte

+ type GroupMemberAssignments map[string]map[string][]int

+ type RangeGroupBalancer struct

+ func (r RangeGroupBalancer) AssignGroups(members []GroupMember, topicPartitions []Partition) GroupMemberAssignments

+ func (r RangeGroupBalancer) ProtocolName() string

+ func (r RangeGroupBalancer) UserData() ([]byte, error)

type ReaderConfig

+ GroupBalancers []GroupBalancer

+ type RoundRobinGroupBalancer struct

+ func (r RoundRobinGroupBalancer) AssignGroups(members []GroupMember, topicPartitions []Partition) GroupMemberAssignments

+ func (r RoundRobinGroupBalancer) ProtocolName() string

+ func (r RoundRobinGroupBalancer) UserData() ([]byte, error)

v0.1.0

Jul 16, 2018

Changes in this version

+ const CompressionNoneCode

+ const DefaultCompressionLevel

+ const SeekAbsolute

+ const SeekCurrent

+ const SeekEnd

+ const SeekStart

+ var DefaultClientID string

+ var DefaultDialer = &Dialer

+ func RegisterCompressionCodec(codec func() CompressionCodec)

+ type Balancer interface

+ Balance func(msg Message, partitions ...int) (partition int)

+ type BalancerFunc func(Message, ...int) int

+ func (f BalancerFunc) Balance(msg Message, partitions ...int) int

+ type Batch struct

+ func (batch *Batch) Close() error

+ func (batch *Batch) HighWaterMark() int64

+ func (batch *Batch) Offset() int64

+ func (batch *Batch) Read(b []byte) (int, error)

+ func (batch *Batch) ReadMessage() (Message, error)

+ func (batch *Batch) Throttle() time.Duration

+ type Broker struct

+ Host string

+ ID int

+ Port int

+ type CompressionCodec interface

+ Code func() int8

+ Decode func(dst, src []byte) (int, error)

+ Encode func(dst, src []byte) (int, error)

+ type ConfigEntry struct

+ ConfigName string

+ ConfigValue string

+ type Conn struct

+ func Dial(network string, address string) (*Conn, error)

+ func DialContext(ctx context.Context, network string, address string) (*Conn, error)

+ func DialLeader(ctx context.Context, network string, address string, topic string, ...) (*Conn, error)

+ func NewConn(conn net.Conn, topic string, partition int) *Conn

+ func NewConnWith(conn net.Conn, config ConnConfig) *Conn

+ func (c *Conn) Close() error

+ func (c *Conn) CreateTopics(topics ...TopicConfig) error

+ func (c *Conn) DeleteTopics(topics ...string) error

+ func (c *Conn) LocalAddr() net.Addr

+ func (c *Conn) Offset() (offset int64, whence int)

+ func (c *Conn) Read(b []byte) (int, error)

+ func (c *Conn) ReadBatch(minBytes int, maxBytes int) *Batch

+ func (c *Conn) ReadFirstOffset() (int64, error)

+ func (c *Conn) ReadLastOffset() (int64, error)

+ func (c *Conn) ReadMessage(maxBytes int) (Message, error)

+ func (c *Conn) ReadOffset(t time.Time) (int64, error)

+ func (c *Conn) ReadOffsets() (first int64, last int64, err error)

+ func (c *Conn) ReadPartitions(topics ...string) (partitions []Partition, err error)

+ func (c *Conn) RemoteAddr() net.Addr

+ func (c *Conn) Seek(offset int64, whence int) (int64, error)

+ func (c *Conn) SetDeadline(t time.Time) error

+ func (c *Conn) SetReadDeadline(t time.Time) error

+ func (c *Conn) SetRequiredAcks(n int) error

+ func (c *Conn) SetWriteDeadline(t time.Time) error

+ func (c *Conn) Write(b []byte) (int, error)

+ func (c *Conn) WriteMessages(msgs ...Message) (int, error)

+ type ConnConfig struct

+ ClientID string

+ Partition int

+ Topic string

+ type Dialer struct

+ ClientID string

+ Deadline time.Time

+ DualStack bool

+ FallbackDelay time.Duration

+ KeepAlive time.Duration

+ LocalAddr net.Addr

+ Resolver Resolver

+ TLS *tls.Config

+ Timeout time.Duration

+ func (d *Dialer) Dial(network string, address string) (*Conn, error)

+ func (d *Dialer) DialContext(ctx context.Context, network string, address string) (*Conn, error)

+ func (d *Dialer) DialLeader(ctx context.Context, network string, address string, topic string, ...) (*Conn, error)

+ func (d *Dialer) LookupLeader(ctx context.Context, network string, address string, topic string, ...) (Broker, error)

+ func (d *Dialer) LookupPartitions(ctx context.Context, network string, address string, topic string) ([]Partition, error)

+ type DurationStats struct

+ Avg time.Duration

+ Max time.Duration

+ Min time.Duration

+ type Error int

+ const BrokerAuthorizationFailed

+ const BrokerNotAvailable

+ const ClusterAuthorizationFailed

+ const ConcurrentTransactions

+ const DuplicateSequenceNumber

+ const GroupAuthorizationFailed

+ const GroupCoordinatorNotAvailable

+ const GroupLoadInProgress

+ const IllegalGeneration

+ const IllegalSASLState

+ const InconsistentGroupProtocol

+ const InvalidCommitOffsetSize

+ const InvalidConfiguration

+ const InvalidGroupId

+ const InvalidMessage

+ const InvalidMessageSize

+ const InvalidPartitionNumber

+ const InvalidProducerEpoch

+ const InvalidProducerIDMapping

+ const InvalidReplicaAssignment

+ const InvalidReplicationFactor

+ const InvalidRequest

+ const InvalidRequiredAcks

+ const InvalidSessionTimeout

+ const InvalidTimestamp

+ const InvalidTopic

+ const InvalidTransactionState

+ const InvalidTransactionTimeout

+ const LeaderNotAvailable

+ const MessageSizeTooLarge

+ const NotController

+ const NotCoordinatorForGroup

+ const NotEnoughReplicas

+ const NotEnoughReplicasAfterAppend

+ const NotLeaderForPartition

+ const OffsetMetadataTooLarge

+ const OffsetOutOfRange

+ const OutOfOrderSequenceNumber

+ const PolicyViolation

+ const RebalanceInProgress

+ const RecordListTooLarge

+ const ReplicaNotAvailable

+ const RequestTimedOut

+ const SecurityDisabled

+ const StaleControllerEpoch

+ const TopicAlreadyExists

+ const TopicAuthorizationFailed

+ const TransactionCoordinatorFenced

+ const TransactionalIDAuthorizationFailed

+ const Unknown

+ const UnknownMemberId

+ const UnknownTopicOrPartition

+ const UnsupportedForMessageFormat

+ const UnsupportedSASLMechanism

+ const UnsupportedVersion

+ func (e Error) Description() string

+ func (e Error) Error() string

+ func (e Error) Temporary() bool

+ func (e Error) Timeout() bool

+ func (e Error) Title() string

+ type Hash struct

+ Hasher hash.Hash32

+ func (h *Hash) Balance(msg Message, partitions ...int) (partition int)

+ type LeastBytes struct

+ func (lb *LeastBytes) Balance(msg Message, partitions ...int) int

+ type ListGroupsResponseGroupV1 struct

+ GroupID string

+ ProtocolType string

+ type Message struct

+ Key []byte

+ Offset int64

+ Partition int

+ Time time.Time

+ Topic string

+ Value []byte

+ type Partition struct

+ ID int

+ Isr []Broker

+ Leader Broker

+ Replicas []Broker

+ Topic string

+ type Reader struct

+ func NewReader(config ReaderConfig) *Reader

+ func (r *Reader) Close() error

+ func (r *Reader) CommitMessages(ctx context.Context, msgs ...Message) error

+ func (r *Reader) Config() ReaderConfig

+ func (r *Reader) FetchMessage(ctx context.Context) (Message, error)

+ func (r *Reader) Lag() int64

+ func (r *Reader) Offset() int64

+ func (r *Reader) ReadLag(ctx context.Context) (lag int64, err error)

+ func (r *Reader) ReadMessage(ctx context.Context) (Message, error)

+ func (r *Reader) SetOffset(offset int64) error

+ func (r *Reader) Stats() ReaderStats

+ type ReaderConfig struct

+ Brokers []string

+ CommitInterval time.Duration

+ Dialer *Dialer

+ ErrorLogger *log.Logger

+ GroupID string

+ HeartbeatInterval time.Duration

+ Logger *log.Logger

+ MaxBytes int

+ MaxWait time.Duration

+ MinBytes int

+ Partition int

+ QueueCapacity int

+ ReadLagInterval time.Duration

+ RebalanceTimeout time.Duration

+ RetentionTime time.Duration

+ SessionTimeout time.Duration

+ Topic string

+ type ReaderStats struct

+ Bytes int64

+ ClientID string

+ DialTime DurationStats

+ Dials int64

+ Errors int64

+ FetchBytes SummaryStats

+ FetchSize SummaryStats

+ Fetches int64

+ Lag int64

+ MaxBytes int64

+ MaxWait time.Duration

+ Messages int64

+ MinBytes int64

+ Offset int64

+ Partition string

+ QueueCapacity int64

+ QueueLength int64

+ ReadTime DurationStats

+ Rebalances int64

+ Timeouts int64

+ Topic string

+ WaitTime DurationStats

+ type ReplicaAssignment struct

+ Partition int

+ Replicas int

+ type Resolver interface

+ LookupHost func(ctx context.Context, host string) (addrs []string, err error)

+ type RoundRobin struct

+ func (rr *RoundRobin) Balance(msg Message, partitions ...int) int

+ type SummaryStats struct

+ Avg int64

+ Max int64

+ Min int64

+ type TopicConfig struct

+ ConfigEntries []ConfigEntry

+ NumPartitions int

+ ReplicaAssignments []ReplicaAssignment

+ ReplicationFactor int

+ Topic string

+ type Writer struct

+ func NewWriter(config WriterConfig) *Writer

+ func (w *Writer) Close() (err error)

+ func (w *Writer) Stats() WriterStats

+ func (w *Writer) WriteMessages(ctx context.Context, msgs ...Message) error

+ type WriterConfig struct

+ Async bool

+ Balancer Balancer

+ BatchSize int

+ BatchTimeout time.Duration

+ Brokers []string

+ Dialer *Dialer

+ MaxAttempts int

+ QueueCapacity int

+ ReadTimeout time.Duration

+ RebalanceInterval time.Duration

+ RequiredAcks int

+ Topic string

+ WriteTimeout time.Duration

+ type WriterStats struct

+ Async bool

+ BatchSize SummaryStats

+ BatchTimeout time.Duration

+ Bytes int64

+ ClientID string

+ DialTime DurationStats

+ Dials int64

+ Errors int64

+ MaxAttempts int64

+ MaxBatchSize int64

+ Messages int64

+ QueueCapacity int64

+ QueueLength int64

+ ReadTimeout time.Duration

+ RebalanceInterval time.Duration

+ Rebalances int64

+ RequiredAcks int64

+ Retries SummaryStats

+ Topic string

+ WaitTime DurationStats

+ WriteTime DurationStats

+ WriteTimeout time.Duration

+ Writes int64

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL