kafka

package

Go to main page

Versions in this module

v1

v1.8.4

Jan 27, 2022

v1.8.3

Jan 27, 2022

v1.8.2

Dec 13, 2021

Changes in this version

type Consumer

+ func (c *Consumer) StoreMessage(m *Message) (storedOffsets []TopicPartition, err error)

v1.7.0

May 10, 2021

Changes in this version

+ const LibrdkafkaLinkInfo — windows/amd64

v1.6.1

Feb 26, 2021

Changes in this version

type Consumer

+ func (c *Consumer) AssignmentLost() bool

+ func (c *Consumer) GetRebalanceProtocol() string

+ func (c *Consumer) IncrementalAssign(partitions []TopicPartition) (err error)

+ func (c *Consumer) IncrementalUnassign(partitions []TopicPartition) (err error)

type ErrorCode

+ const ErrAssignmentLost

+ const ErrAutoOffsetReset

+ const ErrDuplicateResource

+ const ErrFeatureUpdateFailed

+ const ErrInconsistentVoterSet

+ const ErrInvalidUpdateVersion

+ const ErrNoop

+ const ErrPrincipalDeserializationFailure

+ const ErrProducerFenced

+ const ErrResourceNotFound

+ const ErrThrottlingQuotaExceeded

+ const ErrUnacceptableCredential

v1.5.2

Nov 5, 2020

Changes in this version

type ErrorCode

+ const ErrElectionNotNeeded

+ const ErrEligibleLeadersNotAvailable

+ const ErrGroupSubscribedToTopic

+ const ErrInvalidRecord

+ const ErrNoReassignmentInProgress

+ const ErrUnstableOffsetCommit

v1.4.2

May 6, 2020

v1.4.0

Apr 8, 2020

Changes in this version

+ const LibrdkafkaLinkInfo — darwin/amd64, linux/amd64

+ const PurgeInFlight

+ const PurgeNonBlocking

+ const PurgeQueue

+ func WriteErrorCodes(f *os.File)

type AdminClient

+ func (a *AdminClient) ClusterID(ctx context.Context) (clusterID string, err error)

+ func (a *AdminClient) ControllerID(ctx context.Context) (controllerID int32, err error)

type Consumer

+ func (c *Consumer) GetConsumerGroupMetadata() (*ConsumerGroupMetadata, error)

+ func (c *Consumer) Logs() chan LogEvent

+ func (c *Consumer) Position(partitions []TopicPartition) (offsets []TopicPartition, err error)

+ type ConsumerGroupMetadata struct

+ func NewTestConsumerGroupMetadata(groupID string) (*ConsumerGroupMetadata, error)

type Error

+ func (e Error) IsRetriable() bool

+ func (e Error) TxnRequiresAbort() bool

type ErrorCode

+ const ErrApplication

+ const ErrCoordinatorLoadInProgress

+ const ErrCoordinatorNotAvailable

+ const ErrFenced

+ const ErrFencedInstanceID

+ const ErrNotConfigured

+ const ErrNotCoordinator

+ const ErrUnknownBroker

+ type LogEvent struct

+ Level int

+ Message string

+ Name string

+ Tag string

+ Timestamp time.Time

+ func (logEvent LogEvent) String() string

type Producer

+ func (p *Producer) AbortTransaction(ctx context.Context) error

+ func (p *Producer) BeginTransaction() error

+ func (p *Producer) CommitTransaction(ctx context.Context) error

+ func (p *Producer) InitTransactions(ctx context.Context) error

+ func (p *Producer) Logs() chan LogEvent

+ func (p *Producer) Purge(flags int) error

+ func (p *Producer) SendOffsetsToTransaction(ctx context.Context, offsets []TopicPartition, ...) error

v1.1.0

Jul 15, 2019

Changes in this version

type AdminClient

+ func (a *AdminClient) SetOAuthBearerToken(oauthBearerToken OAuthBearerToken) error

+ func (a *AdminClient) SetOAuthBearerTokenFailure(errstr string) error

type Consumer

+ func (c *Consumer) GetWatermarkOffsets(topic string, partition int32) (low, high int64, err error)

