sarama

Jan 10, 2023

Changes in this version

+ var V3_3_0_0 = newKafkaVersion(3, 3, 0, 0)

+ var V3_3_1_0 = newKafkaVersion(3, 3, 1, 0)

type Client

+ LeaderAndEpoch func(topic string, partitionID int32) (*Broker, int32, error)

type MetadataRequest

+ func NewMetadataRequest(version KafkaVersion, topics []string) *MetadataRequest

type PartitionMetadata

+ LeaderEpoch int32

+ Version int16

type TopicMetadata

+ Version int16

Sep 27, 2022

Changes in this version

+ var ErrAddPartitionsToTxn = errors.New("transaction manager: failed to send partitions to transaction")

+ var ErrCannotTransitionNilError = errors.New("transaction manager: cannot transition with a nil error")

+ var ErrNonTransactedProducer = errors.New("transaction manager: you need to add TransactionalID to producer")

+ var ErrTransactionNotReady = errors.New("transaction manager: transaction is not ready")

+ var ErrTransitionNotAllowed = errors.New("transaction manager: invalid transition attempted")

+ var ErrTxnOffsetCommit = errors.New("transaction manager: failed to send offsets to transaction")

+ var ErrTxnUnableToParseResponse = errors.New("transaction manager: unable to parse response")

+ var V2_8_2_0 = newKafkaVersion(2, 8, 2, 0)

+ var V3_0_2_0 = newKafkaVersion(3, 0, 2, 0)

+ var V3_1_1_0 = newKafkaVersion(3, 1, 1, 0)

+ var V3_1_2_0 = newKafkaVersion(3, 1, 2, 0)

+ var V3_2_1_0 = newKafkaVersion(3, 2, 1, 0)

+ var V3_2_2_0 = newKafkaVersion(3, 2, 2, 0)

+ var V3_2_3_0 = newKafkaVersion(3, 2, 3, 0)

type AsyncProducer

+ AbortTxn func() error

+ AddMessageToTxn func(msg *ConsumerMessage, groupId string, metadata *string) error

+ AddOffsetsToTxn func(offsets map[string][]*PartitionOffsetMetadata, groupId string) error

+ BeginTxn func() error

+ CommitTxn func() error

+ IsTransactional func() bool

+ TxnStatus func() ProducerTxnStatusFlag

type Broker

+ func (b *Broker) ResponseSize() int

type Client

+ LeastLoadedBroker func() *Broker

+ RefreshTransactionCoordinator func(transactionID string) error

+ TransactionCoordinator func(transactionID string) (*Broker, error)

type InitProducerIDRequest

+ ProducerEpoch int16

+ ProducerID int64

+ Version int16

type InitProducerIDResponse

+ Version int16

type KError

+ const ErrProducerFenced

+ const ErrThrottlingQuotaExceeded

type MockFetchResponse

+ func (mfr *MockFetchResponse) SetMessageWithKey(topic string, partition int32, offset int64, key, msg Encoder) *MockFetchResponse

+ type ProducerTxnStatusFlag int16

+ const ProducerTxnFlagAbortableError

+ const ProducerTxnFlagAbortingTransaction

+ const ProducerTxnFlagCommittingTransaction

+ const ProducerTxnFlagEndTransaction

+ const ProducerTxnFlagFatalError

+ const ProducerTxnFlagInError

+ const ProducerTxnFlagInTransaction

+ const ProducerTxnFlagInitializing

+ const ProducerTxnFlagReady

+ const ProducerTxnFlagUninitialized

+ func (s ProducerTxnStatusFlag) String() string

type SyncProducer

+ AbortTxn func() error

+ AddMessageToTxn func(msg *ConsumerMessage, groupId string, metadata *string) error

+ AddOffsetsToTxn func(offsets map[string][]*PartitionOffsetMetadata, groupId string) error

+ BeginTxn func() error

+ CommitTxn func() error

+ IsTransactional func() bool

+ TxnStatus func() ProducerTxnStatusFlag

Aug 11, 2022

Changes in this version

+ var V3_1_0_1 = newKafkaVersion(3, 1, 0, 1)

type ProducerMessage

+ func (m *ProducerMessage) ByteSize(version int) int

Jul 22, 2022

Changes in this version

+ var V3_2_0_0 = newKafkaVersion(3, 2, 0, 0)

+ type SyncGroupRequestAssignment struct

+ Assignment []byte

+ MemberId string

Jun 7, 2022

Changes in this version

type GroupDescription

+ AuthorizedOperations int32

+ ErrorCode int16

type GroupMemberDescription

+ MemberId string

May 30, 2022

Changes in this version

+ const MAX_GROUP_INSTANCE_ID_LENGTH

+ var GROUP_INSTANCE_ID_REGEXP = regexp.MustCompile(`^[0-9a-zA-Z\._\-]+$`)

type ClusterAdmin

+ RemoveMemberFromConsumerGroup func(groupId string, groupInstanceIds []string) (*LeaveGroupResponse, error)

type DescribeGroupsRequest

+ IncludeAuthorizedOperations bool

+ Version int16

type DescribeGroupsResponse

+ AuthorizedOperations int32

+ ThrottleTimeMs int32

+ Version int16

type GroupDescription

+ Version int16

+ type GroupMember struct

+ GroupInstanceId *string

+ MemberId string

+ Metadata []byte

type GroupMemberDescription

+ GroupInstanceId *string

+ Version int16

type HeartbeatRequest

+ GroupInstanceId *string

+ Version int16

type HeartbeatResponse

+ ThrottleTime int32

+ Version int16

type JoinGroupRequest

+ GroupInstanceId *string

type LeaveGroupRequest

+ Members []MemberIdentity

+ Version int16

type LeaveGroupResponse

+ Members []MemberResponse

+ ThrottleTime int32

+ Version int16

+ type MemberIdentity struct

+ GroupInstanceId *string

+ MemberId string

+ type MemberResponse struct

+ Err KError

+ GroupInstanceId *string

+ MemberId string

type OffsetCommitRequest

+ GroupInstanceId *string

type SyncGroupRequest

+ GroupInstanceId *string

+ Version int16

type SyncGroupResponse

+ ThrottleTime int32

+ Version int16

May 11, 2022

Changes in this version

+ var ErrCreateACLs = errors.New("kafka server: failed to create one or more ACL rules")

+ var MultiErrorFormat multierror.ErrorFormatFunc = func(es []error) string

+ var V3_0_1_0 = newKafkaVersion(3, 0, 1, 0)

+ func Wrap(sentinel error, wrapped ...error) sentinelError

type ClusterAdmin

+ CreateACLs func([]*ResourceAcls) error

type CompressionCodec

+ func (cc *CompressionCodec) UnmarshalText(text []byte) error

+ func (cc CompressionCodec) MarshalText() ([]byte, error)

+ type MockCreateAclsResponseError struct

+ func NewMockCreateAclsResponseWithError(t TestReporter) *MockCreateAclsResponseError

+ func (mr *MockCreateAclsResponseError) For(reqBody versionedDecoder) encoderWithHeader

type MockSaslAuthenticateResponse

+ func (msar *MockSaslAuthenticateResponse) SetSessionLifetimeMs(sessionLifetimeMs int64) *MockSaslAuthenticateResponse

type SaslAuthenticateRequest

+ Version int16

type SaslAuthenticateResponse

+ SessionLifetimeMs int64

+ Version int16

type TopicError

+ func (t *TopicError) Unwrap() error

type TopicPartitionError

+ func (t *TopicPartitionError) Unwrap() error

Feb 1, 2022

retracted

Changes in this version

+ var V2_6_3_0 = newKafkaVersion(2, 6, 3, 0)

+ var V2_7_2_0 = newKafkaVersion(2, 7, 2, 0)

+ var V3_1_0_0 = newKafkaVersion(3, 1, 0, 0)

type Broker

+ func (b *Broker) TLSConnectionState() (state tls.ConnectionState, ok bool)

type Consumer

+ Pause func(topicPartitions map[string][]int32)

+ PauseAll func()

+ Resume func(topicPartitions map[string][]int32)

+ ResumeAll func()

type ConsumerGroup

+ Pause func(partitions map[string][]int32)

+ PauseAll func()

+ Resume func(partitions map[string][]int32)

+ ResumeAll func()

type MockApiVersionsResponse

+ func (m *MockApiVersionsResponse) SetApiKeys(apiKeys []ApiVersionsResponseKey) *MockApiVersionsResponse

type PartitionConsumer

+ IsPaused func() bool

+ Pause func()

+ Resume func()

Jan 18, 2022

retracted

Changes in this version

type Broker

+ func (b *Broker) AsyncProduce(request *ProduceRequest, cb ProduceCallback) error

type ClusterAdmin

+ IncrementalAlterConfig func(resourceType ConfigResourceType, name string, ...) error

type ConsumerGroupMemberMetadata

+ OwnedPartitions []*OwnedPartition

+ type MockIncrementalAlterConfigsResponse struct

+ func NewMockIncrementalAlterConfigsResponse(t TestReporter) *MockIncrementalAlterConfigsResponse

+ func (mr *MockIncrementalAlterConfigsResponse) For(reqBody versionedDecoder) encoderWithHeader

+ type MockIncrementalAlterConfigsResponseWithErrorCode struct

+ func NewMockIncrementalAlterConfigsResponseWithErrorCode(t TestReporter) *MockIncrementalAlterConfigsResponseWithErrorCode

+ func (mr *MockIncrementalAlterConfigsResponseWithErrorCode) For(reqBody versionedDecoder) encoderWithHeader

+ type OwnedPartition struct

+ Partitions []int32

+ Topic string

+ type ProduceCallback func(*ProduceResponse, error)

Dec 4, 2021

Changes in this version

type FetchResponseBlock

+ LastRecordsBatchOffset *int64

+ type ZstdDecoderParams struct

+ type ZstdEncoderParams struct

+ Level int

Sep 22, 2021

Changes in this version

+ var V0_10_2_2 = newKafkaVersion(0, 10, 2, 2)

+ var V1_0_1_0 = newKafkaVersion(1, 0, 1, 0)

+ var V1_0_2_0 = newKafkaVersion(1, 0, 2, 0)

+ var V2_1_1_0 = newKafkaVersion(2, 1, 1, 0)

+ var V2_2_1_0 = newKafkaVersion(2, 2, 1, 0)

+ var V2_2_2_0 = newKafkaVersion(2, 2, 2, 0)

+ var V2_3_1_0 = newKafkaVersion(2, 3, 1, 0)

+ var V2_4_1_0 = newKafkaVersion(2, 4, 1, 0)

+ var V2_5_1_0 = newKafkaVersion(2, 5, 1, 0)

+ var V2_6_1_0 = newKafkaVersion(2, 6, 1, 0)

+ var V2_6_2_0 = newKafkaVersion(2, 6, 2, 0)

