kafkax

package
v0.0.64 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Aug 10, 2024 License: Apache-2.0 Imports: 15 Imported by: 0

README

Attributions

Code taken from nachogiljaldo's work on watermill-kafka.

Since his pull request made onto the base watermill-kafka repository was not merged yet and there seems to be no activity by the repository owners, we copied the code over to this kafkax package to allow for concurrent consumption of events, which is exactly the feature we needed.

Documentation

Index

Constants

View Source 
const NoSleep time.Duration = -1

NoSleep can be set to SubscriberConfig.NackResendSleep and SubscriberConfig.ReconnectRetrySleep.

View Source 
const UUIDHeaderKey = "_watermill_message_uuid"

Variables

This section is empty.

Functions

func DefaultSaramaSubscriberConfig 

func DefaultSaramaSubscriberConfig() *sarama.Config

DefaultSaramaSubscriberConfig creates default Sarama config used by Watermill.

Custom config can be passed to NewSubscriber and NewPublisher. 

saramaConfig := DefaultSaramaSubscriberConfig()
saramaConfig.Consumer.Offsets.Initial = sarama.OffsetOldest

subscriberConfig.OverwriteSaramaConfig = saramaConfig

subscriber, err := NewSubscriber(subscriberConfig, logger)
// ...

func DefaultSaramaSyncPublisherConfig 

func DefaultSaramaSyncPublisherConfig() *sarama.Config

func MessageKeyFromCtx 

func MessageKeyFromCtx(ctx context.Context) ([]byte, bool)

MessageKeyFromCtx returns Kafka internal key of the consumed message

func MessagePartitionFromCtx 

func MessagePartitionFromCtx(ctx context.Context) (int32, bool)

MessagePartitionFromCtx returns Kafka partition of the consumed message

func MessagePartitionOffsetFromCtx 

func MessagePartitionOffsetFromCtx(ctx context.Context) (int64, bool)

MessagePartitionOffsetFromCtx returns Kafka partition offset of the consumed message

func MessageTimestampFromCtx 

func MessageTimestampFromCtx(ctx context.Context) (time.Time, bool)

MessageTimestampFromCtx returns Kafka internal timestamp of the consumed message

Types

type BatchConsumerConfig 

type BatchConsumerConfig struct {
	// MaxBatchSize max amount of elements the batch will contain.
	// Default value is 100 if nothing is specified.
	MaxBatchSize int16
	// MaxWaitTime max time that it will be waited until MaxBatchSize elements are received.
	// Default value is 100ms if nothing is specified.
	MaxWaitTime time.Duration
}

BatchConsumerConfig configuration to be applied when the selected type of consumption is batch. Batch consumption means that the MaxBatchSize will be read or maxWaitTime waited the messages will then be sent to the output channel. ACK / NACK are handled properly to ensure at-least-once consumption.

type ConsumerModel 

type ConsumerModel int

ConsumerModel indicates the type of consumer model that will be used. 

const (
	// Default is a model when only one message is sent to the customer and customer needs to ACK the message
	// to receive the next.
	Default ConsumerModel = iota
	// Batch works by sending multiple messages in a batch
	Batch
	// PartitionConcurrent has one message sent to the customer per partition and customer needs to ACK the message
	// to receive the next message for the partition.
	PartitionConcurrent
)

type DefaultMarshaler 

type DefaultMarshaler struct{}

func (DefaultMarshaler) Marshal 

func (DefaultMarshaler) Marshal(topic string, msg *message.Message) (*sarama.ProducerMessage, error)

func (DefaultMarshaler) Unmarshal 

func (DefaultMarshaler) Unmarshal(kafkaMsg *sarama.ConsumerMessage) (*message.Message, error)

type GeneratePartitionKey 

type GeneratePartitionKey func(topic string, msg *message.Message) (string, error)

type Marshaler 

type Marshaler interface {
	Marshal(topic string, msg *message.Message) (*sarama.ProducerMessage, error)
}

Marshaler marshals Watermill's message to Kafka message.

type MarshalerUnmarshaler 

type MarshalerUnmarshaler interface {
	Marshaler
	Unmarshaler
}

func NewWithPartitioningMarshaler 

func NewWithPartitioningMarshaler(generatePartitionKey GeneratePartitionKey) MarshalerUnmarshaler

type MessageHandler 

type MessageHandler interface {
	// Setup is called at the beginning of a new session, before ConsumeClaim (and thus ProcessMessages) is called.
	// It may have a consumer group session in the case of a consumer group.
	Setup(*sarama.ConsumerGroupSession) error
	Cleanup(*sarama.ConsumerGroupSession) error
	ProcessMessages(
		ctx context.Context,
		kafkaMessages <-chan *sarama.ConsumerMessage,
		sess sarama.ConsumerGroupSession,
		messageLogFields watermill.LogFields,
	) error
}

MessageHandler an message processor that is able to receive a ConsumerMessage and perform some task with it. Once consumed, if there is a session, it will the offset will be marked as processed.

func NewBatchedMessageHandler 

func NewBatchedMessageHandler(
	outputChannel chan<- *message.Message,
	unmarshaler Unmarshaler,
	logger watermill.LoggerAdapter,
	closing chan struct{},
	maxBatchSize int16,
	maxWaitTime time.Duration,
	nackResendSleep time.Duration,
) MessageHandler

func NewMessageHandler 

func NewMessageHandler(
	outputChannel chan<- *message.Message,
	unmarshaler Unmarshaler,
	logger watermill.LoggerAdapter,
	closing chan struct{},
	nackResendSleep time.Duration,
) MessageHandler