+ func (c *Consumer) SetOAuthBearerToken(oauthBearerToken OAuthBearerToken) error

+ func (c *Consumer) SetOAuthBearerTokenFailure(errstr string) error

type ErrorCode

+ const ErrFencedLeaderEpoch

+ const ErrGroupMaxSizeReached

+ const ErrMemberIDRequired

+ const ErrOffsetNotAvailable

+ const ErrPreferredLeaderNotAvailable

+ const ErrStaleBrokerEpoch

+ const ErrUnknownLeaderEpoch

type Handle

+ SetOAuthBearerToken func(oauthBearerToken OAuthBearerToken) error

+ SetOAuthBearerTokenFailure func(errstr string) error

+ type OAuthBearerToken struct

+ Expiration time.Time

+ Extensions map[string]string

+ Principal string

+ TokenValue string

+ type OAuthBearerTokenRefresh struct

+ Config string

+ func (o OAuthBearerTokenRefresh) String() string

type Producer

+ func (p *Producer) SetOAuthBearerToken(oauthBearerToken OAuthBearerToken) error

+ func (p *Producer) SetOAuthBearerTokenFailure(errstr string) error

type TopicPartition

+ Metadata *string

v1.0.0

Mar 27, 2019

Changes in this version

type Error

+ func NewError(code ErrorCode, str string, fatal bool) (err Error)

+ func (e Error) IsFatal() bool

type ErrorCode

+ const ErrDelegationTokenAuthDisabled

+ const ErrDelegationTokenAuthorizationFailed

+ const ErrDelegationTokenExpired

+ const ErrDelegationTokenNotFound

+ const ErrDelegationTokenOwnerMismatch

+ const ErrDelegationTokenRequestNotAllowed

+ const ErrFatal

+ const ErrFetchSessionIDNotFound

+ const ErrGaplessGuarantee

+ const ErrGroupIDNotFound

+ const ErrInconsistent

+ const ErrInvalidFetchSessionEpoch

+ const ErrInvalidPrincipalType

+ const ErrKafkaStorageError

+ const ErrListenerNotFound

+ const ErrLogDirNotFound

+ const ErrMaxPollExceeded

+ const ErrNonEmptyGroup

+ const ErrPurgeInflight

+ const ErrPurgeQueue

+ const ErrReassignmentInProgress

+ const ErrRetry

+ const ErrSaslAuthenticationFailed

+ const ErrTopicDeletionDisabled

+ const ErrUnknownProducerID

+ const ErrUnsupportedCompressionType

type Producer

+ func (p *Producer) GetFatalError() error

+ func (p *Producer) TestFatalError(code ErrorCode, str string) ErrorCode

v0

v0.11.6

Oct 24, 2018

Changes in this version

+ const AlterOperationSet

+ const ConfigSourceDefault

+ const ConfigSourceDynamicBroker

+ const ConfigSourceDynamicDefaultBroker

+ const ConfigSourceDynamicTopic

+ const ConfigSourceStaticBroker

+ const ConfigSourceUnknown

+ const ResourceAny

+ const ResourceBroker

+ const ResourceGroup

+ const ResourceTopic

+ const ResourceUnknown

+ type AdminClient struct

+ func NewAdminClient(conf *ConfigMap) (*AdminClient, error)

+ func NewAdminClientFromConsumer(c *Consumer) (a *AdminClient, err error)

+ func NewAdminClientFromProducer(p *Producer) (a *AdminClient, err error)

+ func (a *AdminClient) AlterConfigs(ctx context.Context, resources []ConfigResource, ...) (result []ConfigResourceResult, err error)

+ func (a *AdminClient) Close()

+ func (a *AdminClient) CreatePartitions(ctx context.Context, partitions []PartitionsSpecification, ...) (result []TopicResult, err error)

+ func (a *AdminClient) CreateTopics(ctx context.Context, topics []TopicSpecification, ...) (result []TopicResult, err error)

+ func (a *AdminClient) DeleteTopics(ctx context.Context, topics []string, options ...DeleteTopicsAdminOption) (result []TopicResult, err error)

+ func (a *AdminClient) DescribeConfigs(ctx context.Context, resources []ConfigResource, ...) (result []ConfigResourceResult, err error)