+ var V2_7_1_0 = newKafkaVersion(2, 7, 1, 0)

+ var V2_8_1_0 = newKafkaVersion(2, 8, 1, 0)

+ var V3_0_0_0 = newKafkaVersion(3, 0, 0, 0)

+ type AlterClientQuotasEntry struct

+ Entity []QuotaEntityComponent

+ Ops []ClientQuotasOp

+ type AlterClientQuotasEntryResponse struct

+ Entity []QuotaEntityComponent

+ ErrorCode KError

+ ErrorMsg *string

+ type AlterClientQuotasRequest struct

+ Entries []AlterClientQuotasEntry

+ ValidateOnly bool

+ type AlterClientQuotasResponse struct

+ Entries []AlterClientQuotasEntryResponse

+ ThrottleTime time.Duration

type ApiVersionsRequest

+ ClientSoftwareName string

+ ClientSoftwareVersion string

+ Version int16

type ApiVersionsResponse

+ ApiKeys []ApiVersionsResponseKey

+ ErrorCode int16

+ ThrottleTimeMs int32

+ Version int16

+ type ApiVersionsResponseKey struct

+ ApiKey int16

+ MaxVersion int16

+ MinVersion int16

+ Version int16

type Broker

+ func (b *Broker) AlterClientQuotas(request *AlterClientQuotasRequest) (*AlterClientQuotasResponse, error)

+ func (b *Broker) DeleteOffsets(request *DeleteOffsetsRequest) (*DeleteOffsetsResponse, error)

+ func (b *Broker) DescribeClientQuotas(request *DescribeClientQuotasRequest) (*DescribeClientQuotasResponse, error)

+ type ClientQuotasOp struct

+ Key string

+ Remove bool

+ Value float64

type ClusterAdmin

+ AlterClientQuotas func(entity []QuotaEntityComponent, op ClientQuotasOp, validateOnly bool) error

+ Controller func() (*Broker, error)

+ DeleteConsumerGroupOffset func(group string, topic string, partition int32) error

+ DescribeClientQuotas func(components []QuotaFilterComponent, strict bool) ([]DescribeClientQuotasEntry, error)

type Config

+ ApiVersionsRequest bool

+ type DeleteOffsetsRequest struct

+ Group string

+ func (r *DeleteOffsetsRequest) AddPartition(topic string, partitionID int32)

+ type DeleteOffsetsResponse struct

+ ErrorCode KError

+ Errors map[string]map[int32]KError

+ ThrottleTime time.Duration

+ func (r *DeleteOffsetsResponse) AddError(topic string, partition int32, errorCode KError)

+ type DescribeClientQuotasEntry struct

+ Entity []QuotaEntityComponent

+ Values map[string]float64

+ type DescribeClientQuotasRequest struct

+ Components []QuotaFilterComponent

+ Strict bool

+ type DescribeClientQuotasResponse struct

+ Entries []DescribeClientQuotasEntry

+ ErrorCode KError

+ ErrorMsg *string

+ ThrottleTime time.Duration

+ type MockApiVersionsResponse struct

+ func NewMockApiVersionsResponse(t TestReporter) *MockApiVersionsResponse

+ func (mr *MockApiVersionsResponse) For(reqBody versionedDecoder) encoderWithHeader

+ type MockDeleteOffsetResponse struct

+ func NewMockDeleteOffsetRequest(t TestReporter) *MockDeleteOffsetResponse

+ func (m *MockDeleteOffsetResponse) For(reqBody versionedDecoder) encoderWithHeader

+ func (m *MockDeleteOffsetResponse) SetDeletedOffset(errorCode KError, topic string, partition int32, errorPartition KError) *MockDeleteOffsetResponse

+ type QuotaEntityComponent struct

+ EntityType QuotaEntityType

+ MatchType QuotaMatchType

+ Name string

+ type QuotaEntityType string

+ const QuotaEntityClientID

+ const QuotaEntityIP

+ const QuotaEntityUser

+ type QuotaFilterComponent struct

+ EntityType QuotaEntityType

+ Match string

+ MatchType QuotaMatchType

+ type QuotaMatchType int

+ const QuotaMatchAny

+ const QuotaMatchDefault

+ const QuotaMatchExact

type StdLogger

+ var DebugLogger StdLogger = &debugLogger{}

Jun 22, 2021

Changes in this version

type AclResourceType

+ const AclResourceDelegationToken

type Broker

+ func (b *Broker) IncrementalAlterConfigs(request *IncrementalAlterConfigsRequest) (*IncrementalAlterConfigsResponse, error)

+ type IncrementalAlterConfigsEntry struct

+ Operation IncrementalAlterConfigsOperation

+ Value *string

+ type IncrementalAlterConfigsOperation int8

+ const IncrementalAlterConfigsOperationAppend

+ const IncrementalAlterConfigsOperationDelete

+ const IncrementalAlterConfigsOperationSet

+ const IncrementalAlterConfigsOperationSubtract

+ type IncrementalAlterConfigsRequest struct

+ Resources []*IncrementalAlterConfigsResource

+ ValidateOnly bool

+ type IncrementalAlterConfigsResource struct

+ ConfigEntries map[string]IncrementalAlterConfigsEntry

+ Name string

+ Type ConfigResourceType

+ type IncrementalAlterConfigsResponse struct

+ Resources []*AlterConfigsResourceResponse

+ ThrottleTime time.Duration

May 7, 2021

Changes in this version

+ var ErrUnknownScramMechanism = errors.New("kafka: unknown SCRAM mechanism provided")

+ var V2_8_0_0 = newKafkaVersion(2, 8, 0, 0)

type AclOperation

+ func (a *AclOperation) MarshalText() ([]byte, error)

+ func (a *AclOperation) String() string

+ func (a *AclOperation) UnmarshalText(text []byte) error

type AclPermissionType

+ func (a *AclPermissionType) MarshalText() ([]byte, error)

+ func (a *AclPermissionType) String() string

+ func (a *AclPermissionType) UnmarshalText(text []byte) error

type AclResourcePatternType

+ func (a *AclResourcePatternType) MarshalText() ([]byte, error)

+ func (a *AclResourcePatternType) String() string

+ func (a *AclResourcePatternType) UnmarshalText(text []byte) error

type AclResourceType

+ func (a *AclResourceType) MarshalText() ([]byte, error)

+ func (a *AclResourceType) String() string

+ func (a *AclResourceType) UnmarshalText(text []byte) error

+ type AlterUserScramCredentialsDelete struct

+ Mechanism ScramMechanismType

+ Name string

+ type AlterUserScramCredentialsRequest struct

+ Deletions []AlterUserScramCredentialsDelete

+ Upsertions []AlterUserScramCredentialsUpsert

+ Version int16

+ type AlterUserScramCredentialsResponse struct

+ Results []*AlterUserScramCredentialsResult

+ ThrottleTime time.Duration

+ Version int16

+ type AlterUserScramCredentialsResult struct

+ ErrorCode KError

+ ErrorMessage *string

+ User string

+ type AlterUserScramCredentialsUpsert struct

+ Iterations int32

+ Mechanism ScramMechanismType

+ Name string

+ Password []byte

+ Salt []byte

type Broker

+ func (b *Broker) AlterUserScramCredentials(req *AlterUserScramCredentialsRequest) (*AlterUserScramCredentialsResponse, error)

+ func (b *Broker) DescribeUserScramCredentials(req *DescribeUserScramCredentialsRequest) (*DescribeUserScramCredentialsResponse, error)

type ClusterAdmin

+ DeleteUserScramCredentials func(delete []AlterUserScramCredentialsDelete) ([]*AlterUserScramCredentialsResult, error)

+ DescribeUserScramCredentials func(users []string) ([]*DescribeUserScramCredentialsResult, error)

+ UpsertUserScramCredentials func(upsert []AlterUserScramCredentialsUpsert) ([]*AlterUserScramCredentialsResult, error)

+ type DescribeUserScramCredentialsRequest struct

+ DescribeUsers []DescribeUserScramCredentialsRequestUser

+ Version int16

+ type DescribeUserScramCredentialsRequestUser struct

+ Name string

+ type DescribeUserScramCredentialsResponse struct

+ ErrorCode KError

+ ErrorMessage *string

+ Results []*DescribeUserScramCredentialsResult

+ ThrottleTime time.Duration

+ Version int16

+ type DescribeUserScramCredentialsResult struct

+ CredentialInfos []*UserScramCredentialsResponseInfo

+ ErrorCode KError

+ ErrorMessage *string

+ User string

type KError

+ const ErrElectionNotNeeded

+ const ErrEligibleLeadersNotAvailable

+ const ErrGroupSubscribedToTopic

+ const ErrInvalidRecord

+ const ErrNoReassignmentInProgress

+ const ErrUnstableOffsetCommit

type OffsetFetchRequest

+ RequireStable bool

type OffsetRequest

+ IsolationLevel IsolationLevel

type OffsetResponse

+ ThrottleTimeMs int32

+ type ScramMechanismType int8

+ const SCRAM_MECHANISM_SHA_256

+ const SCRAM_MECHANISM_SHA_512

+ const SCRAM_MECHANISM_UNKNOWN

+ func (s ScramMechanismType) String() string

+ type UserScramCredentialsResponseInfo struct

+ Iterations int32

+ Mechanism ScramMechanismType

Feb 15, 2021

Changes in this version

+ var ErrBrokerNotFound = errors.New("kafka: broker for ID is not found")

+ var V2_7_0_0 = newKafkaVersion(2, 7, 0, 0)

type Client

+ Broker func(brokerID int32) (*Broker, error)

Oct 21, 2020

Changes in this version

+ type MockHeartbeatResponse struct

+ Err KError

+ func NewMockHeartbeatResponse(t TestReporter) *MockHeartbeatResponse

+ func (m *MockHeartbeatResponse) For(reqBody versionedDecoder) encoderWithHeader

+ func (m *MockHeartbeatResponse) SetError(kerr KError) *MockHeartbeatResponse

+ type MockJoinGroupResponse struct

+ Err KError

+ GenerationId int32

+ GroupProtocol string

+ LeaderId string

+ MemberId string

+ Members map[string][]byte

+ ThrottleTime int32

+ func NewMockJoinGroupResponse(t TestReporter) *MockJoinGroupResponse

+ func (m *MockJoinGroupResponse) For(reqBody versionedDecoder) encoderWithHeader

+ func (m *MockJoinGroupResponse) SetError(kerr KError) *MockJoinGroupResponse

+ func (m *MockJoinGroupResponse) SetGenerationId(id int32) *MockJoinGroupResponse

+ func (m *MockJoinGroupResponse) SetGroupProtocol(proto string) *MockJoinGroupResponse

+ func (m *MockJoinGroupResponse) SetLeaderId(id string) *MockJoinGroupResponse

+ func (m *MockJoinGroupResponse) SetMember(id string, meta *ConsumerGroupMemberMetadata) *MockJoinGroupResponse

