Versions in this module Expand all Collapse all v1 v1.1.0 Sep 9, 2026 v1.0.0 Aug 25, 2026 Changes in this version + var ErrBatchDeliveryFailed = errors.New("kafka: one or more batch records failed delivery") + var ErrBatchHandlerRequired = errors.New("kafka: consumer batch handler is required") + var ErrBatchTooLarge = errors.New("kafka: producer batch exceeds configured byte limit") + var ErrBrokersRequired = errors.New("kafka: at least one broker is required") + var ErrClientIDRequired = errors.New("kafka: client ID is required") + var ErrClientIDTooLarge = errors.New("kafka: client ID exceeds configured limit") + var ErrConsumerBusy = errors.New("kafka: consumer runner is already active") + var ErrConsumerClosed = errors.New("kafka: consumer is closed") + var ErrConsumerClosing = errors.New("kafka: consumer is shutting down") + var ErrConsumerDrainActive = errors.New("kafka: consumer drain is already active") + var ErrConsumerDrainIncomplete = errors.New("kafka: consumer drain is incomplete") + var ErrConsumerDraining = errors.New("kafka: consumer is draining") + var ErrConsumerFailureStopped = errors.New("kafka: consumer failure stopped without settlement") + var ErrConsumerFatal = errors.New("kafka: consumer entered a fatal state") + var ErrConsumerInstanceFenced = errors.New("kafka: consumer static instance was fenced") + var ErrConsumerOwnershipLost = errors.New("kafka: consumer partition ownership was lost") + var ErrConsumerRebalance = errors.New("kafka: consumer handler canceled for a pending rebalance") + var ErrConsumerShutdownActive = errors.New("kafka: consumer shutdown is already active") + var ErrConsumerShutdownIncomplete = errors.New("kafka: consumer shutdown is incomplete") + var ErrContextRequired = errors.New("kafka: context is required") + var ErrCredentialProviderFailed = errors.New("kafka: credential provider failed") + var ErrCredentialProviderPanic = errors.New("kafka: credential provider panicked") + var ErrDeliveryResultInvalid = errors.New("kafka: producer returned inconsistent delivery results") + var ErrDeliveryResultMissing = errors.New("kafka: producer omitted a delivery result") + var ErrDrainIncomplete = errors.New("kafka: producer drain is incomplete") + var ErrDuplicateBroker = errors.New("kafka: broker address is duplicated") + var ErrDuplicateInspectionTarget = errors.New("kafka: inspection target is duplicated") + var ErrDuplicatePausePartition = errors.New("kafka: pause partition is duplicated") + var ErrDuplicateReplayCheckpoint = errors.New("kafka: replay checkpoint position is duplicated") + var ErrDuplicateReplayRange = errors.New("kafka: replay range is duplicated") + var ErrDuplicateTopic = errors.New("kafka: topic is duplicated") + var ErrExpiredOAuthBearerToken = errors.New("kafka: credential provider returned an expired OAuth bearer token") + var ErrFailureAttemptsExhausted = errors.New("kafka: consumer failure attempts exhausted") + var ErrFailureBackoff = errors.New("kafka: consumer failure retry backoff interrupted") + var ErrFailureCallbackPanic = errors.New("kafka: consumer failure callback panicked") + var ErrFailureDelegate = errors.New("kafka: consumer failure delegate failed") + var ErrFailureDelegateRequired = errors.New("kafka: consumer failure delegate is required") + var ErrFailurePublish = errors.New("kafka: consumer failure publication failed") + var ErrFailurePublisherRequired = errors.New("kafka: consumer failure publisher is required") + var ErrFailureRecordInvalid = errors.New("kafka: consumer failure record is invalid") + var ErrFetchBatchMalformed = errors.New("kafka: fetched record batch compression is malformed") + var ErrFetchBatchTooLarge = errors.New("kafka: fetched record batch exceeds configured decoded byte limit") + var ErrFetchDecompressedBufferFull = errors.New("kafka: fetched record batches exceed configured decoded buffer limit") + var ErrGroupIDRequired = errors.New("kafka: consumer group ID is required") + var ErrGroupIDTooLarge = errors.New("kafka: consumer group ID exceeds configured limit") + var ErrHandlerPanic = errors.New("kafka: consumer handler panicked") + var ErrHandlerRequired = errors.New("kafka: consumer handler is required") + var ErrHeaderKeyRequired = errors.New("kafka: header key is required") + var ErrHeaderKeyTooLarge = errors.New("kafka: header key exceeds configured limit") + var ErrHeaderValueTooLarge = errors.New("kafka: header value exceeds configured limit") + var ErrHeadersTooLarge = errors.New("kafka: headers exceed aggregate configured limit") + var ErrInspectionResponseTooLarge = errors.New("kafka: broker inspection response exceeds configured limits") + var ErrInspectionTargetsFailed = errors.New("kafka: one or more inspection targets failed") + var ErrInspectionTargetsRequired = errors.New("kafka: inspection targets are required") + var ErrInspectorClosed = errors.New("kafka: inspector is closed") + var ErrInvalidAssignment = errors.New("kafka: consumer assignment is invalid") + var ErrInvalidBalancePolicy = errors.New("kafka: consumer balance policy is invalid") + var ErrInvalidBroker = errors.New("kafka: broker address is invalid") + var ErrInvalidClientCertificateRequest = errors.New("kafka: TLS client certificate request exceeds policy limits") + var ErrInvalidClientID = errors.New("kafka: client ID is invalid") + var ErrInvalidCompressionPreference = errors.New("kafka: producer compression preference is invalid") + var ErrInvalidConsumerConfig = errors.New("kafka: consumer configuration is outside bounded limits") + var ErrInvalidCredentials = errors.New("kafka: credential provider returned invalid credentials") + var ErrInvalidFailureBatch = errors.New("kafka: consumer failure batch is invalid") + var ErrInvalidFailureClassification = errors.New("kafka: consumer failure classification is invalid") + var ErrInvalidFailurePolicy = errors.New("kafka: consumer failure policy is invalid") + var ErrInvalidFailureTarget = errors.New("kafka: consumer failure target is invalid") + var ErrInvalidGroupID = errors.New("kafka: consumer group ID is invalid") + var ErrInvalidInspectionResponse = errors.New("kafka: broker inspection response is invalid") + var ErrInvalidInspectionTarget = errors.New("kafka: inspection target is invalid") + var ErrInvalidInspectorConfig = errors.New("kafka: inspector configuration is outside bounded limits") + var ErrInvalidInstanceID = errors.New("kafka: consumer instance ID is invalid") + var ErrInvalidMessageLimits = errors.New("kafka: all message limits must be positive") + var ErrInvalidObservation = errors.New("kafka: observation is invalid") + var ErrInvalidObserverPolicy = errors.New("kafka: observer policy is outside bounded limits") + var ErrInvalidOffsetPolicy = errors.New("kafka: consumer offset policy is invalid") + var ErrInvalidPartitionSelection = errors.New("kafka: producer partition selection is invalid") + var ErrInvalidPausePartition = errors.New("kafka: pause partition is invalid") + var ErrInvalidProducerConfig = errors.New("kafka: producer configuration is outside bounded limits") + var ErrInvalidProtocolPolicy = errors.New("kafka: protocol policy is invalid") + var ErrInvalidRack = errors.New("kafka: consumer rack is invalid") + var ErrInvalidReadinessPolicy = errors.New("kafka: inspector readiness policy is outside bounded limits") + var ErrInvalidRebalanceHandlerPolicy = errors.New("kafka: consumer rebalance handler policy is invalid") + var ErrInvalidReplayCheckpoint = errors.New("kafka: replay checkpoint is invalid") + var ErrInvalidReplayConfig = errors.New("kafka: replay configuration is outside bounded limits") + var ErrInvalidReplayRange = errors.New("kafka: replay range is invalid") + var ErrInvalidReplayTimestampWindow = errors.New("kafka: replay timestamp window is invalid") + var ErrInvalidSecurityConfig = errors.New("kafka: client security configuration is invalid") + var ErrInvalidTopic = errors.New("kafka: topic name is invalid") + var ErrInvalidTransactionProcessorConfig = errors.New("kafka: transaction processor configuration is invalid") + var ErrInvalidTrustAnchors = errors.New("kafka: trust-anchor provider returned invalid certificates") + var ErrKeyRequired = errors.New("kafka: record key is required by producer policy") + var ErrKeyTooLarge = errors.New("kafka: key exceeds configured limit") + var ErrObserverFailureHandlerRequired = errors.New("kafka: observer failure handler is required") + var ErrObserverPanic = errors.New("kafka: observer panicked") + var ErrObserverReentry = errors.New("kafka: observer callback cannot re-enter client") + var ErrPausePartitionsRequired = errors.New("kafka: at least one pause partition is required") + var ErrPauseTopicNotSubscribed = errors.New("kafka: pause topic is not subscribed") + var ErrProducerBusy = errors.New("kafka: producer has in-flight operations") + var ErrProducerClosed = errors.New("kafka: producer is closed") + var ErrProducerFatal = errors.New("kafka: producer entered a fatal state") + var ErrRecordsRequired = errors.New("kafka: at least one producer record is required") + var ErrReplayAlreadyRun = errors.New("kafka: replay reader has already run") + var ErrReplayBoundsUnavailable = errors.New("kafka: replay broker offset bounds are unavailable") + var ErrReplayBusy = errors.New("kafka: replay reader is already running") + var ErrReplayClosed = errors.New("kafka: replay reader is closed") + var ErrReplayClosing = errors.New("kafka: replay reader is shutting down") + var ErrReplayOffsetGap = errors.New("kafka: replay range contains an offset gap") + var ErrReplayOffsetOutOfRange = errors.New("kafka: replay offset is outside broker retention bounds") + var ErrReplayRangesRequired = errors.New("kafka: at least one replay range is required") + var ErrReplayShutdownActive = errors.New("kafka: replay shutdown is already active") + var ErrReplayShutdownIncomplete = errors.New("kafka: replay shutdown is incomplete") + var ErrReplaySideEffectsDenied = errors.New("kafka: replay side effects require explicit opt-in") + var ErrReplayStalled = errors.New("kafka: replay made no progress before its bounded deadline") + var ErrReplayTimestampRangeIncomplete = errors.New("kafka: replay timestamp window may precede retained records") + var ErrTooManyAssignedPartitions = errors.New("kafka: assigned partition count exceeds configured limit") + var ErrTooManyBatchRecords = errors.New("kafka: producer batch record count exceeds configured limit") + var ErrTooManyBrokers = errors.New("kafka: broker count exceeds configured limit") + var ErrTooManyFetchedRecords = errors.New("kafka: fetched record count exceeds configured limit") + var ErrTooManyHeaders = errors.New("kafka: header count exceeds configured limit") + var ErrTooManyInspectionTargets = errors.New("kafka: inspection target count exceeds configured limit") + var ErrTooManyPausedPartitions = errors.New("kafka: paused partition count exceeds configured limit") + var ErrTooManyReplayRanges = errors.New("kafka: replay range count exceeds configured limit") + var ErrTooManyTopics = errors.New("kafka: topic count exceeds configured limit") + var ErrTooManyTransactionOutputRecords = errors.New("kafka: transaction output record count exceeds configured limit") + var ErrTopicNotAllowed = errors.New("kafka: topic is outside producer allowlist") + var ErrTopicRequired = errors.New("kafka: topic is required") + var ErrTopicTooLarge = errors.New("kafka: topic exceeds configured limit") + var ErrTopicsRequired = errors.New("kafka: at least one topic is required") + var ErrTransactionClosed = errors.New("kafka: transaction is closed") + var ErrTransactionHandlerRequired = errors.New("kafka: transaction processor handler is required") + var ErrTransactionInProgress = errors.New("kafka: producer transaction is in progress") + var ErrTransactionNotCommitted = errors.New("kafka: consume-transform-produce transaction was not committed") + var ErrTransactionOutcomeUnknown = errors.New("kafka: transaction commit outcome is unknown") + var ErrTransactionOutputTooLarge = errors.New("kafka: transaction output bytes exceed configured limit") + var ErrTransactionPanic = errors.New("kafka: transaction callback panicked") + var ErrTransactionProcessorBusy = errors.New("kafka: transaction processor runner is already active") + var ErrTransactionProcessorClosed = errors.New("kafka: transaction processor is closed") + var ErrTransactionProcessorClosing = errors.New("kafka: transaction processor is shutting down") + var ErrTransactionProcessorFatal = errors.New("kafka: transaction processor entered a fatal state") + var ErrTransactionProcessorShutdownActive = errors.New("kafka: transaction processor shutdown is already active") + var ErrTransactionProcessorShutdownIncomplete = errors.New("kafka: transaction processor shutdown is incomplete") + var ErrTransactionRequired = errors.New("kafka: transaction callback is required") + var ErrTransactionsDisabled = errors.New("kafka: producer transactions are disabled") + var ErrUnexpectedReplayRecord = errors.New("kafka: replay returned a record outside the requested ranges") + var ErrValueTooLarge = errors.New("kafka: value exceeds configured limit") + type Authentication struct + func NewOAuthBearerAuthentication(provider OAuthBearerProvider) Authentication + func NewPlainAuthentication(provider UsernamePasswordProvider) Authentication + func NewSCRAMSHA256Authentication(provider UsernamePasswordProvider) Authentication + func NewSCRAMSHA512Authentication(provider UsernamePasswordProvider) Authentication + func (authentication Authentication) GoString() string + func (authentication Authentication) Method() AuthenticationMethod + func (authentication Authentication) String() string + type AuthenticationMethod uint8 + const AuthenticationNone + const AuthenticationOAuthBearer + const AuthenticationPlain + const AuthenticationSCRAMSHA256 + const AuthenticationSCRAMSHA512 + func (method AuthenticationMethod) String() string + type BatchFailure struct + Attempt int + Batch ConsumedBatch + Category ErrorCategory + func (failure BatchFailure) Cause() error + func (failure BatchFailure) Retain() BatchFailure + type BatchFailureDelegate interface + HandleBatchFailure func(context.Context, BatchFailure) error + type BatchFailureDelegateFunc func(context.Context, BatchFailure) error + func (delegate BatchFailureDelegateFunc) HandleBatchFailure(ctx context.Context, failure BatchFailure) error + type BatchFailureHandlerConfig struct + Classifier FailureClassifier + Delegate BatchFailureDelegate + Handler BatchHandler + Limits MessageLimits + MaxBatchBytes int64 + MaxBatchRecords int + Mode FailureMode + PublishTimeout time.Duration + Publisher BatchFailurePublisher + Retry FailureRetryPolicy + Target FailureTarget + func (config BatchFailureHandlerConfig) Validate() error + type BatchFailurePublisher interface + PublishBatch func(context.Context, []ProducerRecord) ([]DeliveryResult, error) + type BatchFailurePublisherFunc func(context.Context, []ProducerRecord) ([]DeliveryResult, error) + func (publisher BatchFailurePublisherFunc) PublishBatch(ctx context.Context, records []ProducerRecord) ([]DeliveryResult, error) + type BatchHandler interface + HandleBatch func(context.Context, ConsumedBatch) error + func NewBatchFailureHandler(config BatchFailureHandlerConfig) (BatchHandler, error) + type BatchHandlerFunc func(context.Context, ConsumedBatch) error + func (handler BatchHandlerFunc) HandleBatch(ctx context.Context, batch ConsumedBatch) error + type BrokerState struct + Host string + NodeID int32 + Port int32 + Rack string + type ClientCertificateProvider interface + ClientCertificate func(context.Context, ClientCertificateRequest) (tls.Certificate, error) + type ClientCertificateProviderFunc func(context.Context, ClientCertificateRequest) (tls.Certificate, error) + func (provider ClientCertificateProviderFunc) ClientCertificate(ctx context.Context, request ClientCertificateRequest) (tls.Certificate, error) + type ClientCertificateRequest struct + AcceptableCAs [][]byte + SignatureSchemes []tls.SignatureScheme + Version uint16 + type ClientSecurity struct + Authentication Authentication + ClientCertificateProvider ClientCertificateProvider + CredentialTimeout time.Duration + TLS *tls.Config + Transport TransportSecurity + TrustAnchorProvider TrustAnchorProvider + func DevelopmentPlaintextSecurity() ClientSecurity + func (security ClientSecurity) GoString() string + func (security ClientSecurity) String() string + func (security ClientSecurity) Validate() error + type ClusterState struct + Brokers []BrokerState + ControllerID int32 + ControllerVisible bool + ID string + IDVisible bool + type CompressionCodec uint8 + const CompressionGzip + const CompressionLz4 + const CompressionNone + const CompressionSnappy + const CompressionZstd + func (codec CompressionCodec) String() string + type ConsumedBatch struct + Partition int32 + Records []ConsumedRecord + Topic string + func (batch ConsumedBatch) Retain() ConsumedBatch + type ConsumedMessage = ConsumedRecord + type ConsumedRecord struct + Headers []Header + Key []byte + LeaderEpoch int32 + Offset int64 + Partition int32 + Timestamp time.Time + TimestampType TimestampType + Topic string + Value []byte + func (record ConsumedRecord) Retain() ConsumedRecord + type Consumer struct + func NewConsumer(config ConsumerConfig) (*Consumer, error) + func (consumer *Consumer) Assignment() (ConsumerAssignment, error) + func (consumer *Consumer) Close() error + func (consumer *Consumer) Drain(ctx context.Context) error + func (consumer *Consumer) PausePartitions(partitions ...TopicPartition) error + func (consumer *Consumer) PausedPartitions() []TopicPartition + func (consumer *Consumer) ResumePartitions(partitions ...TopicPartition) error + func (consumer *Consumer) Run(ctx context.Context, handler Handler) error + func (consumer *Consumer) RunBatchOnce(ctx context.Context, handler BatchHandler) (PollResult, error) + func (consumer *Consumer) RunOnce(ctx context.Context, handler Handler) (PollResult, error) + func (consumer *Consumer) Shutdown(ctx context.Context) (err error) + type ConsumerAssignment struct + Epoch uint64 + Lost bool + Partitions []TopicPartition + type ConsumerConfig struct + BalancePolicy GroupBalancePolicy + BrokerMaxReadBytes int32 + Brokers []string + ClientID string + CommitTimeout time.Duration + DialTimeout time.Duration + FetchMaxBytes int32 + FetchMaxPartitionBytes int32 + FetchMaxWait time.Duration + FetchMinBytes int32 + GroupID string + HandlerTimeout time.Duration + HeartbeatInterval time.Duration + InstanceID string + Limits MessageLimits + MaxAssignedPartitions int + MaxBufferedDecompressedBytes int64 + MaxConcurrentFetches int + MaxConcurrentHandlers int + MaxDecompressedBatchBytes int64 + MaxPausedPartitions int + MaxPollRecords int + Observers ObserverPolicy + Protocol ProtocolPolicy + Rack string + RebalanceHandler RebalanceHandlerPolicy + RebalanceTimeout time.Duration + ResetOffset OffsetPolicy + Security ClientSecurity + SessionTimeout time.Duration + ShutdownTimeout time.Duration + Topics []string + func (config ConsumerConfig) Validate() error + type ConsumerError struct + func (err *ConsumerError) Category() ErrorCategory + func (err *ConsumerError) Error() string + func (err *ConsumerError) Operation() ConsumerOperation + func (err *ConsumerError) Retryable() bool + func (err *ConsumerError) Unwrap() error + type ConsumerGroupInspectionResult struct + Category ErrorCategory + Err error + Group string + State ConsumerGroupState + type ConsumerGroupMemberState struct + Assignments []TopicPartition + ClientHost string + ClientID string + InstanceID string + InstanceIDVisible bool + MemberID string + type ConsumerGroupPartitionLag struct + CommittedOffset int64 + EndOffset int64 + Lag int64 + Partition int32 + StartOffset int64 + Topic string + type ConsumerGroupState struct + CoordinatorID int32 + Group string + Members []ConsumerGroupMemberState + Partitions []ConsumerGroupPartitionLag + Protocol string + ProtocolType string + State string + type ConsumerOperation uint8 + const ConsumerOperationCommit + const ConsumerOperationLeave + const ConsumerOperationPoll + func (operation ConsumerOperation) String() string + type ConsumerProtocolGroupInspectionResult struct + Category ErrorCategory + Err error + Group string + State ConsumerProtocolGroupState + type ConsumerProtocolGroupMemberState struct + Assignments []TopicPartition + ClientHost string + ClientID string + InstanceID string + InstanceIDVisible bool + MemberEpoch int32 + MemberID string + MemberType ConsumerProtocolMemberType + RackID string + RackIDVisible bool + SubscribedTopicRegex string + SubscribedTopicRegexVisible bool + SubscribedTopics []string + TargetAssignments []TopicPartition + type ConsumerProtocolGroupState struct + AssignmentEpoch int32 + Assignor string + CoordinatorID int32 + Epoch int32 + Group string + Members []ConsumerProtocolGroupMemberState + Partitions []ConsumerGroupPartitionLag + State string + type ConsumerProtocolMemberType int8 + const ConsumerProtocolMemberTypeClassic + const ConsumerProtocolMemberTypeConsumer + const ConsumerProtocolMemberTypeUnknown + type DeliveryError struct + func (err *DeliveryError) Category() ErrorCategory + func (err *DeliveryError) Error() string + func (err *DeliveryError) Retryable() bool + func (err *DeliveryError) Unwrap() error + type DeliveryResult struct + Err error + Offset int64 + Partition int32 + Timestamp time.Time + Topic string + type ErrorCategory uint8 + const ErrorAmbiguous + const ErrorAuthorization + const ErrorCanceled + const ErrorFatal + const ErrorFenced + const ErrorOversized + const ErrorPermanent + const ErrorRetryable + const ErrorShutdown + const ErrorTimeout + const ErrorUnknown + func (category ErrorCategory) String() string + type FailureClassifier interface + ClassifyFailure func(error) ErrorCategory + type FailureClassifierFunc func(error) ErrorCategory + func (classifier FailureClassifierFunc) ClassifyFailure(err error) ErrorCategory + type FailureDelegate interface + HandleFailure func(context.Context, HandlerFailure) error + type FailureDelegateFunc func(context.Context, HandlerFailure) error + func (delegate FailureDelegateFunc) HandleFailure(ctx context.Context, failure HandlerFailure) error + type FailureHandlerConfig struct + Classifier FailureClassifier + Delegate FailureDelegate + Handler Handler + Limits MessageLimits + Mode FailureMode + PublishTimeout time.Duration + Publisher FailurePublisher + Retry FailureRetryPolicy + Target FailureTarget + func (config FailureHandlerConfig) Validate() error + type FailureHandlingError struct + func (err *FailureHandlingError) Attempt() int + func (err *FailureHandlingError) Category() ErrorCategory + func (err *FailureHandlingError) DeliveryResults() []DeliveryResult + func (err *FailureHandlingError) Error() string + func (err *FailureHandlingError) Stage() FailureStage + func (err *FailureHandlingError) Unwrap() []error + type FailureMode uint8 + const FailureModeDeadLetter + const FailureModeDelegate + const FailureModeRetryTopic + const FailureModeStop + type FailurePublisher interface + PublishRecord func(context.Context, ProducerRecord) DeliveryResult + type FailureRetryPolicy struct + Categories []ErrorCategory + InitialBackoff time.Duration + MaxAttempts int + MaxBackoff time.Duration + type FailureStage uint8 + const FailureStageBackoff + const FailureStageClassify + const FailureStageDelegate + const FailureStagePublish + const FailureStageStop + func (stage FailureStage) String() string + type FailureTarget struct + Topic string + Version uint16 + type GroupBalancePolicy uint8 + const BalanceCooperativeSticky + const BalanceEagerSticky + const BalanceEagerToCooperative + type Handler interface + Handle func(context.Context, ConsumedMessage) error + func NewFailureHandler(config FailureHandlerConfig) (Handler, error) + type HandlerFailure struct + Attempt int + Category ErrorCategory + Record ConsumedRecord + func (failure HandlerFailure) Cause() error + func (failure HandlerFailure) Retain() HandlerFailure + type HandlerFunc func(context.Context, ConsumedMessage) error + func (handler HandlerFunc) Handle(ctx context.Context, message ConsumedMessage) error + type Header struct + Key string + Value []byte + type Inspector struct + func NewInspector(config InspectorConfig) (*Inspector, error) + func (inspector *Inspector) Close() error + func (inspector *Inspector) Cluster(ctx context.Context) (result ClusterState, resultErr error) + func (inspector *Inspector) ConsumerGroupLag(ctx context.Context, groups ...string) (result []ConsumerGroupState, resultErr error) + func (inspector *Inspector) ConsumerProtocolGroupLag(ctx context.Context, groups ...string) (result []ConsumerProtocolGroupState, resultErr error) + func (inspector *Inspector) DependencyHealth(ctx context.Context) (resultErr error) + func (inspector *Inspector) Health(ctx context.Context) error + func (inspector *Inspector) InspectConsumerGroups(ctx context.Context, groups ...string) ([]ConsumerGroupInspectionResult, error) + func (inspector *Inspector) InspectConsumerProtocolGroups(ctx context.Context, groups ...string) ([]ConsumerProtocolGroupInspectionResult, error) + func (inspector *Inspector) InspectTopics(ctx context.Context, topics ...string) ([]TopicInspectionResult, error) + func (inspector *Inspector) Liveness() LivenessState + func (inspector *Inspector) PlanReplayByTimestamp(ctx context.Context, request ReplayTimestampRequest) (ReplayTimestampPlan, error) + func (inspector *Inspector) Readiness(ctx context.Context) (state ReadinessState, resultErr error) + func (inspector *Inspector) Topics(ctx context.Context, topics ...string) (result []TopicState, resultErr error) + type InspectorConfig struct + Brokers []string + ClientID string + DialTimeout time.Duration + MaxConcurrentInspections int + MaxGroupMembers int + MaxMetadataBrokers int + MaxMetadataPartitions int + Observers ObserverPolicy + Protocol ProtocolPolicy + Readiness ReadinessPolicy + RequestTimeout time.Duration + Security ClientSecurity + type KeyPolicy uint8 + const KeyRequired + const UnkeyedAllowed + type LivenessState struct + Live bool + type Message = ProducerRecord + type MessageLimits struct + MaxHeaderBytes int + MaxHeaderKeyBytes int + MaxHeaderValueBytes int + MaxHeaders int + MaxKeyBytes int + MaxTopicBytes int + MaxValueBytes int + func DefaultMessageLimits() MessageLimits + func (limits MessageLimits) Validate() error + type OAuthBearerProvider interface + Token func(context.Context) (OAuthBearerToken, error) + type OAuthBearerProviderFunc func(context.Context) (OAuthBearerToken, error) + func (provider OAuthBearerProviderFunc) Token(ctx context.Context) (OAuthBearerToken, error) + type OAuthBearerToken struct + AuthorizationID string + ExpiresAt time.Time + Extensions map[string]string + Token []byte + func (token OAuthBearerToken) GoString() string + func (token OAuthBearerToken) String() string + type Observation struct + APIKey int16 + APIKeyKnown bool + AuthenticationMethod AuthenticationMethod + BrokerCount int + BrokerID int32 + BrokerKnown bool + Category ErrorCategory + ClientID string + CommittedCount int + ConsecutiveFailures int + ConsecutiveSuccesses int + DependencyHealthy bool + Duration time.Duration + GroupCount int + GroupID string + GroupMemberCount int + Kind ObservationKind + Offset int64 + OffsetKnown bool + Partition int32 + PartitionCount int + PartitionKnown bool + ProcessedCount int + QueueDuration time.Duration + Ready bool + RecordBytes int64 + RecordCount int + ReplayFailed int64 + ReplayProcessed int64 + ReplayRemaining int64 + ReplaySkipped int64 + RequestBytes int64 + ResponseBytes int64 + StartedAt time.Time + Succeeded bool + ThrottleDuration time.Duration + ThrottledAfterResponse bool + Timestamp time.Time + Topic string + TopicCount int + Truncated bool + func (observation Observation) Validate() error + type ObservationFailure struct + Kind ObservationKind + ObserverIndex int + Panicked bool + TimedOut bool + func (failure ObservationFailure) Cause() error + func (failure ObservationFailure) Error() string + type ObservationFailureFunc func(context.Context, ObservationFailure) + type ObservationKind uint8 + const ObservationBrokerConnect + const ObservationBrokerDisconnect + const ObservationBrokerRequest + const ObservationBrokerThrottle + const ObservationConsumeAssigned + const ObservationConsumeBatch + const ObservationConsumeBlocked + const ObservationConsumeCommit + const ObservationConsumeGroupError + const ObservationConsumeLost + const ObservationConsumePoll + const ObservationConsumeRebalanceWait + const ObservationConsumeRecord + const ObservationConsumeRetryScheduled + const ObservationConsumeRevoked + const ObservationConsumerShutdown + const ObservationDependencyHealth + const ObservationInspectorCluster + const ObservationInspectorConsumerGroups + const ObservationInspectorShutdown + const ObservationInspectorTopics + const ObservationProduceAsync + const ObservationProduceBatch + const ObservationProduceRecord + const ObservationProducerShutdown + const ObservationReadiness + const ObservationReplayPlan + const ObservationReplayRecord + const ObservationReplayRun + const ObservationReplayShutdown + const ObservationTransactionAbort + const ObservationTransactionBegin + const ObservationTransactionCommit + const ObservationTransactionProcessorShutdown + func (kind ObservationKind) String() string + type ObserverFunc func(context.Context, Observation) error + type ObserverPolicy struct + FailureHandler ObservationFailureFunc + Observers []ObserverFunc + Timeout time.Duration + func (policy ObserverPolicy) Validate() error + type OffsetPolicy uint8 + const OffsetEarliest + const OffsetLatest + type PartitionSelection struct + Mode PartitionSelectionMode + Partition int32 + func ExplicitPartition(partition int32) PartitionSelection + type PartitionSelectionMode uint8 + const PartitionAutomatic + const PartitionExplicit + type PollResult struct + Committed int + Polled int + Processed int + type Producer struct + func NewProducer(config ProducerConfig) (*Producer, error) + func (producer *Producer) Abort(ctx context.Context) error + func (producer *Producer) Close() error + func (producer *Producer) Diagnostic() ProducerDiagnostic + func (producer *Producer) Drain(ctx context.Context) error + func (producer *Producer) Health(ctx context.Context) error + func (producer *Producer) Publish(ctx context.Context, message Message) error + func (producer *Producer) PublishAsync(ctx context.Context, record ProducerRecord) (<-chan DeliveryResult, error) + func (producer *Producer) PublishBatch(ctx context.Context, records []ProducerRecord) (results []DeliveryResult, resultErr error) + func (producer *Producer) PublishRecord(ctx context.Context, record ProducerRecord) DeliveryResult + func (producer *Producer) RunTransaction(ctx context.Context, callback func(Transaction) error) error + func (producer *Producer) Shutdown(ctx context.Context) (resultErr error) + type ProducerConfig struct + AllowedTopics []string + Brokers []string + ClientID string + CompressionPreferences []CompressionCodec + DeliveryTimeout time.Duration + DialTimeout time.Duration + KeyPolicy KeyPolicy + Limits MessageLimits + Linger time.Duration + MaxBatchBytes int32 + MaxBatchRecords int + MaxBufferedBytes int + MaxBufferedRecords int + Observers ObserverPolicy + Protocol ProtocolPolicy + RecordRetries int + RequestTimeout time.Duration + RetryBackoffMax time.Duration + RetryBackoffMin time.Duration + Security ClientSecurity + ShutdownTimeout time.Duration + TransactionEndTimeout time.Duration + TransactionTimeout time.Duration + TransactionalID string + func (config ProducerConfig) Validate() error + type ProducerDiagnostic struct + Accepting bool + AdmissionsInProgress int + BufferedBytes int64 + BufferedRecords int64 + Closed bool + Fatal bool + FatalCategory ErrorCategory + InFlightOperations int + MaintenanceActive bool + ShutdownComplete bool + TransactionActive bool + TransactionsEnabled bool + type ProducerRecord struct + Headers []Header + Key []byte + Partition PartitionSelection + Timestamp time.Time + Topic string + Value []byte + func (record ProducerRecord) Validate(limits MessageLimits) error + type ProtocolPolicy struct + MinimumVersion string + func (policy ProtocolPolicy) Validate() error + type ReadinessPolicy struct + FailureThreshold int + RecoveryThreshold int + func (policy ReadinessPolicy) Validate() error + type ReadinessState struct + ConsecutiveFailures int + ConsecutiveSuccesses int + DependencyHealthy bool + Ready bool + type RebalanceHandlerPolicy uint8 + const RebalanceCancelHandler + const RebalanceDrainHandler + type ReplayCheckpoint struct + Positions []ReplayPosition + func (checkpoint ReplayCheckpoint) Retain() ReplayCheckpoint + type ReplayConfig struct + BrokerMaxReadBytes int32 + Brokers []string + Checkpoint ReplayCheckpoint + ClientID string + DialTimeout time.Duration + FetchMaxBytes int32 + FetchMaxPartitionBytes int32 + FetchMaxWait time.Duration + FetchMinBytes int32 + HandlerTimeout time.Duration + Limits MessageLimits + MaxBufferedDecompressedBytes int64 + MaxConcurrentFetches int + MaxConcurrentHandlers int + MaxDecompressedBatchBytes int64 + MaxPollRecords int + Observers ObserverPolicy + PlanningTimeout time.Duration + ProgressTimeout time.Duration + Protocol ProtocolPolicy + Ranges []ReplayRange + Security ClientSecurity + ShutdownTimeout time.Duration + SideEffects ReplaySideEffectPolicy + func (config ReplayConfig) Validate() error + type ReplayHandler interface + HandleReplay func(context.Context, ReplayRecord) error + type ReplayHandlerFunc func(context.Context, ReplayRecord) error + func (handler ReplayHandlerFunc) HandleReplay(ctx context.Context, record ReplayRecord) error + type ReplayMetadata struct + EffectiveStartOffset int64 + Range ReplayRange + type ReplayPlan struct + Ranges []ReplayPlannedRange + TotalRemaining int64 + type ReplayPlannedRange struct + NextOffset int64 + Remaining int64 + type ReplayPosition struct + NextOffset int64 + Partition int32 + Topic string + type ReplayRange struct + EndOffset int64 + Partition int32 + StartOffset int64 + Topic string + type ReplayRangeResult struct + Complete bool + Failed int64 + NextOffset int64 + Processed int64 + Skipped int64 + type ReplayReader struct + func NewReplayReader(config ReplayConfig) (*ReplayReader, error) + func (reader *ReplayReader) Close() error + func (reader *ReplayReader) Plan() ReplayPlan + func (reader *ReplayReader) PlanAgainstBroker(ctx context.Context) (plan ReplayPlan, resultErr error) + func (reader *ReplayReader) Replay(ctx context.Context, handler ReplayHandler) (result ReplayResult, resultErr error) + func (reader *ReplayReader) Shutdown(ctx context.Context) (resultErr error) + type ReplayRecord struct + Metadata ReplayMetadata + func (record ReplayRecord) Retain() ReplayRecord + type ReplayResult struct + CompletedRanges int + Failed int64 + IncompleteRanges int + Polled int64 + Processed int64 + Ranges []ReplayRangeResult + Skipped int64 + func (result ReplayResult) Checkpoint() ReplayCheckpoint + type ReplaySideEffectPolicy uint8 + const ReplaySideEffectsAllowed + const ReplaySideEffectsDenied + type ReplayTimestampPartition struct + EndOffset int64 + Partition int32 + Remaining int64 + StartOffset int64 + Topic string + type ReplayTimestampPlan struct + EndExclusive time.Time + Partitions []ReplayTimestampPartition + StartInclusive time.Time + TotalRemaining int64 + func (plan ReplayTimestampPlan) ReplayRanges() []ReplayRange + type ReplayTimestampRequest struct + EndExclusive time.Time + Partitions []TopicPartition + StartInclusive time.Time + func (request ReplayTimestampRequest) Validate() error + type TimestampType int8 + const TimestampCreateTime + const TimestampLogAppendTime + const TimestampUnknown + type TopicCleanupPolicy uint8 + const TopicCleanupCompact + const TopicCleanupDelete + type TopicInspectionResult struct + Category ErrorCategory + Err error + State TopicState + Topic string + type TopicPartition struct + Partition int32 + Topic string + type TopicPartitionState struct + BeginningOffset int64 + EndOffset int64 + InSyncReplicaIDs []int32 + InSyncReplicas int + Leader int32 + LeaderEpoch int32 + OfflineReplicaIDs []int32 + OfflineReplicas int + Partition int32 + Replicas []int32 + ReplicationFactor int + type TopicState struct + CleanupPolicy TopicCleanupPolicy + DeleteRetentionMilliseconds int64 + Internal bool + LocalRetentionBytesPerPartition int64 + LocalRetentionMilliseconds int64 + LocalRetentionVisible bool + MaximumCompactionLagMilliseconds int64 + MinInSyncReplicas int + MinimumCleanableDirtyRatio float64 + MinimumCompactionLagMilliseconds int64 + Name string + Partitions []TopicPartitionState + RemoteLogCopyDisabled bool + RemoteLogCopyDisabledVisible bool + RemoteStorageEnabled bool + RemoteStorageEnabledVisible bool + RetentionBytesPerPartition int64 + RetentionMilliseconds int64 + SegmentBytes int64 + SegmentMilliseconds int64 + UncleanLeaderElectionEnabled bool + type Transaction struct + func (transaction Transaction) Publish(ctx context.Context, message Message) error + type TransactionConnectionConfig struct + Brokers []string + ClientID string + DialTimeout time.Duration + Protocol ProtocolPolicy + Security ClientSecurity + type TransactionError struct + func (err *TransactionError) Abortable() bool + func (err *TransactionError) Category() ErrorCategory + func (err *TransactionError) Error() string + func (err *TransactionError) Operation() TransactionOperation + func (err *TransactionError) OutcomeKnown() bool + func (err *TransactionError) Unwrap() error + type TransactionGroupConfig struct + BalancePolicy GroupBalancePolicy + BrokerMaxReadBytes int32 + FetchMaxBytes int32 + FetchMaxPartitionBytes int32 + FetchMaxWait time.Duration + FetchMinBytes int32 + GroupID string + HeartbeatInterval time.Duration + InstanceID string + MaxBufferedDecompressedBytes int64 + MaxConcurrentFetches int + MaxDecompressedBatchBytes int64 + MaxPollRecords int + ProcessingTimeout time.Duration + Rack string + RebalanceTimeout time.Duration + ResetOffset OffsetPolicy + SessionTimeout time.Duration + Topics []string + type TransactionHandler interface + Handle func(context.Context, ConsumedRecord, Transaction) error + type TransactionHandlerFunc func(context.Context, ConsumedRecord, Transaction) error + func (handler TransactionHandlerFunc) Handle(ctx context.Context, record ConsumedRecord, transaction Transaction) error + type TransactionOperation uint8 + const TransactionOperationAbort + const TransactionOperationBegin + const TransactionOperationCommit + func (operation TransactionOperation) String() string + type TransactionOutputConfig struct + AllowedTopics []string + CompressionPreferences []CompressionCodec + DeliveryTimeout time.Duration + KeyPolicy KeyPolicy + Linger time.Duration + MaxBatchBytes int32 + MaxBufferedBytes int + MaxBufferedRecords int + MaxOutputBytes int64 + MaxOutputRecords int + RecordRetries int + RequestTimeout time.Duration + RetryBackoffMax time.Duration + RetryBackoffMin time.Duration + TransactionEndTimeout time.Duration + TransactionTimeout time.Duration + TransactionalID string + type TransactionPollResult struct + Committed bool + Polled int + Processed int + Published int + type TransactionProcessor struct + func NewTransactionProcessor(config TransactionProcessorConfig) (*TransactionProcessor, error) + func (processor *TransactionProcessor) Close() error + func (processor *TransactionProcessor) Diagnostic() TransactionProcessorDiagnostic + func (processor *TransactionProcessor) Run(ctx context.Context, handler TransactionHandler) error + func (processor *TransactionProcessor) RunOnce(ctx context.Context, handler TransactionHandler) (TransactionPollResult, error) + func (processor *TransactionProcessor) Shutdown(ctx context.Context) (err error) + type TransactionProcessorConfig struct + Connection TransactionConnectionConfig + Group TransactionGroupConfig + Limits MessageLimits + Observers ObserverPolicy + Output TransactionOutputConfig + ShutdownTimeout time.Duration + func (config TransactionProcessorConfig) Validate() error + type TransactionProcessorDiagnostic struct + Accepting bool + BufferedBytes int64 + BufferedRecords int64 + ClientTerminated bool + Closed bool + Closing bool + Fatal bool + FatalCategory ErrorCategory + Running bool + ShutdownActive bool + TransactionActive bool + type TransportSecurity uint8 + const TransportDevelopmentPlaintext + const TransportTLS + func (transport TransportSecurity) String() string + type TrustAnchorProvider interface + TrustAnchors func(context.Context) (TrustAnchors, error) + type TrustAnchorProviderFunc func(context.Context) (TrustAnchors, error) + func (provider TrustAnchorProviderFunc) TrustAnchors(ctx context.Context) (TrustAnchors, error) + type TrustAnchors struct + Certificates [][]byte + func (anchors TrustAnchors) GoString() string + func (anchors TrustAnchors) String() string + type UsernamePassword struct + AuthorizationID string + Password []byte + Username string + func (credentials UsernamePassword) GoString() string + func (credentials UsernamePassword) String() string + type UsernamePasswordProvider interface + Credentials func(context.Context) (UsernamePassword, error) + type UsernamePasswordProviderFunc func(context.Context) (UsernamePassword, error) + func (provider UsernamePasswordProviderFunc) Credentials(ctx context.Context) (UsernamePassword, error)
Go to main page