+ func (a *AdminClient) GetMetadata(topic *string, allTopics bool, timeoutMs int) (*Metadata, error)

+ func (a *AdminClient) String() string

+ type AdminOption interface

+ type AdminOptionOperationTimeout struct

+ func SetAdminOperationTimeout(t time.Duration) (ao AdminOptionOperationTimeout)

+ type AdminOptionRequestTimeout struct

+ func SetAdminRequestTimeout(t time.Duration) (ao AdminOptionRequestTimeout)

+ type AdminOptionValidateOnly struct

+ func SetAdminValidateOnly(validateOnly bool) (ao AdminOptionValidateOnly)

+ type AlterConfigsAdminOption interface

+ type AlterOperation int

+ func (o AlterOperation) String() string

+ type ConfigEntry struct

+ Name string

+ Operation AlterOperation

+ Value string

+ func StringMapToConfigEntries(stringMap map[string]string, operation AlterOperation) []ConfigEntry

+ func (c ConfigEntry) String() string

+ type ConfigEntryResult struct

+ IsReadOnly bool

+ IsSensitive bool

+ IsSynonym bool

+ Name string

+ Source ConfigSource

+ Synonyms map[string]ConfigEntryResult

+ Value string

+ func (c ConfigEntryResult) String() string

+ type ConfigResource struct

+ Config []ConfigEntry

+ Name string

+ Type ResourceType

+ func (c ConfigResource) String() string

+ type ConfigResourceResult struct

+ Config map[string]ConfigEntryResult

+ Error Error

+ Name string

+ Type ResourceType

+ func (c ConfigResourceResult) String() string

+ type ConfigSource int

+ func (t ConfigSource) String() string

+ type CreatePartitionsAdminOption interface

+ type CreateTopicsAdminOption interface

+ type DeleteTopicsAdminOption interface

+ type DescribeConfigsAdminOption interface

type ErrorCode

+ const ErrInvalidType

+ type PartitionsSpecification struct

+ IncreaseTo int

+ ReplicaAssignment [][]int32

+ Topic string

+ type ResourceType int

+ func ResourceTypeFromString(typeString string) (ResourceType, error)

+ func (t ResourceType) String() string

+ type TopicResult struct

+ Error Error

+ Topic string

+ func (t TopicResult) String() string

+ type TopicSpecification struct

+ Config map[string]string

+ NumPartitions int

+ ReplicaAssignment [][]int32

+ ReplicationFactor int

+ Topic string

v0.11.4

Mar 13, 2018

Changes in this version

type ConfigMap

+ func (m ConfigMap) Get(key string, defval ConfigValue) (ConfigValue, error)

type Consumer

+ func (c *Consumer) Committed(partitions []TopicPartition, timeoutMs int) (offsets []TopicPartition, err error)

+ func (c *Consumer) OffsetsForTimes(times []TopicPartition, timeoutMs int) (offsets []TopicPartition, err error)

+ func (c *Consumer) Pause(partitions []TopicPartition) (err error)

+ func (c *Consumer) ReadMessage(timeout time.Duration) (*Message, error)

+ func (c *Consumer) Resume(partitions []TopicPartition) (err error)

+ func (c *Consumer) StoreOffsets(offsets []TopicPartition) (storedOffsets []TopicPartition, err error)

type ErrorCode

+ const ErrNoent

+ const ErrPartial

+ const ErrReadOnly

+ const ErrUnderflow

+ type Header struct

+ Key string

+ Value []byte

+ func (h Header) String() string

type Message

+ Headers []Header

type Producer

+ func (p *Producer) OffsetsForTimes(times []TopicPartition, timeoutMs int) (offsets []TopicPartition, err error)

v0.11.0

Jun 20, 2017

Changes in this version

type ErrorCode

+ const ErrConcurrentTransactions

+ const ErrDuplicateSequenceNumber

+ const ErrIntr

+ const ErrInvalidConfig

+ const ErrInvalidPartitions

+ const ErrInvalidProducerEpoch

+ const ErrInvalidProducerIDMapping

+ const ErrInvalidReplicaAssignment

+ const ErrInvalidReplicationFactor

+ const ErrInvalidRequest