+ func (m *MockJoinGroupResponse) SetMemberId(id string) *MockJoinGroupResponse

+ func (m *MockJoinGroupResponse) SetThrottleTime(t int32) *MockJoinGroupResponse

+ type MockLeaveGroupResponse struct

+ Err KError

+ func NewMockLeaveGroupResponse(t TestReporter) *MockLeaveGroupResponse

+ func (m *MockLeaveGroupResponse) For(reqBody versionedDecoder) encoderWithHeader

+ func (m *MockLeaveGroupResponse) SetError(kerr KError) *MockLeaveGroupResponse

+ type MockSyncGroupResponse struct

+ Err KError

+ MemberAssignment []byte

+ func NewMockSyncGroupResponse(t TestReporter) *MockSyncGroupResponse

+ func (m *MockSyncGroupResponse) For(reqBody versionedDecoder) encoderWithHeader

+ func (m *MockSyncGroupResponse) SetError(kerr KError) *MockSyncGroupResponse

+ func (m *MockSyncGroupResponse) SetMemberAssignment(assignment *ConsumerGroupMemberAssignment) *MockSyncGroupResponse

Oct 7, 2020

Changes in this version

+ var DefaultVersion = V1_0_0_0

type Client

+ RefreshBrokers func(addrs []string) error

type ConsumerError

+ func (ce ConsumerError) Unwrap() error

type ProducerError

+ func (pe ProducerError) Unwrap() error

Aug 11, 2020

Changes in this version

+ var V2_6_0_0 = newKafkaVersion(2, 6, 0, 0)

type ConsumerGroupSession

+ Commit func()

+ type ConsumerInterceptor interface

+ OnConsume func(*ConsumerMessage)

type GSSAPIConfig

+ DisablePAFXFAST bool

type OffsetManager

+ Commit func()

+ type ProducerInterceptor interface

+ OnSend func(*ProducerMessage)

May 6, 2020

Changes in this version

+ var V2_5_0_0 = newKafkaVersion(2, 5, 0, 0)

+ type AlterPartitionReassignmentsRequest struct

+ TimeoutMs int32

+ Version int16

+ func (r *AlterPartitionReassignmentsRequest) AddBlock(topic string, partitionID int32, replicas []int32)

+ type AlterPartitionReassignmentsResponse struct

+ ErrorCode KError

+ ErrorMessage *string

+ Errors map[string]map[int32]*alterPartitionReassignmentsErrorBlock

+ ThrottleTimeMs int32

+ Version int16

+ func (r *AlterPartitionReassignmentsResponse) AddError(topic string, partition int32, kerror KError, message *string)

type Broker

+ func (b *Broker) AlterPartitionReassignments(request *AlterPartitionReassignmentsRequest) (*AlterPartitionReassignmentsResponse, error)

+ func (b *Broker) ListPartitionReassignments(request *ListPartitionReassignmentsRequest) (*ListPartitionReassignmentsResponse, error)

type ClusterAdmin

+ AlterPartitionReassignments func(topic string, assignment [][]int32) error

+ DescribeLogDirs func(brokers []int32) (map[int32][]DescribeLogDirsResponseDirMetadata, error)

+ ListPartitionReassignments func(topics string, partitions []int32) (topicStatus map[string]map[int32]*PartitionReplicaReassignmentsStatus, ...)

type Config

+ RackID string

+ type ErrReassignPartitions struct

+ func (err ErrReassignPartitions) Error() string

type FetchRequest

+ RackID string

type FetchResponseBlock

+ PreferredReadReplica int32

+ type ListPartitionReassignmentsRequest struct

+ TimeoutMs int32

+ Version int16

+ func (r *ListPartitionReassignmentsRequest) AddBlock(topic string, partitionIDs []int32)

+ type ListPartitionReassignmentsResponse struct

+ ErrorCode KError

+ ErrorMessage *string

+ ThrottleTimeMs int32

+ TopicStatus map[string]map[int32]*PartitionReplicaReassignmentsStatus

+ Version int16

+ func (r *ListPartitionReassignmentsResponse) AddBlock(topic string, partition int32, ...)

+ type MockAlterConfigsResponseWithErrorCode struct

+ func NewMockAlterConfigsResponseWithErrorCode(t TestReporter) *MockAlterConfigsResponseWithErrorCode

+ func (mr *MockAlterConfigsResponseWithErrorCode) For(reqBody versionedDecoder) encoderWithHeader

+ type MockAlterPartitionReassignmentsResponse struct

+ func NewMockAlterPartitionReassignmentsResponse(t TestReporter) *MockAlterPartitionReassignmentsResponse

+ func (mr *MockAlterPartitionReassignmentsResponse) For(reqBody versionedDecoder) encoderWithHeader

+ type MockDescribeConfigsResponseWithErrorCode struct

+ func NewMockDescribeConfigsResponseWithErrorCode(t TestReporter) *MockDescribeConfigsResponseWithErrorCode

+ func (mr *MockDescribeConfigsResponseWithErrorCode) For(reqBody versionedDecoder) encoderWithHeader

+ type MockDescribeLogDirsResponse struct

+ func NewMockDescribeLogDirsResponse(t TestReporter) *MockDescribeLogDirsResponse

+ func (m *MockDescribeLogDirsResponse) For(reqBody versionedDecoder) encoderWithHeader

+ func (m *MockDescribeLogDirsResponse) SetLogDirs(logDirPath string, topicPartitions map[string]int) *MockDescribeLogDirsResponse

+ type MockListPartitionReassignmentsResponse struct

+ func NewMockListPartitionReassignmentsResponse(t TestReporter) *MockListPartitionReassignmentsResponse

+ func (mr *MockListPartitionReassignmentsResponse) For(reqBody versionedDecoder) encoderWithHeader

type MultiError

+ func (mErr MultiError) PrettyError() string

+ type PartitionReplicaReassignmentsStatus struct

+ AddingReplicas []int32

+ RemovingReplicas []int32

+ Replicas []int32

Feb 4, 2020

retracted

Changes in this version

type BalanceStrategy

+ AssignmentData func(memberID string, topics map[string][]int32, generationID int32) ([]byte, error)

Jan 24, 2020

retracted

Changes in this version

type Client

+ RefreshController func() (*Broker, error)

type ConfigResourceType

+ const BrokerLoggerResource

type FetchRequest

+ SessionEpoch int32

+ SessionID int32

type FetchResponse

+ ErrorCode int16

+ SessionID int32

type FetchResponseBlock

+ LogStartOffset int64

type ProduceResponseBlock

+ StartOffset int64

Jan 13, 2020

retracted

Changes in this version

+ var V2_4_0_0 = newKafkaVersion(2, 4, 0, 0)

Oct 31, 2019

retracted

Changes in this version

type Broker

+ func (b *Broker) DescribeLogDirs(request *DescribeLogDirsRequest) (*DescribeLogDirsResponse, error)

+ type DescribeLogDirsRequest struct

+ DescribeTopics []DescribeLogDirsRequestTopic

+ Version int16

+ type DescribeLogDirsRequestTopic struct

+ PartitionIDs []int32

+ Topic string

+ type DescribeLogDirsResponse struct

+ LogDirs []DescribeLogDirsResponseDirMetadata

+ ThrottleTime time.Duration

+ Version int16

+ type DescribeLogDirsResponseDirMetadata struct

+ ErrorCode KError

+ Path string

+ Topics []DescribeLogDirsResponseTopic

+ type DescribeLogDirsResponsePartition struct

+ IsTemporary bool

+ OffsetLag int64

+ PartitionID int32

+ Size int64

+ type DescribeLogDirsResponseTopic struct

+ Partitions []DescribeLogDirsResponsePartition

+ Topic string

Oct 9, 2019

Changes in this version

+ const RangeBalanceStrategyName

+ const RoundRobinBalanceStrategyName

+ const StickyBalanceStrategyName

+ var BalanceStrategySticky = &stickyBalanceStrategy

type ClusterAdmin

+ func NewClusterAdminFromClient(client Client) (ClusterAdmin, error)

type KError

+ const ErrFencedInstancedId

+ type StickyAssignorUserData interface

+ type StickyAssignorUserDataV0 struct

+ Topics map[string][]int32

+ type StickyAssignorUserDataV1 struct

+ Generation int32

+ Topics map[string][]int32

Jul 22, 2019

Changes in this version

+ type ErrDeleteRecords struct

+ func (err ErrDeleteRecords) Error() string

+ type MultiError struct

+ Errors *[]error

+ func (mErr MultiError) Error() string

Jul 2, 2019

Changes in this version

+ const GSS_API_FINISH

+ const GSS_API_GENERIC_TAG

+ const GSS_API_INITIAL

+ const GSS_API_VERIFY

+ const KRB5_KEYTAB_AUTH

+ const KRB5_USER_AUTH

+ const SASLTypeGSSAPI

+ const TOK_ID_KRB_AP_REQ

+ var V2_3_0_0 = newKafkaVersion(2, 3, 0, 0)

type ClusterAdmin

+ DeleteConsumerGroup func(group string) error

+ type GSSAPIConfig struct

+ AuthType int

+ KerberosConfigPath string

+ KeyTabPath string

+ Password string

+ Realm string

+ ServiceName string

+ Username string

+ type GSSAPIKerberosAuth struct

+ Config *GSSAPIConfig

+ NewKerberosClientFunc func(config *GSSAPIConfig) (KerberosClient, error)

+ func (krbAuth *GSSAPIKerberosAuth) Authorize(broker *Broker) error

+ type GSSApiHandlerFunc func([]byte) []byte

type KError

+ const ErrGroupMaxSizeReached

+ const ErrMemberIdRequired

+ const ErrOffsetNotAvailable

+ const ErrPreferredLeaderNotAvailable

+ const ErrStaleBrokerEpoch

+ type KafkaGSSAPIHandler struct

+ func (h *KafkaGSSAPIHandler) MockKafkaGSSAPI(buffer []byte) []byte

+ type KerberosClient interface

+ CName func() types.PrincipalName

+ Destroy func()

+ Domain func() string

+ GetServiceTicket func(spn string) (messages.Ticket, types.EncryptionKey, error)

+ Login func() error

+ func NewKerberosClient(config *GSSAPIConfig) (KerberosClient, error)

+ type KerberosGoKrb5Client struct

+ func (c *KerberosGoKrb5Client) CName() types.PrincipalName

+ func (c *KerberosGoKrb5Client) Domain() string

type MockBroker

+ func (b *MockBroker) SetGSSAPIHandler(handler GSSApiHandlerFunc)

+ type MockDeleteGroupsResponse struct

+ func NewMockDeleteGroupsRequest(t TestReporter) *MockDeleteGroupsResponse

+ func (m *MockDeleteGroupsResponse) For(reqBody versionedDecoder) encoder

