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

kafka

package

Go to main page

Versions in this module

v1

v1.13.1

Apr 1, 2024

v1.13.0

Feb 21, 2024

Changes in this version

+ const DefaultMaxBulkSubAwaitDurationMs

+ const DefaultMaxBulkSubCount

+ var SHA256 scram.HashGeneratorFcn = sha256.New

+ var SHA512 scram.HashGeneratorFcn = sha512.New

+ func GetEventMetadata(message *sarama.ConsumerMessage) map[string]string

+ type BulkEventHandler func(ctx context.Context, msg *KafkaBulkMessage) ([]pubsub.BulkSubscribeResponseEntry, error)

+ type EventHandler func(ctx context.Context, msg *NewEvent) error

+ type Kafka struct

+ DefaultConsumeRetryEnabled bool

+ func NewKafka(logger logger.Logger) *Kafka

+ func (k *Kafka) AddTopicHandler(topic string, handlerConfig SubscriptionHandlerConfig)

+ func (k *Kafka) BulkPublish(_ context.Context, topic string, entries []pubsub.BulkMessageEntry, ...) (pubsub.BulkPublishResponse, error)

+ func (k *Kafka) Close() (err error)

+ func (k *Kafka) CloseSubscriptionResources()

+ func (k *Kafka) DeserializeValue(message *sarama.ConsumerMessage, config SubscriptionHandlerConfig) ([]byte, error)

+ func (k *Kafka) GetTopicHandlerConfig(topic string) (SubscriptionHandlerConfig, error)

+ func (k *Kafka) Init(ctx context.Context, metadata map[string]string) error

+ func (k *Kafka) Publish(_ context.Context, topic string, data []byte, metadata map[string]string) error

+ func (k *Kafka) RemoveTopicHandler(topic string)

+ func (k *Kafka) SerializeValue(topic string, data []byte, metadata map[string]string) ([]byte, error)

+ func (k *Kafka) Subscribe(ctx context.Context) error

+ type KafkaBulkMessage struct

+ Entries []KafkaBulkMessageEntry

+ Metadata map[string]string

+ Topic string

+ type KafkaBulkMessageEntry struct

+ ContentType string

+ EntryId string

+ Event []byte

+ Metadata map[string]string

+ type KafkaMetadata struct

+ AWSAccessKey string

+ AWSIamRoleArn string

+ AWSRegion string

+ AWSSecretKey string

+ AWSSessionToken string

+ AWSStsSessionName string

+ AuthType string

+ Brokers string

+ ClientID string

+ ConsumeRetryEnabled bool

+ ConsumeRetryInterval time.Duration

+ ConsumerGroup string

+ InitialOffset string

+ MaxMessageBytes int

+ OidcClientID string

+ OidcClientSecret string

+ OidcExtensions string

+ OidcScopes string

+ OidcTokenEndpoint string

+ SaslMechanism string

+ SaslPassword string

+ SaslUsername string

+ SchemaCachingEnabled bool

+ SchemaLatestVersionCacheTTL time.Duration

+ SchemaRegistryAPIKey string

+ SchemaRegistryAPISecret string

+ SchemaRegistryURL string

+ TLSCaCert string

+ TLSClientCert string

+ TLSClientKey string

+ TLSDisable bool

+ TLSSkipVerify bool

+ Version string

+ type NewEvent struct

+ ContentType *string

+ Data []byte

+ Metadata map[string]string

+ Topic string

+ type OAuthTokenSource struct

+ CachedToken oauth2.Token

+ ClientID string

+ ClientSecret string

+ Extensions map[string]string

+ Scopes []string

+ TokenEndpoint oauth2.Endpoint

+ func (ts *OAuthTokenSource) Token() (*sarama.AccessToken, error)

+ type SaramaLogBridge struct

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

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

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

+ type SchemaCacheEntry struct

+ type SchemaType int

+ const Avro

+ const None

+ func GetValueSchemaType(metadata map[string]string) (SchemaType, error)

+ type SubscriptionHandlerConfig struct

+ BulkHandler BulkEventHandler

+ Handler EventHandler

+ IsBulkSubscribe bool

+ SubscribeConfig pubsub.BulkSubscribeConfig

+ ValueSchemaType SchemaType

+ type TopicHandlerConfig map[string]SubscriptionHandlerConfig

+ func (tbh TopicHandlerConfig) TopicList() []string

+ type XDGSCRAMClient struct

+ func (x *XDGSCRAMClient) Begin(userName, password, authzID string) (err error)

+ func (x *XDGSCRAMClient) Done() bool

+ func (x *XDGSCRAMClient) Step(challenge string) (response string, err error)

v1.13.0-rc.10

Mar 1, 2024

v1.13.0-rc.9

Mar 1, 2024

v1.13.0-rc.8

Feb 21, 2024

v1.13.0-rc.7

Feb 26, 2024

v1.13.0-rc.6

Feb 21, 2024

v1.13.0-rc.5

Feb 21, 2024

v1.13.0-rc.4

Feb 9, 2024

v1.13.0-rc.3

Feb 7, 2024

v1.13.0-rc.2

Feb 2, 2024

v1.13.0-rc.1

Jan 30, 2024