+ const ErrInvalidTransactionTimeout

+ const ErrInvalidTxnState

+ const ErrKeyDeserialization

+ const ErrKeySerialization

+ const ErrNotController

+ const ErrOperationNotAttempted

+ const ErrOutOfOrderSequenceNumber

+ const ErrPolicyViolation

+ const ErrSecurityDisabled

+ const ErrTopicAlreadyExists

+ const ErrTransactionCoordinatorFenced

+ const ErrTransactionalIDAuthorizationFailed

+ const ErrUnsupportedFeature

+ const ErrUnsupportedForMessageFormat

+ const ErrValueDeserialization

+ const ErrValueSerialization

+ const ErrWaitCache

+ type Stats struct

+ func (e Stats) String() string

v0.9.4

Feb 8, 2017

Changes in this version

type Consumer

+ func (c *Consumer) Assignment() (partitions []TopicPartition, err error)

+ func (c *Consumer) Seek(partition TopicPartition, timeoutMs int) error

+ func (c *Consumer) Subscription() (topics []string, err error)

+ type TopicPartitions []TopicPartition

+ func (tps TopicPartitions) Len() int

+ func (tps TopicPartitions) Less(i, j int) bool

+ func (tps TopicPartitions) Swap(i, j int)

v0.9.2

Oct 19, 2016

Changes in this version

+ const OffsetBeginning

+ const OffsetEnd

+ const OffsetInvalid

+ const OffsetStored

+ const PartitionAny

+ const TimestampCreateTime

+ const TimestampLogAppendTime

+ const TimestampNotAvailable

+ func LibraryVersion() (int, string)

+ type AssignedPartitions struct

+ Partitions []TopicPartition

+ func (e AssignedPartitions) String() string

+ type BrokerMetadata struct

+ Host string

+ ID int32

+ Port int

+ type ConfigMap map[string]ConfigValue

+ func (m ConfigMap) Set(kv string) error

+ func (m ConfigMap) SetKey(key string, value ConfigValue) error

+ type ConfigValue interface

+ type Consumer struct

+ func NewConsumer(conf *ConfigMap) (*Consumer, error)

+ func (c *Consumer) Assign(partitions []TopicPartition) (err error)

+ func (c *Consumer) Close() (err error)

+ func (c *Consumer) Commit() ([]TopicPartition, error)

+ func (c *Consumer) CommitMessage(m *Message) ([]TopicPartition, error)

+ func (c *Consumer) CommitOffsets(offsets []TopicPartition) ([]TopicPartition, error)

+ func (c *Consumer) Events() chan Event

+ func (c *Consumer) GetMetadata(topic *string, allTopics bool, timeoutMs int) (*Metadata, error)

+ func (c *Consumer) Poll(timeoutMs int) (event Event)

+ func (c *Consumer) QueryWatermarkOffsets(topic string, partition int32, timeoutMs int) (low, high int64, err error)

+ func (c *Consumer) String() string

+ func (c *Consumer) Subscribe(topic string, rebalanceCb RebalanceCb) error

+ func (c *Consumer) SubscribeTopics(topics []string, rebalanceCb RebalanceCb) (err error)

+ func (c *Consumer) Unassign() (err error)

+ func (c *Consumer) Unsubscribe() (err error)

+ type Error struct

+ func (e Error) Code() ErrorCode

+ func (e Error) Error() string

+ func (e Error) String() string

+ type ErrorCode int

+ const ErrAllBrokersDown

+ const ErrAssignPartitions

+ const ErrAuthentication

+ const ErrBadCompression

+ const ErrBadMsg

+ const ErrBrokerNotAvailable

+ const ErrClusterAuthorizationFailed

+ const ErrConflict

+ const ErrCritSysResource

+ const ErrDestroy

+ const ErrExistingSubscription

+ const ErrFail

+ const ErrFs

+ const ErrGroupAuthorizationFailed

+ const ErrGroupCoordinatorNotAvailable

+ const ErrGroupLoadInProgress

+ const ErrIllegalGeneration

+ const ErrIllegalSaslState

+ const ErrInProgress

+ const ErrInconsistentGroupProtocol

+ const ErrInvalidArg

