offsetmgr

Documentation

Index

• type Factory
  • func SpawnFactory(parentActDesc *actor.Descriptor, cfg *config.Proxy, kafkaClt sarama.Client) Factory
• type Offset
• type T

Types

type Factory

type Factory interface {
	// Span creates and starts an offset manager for a group-topic-partition.
	// It returns an error if given group-topic-partition has a running
	// OffsetManager instance already. After an old offset manager instance is
	// stopped a new one can be started.
	Spawn(parentActDesc *actor.Descriptor, group, topic string, partition int32) (T, error)

	// Stop waits for the spawned offset managers to stop and then terminates. Note
	// that all spawned offset managers has to be explicitly stopped by calling
	// their Stop method.
	Stop()
}

Factory provides a method to spawn offset manager instances to commit offsets for a particular group-topic-partition. It makes sure that there is only one running manager instance for a particular group-topic-partition combination.

One Factory instance per application is usually more then enough, but it is possible to create many of them.

Factory spawns background goroutines so it must be explicitly stopped by the application. But first it should explicitly stop all spawned offset manager instances.

func SpawnFactory

func SpawnFactory(parentActDesc *actor.Descriptor, cfg *config.Proxy, kafkaClt sarama.Client) Factory

SpawnFactory creates a new offset manager factory from the given client.

type Offset

type Offset struct {
	Val  int64
	Meta string
}

Offset represents an offset data as it is stored in Kafka, that is an offset value decorated with a metadata string.

type T

type T interface {
	// SubmitOffset triggers saving of the specified offset in Kafka. Commits are
	// performed periodically in a background goroutine. The commit interval is
	// configured by `Config.Consumer.Offsets.CommitInterval`. Note that not every
	// submitted offset gets committed. Committed offsets are sent down to the
	// `CommittedOffsets()` channel. The `CommittedOffsets()` channel has to be
	// read alongside with submitting offsets, otherwise the partition offset
	// manager will block.
	SubmitOffset(offset Offset)

	// CommittedOffsets returns a channel that offsets committed to Kafka are
	// sent to. The first offset sent to this channel is the initial offset
	// fetched from Kafka for the group-topic-partition. The user must read
	// from this channel otherwise the `SubmitOffset` function will eventually
	// block forever.
	CommittedOffsets() <-chan Offset

	// Stop stops the offset manager. It is required to stop all spawned offset
	// managers before their parent factory can be stopped.
	//
	// It is guaranteed that the most recent offset is committed before `Stop`
	// returns.
	Stop()
}

T provides interface to store and retrieve offsets for a particular group-topic-partition in Kafka.