+ func (m *MockDeleteGroupsResponse) SetDeletedGroups(groups []string) *MockDeleteGroupsResponse

+ type MockKerberosClient struct

+ ASRep messages.ASRep

+ func (c *MockKerberosClient) CName() types.PrincipalName

+ func (c *MockKerberosClient) Destroy()

+ func (c *MockKerberosClient) Domain() string

+ func (c *MockKerberosClient) GetServiceTicket(spn string) (messages.Ticket, types.EncryptionKey, error)

+ func (c *MockKerberosClient) Login() error

type MockOffsetFetchResponse

+ func (mr *MockOffsetFetchResponse) SetError(kerror KError) *MockOffsetFetchResponse

type Resource

+ ResourcePatternType AclResourcePatternType

Apr 9, 2019

Changes in this version

+ const SASLTypeSCRAMSHA256

+ const SASLTypeSCRAMSHA512

+ var V2_2_0_0 = newKafkaVersion(2, 2, 0, 0)

type Client

+ OfflineReplicas func(topic string, partitionID int32) ([]int32, error)

+ type ControlRecord struct

+ CoordinatorEpoch int32

+ Type ControlRecordType

+ Version int16

+ type ControlRecordType int

+ const ControlRecordAbort

+ const ControlRecordCommit

+ const ControlRecordUnknown

type FetchResponse

+ func (r *FetchResponse) AddControlRecord(topic string, partition int32, offset int64, producerID int64, ...)

+ func (r *FetchResponse) AddControlRecordWithTimestamp(topic string, partition int32, offset int64, producerID int64, ...)

+ func (r *FetchResponse) AddRecordBatch(topic string, partition int32, key, value Encoder, offset int64, ...)

+ func (r *FetchResponse) AddRecordBatchWithTimestamp(topic string, partition int32, key, value Encoder, offset int64, ...)

type RecordBatch

+ IsTransactional bool

+ func (b *RecordBatch) LastOffset() int64

+ type SCRAMClient interface

+ Begin func(userName, password, authzID string) error

+ Done func() bool

+ Step func(challenge string) (response string, err error)

type TopicError

+ func (t *TopicError) Error() string

type TopicPartitionError

+ func (t *TopicPartitionError) Error() string

Feb 24, 2019

Changes in this version

+ const APIKeySASLAuth

+ const SASLExtKeyAuth

+ const SASLHandshakeV0

+ const SASLHandshakeV1

+ const SASLTypeOAuth

+ const SASLTypePlaintext

+ type AccessToken struct

+ Extensions map[string]string

+ Token string

+ type AccessTokenProvider interface

+ Token func() (*AccessToken, error)

type AclFilter

+ ResourcePatternTypeFilter AclResourcePatternType

+ Version int

+ type AclResourcePatternType int

+ const AclPatternAny

+ const AclPatternLiteral

+ const AclPatternMatch

+ const AclPatternPrefixed

+ const AclPatternUnknown

type ClusterAdmin

+ DescribeCluster func() (brokers []*Broker, controllerID int32, err error)

+ DescribeConsumerGroups func(groups []string) ([]*GroupDescription, error)

+ DescribeTopics func(topics []string) (metadata []*TopicMetadata, err error)

+ ListConsumerGroupOffsets func(group string, topicPartitions map[string][]int32) (*OffsetFetchResponse, error)

+ ListConsumerGroups func() (map[string]string, error)

+ ListTopics func() (map[string]TopicDetail, error)

type CreateAclsRequest

+ Version int16

type DeleteAclsRequest

+ Version int

type DeleteAclsResponse

+ Version int16

type DescribeAclsRequest

+ Version int

type DescribeAclsResponse

+ Version int16

type FetchResponse

+ LogAppendTime bool

+ Timestamp time.Time

+ func (r *FetchResponse) AddMessageWithTimestamp(topic string, partition int32, key, value Encoder, offset int64, ...)

+ func (r *FetchResponse) AddRecordWithTimestamp(topic string, partition int32, key, value Encoder, offset int64, ...)

type KError

+ const ErrFencedLeaderEpoch

+ const ErrTopicDeletionDisabled

+ const ErrUnknownLeaderEpoch

+ const ErrUnsupportedCompressionType

type Message

+ LogAppendTime bool

+ type MockDescribeGroupsResponse struct

+ func NewMockDescribeGroupsResponse(t TestReporter) *MockDescribeGroupsResponse

+ func (m *MockDescribeGroupsResponse) AddGroupDescription(groupID string, description *GroupDescription) *MockDescribeGroupsResponse

+ func (m *MockDescribeGroupsResponse) For(reqBody versionedDecoder) encoder

+ type MockListGroupsResponse struct

+ func NewMockListGroupsResponse(t TestReporter) *MockListGroupsResponse

+ func (m *MockListGroupsResponse) AddGroup(groupID, protocolType string) *MockListGroupsResponse

+ func (m *MockListGroupsResponse) For(reqBody versionedDecoder) encoder

+ type MockSaslAuthenticateResponse struct

+ func NewMockSaslAuthenticateResponse(t TestReporter) *MockSaslAuthenticateResponse

+ func (msar *MockSaslAuthenticateResponse) For(reqBody versionedDecoder) encoder

+ func (msar *MockSaslAuthenticateResponse) SetAuthBytes(saslAuthBytes []byte) *MockSaslAuthenticateResponse

+ func (msar *MockSaslAuthenticateResponse) SetError(kerror KError) *MockSaslAuthenticateResponse

+ type MockSaslHandshakeResponse struct

+ func NewMockSaslHandshakeResponse(t TestReporter) *MockSaslHandshakeResponse

+ func (mshr *MockSaslHandshakeResponse) For(reqBody versionedDecoder) encoder

+ func (mshr *MockSaslHandshakeResponse) SetEnabledMechanisms(enabledMechanisms []string) *MockSaslHandshakeResponse

+ func (mshr *MockSaslHandshakeResponse) SetError(kerror KError) *MockSaslHandshakeResponse

type RecordBatch

+ LogAppendTime bool

type Resource

+ ResoucePatternType AclResourcePatternType

+ type SASLMechanism string

+ type SaslAuthenticateRequest struct

+ SaslAuthBytes []byte

+ type SaslAuthenticateResponse struct

+ Err KError

+ ErrorMessage *string

+ SaslAuthBytes []byte

type SaslHandshakeRequest

+ Version int16

Jan 10, 2019

Changes in this version

type ConfigEntry

+ Source ConfigSource

+ Synonyms []*ConfigSynonym

+ type ConfigSource int8

+ const SourceDefault

+ const SourceDynamicBroker

+ const SourceDynamicDefaultBroker

+ const SourceStaticBroker

+ const SourceTopic

+ const SourceUnknown

+ func (s ConfigSource) String() string

+ type ConfigSynonym struct

+ ConfigName string

+ ConfigValue string

+ Source ConfigSource

type DescribeConfigsRequest

+ IncludeSynonyms bool

+ Version int16

type DescribeConfigsResponse

+ Version int16

type OffsetRequest

+ func (r *OffsetRequest) ReplicaID() int32

+ func (r *OffsetRequest) SetReplicaID(id int32)

Dec 10, 2018

Changes in this version

+ var V1_1_1_0 = newKafkaVersion(1, 1, 1, 0)

+ var V2_0_1_0 = newKafkaVersion(2, 0, 1, 0)

+ var V2_1_0_0 = newKafkaVersion(2, 1, 0, 0)

type Broker

+ func (b *Broker) Rack() string

type Client

+ InitProducerID func() (*InitProducerIDResponse, error)

type CompressionCodec

+ const CompressionZSTD

type KError

+ const ErrDelegationTokenAuthDisabled

+ const ErrDelegationTokenAuthorizationFailed

+ const ErrDelegationTokenExpired

+ const ErrDelegationTokenNotFound

+ const ErrDelegationTokenOwnerMismatch

+ const ErrDelegationTokenRequestNotAllowed

+ const ErrFetchSessionIDNotFound

+ const ErrGroupIDNotFound

+ const ErrInvalidFetchSessionEpoch

+ const ErrInvalidPrincipalType

+ const ErrListenerNotFound

+ const ErrNonEmptyGroup

type OffsetCommitResponse

+ ThrottleTimeMs int32

+ Version int16

type OffsetFetchRequest

+ func (r *OffsetFetchRequest) ZeroPartitions()

type OffsetFetchResponse

+ Err KError

+ ThrottleTimeMs int32

+ Version int16

type OffsetFetchResponseBlock

+ LeaderEpoch int32

Sep 27, 2018

Changes in this version

+ var BalanceStrategyRange = &balanceStrategy

+ var BalanceStrategyRoundRobin = &balanceStrategy

+ var ErrClosedConsumerGroup = errors.New("kafka: tried to use a consumer group that was closed")

+ type BalanceStrategy interface

+ Name func() string

+ Plan func(members map[string]ConsumerGroupMemberMetadata, topics map[string][]int32) (BalanceStrategyPlan, error)

+ type BalanceStrategyPlan map[string]map[string][]int32

+ func (p BalanceStrategyPlan) Add(memberID, topic string, partitions ...int32)

+ type ConsumerGroup interface

+ Close func() error

+ Consume func(ctx context.Context, topics []string, handler ConsumerGroupHandler) error

+ Errors func() <-chan error

+ func NewConsumerGroup(addrs []string, groupID string, config *Config) (ConsumerGroup, error)

+ func NewConsumerGroupFromClient(groupID string, client Client) (ConsumerGroup, error)

+ type ConsumerGroupClaim interface

+ HighWaterMarkOffset func() int64

+ InitialOffset func() int64

+ Messages func() <-chan *ConsumerMessage

+ Partition func() int32

+ Topic func() string

+ type ConsumerGroupHandler interface

+ Cleanup func(ConsumerGroupSession) error

+ ConsumeClaim func(ConsumerGroupSession, ConsumerGroupClaim) error

+ Setup func(ConsumerGroupSession) error

+ type ConsumerGroupSession interface

+ Claims func() map[string][]int32

+ Context func() context.Context

+ GenerationID func() int32

+ MarkMessage func(msg *ConsumerMessage, metadata string)

+ MarkOffset func(topic string, partition int32, offset int64, metadata string)

+ MemberID func() string

+ ResetOffset func(topic string, partition int32, offset int64, metadata string)

Sep 7, 2018

Changes in this version

+ var V2_0_0_0 = newKafkaVersion(2, 0, 0, 0)

+ type ClusterAdmin interface

+ AlterConfig func(resourceType ConfigResourceType, name string, entries map[string]*string, ...) error

+ Close func() error

+ CreateACL func(resource Resource, acl Acl) error

+ CreatePartitions func(topic string, count int32, assignment [][]int32, validateOnly bool) error

+ CreateTopic func(topic string, detail *TopicDetail, validateOnly bool) error