+ const ErrInvalidCommitOffsetSize

+ const ErrInvalidGroupID

+ const ErrInvalidMsg

+ const ErrInvalidMsgSize

+ const ErrInvalidRequiredAcks

+ const ErrInvalidSessionTimeout

+ const ErrInvalidTimestamp

+ const ErrIsrInsuff

+ const ErrLeaderNotAvailable

+ const ErrMsgSizeTooLarge

+ const ErrMsgTimedOut

+ const ErrNetworkException

+ const ErrNoError

+ const ErrNoOffset

+ const ErrNodeUpdate

+ const ErrNotCoordinatorForGroup

+ const ErrNotEnoughReplicas

+ const ErrNotEnoughReplicasAfterAppend

+ const ErrNotImplemented

+ const ErrNotLeaderForPartition

+ const ErrOffsetMetadataTooLarge

+ const ErrOffsetOutOfRange

+ const ErrOutdated

+ const ErrPartitionEOF

+ const ErrPrevInProgress

+ const ErrQueueFull

+ const ErrRebalanceInProgress

+ const ErrRecordListTooLarge

+ const ErrReplicaNotAvailable

+ const ErrRequestTimedOut

+ const ErrResolve

+ const ErrRevokePartitions

+ const ErrSsl

+ const ErrStaleCtrlEpoch

+ const ErrState

+ const ErrTimedOut

+ const ErrTimedOutQueue

+ const ErrTopicAuthorizationFailed

+ const ErrTopicException

+ const ErrTransport

+ const ErrUnknown

+ const ErrUnknownGroup

+ const ErrUnknownMemberID

+ const ErrUnknownPartition

+ const ErrUnknownProtocol

+ const ErrUnknownTopic

+ const ErrUnknownTopicOrPart

+ const ErrUnsupportedSaslMechanism

+ const ErrUnsupportedVersion

+ const ErrWaitCoord

+ func (c ErrorCode) String() string

+ type Event interface

+ String func() string

+ type Handle interface

+ type Message struct

+ Key []byte

+ Opaque interface{}

+ Timestamp time.Time

+ TimestampType TimestampType

+ TopicPartition TopicPartition

+ Value []byte

+ func (m *Message) String() string

+ type Metadata struct

+ Brokers []BrokerMetadata

+ OriginatingBroker BrokerMetadata

+ Topics map[string]TopicMetadata

+ type Offset int64

+ func NewOffset(offset interface{}) (Offset, error)

+ func OffsetTail(relativeOffset Offset) Offset

+ func (o Offset) Set(offset interface{}) error

+ func (o Offset) String() string

+ type OffsetsCommitted struct

+ Error error

+ Offsets []TopicPartition

+ func (o OffsetsCommitted) String() string

+ type PartitionEOF TopicPartition

+ func (p PartitionEOF) String() string

+ type PartitionMetadata struct

+ Error Error

+ ID int32

+ Isrs []int32

+ Leader int32

+ Replicas []int32

+ type Producer struct

+ func NewProducer(conf *ConfigMap) (*Producer, error)

+ func (p *Producer) Close()

+ func (p *Producer) Events() chan Event

+ func (p *Producer) Flush(timeoutMs int) int

+ func (p *Producer) GetMetadata(topic *string, allTopics bool, timeoutMs int) (*Metadata, error)

+ func (p *Producer) Len() int

+ func (p *Producer) Produce(msg *Message, deliveryChan chan Event) error

+ func (p *Producer) ProduceChannel() chan *Message

+ func (p *Producer) QueryWatermarkOffsets(topic string, partition int32, timeoutMs int) (low, high int64, err error)

+ func (p *Producer) String() string

+ type RebalanceCb func(*Consumer, Event) error

+ type RevokedPartitions struct

+ Partitions []TopicPartition

+ func (e RevokedPartitions) String() string

+ type TimestampType int

+ func (t TimestampType) String() string

+ type TopicMetadata struct

+ Error Error

+ Partitions []PartitionMetadata

+ Topic string

+ type TopicPartition struct

+ Error error

+ Offset Offset

+ Partition int32

+ Topic *string

+ func (p TopicPartition) String() string

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