func NewPartitionConcurrentMessageHandler 

func NewPartitionConcurrentMessageHandler(
	outputChannel chan<- *message.Message,
	unmarshaler Unmarshaler,
	logger watermill.LoggerAdapter,
	closing chan struct{},
	nackResendSleep time.Duration,
) MessageHandler

type OTELSaramaTracer 

type OTELSaramaTracer struct {
	// contains filtered or unexported fields
}

func (OTELSaramaTracer) WrapConsumer 

func (t OTELSaramaTracer) WrapConsumer(c sarama.Consumer, consumerInfo otelsaramax.ConsumerInfo) sarama.Consumer

func (OTELSaramaTracer) WrapConsumerGroupHandler 

func (t OTELSaramaTracer) WrapConsumerGroupHandler(h sarama.ConsumerGroupHandler, consumerInfo otelsaramax.ConsumerInfo) sarama.ConsumerGroupHandler

func (OTELSaramaTracer) WrapSyncProducer 

func (t OTELSaramaTracer) WrapSyncProducer(cfg *sarama.Config, p sarama.SyncProducer) sarama.SyncProducer

type PartitionOffset 

type PartitionOffset map[int32]int64

type Publisher 

type Publisher struct {
	// contains filtered or unexported fields
}

func NewPublisher 

func NewPublisher(
	config PublisherConfig,
	logger watermill.LoggerAdapter,
) (*Publisher, error)

NewPublisher creates a new Kafka Publisher.

func (*Publisher) BulkPublish added in v0.0.44 

func (p *Publisher) BulkPublish(topic string, msgs ...*message.Message) error

func (*Publisher) Close 

func (p *Publisher) Close() error

func (*Publisher) Publish 

func (p *Publisher) Publish(topic string, msgs ...*message.Message) error

Publish publishes message to Kafka.

Publish is blocking and wait for ack from Kafka. When one of messages delivery fails - function is interrupted.

type PublisherConfig 

type PublisherConfig struct {
	// Kafka brokers list.
	Brokers []string

	// Marshaler is used to marshal messages from Watermill format into Kafka format.
	Marshaler Marshaler

	// OverwriteSaramaConfig holds additional sarama settings.
	OverwriteSaramaConfig *sarama.Config

	// If true then each sent message will be wrapped with Opentelemetry tracing, provided by otelsarama.
	OTELEnabled bool

	// Tracer is used to trace Kafka messages.
	// If nil, then no tracing will be used.
	Tracer SaramaTracer
}

func (PublisherConfig) Validate 

func (c PublisherConfig) Validate() error

type SaramaTracer 

type SaramaTracer interface {
	WrapConsumer(sarama.Consumer, otelsaramax.ConsumerInfo) sarama.Consumer
	// WrapPartitionConsumer(sarama.PartitionConsumer) sarama.PartitionConsumer
	WrapConsumerGroupHandler(sarama.ConsumerGroupHandler, otelsaramax.ConsumerInfo) sarama.ConsumerGroupHandler
	WrapSyncProducer(*sarama.Config, sarama.SyncProducer) sarama.SyncProducer
}

func NewOTELSaramaTracer 

func NewOTELSaramaTracer(option ...otelsaramax.Option) SaramaTracer

type Subscriber 

type Subscriber struct {
	// contains filtered or unexported fields
}

func NewSubscriber 

func NewSubscriber(
	config SubscriberConfig,
	logger watermill.LoggerAdapter,
) (*Subscriber, error)

NewSubscriber creates a new Kafka Subscriber.

func (*Subscriber) Close 

func (s *Subscriber) Close() error

func (*Subscriber) PartitionOffset 

func (s *Subscriber) PartitionOffset(topic string) (PartitionOffset, error)

func (*Subscriber) Subscribe 

func (s *Subscriber) Subscribe(ctx context.Context, topic string) (<-chan *message.Message, error)

Subscribe subscribers for messages in Kafka.

There are multiple subscribers spawned

func (*Subscriber) SubscribeInitialize 

func (s *Subscriber) SubscribeInitialize(topic string) (err error)

type SubscriberConfig 

type SubscriberConfig struct {
	// Kafka brokers list.
	Brokers []string

	// Unmarshaler is used to unmarshal messages from Kafka format into Watermill format.
	Unmarshaler Unmarshaler

	// OverwriteSaramaConfig holds additional sarama settings.
	OverwriteSaramaConfig *sarama.Config

	// Kafka consumer group.
	// When empty, all messages from all partitions will be returned.
	ConsumerGroup string

	// How long after Nack message should be redelivered.
	NackResendSleep time.Duration

	// How long about unsuccessful reconnecting next reconnect will occur.
	ReconnectRetrySleep time.Duration

	InitializeTopicDetails *sarama.TopicDetail

	// If true then each consumed message will be wrapped with Opentelemetry tracing, provided by otelsarama.
	//
	// Deprecated: pass OTELSaramaTracer to Tracer field instead.
	OTELEnabled bool

	// Tracer is used to trace Kafka messages.
	// If nil, then no tracing will be used.
	Tracer SaramaTracer

	// ConsumerModel indicates which type of consumer should be used
	ConsumerModel ConsumerModel
	// When set to not nil, consumption will be performed in batches and this configuration will be used
	BatchConsumerConfig *BatchConsumerConfig
}

func (SubscriberConfig) Validate 

func (c SubscriberConfig) Validate() error

type Unmarshaler 

type Unmarshaler interface {
	Unmarshal(*sarama.ConsumerMessage) (*message.Message, error)
}

Unmarshaler unmarshals Kafka's message to Watermill's message.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.