+ DeleteACL func(filter AclFilter, validateOnly bool) ([]MatchingAcl, error)

+ DeleteRecords func(topic string, partitionOffsets map[int32]int64) error

+ DeleteTopic func(topic string) error

+ DescribeConfig func(resource ConfigResource) ([]ConfigEntry, error)

+ ListAcls func(filter AclFilter) ([]ResourceAcls, error)

+ func NewClusterAdmin(addrs []string, conf *Config) (ClusterAdmin, error)

type Config

+ Admin struct{ ... }

+ type DynamicConsistencyPartitioner interface

+ MessageRequiresConsistency func(message *ProducerMessage) bool

+ type HashPartitionerOption func(*hashPartitioner)

+ func WithAbsFirst() HashPartitionerOption

+ func WithCustomFallbackPartitioner(randomHP *hashPartitioner) HashPartitionerOption

+ func WithCustomHashFunction(hasher func() hash.Hash32) HashPartitionerOption

type MessageSet

+ OverflowMessage bool

+ type MockAlterConfigsResponse struct

+ func NewMockAlterConfigsResponse(t TestReporter) *MockAlterConfigsResponse

+ func (mr *MockAlterConfigsResponse) For(reqBody versionedDecoder) encoder

+ type MockCreateAclsResponse struct

+ func NewMockCreateAclsResponse(t TestReporter) *MockCreateAclsResponse

+ func (mr *MockCreateAclsResponse) For(reqBody versionedDecoder) encoder

+ type MockCreatePartitionsResponse struct

+ func NewMockCreatePartitionsResponse(t TestReporter) *MockCreatePartitionsResponse

+ func (mr *MockCreatePartitionsResponse) For(reqBody versionedDecoder) encoder

+ type MockCreateTopicsResponse struct

+ func NewMockCreateTopicsResponse(t TestReporter) *MockCreateTopicsResponse

+ func (mr *MockCreateTopicsResponse) For(reqBody versionedDecoder) encoder

+ type MockDeleteAclsResponse struct

+ func NewMockDeleteAclsResponse(t TestReporter) *MockDeleteAclsResponse

+ func (mr *MockDeleteAclsResponse) For(reqBody versionedDecoder) encoder

+ type MockDeleteRecordsResponse struct

+ func NewMockDeleteRecordsResponse(t TestReporter) *MockDeleteRecordsResponse

+ func (mr *MockDeleteRecordsResponse) For(reqBody versionedDecoder) encoder

+ type MockDeleteTopicsResponse struct

+ func NewMockDeleteTopicsResponse(t TestReporter) *MockDeleteTopicsResponse

+ func (mr *MockDeleteTopicsResponse) For(reqBody versionedDecoder) encoder

+ type MockDescribeConfigsResponse struct

+ func NewMockDescribeConfigsResponse(t TestReporter) *MockDescribeConfigsResponse

+ func (mr *MockDescribeConfigsResponse) For(reqBody versionedDecoder) encoder

+ type MockListAclsResponse struct

+ func NewMockListAclsResponse(t TestReporter) *MockListAclsResponse

+ func (mr *MockListAclsResponse) For(reqBody versionedDecoder) encoder

type Partitioner

+ func NewReferenceHashPartitioner(topic string) Partitioner

type PartitionerConstructor

+ func NewCustomPartitioner(options ...HashPartitionerOption) PartitionerConstructor

May 30, 2018

Changes in this version

+ const CompressionLevelDefault

+ var ErrControllerNotAvailable = errors.New("kafka: controller is not available")

+ var ErrNoTopicsToUpdateMetadata = errors.New("kafka: no specific topics to update metadata")

+ var MaxVersion = V1_1_0_0

+ var MinVersion = V0_8_2_0

+ var NoNode = &Broker

+ var SupportedVersions = []KafkaVersion

+ var V0_10_1_1 = newKafkaVersion(0, 10, 1, 1)

+ var V0_10_2_1 = newKafkaVersion(0, 10, 2, 1)

+ var V0_11_0_1 = newKafkaVersion(0, 11, 0, 1)

+ var V0_11_0_2 = newKafkaVersion(0, 11, 0, 2)

+ var V1_1_0_0 = newKafkaVersion(1, 1, 0, 0)

type Broker

+ func (b *Broker) CreatePartitions(request *CreatePartitionsRequest) (*CreatePartitionsResponse, error)

+ func (b *Broker) DeleteGroups(request *DeleteGroupsRequest) (*DeleteGroupsResponse, error)

+ func (b *Broker) DeleteRecords(request *DeleteRecordsRequest) (*DeleteRecordsResponse, error)

+ func (b *Broker) FindCoordinator(request *FindCoordinatorRequest) (*FindCoordinatorResponse, error)

type Client

+ Controller func() (*Broker, error)

type CompressionCodec

+ func (cc CompressionCodec) String() string

+ type CoordinatorType int8

+ const CoordinatorGroup

+ const CoordinatorTransaction

+ type DeleteGroupsRequest struct

+ Groups []string

+ func (r *DeleteGroupsRequest) AddGroup(group string)

+ type DeleteGroupsResponse struct

+ GroupErrorCodes map[string]KError

+ ThrottleTime time.Duration

+ type DeleteRecordsRequest struct

+ Timeout time.Duration

+ Topics map[string]*DeleteRecordsRequestTopic

+ type DeleteRecordsRequestTopic struct

+ PartitionOffsets map[int32]int64

+ type DeleteRecordsResponse struct

+ ThrottleTime time.Duration

+ Topics map[string]*DeleteRecordsResponseTopic

+ Version int16

+ type DeleteRecordsResponsePartition struct

+ Err KError

+ LowWatermark int64

+ type DeleteRecordsResponseTopic struct

+ Partitions map[int32]*DeleteRecordsResponsePartition

type DeleteTopicsRequest

+ Version int16

+ type FindCoordinatorRequest struct

+ CoordinatorKey string

+ CoordinatorType CoordinatorType

+ Version int16

+ type FindCoordinatorResponse struct

+ Coordinator *Broker

+ Err KError

+ ErrMsg *string

+ ThrottleTime time.Duration

+ Version int16

type JoinGroupRequest

+ RebalanceTimeout int32

+ Version int16

type JoinGroupResponse

+ ThrottleTime int32

+ Version int16

type Message

+ CompressionLevel int

type MetadataRequest

+ AllowAutoTopicCreation bool

+ Version int16

type MetadataResponse

+ ClusterID *string

+ ControllerID int32

+ ThrottleTimeMs int32

+ Version int16

+ type MockFindCoordinatorResponse struct

+ func NewMockFindCoordinatorResponse(t TestReporter) *MockFindCoordinatorResponse

+ func (mr *MockFindCoordinatorResponse) For(reqBody versionedDecoder) encoder

+ func (mr *MockFindCoordinatorResponse) SetCoordinator(coordinatorType CoordinatorType, group string, broker *MockBroker) *MockFindCoordinatorResponse

+ func (mr *MockFindCoordinatorResponse) SetError(coordinatorType CoordinatorType, group string, kerror KError) *MockFindCoordinatorResponse

type MockMetadataResponse

+ func (mmr *MockMetadataResponse) SetController(brokerID int32) *MockMetadataResponse

type OffsetCommitRequest

+ func (r *OffsetCommitRequest) Offset(topic string, partitionID int32) (int64, string, error)

type PartitionMetadata

+ OfflineReplicas []int32

type RecordBatch

+ CompressionLevel int

type Records

+ MsgSet *MessageSet

+ RecordBatch *RecordBatch

type TopicMetadata

+ IsInternal bool

Feb 12, 2018

Changes in this version

+ var ErrConsumerOffsetNotAdvanced = errors.New("kafka: consumer offset was not advanced after a RecordBatch")

+ type Acl struct

+ Host string

+ Operation AclOperation

+ PermissionType AclPermissionType

+ Principal string

+ type AclCreation struct

+ type AclCreationResponse struct

+ Err KError

+ ErrMsg *string

+ type AclFilter struct

+ Host *string

+ Operation AclOperation

+ PermissionType AclPermissionType

+ Principal *string

+ ResourceName *string

+ ResourceType AclResourceType

+ type AclOperation int

+ const AclOperationAll

+ const AclOperationAlter

+ const AclOperationAlterConfigs

+ const AclOperationAny

+ const AclOperationClusterAction

+ const AclOperationCreate

+ const AclOperationDelete

+ const AclOperationDescribe

+ const AclOperationDescribeConfigs

+ const AclOperationIdempotentWrite

+ const AclOperationRead

+ const AclOperationUnknown

+ const AclOperationWrite

+ type AclPermissionType int

+ const AclPermissionAllow

+ const AclPermissionAny

+ const AclPermissionDeny

+ const AclPermissionUnknown

+ type AclResourceType int

+ const AclResourceAny

+ const AclResourceCluster

+ const AclResourceGroup

+ const AclResourceTopic

+ const AclResourceTransactionalID

+ const AclResourceUnknown

+ type AddOffsetsToTxnRequest struct

+ GroupID string

+ ProducerEpoch int16

+ ProducerID int64

+ TransactionalID string

+ type AddOffsetsToTxnResponse struct

+ Err KError

+ ThrottleTime time.Duration

+ type AddPartitionsToTxnRequest struct

+ ProducerEpoch int16

+ ProducerID int64

+ TopicPartitions map[string][]int32

+ TransactionalID string

+ type AddPartitionsToTxnResponse struct

+ Errors map[string][]*PartitionError

+ ThrottleTime time.Duration

+ type AlterConfigsRequest struct

+ Resources []*AlterConfigsResource

+ ValidateOnly bool

+ type AlterConfigsResource struct

+ ConfigEntries map[string]*string

+ Name string

+ Type ConfigResourceType

+ type AlterConfigsResourceResponse struct

+ ErrorCode int16

+ ErrorMsg string

+ Name string

+ Type ConfigResourceType

+ type AlterConfigsResponse struct

+ Resources []*AlterConfigsResourceResponse

+ ThrottleTime time.Duration

type Broker

+ func (b *Broker) AddOffsetsToTxn(request *AddOffsetsToTxnRequest) (*AddOffsetsToTxnResponse, error)

+ func (b *Broker) AddPartitionsToTxn(request *AddPartitionsToTxnRequest) (*AddPartitionsToTxnResponse, error)

+ func (b *Broker) AlterConfigs(request *AlterConfigsRequest) (*AlterConfigsResponse, error)

+ func (b *Broker) CreateAcls(request *CreateAclsRequest) (*CreateAclsResponse, error)

+ func (b *Broker) CreateTopics(request *CreateTopicsRequest) (*CreateTopicsResponse, error)

+ func (b *Broker) DeleteAcls(request *DeleteAclsRequest) (*DeleteAclsResponse, error)

+ func (b *Broker) DeleteTopics(request *DeleteTopicsRequest) (*DeleteTopicsResponse, error)

