redelivery

type AssignedPartition ¶

type AssignedPartition interface {
	// AddMessage adds a marker message to this partition.
	// `now` is the local, monotonic clock time
	AddMessage(msg *confluentKafka.Message, now time.Time) error

	// SmallestMarkerOffset returns the smallest offset of all marker messages currently being tracked
	SmallestMarkerOffset() (int64, bool)

	// Redeliver determines which messages have expired redelivery deadlines and initiates redelivery.
	// `now` is the local, monotonic clock time
	Redeliver(now time.Time)

	// Close releases the resources consumed by this partition instance
	Close()
}

AssignedPartition allows the marker consumer to track consumption progress as well as which messages require redelivery for a specific partition in the marker topic

func NewAssignedPartition(markersQueue MarkersQueue, redeliverer Redeliverer, durationUseNowIfNoMarkerSeen time.Duration) AssignedPartition

type MarkerConsumer ¶

type MarkerConsumer interface {
	// Start the `MarkerConsumer`
	Start()

	// Stop the `MarkerConsumer`, it is unusable after it is stopped
	Stop()
}

MarkerConsumer encapsulates the ability to consume markers from the markers topic and redeliver messages redelivery deadlines have expired

func NewMarkerConsumer(config queue.Config, clients kafka.ClientFactory) (MarkerConsumer, error)

type MarkerDeadline ¶

type MarkerDeadline struct {
	MessageID          MessageID
	RedeliveryDeadline time.Time
	// contains filtered or unexported fields
}

MarkerDeadline associates a message with its redelivery deadline

type MarkerDeadlinesPQ ¶

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

MarkerDeadlinesPQ is a priority queue to track messages by their redelivery deadline in ascending order

func NewMarkerDeadlinesPQ() *MarkerDeadlinesPQ

func (pq *MarkerDeadlinesPQ) Dequeue() (MarkerDeadline, bool)

func (pq *MarkerDeadlinesPQ) Enqueue(markerDeadline MarkerDeadline) bool

func (pq *MarkerDeadlinesPQ) Head() (MarkerDeadline, bool)

func (pq *MarkerDeadlinesPQ) Update(messageID MessageID, redeliveryDeadline time.Time) bool

type MarkerID ¶

type MarkerID = RecordID

MarkerID uniquely identifies a record from the marker topic

func NewMarkerID(partition int32, offset int64) MarkerID

type MarkerOffset ¶

type MarkerOffset struct {
	MessageID      MessageID
	OffsetOfMarker int64
}

MarkerOffset associates a message with the location of the marker that tracks it in the maker topic

type MarkerOffsetsPQ ¶

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

MarkerOffsetsPQ is a priority queue to track markers by their offset in the marker topic in ascending order

func NewMarkerOffsetsPQ() *MarkerOffsetsPQ

func (pq *MarkerOffsetsPQ) Dequeue() (MarkerOffset, bool)

func (pq *MarkerOffsetsPQ) Enqueue(markerOffset MarkerOffset) bool

func (pq *MarkerOffsetsPQ) Head() (MarkerOffset, bool)

type MarkersQueue ¶

type MarkersQueue interface {
	// Partition returns the ordinal of the partition that the queue is tracking
	Partition() int32

	// AddMarker adds a marker to the queue.
	AddMarker(offset int64, timestamp time.Time, marker *kafkamq.Marker)

	// MarkersToRedelivery returns the markers that require redelivery given the specified timestamp.
	MarkersToRedeliver(now time.Time) []*kafkamq.Marker

	// SmallestMarkerOffset returns the smallest marker offset tracked by the queue.
	// Returns `(0, false)` if the queue is empty.
	SmallestMarkerOffset() (int64, bool)
}

MarkersQueue is the datastructure used to track Markers consumed from the markers topic that determines:

1. Which messages currently are in-progress, including their payload.
2. Which messages require redelivery (based on marker redelivery deadlines)
3. The offset up to which markers have been consumed from the markers topic; this offset is used when committing marker consumer offsets to Kafka.

Note that the MarkersQueue is not thread-safe as it is only meant to be used by a single Kafka consumer thread.

func NewMarkersQueue(partition int32) MarkersQueue

type MessageID ¶

type MessageID = RecordID

MessageID uniquely identifies a record from the queue topic

func MessageIDFromMarker(marker *kafkamq.Marker) MessageID

func NewMessageID(partition int32, offset int64) MessageID

type OffsetCommitter ¶

type OffsetCommitter interface {
	// AddMarker registers a marker to be committed
	AddMarker(MarkerID)

	// DropOffsets drops any cached offsets for the specified partition
	DropOffsets(partition int32)
}

OffsetCommitter allows the redelivery tracker to periodically commit batches of marker offsets to Kafka

func NewOffsetCommitter(markerTopic string, numOffsetsPerCommit uint, kConsumer kafka.Consumer) OffsetCommitter

type RecordID ¶

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

RecordID uniqely identifies a record in a kafka topic by using the record's location (partition:offset)

type Redeliverer ¶

type Redeliverer interface {
	Redeliver(markers []*kafkamq.Marker)
	Stop()
}

Redeliverer provides functionality to redeliver messages associated with markers whose deadlines have expired.

with the mechanics of message redelivery.

func NewRedeliverer(partition int32, messageSender internal.MessageSender, markerProducer internal.MarkerProducer, backoffPolicy backoff.BackOff) Redeliverer