+ func (b *Broker) DescribeAcls(request *DescribeAclsRequest) (*DescribeAclsResponse, error)

+ func (b *Broker) DescribeConfigs(request *DescribeConfigsRequest) (*DescribeConfigsResponse, error)

+ func (b *Broker) EndTxn(request *EndTxnRequest) (*EndTxnResponse, error)

+ func (b *Broker) InitProducerID(request *InitProducerIDRequest) (*InitProducerIDResponse, error)

+ func (b *Broker) TxnOffsetCommit(request *TxnOffsetCommitRequest) (*TxnOffsetCommitResponse, error)

+ type ConfigEntry struct

+ Default bool

+ Name string

+ ReadOnly bool

+ Sensitive bool

+ Value string

+ type ConfigResource struct

+ ConfigNames []string

+ Name string

+ Type ConfigResourceType

+ type ConfigResourceType int8

+ const AnyResource

+ const BrokerResource

+ const ClusterResource

+ const GroupResource

+ const TopicResource

+ const UnknownResource

+ type CreateAclsRequest struct

+ AclCreations []*AclCreation

+ type CreateAclsResponse struct

+ AclCreationResponses []*AclCreationResponse

+ ThrottleTime time.Duration

+ type CreateTopicsRequest struct

+ Timeout time.Duration

+ TopicDetails map[string]*TopicDetail

+ ValidateOnly bool

+ Version int16

+ type CreateTopicsResponse struct

+ ThrottleTime time.Duration

+ TopicErrors map[string]*TopicError

+ Version int16

+ type DeleteAclsRequest struct

+ Filters []*AclFilter

+ type DeleteAclsResponse struct

+ FilterResponses []*FilterResponse

+ ThrottleTime time.Duration

+ type DeleteTopicsRequest struct

+ Timeout time.Duration

+ Topics []string

+ type DeleteTopicsResponse struct

+ ThrottleTime time.Duration

+ TopicErrorCodes map[string]KError

+ Version int16

+ type DescribeAclsRequest struct

+ type DescribeAclsResponse struct

+ Err KError

+ ErrMsg *string

+ ResourceAcls []*ResourceAcls

+ ThrottleTime time.Duration

+ type DescribeConfigsRequest struct

+ Resources []*ConfigResource

+ type DescribeConfigsResponse struct

+ Resources []*ResourceResponse

+ ThrottleTime time.Duration

+ type EndTxnRequest struct

+ ProducerEpoch int16

+ ProducerID int64

+ TransactionResult bool

+ TransactionalID string

+ type EndTxnResponse struct

+ Err KError

+ ThrottleTime time.Duration

type FetchResponse

+ func (r *FetchResponse) SetLastOffsetDelta(topic string, partition int32, offset int32)

type FetchResponseBlock

+ Partial bool

+ RecordsSet []*Records

+ type FilterResponse struct

+ Err KError

+ ErrMsg *string

+ MatchingAcls []*MatchingAcl

+ type InitProducerIDRequest struct

+ TransactionTimeout time.Duration

+ TransactionalID *string

+ type InitProducerIDResponse struct

+ Err KError

+ ProducerEpoch int16

+ ProducerID int64

+ ThrottleTime time.Duration

+ type MatchingAcl struct

+ Err KError

+ ErrMsg *string

type MockBroker

+ func NewMockBrokerListener(t TestReporter, brokerID int32, listener net.Listener) *MockBroker

type MockProduceResponse

+ func (mr *MockProduceResponse) SetVersion(version int16) *MockProduceResponse

+ type PartitionError struct

+ Err KError

+ Partition int32

+ type PartitionOffsetMetadata struct

+ Metadata *string

+ Offset int64

+ Partition int32

+ type Resource struct

+ ResourceName string

+ ResourceType AclResourceType

+ type ResourceAcls struct

+ Acls []*Acl

+ type ResourceResponse struct

+ Configs []*ConfigEntry

+ ErrorCode int16

+ ErrorMsg string

+ Name string

+ Type ConfigResourceType

+ type TopicDetail struct

+ ConfigEntries map[string]*string

+ NumPartitions int32

+ ReplicaAssignment map[int32][]int32

+ ReplicationFactor int16

+ type TopicError struct

+ Err KError

+ ErrMsg *string

+ type TxnOffsetCommitRequest struct

+ GroupID string

+ ProducerEpoch int16

+ ProducerID int64

+ Topics map[string][]*PartitionOffsetMetadata

+ TransactionalID string

+ type TxnOffsetCommitResponse struct

+ ThrottleTime time.Duration

+ Topics map[string][]*PartitionError

Dec 8, 2017

Changes in this version

+ var V1_0_0_0 = newKafkaVersion(1, 0, 0, 0)

+ type CreatePartitionsRequest struct

+ Timeout time.Duration

+ TopicPartitions map[string]*TopicPartition

+ ValidateOnly bool

+ type CreatePartitionsResponse struct

+ ThrottleTime time.Duration

+ TopicPartitionErrors map[string]*TopicPartitionError

type KError

+ const ErrConcurrentTransactions

+ const ErrDuplicateSequenceNumber

+ const ErrInvalidProducerEpoch

+ const ErrInvalidProducerIDMapping

+ const ErrInvalidTransactionTimeout

+ const ErrInvalidTxnState

+ const ErrKafkaStorageError

+ const ErrLogDirNotFound

+ const ErrOperationNotAttempted

+ const ErrOutOfOrderSequenceNumber

+ const ErrReassignmentInProgress

+ const ErrSASLAuthenticationFailed

+ const ErrSecurityDisabled

+ const ErrTransactionCoordinatorFenced

+ const ErrTransactionalIDAuthorizationFailed

+ const ErrUnknownProducerID

type KafkaVersion

+ func ParseKafkaVersion(s string) (KafkaVersion, error)

+ func (v KafkaVersion) String() string

+ type TopicPartition struct

+ Assignment [][]int32

+ Count int32

+ type TopicPartitionError struct

+ Err KError

+ ErrMsg *string

Nov 13, 2017

Changes in this version

+ var V0_11_0_0 = newKafkaVersion(0, 11, 0, 0)

+ type AbortedTransaction struct

+ FirstOffset int64

+ ProducerID int64

type ConsumerMessage

+ Headers []*RecordHeader

type FetchRequest

+ Isolation IsolationLevel

type FetchResponse

+ func (r *FetchResponse) AddRecord(topic string, partition int32, key, value Encoder, offset int64)

+ func (r *FetchResponse) SetLastStableOffset(topic string, partition int32, offset int64)

type FetchResponseBlock

+ AbortedTransactions []*AbortedTransaction

+ LastStableOffset int64

+ Records Records

+ type IsolationLevel int8

+ const ReadCommitted

+ const ReadUncommitted

type MockOffsetResponse

+ func (mor *MockOffsetResponse) SetVersion(version int16) *MockOffsetResponse

type ProduceRequest

+ TransactionalID *string

+ func (r *ProduceRequest) AddBatch(topic string, partition int32, batch *RecordBatch)

type ProducerMessage

+ Headers []RecordHeader

+ type Record struct

+ Attributes int8

+ Headers []*RecordHeader

+ Key []byte

+ OffsetDelta int64

+ TimestampDelta time.Duration

+ Value []byte

+ type RecordBatch struct

+ Codec CompressionCodec

+ Control bool

+ FirstOffset int64

+ FirstSequence int32

+ FirstTimestamp time.Time

+ LastOffsetDelta int32

+ MaxTimestamp time.Time

+ PartialTrailingRecord bool

+ PartitionLeaderEpoch int32

+ ProducerEpoch int16

+ ProducerID int64

+ Records []*Record

+ Version int8

+ type RecordHeader struct

+ Key []byte

+ Value []byte

+ type Records struct

+ type Timestamp struct

Oct 4, 2017

Changes in this version

type ConsumerMessage

+ BlockTimestamp time.Time

type FetchRequest

+ MaxBytes int32

type MockFetchResponse

+ func (mfr *MockFetchResponse) SetVersion(version int16) *MockFetchResponse

type PartitionOffsetManager

+ ResetOffset func(offset int64, metadata string)

May 8, 2017

Changes in this version

+ var V0_10_2_0 = newKafkaVersion(0, 10, 2, 0)

type Broker

+ func (b *Broker) ApiVersions(request *ApiVersionsRequest) (*ApiVersionsResponse, error)

type Client

+ Brokers func() []*Broker

+ InSyncReplicas func(topic string, partitionID int32) ([]int32, error)

+ type GroupProtocol struct

+ Metadata []byte

+ Name string

type JoinGroupRequest

+ OrderedGroupProtocols []*GroupProtocol

type KError

+ const ErrInvalidConfig

+ const ErrInvalidPartitions

+ const ErrInvalidReplicaAssignment

+ const ErrInvalidReplicationFactor

+ const ErrInvalidRequest

+ const ErrNotController

+ const ErrPolicyViolation

+ const ErrTopicAlreadyExists

type PartitionerConstructor

+ func NewCustomHashPartitioner(hasher func() hash.Hash32) PartitionerConstructor

Dec 20, 2016

Changes in this version

+ var V0_10_0_1 = newKafkaVersion(0, 10, 0, 1)

+ var V0_10_1_0 = newKafkaVersion(0, 10, 1, 0)

type CompressionCodec

+ const CompressionLZ4

type Config

+ MetricRegistry metrics.Registry

type Consumer

+ HighWaterMarks func() map[string]map[int32]int64

type GroupMemberDescription

+ func (gmd *GroupMemberDescription) GetMemberAssignment() (*ConsumerGroupMemberAssignment, error)

+ func (gmd *GroupMemberDescription) GetMemberMetadata() (*ConsumerGroupMemberMetadata, error)

type KError

+ const ErrUnsupportedForMessageFormat

type MockBroker

+ func (b *MockBroker) SetNotifier(notifier RequestNotifierFunc)

type OffsetRequest

+ Version int16

type OffsetResponse

+ Version int16

type OffsetResponseBlock

+ Offset int64

+ Timestamp int64

+ type RequestNotifierFunc func(bytesRead, bytesWritten int)

Aug 2, 2016

Changes in this version

+ var V0_10_0_0 = newKafkaVersion(0, 10, 0, 0)

+ var V0_8_2_0 = newKafkaVersion(0, 8, 2, 0)

+ var V0_8_2_1 = newKafkaVersion(0, 8, 2, 1)

+ var V0_8_2_2 = newKafkaVersion(0, 8, 2, 2)

+ var V0_9_0_0 = newKafkaVersion(0, 9, 0, 0)

+ var V0_9_0_1 = newKafkaVersion(0, 9, 0, 1)

+ type ApiVersionsRequest struct

+ type ApiVersionsResponse struct

+ ApiVersions []*ApiVersionsResponseBlock

+ Err KError

+ type ApiVersionsResponseBlock struct

+ ApiKey int16

+ MaxVersion int16

+ MinVersion int16

type Config

+ Version KafkaVersion

type ConsumerMessage

+ Timestamp time.Time

type FetchRequest

+ Version int16

type FetchResponse

+ ThrottleTime time.Duration

+ Version int16

type KError

+ const ErrIllegalSASLState

+ const ErrInvalidTimestamp

+ const ErrNetworkException

+ const ErrUnsupportedSASLMechanism

+ const ErrUnsupportedVersion

+ type KafkaVersion struct

+ func (v KafkaVersion) IsAtLeast(other KafkaVersion) bool

type Message

+ Timestamp time.Time

+ Version int8

type ProduceRequest

+ Version int16

type ProduceResponse

+ ThrottleTime time.Duration

+ Version int16

type ProduceResponseBlock

+ Timestamp time.Time

type ProducerMessage

+ Timestamp time.Time

+ type SaslHandshakeRequest struct

+ Mechanism string

+ type SaslHandshakeResponse struct

+ EnabledMechanisms []string

+ Err KError

type SyncProducer

+ SendMessages func(msgs []*ProducerMessage) error

May 16, 2016

Changes in this version

+ type MockBroker struct

+ func NewMockBroker(t TestReporter, brokerID int32) *MockBroker

+ func NewMockBrokerAddr(t TestReporter, brokerID int32, addr string) *MockBroker

+ func (b *MockBroker) Addr() string

+ func (b *MockBroker) BrokerID() int32

+ func (b *MockBroker) Close()

+ func (b *MockBroker) History() []RequestResponse

+ func (b *MockBroker) Port() int32

+ func (b *MockBroker) Returns(e encoder)

+ func (b *MockBroker) SetHandlerByMap(handlerMap map[string]MockResponse)

+ func (b *MockBroker) SetLatency(latency time.Duration)

+ type MockConsumerMetadataResponse struct

+ func NewMockConsumerMetadataResponse(t TestReporter) *MockConsumerMetadataResponse

+ func (mr *MockConsumerMetadataResponse) For(reqBody decoder) encoder

+ func (mr *MockConsumerMetadataResponse) SetCoordinator(group string, broker *MockBroker) *MockConsumerMetadataResponse

+ func (mr *MockConsumerMetadataResponse) SetError(group string, kerror KError) *MockConsumerMetadataResponse

+ type MockFetchResponse struct

+ func NewMockFetchResponse(t TestReporter, batchSize int) *MockFetchResponse

+ func (mfr *MockFetchResponse) For(reqBody decoder) encoder

+ func (mfr *MockFetchResponse) SetHighWaterMark(topic string, partition int32, offset int64) *MockFetchResponse

+ func (mfr *MockFetchResponse) SetMessage(topic string, partition int32, offset int64, msg Encoder) *MockFetchResponse

+ type MockMetadataResponse struct

+ func NewMockMetadataResponse(t TestReporter) *MockMetadataResponse

+ func (mmr *MockMetadataResponse) For(reqBody decoder) encoder

+ func (mmr *MockMetadataResponse) SetBroker(addr string, brokerID int32) *MockMetadataResponse

+ func (mmr *MockMetadataResponse) SetLeader(topic string, partition, brokerID int32) *MockMetadataResponse

+ type MockOffsetCommitResponse struct

+ func NewMockOffsetCommitResponse(t TestReporter) *MockOffsetCommitResponse

+ func (mr *MockOffsetCommitResponse) For(reqBody decoder) encoder

+ func (mr *MockOffsetCommitResponse) SetError(group, topic string, partition int32, kerror KError) *MockOffsetCommitResponse

+ type MockOffsetFetchResponse struct

+ func NewMockOffsetFetchResponse(t TestReporter) *MockOffsetFetchResponse

+ func (mr *MockOffsetFetchResponse) For(reqBody decoder) encoder

+ func (mr *MockOffsetFetchResponse) SetOffset(group, topic string, partition int32, offset int64, metadata string, ...) *MockOffsetFetchResponse

+ type MockOffsetResponse struct

+ func NewMockOffsetResponse(t TestReporter) *MockOffsetResponse

+ func (mor *MockOffsetResponse) For(reqBody decoder) encoder

+ func (mor *MockOffsetResponse) SetOffset(topic string, partition int32, time, offset int64) *MockOffsetResponse

+ type MockProduceResponse struct

+ func NewMockProduceResponse(t TestReporter) *MockProduceResponse

+ func (mr *MockProduceResponse) For(reqBody decoder) encoder

+ func (mr *MockProduceResponse) SetError(topic string, partition int32, kerror KError) *MockProduceResponse

+ type MockResponse interface

+ For func(reqBody decoder) (res encoder)

+ type MockSequence struct

+ func NewMockSequence(responses ...interface{}) *MockSequence

+ func (mc *MockSequence) For(reqBody decoder) (res encoder)

+ type MockWrapper struct

+ func NewMockWrapper(res encoder) *MockWrapper

+ func (mw *MockWrapper) For(reqBody decoder) (res encoder)

+ type RequestResponse struct

+ Request requestBody

+ Response encoder

+ type TestReporter interface

+ Error func(...interface{})

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

+ Fatal func(...interface{})

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

Feb 1, 2016

Changes in this version

+ const GroupGenerationUndefined

type Broker

+ func (b *Broker) DescribeGroups(request *DescribeGroupsRequest) (*DescribeGroupsResponse, error)

+ func (b *Broker) Heartbeat(request *HeartbeatRequest) (*HeartbeatResponse, error)

+ func (b *Broker) JoinGroup(request *JoinGroupRequest) (*JoinGroupResponse, error)

+ func (b *Broker) LeaveGroup(request *LeaveGroupRequest) (*LeaveGroupResponse, error)

+ func (b *Broker) ListGroups(request *ListGroupsRequest) (*ListGroupsResponse, error)

+ func (b *Broker) SyncGroup(request *SyncGroupRequest) (*SyncGroupResponse, error)

+ type ConsumerGroupMemberAssignment struct

+ Topics map[string][]int32

+ UserData []byte

+ Version int16

+ type ConsumerGroupMemberMetadata struct

+ Topics []string

+ UserData []byte

+ Version int16

type JoinGroupRequest

+ func (r *JoinGroupRequest) AddGroupProtocolMetadata(name string, metadata *ConsumerGroupMemberMetadata) error

type JoinGroupResponse

+ func (r *JoinGroupResponse) GetMembers() (map[string]ConsumerGroupMemberMetadata, error)

type SyncGroupRequest

+ func (r *SyncGroupRequest) AddGroupAssignmentMember(memberId string, memberAssignment *ConsumerGroupMemberAssignment) error

type SyncGroupResponse

+ func (r *SyncGroupResponse) GetMemberAssignment() (*ConsumerGroupMemberAssignment, error)

Dec 11, 2015

Changes in this version

+ type DescribeGroupsRequest struct

+ Groups []string

+ func (r *DescribeGroupsRequest) AddGroup(group string)

+ type DescribeGroupsResponse struct

+ Groups []*GroupDescription

+ type GroupDescription struct

+ Err KError

+ GroupId string

+ Members map[string]*GroupMemberDescription

+ Protocol string

+ ProtocolType string

+ State string

+ type GroupMemberDescription struct

+ ClientHost string

+ ClientId string

+ MemberAssignment []byte

+ MemberMetadata []byte

+ type HeartbeatRequest struct

+ GenerationId int32

+ GroupId string

+ MemberId string

+ type HeartbeatResponse struct

+ Err KError

+ type JoinGroupRequest struct

+ GroupId string

+ GroupProtocols map[string][]byte

+ MemberId string

+ ProtocolType string

+ SessionTimeout int32

+ func (r *JoinGroupRequest) AddGroupProtocol(name string, metadata []byte)

+ type JoinGroupResponse struct

+ Err KError

+ GenerationId int32

+ GroupProtocol string

+ LeaderId string

+ MemberId string

+ Members map[string][]byte

type KError

+ const ErrClusterAuthorizationFailed

+ const ErrGroupAuthorizationFailed

+ const ErrIllegalGeneration

+ const ErrInconsistentGroupProtocol

+ const ErrInvalidCommitOffsetSize

+ const ErrInvalidGroupId

+ const ErrInvalidRequiredAcks

+ const ErrInvalidSessionTimeout

+ const ErrRebalanceInProgress

+ const ErrTopicAuthorizationFailed

+ const ErrUnknownMemberId

+ type LeaveGroupRequest struct

+ GroupId string

+ MemberId string

+ type LeaveGroupResponse struct

+ Err KError

+ type ListGroupsRequest struct

+ type ListGroupsResponse struct

+ Err KError

+ Groups map[string]string

+ type SyncGroupRequest struct

+ GenerationId int32

+ GroupAssignments map[string][]byte

+ GroupId string

+ MemberId string

+ func (r *SyncGroupRequest) AddGroupAssignment(memberId string, memberAssignment []byte)

+ type SyncGroupResponse struct

+ Err KError

+ MemberAssignment []byte

Sep 4, 2015

Changes in this version

+ type OffsetManager interface

+ Close func() error

+ ManagePartition func(topic string, partition int32) (PartitionOffsetManager, error)

+ func NewOffsetManagerFromClient(group string, client Client) (OffsetManager, error)

+ type PartitionOffsetManager interface

+ AsyncClose func()

+ Close func() error

+ Errors func() <-chan *ConsumerError

+ MarkOffset func(offset int64, metadata string)

+ NextOffset func() (int64, string)

Aug 17, 2015

Changes in this version

type OffsetCommitResponse

+ func (r *OffsetCommitResponse) AddError(topic string, partition int32, kerror KError)

type OffsetFetchResponse

+ func (r *OffsetFetchResponse) AddBlock(topic string, partition int32, block *OffsetFetchResponseBlock)

+ func (r *OffsetFetchResponse) GetBlock(topic string, partition int32) *OffsetFetchResponseBlock

May 1, 2015

Changes in this version

type Consumer

+ Partitions func(topic string) ([]int32, error)

+ Topics func() ([]string, error)

type PartitionConsumer

+ HighWaterMarkOffset func() int64

Apr 16, 2015

Changes in this version

type Client

+ Coordinator func(consumerGroup string) (*Broker, error)

+ RefreshCoordinator func(consumerGroup string) error

type ConsumerMetadataResponse

+ Coordinator *Broker

Apr 7, 2015

Changes in this version

type OffsetCommitRequest

+ ConsumerGroupGeneration int32

+ ConsumerID string

+ RetentionTime int64

+ Version int16

type OffsetFetchRequest

+ Version int16

Mar 20, 2015

Changes in this version

type MetadataResponse

+ func (m *MetadataResponse) AddTopic(topic string, err KError) *TopicMetadata

Mar 17, 2015

Changes in this version

+ const OffsetNewest

+ const OffsetOldest

+ const ReceiveTime

+ var ErrAlreadyConnected = errors.New("kafka: broker connection already initiated")

+ var ErrClosedClient = errors.New("kafka: tried to use a client that was closed")

+ var ErrIncompleteResponse = errors.New("kafka: response did not contain all the expected topic/partition blocks")

+ var ErrInsufficientData = errors.New("kafka: insufficient data to decode packet, more bytes expected")

+ var ErrInvalidPartition = errors.New("kafka: partitioner returned an invalid partition index")

+ var ErrMessageTooLarge = errors.New("kafka: message is larger than Consumer.Fetch.Max")

+ var ErrNotConnected = errors.New("kafka: broker not connected")

+ var ErrOutOfBrokers = errors.New(...)

+ var ErrShuttingDown = errors.New("kafka: message received by producer in process of shutting down")

+ var MaxRequestSize int32 = 100 * 1024 * 1024

+ var MaxResponseSize int32 = 100 * 1024 * 1024

+ var PanicHandler func(interface{})

+ type AsyncProducer interface

+ AsyncClose func()

+ Close func() error

+ Errors func() <-chan *ProducerError

+ Input func() chan<- *ProducerMessage

+ Successes func() <-chan *ProducerMessage

+ func NewAsyncProducer(addrs []string, conf *Config) (AsyncProducer, error)

+ func NewAsyncProducerFromClient(client Client) (AsyncProducer, error)

+ type Broker struct

+ func NewBroker(addr string) *Broker

+ func (b *Broker) Addr() string

+ func (b *Broker) Close() error

+ func (b *Broker) CommitOffset(request *OffsetCommitRequest) (*OffsetCommitResponse, error)

+ func (b *Broker) Connected() (bool, error)

+ func (b *Broker) Fetch(request *FetchRequest) (*FetchResponse, error)

+ func (b *Broker) FetchOffset(request *OffsetFetchRequest) (*OffsetFetchResponse, error)

+ func (b *Broker) GetAvailableOffsets(request *OffsetRequest) (*OffsetResponse, error)

+ func (b *Broker) GetConsumerMetadata(request *ConsumerMetadataRequest) (*ConsumerMetadataResponse, error)

+ func (b *Broker) GetMetadata(request *MetadataRequest) (*MetadataResponse, error)

+ func (b *Broker) ID() int32

+ func (b *Broker) Open(conf *Config) error

+ func (b *Broker) Produce(request *ProduceRequest) (*ProduceResponse, error)

+ type ByteEncoder []byte

+ func (b ByteEncoder) Encode() ([]byte, error)

+ func (b ByteEncoder) Length() int

+ type Client interface

+ Close func() error

+ Closed func() bool

+ Config func() *Config

+ GetOffset func(topic string, partitionID int32, time int64) (int64, error)

+ Leader func(topic string, partitionID int32) (*Broker, error)

+ Partitions func(topic string) ([]int32, error)

+ RefreshMetadata func(topics ...string) error

+ Replicas func(topic string, partitionID int32) ([]int32, error)

+ Topics func() ([]string, error)

+ WritablePartitions func(topic string) ([]int32, error)

+ func NewClient(addrs []string, conf *Config) (Client, error)

+ type CompressionCodec int8

+ const CompressionGZIP

+ const CompressionNone

+ const CompressionSnappy

+ type Config struct

+ ChannelBufferSize int

+ ClientID string

+ Consumer struct{ ... }

+ Metadata struct{ ... }

+ Net struct{ ... }

+ Producer struct{ ... }

+ func NewConfig() *Config

+ func (c *Config) Validate() error

+ type ConfigurationError string

+ func (err ConfigurationError) Error() string

+ type Consumer interface

+ Close func() error

+ ConsumePartition func(topic string, partition int32, offset int64) (PartitionConsumer, error)

+ func NewConsumer(addrs []string, config *Config) (Consumer, error)

+ func NewConsumerFromClient(client Client) (Consumer, error)

+ type ConsumerError struct

+ Err error

+ Partition int32

+ Topic string

+ func (ce ConsumerError) Error() string

+ type ConsumerErrors []*ConsumerError

+ func (ce ConsumerErrors) Error() string

+ type ConsumerMessage struct

+ Key []byte

+ Offset int64

+ Partition int32

+ Topic string

+ Value []byte

+ type ConsumerMetadataRequest struct

+ ConsumerGroup string

+ type ConsumerMetadataResponse struct

+ CoordinatorHost string

+ CoordinatorID int32

+ CoordinatorPort int32

+ Err KError

+ type Encoder interface

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

+ Length func() int

+ type FetchRequest struct

+ MaxWaitTime int32

+ MinBytes int32

+ func (f *FetchRequest) AddBlock(topic string, partitionID int32, fetchOffset int64, maxBytes int32)

+ type FetchResponse struct

+ Blocks map[string]map[int32]*FetchResponseBlock

+ func (fr *FetchResponse) AddError(topic string, partition int32, err KError)

+ func (fr *FetchResponse) AddMessage(topic string, partition int32, key, value Encoder, offset int64)

+ func (fr *FetchResponse) GetBlock(topic string, partition int32) *FetchResponseBlock

+ type FetchResponseBlock struct

+ Err KError

+ HighWaterMarkOffset int64

+ MsgSet MessageSet

+ type KError int16

+ const ErrBrokerNotAvailable

+ const ErrConsumerCoordinatorNotAvailable

+ const ErrInvalidMessage

+ const ErrInvalidMessageSize

+ const ErrInvalidTopic

+ const ErrLeaderNotAvailable

+ const ErrMessageSetSizeTooLarge

+ const ErrMessageSizeTooLarge

+ const ErrNoError

+ const ErrNotCoordinatorForConsumer

+ const ErrNotEnoughReplicas

+ const ErrNotEnoughReplicasAfterAppend

+ const ErrNotLeaderForPartition

+ const ErrOffsetMetadataTooLarge

+ const ErrOffsetOutOfRange

+ const ErrOffsetsLoadInProgress

+ const ErrReplicaNotAvailable

+ const ErrRequestTimedOut

+ const ErrStaleControllerEpochCode

+ const ErrUnknown

+ const ErrUnknownTopicOrPartition

+ func (err KError) Error() string

+ type Message struct

+ Codec CompressionCodec

+ Key []byte

+ Set *MessageSet

+ Value []byte

+ type MessageBlock struct

+ Msg *Message

+ Offset int64

+ func (msb *MessageBlock) Messages() []*MessageBlock

+ type MessageSet struct

+ Messages []*MessageBlock

+ PartialTrailingMessage bool

+ type MetadataRequest struct

+ Topics []string

+ type MetadataResponse struct

+ Brokers []*Broker

+ Topics []*TopicMetadata

+ func (m *MetadataResponse) AddBroker(addr string, id int32)

+ func (m *MetadataResponse) AddTopicPartition(topic string, partition, brokerID int32, replicas, isr []int32, err KError)

+ type OffsetCommitRequest struct

+ ConsumerGroup string

+ func (r *OffsetCommitRequest) AddBlock(topic string, partitionID int32, offset int64, timestamp int64, ...)

+ type OffsetCommitResponse struct

+ Errors map[string]map[int32]KError

+ type OffsetFetchRequest struct

+ ConsumerGroup string

+ func (r *OffsetFetchRequest) AddPartition(topic string, partitionID int32)

+ type OffsetFetchResponse struct

+ Blocks map[string]map[int32]*OffsetFetchResponseBlock

+ type OffsetFetchResponseBlock struct

+ Err KError

+ Metadata string

+ Offset int64

+ type OffsetRequest struct

+ func (r *OffsetRequest) AddBlock(topic string, partitionID int32, time int64, maxOffsets int32)

+ type OffsetResponse struct

+ Blocks map[string]map[int32]*OffsetResponseBlock

+ func (r *OffsetResponse) AddTopicPartition(topic string, partition int32, offset int64)

+ func (r *OffsetResponse) GetBlock(topic string, partition int32) *OffsetResponseBlock

+ type OffsetResponseBlock struct

+ Err KError

+ Offsets []int64

+ type PacketDecodingError struct

+ Info string

+ func (err PacketDecodingError) Error() string

+ type PacketEncodingError struct

+ Info string

+ func (err PacketEncodingError) Error() string

+ type PartitionConsumer interface

+ AsyncClose func()

+ Close func() error

+ Errors func() <-chan *ConsumerError

+ Messages func() <-chan *ConsumerMessage

+ type PartitionMetadata struct

+ Err KError

+ ID int32

+ Isr []int32

+ Leader int32

+ Replicas []int32

+ type Partitioner interface

+ Partition func(message *ProducerMessage, numPartitions int32) (int32, error)

+ RequiresConsistency func() bool

+ func NewHashPartitioner(topic string) Partitioner

+ func NewManualPartitioner(topic string) Partitioner

+ func NewRandomPartitioner(topic string) Partitioner

+ func NewRoundRobinPartitioner(topic string) Partitioner

+ type PartitionerConstructor func(topic string) Partitioner

+ type ProduceRequest struct

+ RequiredAcks RequiredAcks

+ Timeout int32

+ func (p *ProduceRequest) AddMessage(topic string, partition int32, msg *Message)

+ func (p *ProduceRequest) AddSet(topic string, partition int32, set *MessageSet)

+ type ProduceResponse struct

+ Blocks map[string]map[int32]*ProduceResponseBlock

+ func (pr *ProduceResponse) AddTopicPartition(topic string, partition int32, err KError)

+ func (pr *ProduceResponse) GetBlock(topic string, partition int32) *ProduceResponseBlock

+ type ProduceResponseBlock struct

+ Err KError

+ Offset int64

+ type ProducerError struct

+ Err error

+ Msg *ProducerMessage

+ func (pe ProducerError) Error() string

+ type ProducerErrors []*ProducerError

+ func (pe ProducerErrors) Error() string

+ type ProducerMessage struct

+ Key Encoder

+ Metadata interface{}

+ Offset int64

+ Partition int32

+ Topic string

+ Value Encoder

+ type RequiredAcks int16

+ const NoResponse

+ const WaitForAll

+ const WaitForLocal

+ type StdLogger interface

+ Print func(v ...interface{})

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

+ Println func(v ...interface{})

+ var Logger StdLogger = log.New(ioutil.Discard, "[Sarama] ", log.LstdFlags)

+ type StringEncoder string

+ func (s StringEncoder) Encode() ([]byte, error)

+ func (s StringEncoder) Length() int

+ type SyncProducer interface

+ Close func() error

+ SendMessage func(msg *ProducerMessage) (partition int32, offset int64, err error)

+ func NewSyncProducer(addrs []string, config *Config) (SyncProducer, error)

+ func NewSyncProducerFromClient(client Client) (SyncProducer, error)

+ type TopicMetadata struct

+ Err KError

+ Name string

+ Partitions []*PartitionMetadata