info

Documentation

Overview

Implements a go routine that collects topic info periodically outside of the broker.

Index

• Constants
• type TopicInfoGatherer
  • func New(events store.EventStore, topics store.TopicInfoStore) *TopicInfoGatherer
  • func (t *TopicInfoGatherer) Gather(wg *sync.WaitGroup) error
  • func (t *TopicInfoGatherer) Run()
  • func (t *TopicInfoGatherer) Shutdown() error

Constants

const (
	// InfoInterval specifies the delay between topic info gathering runs. This interval
	// is approximate because the time it takes to perform an info gathering run is
	// added to the interval; the more topics in the system, the longer the info
	// gathering interval.
	InfoInterval = 75 * time.Second

	// InfoWorkers specifes the number of workers that loop through each topic. The goal
	// of parallelization is to ensure that a large topic does not dominate the info
	// gathering process, but to minimize the amount of CPU needed to perform topic info
	// gathering in favor of publisher and subscriber routines.
	InfoWorkers = 4
)

Types

type TopicInfoGatherer

type TopicInfoGatherer struct {
	sync.Mutex
	// contains filtered or unexported fields
}

TopicInfoGatherer runs a go routine that periodically lists all of the topics on the node and updates the topic info. This routine is designed to be outside of the broker process so that updating topic info does not slow down the evening process. It does mean that topic info may be behind the actual state of the server, but given a routine enough periodicity, should quickly be resolved with eventual consistency; e.g. in the absence of writes, the topic info will eventually become consistent.

NOTE: This routine works as a single thread and guarantees consistency in topic info -- no other go routine should write to the topic info, only read from it.

func New

func New(events store.EventStore, topics store.TopicInfoStore) *TopicInfoGatherer

func (*TopicInfoGatherer) Gather

func (t *TopicInfoGatherer) Gather(wg *sync.WaitGroup) error

Loops through all topics currently stored in the database and gathers topic info from the events, then saves that topic info back to disk. Any errors returned from this method are fatal; e.g. the gatherer cannot access the database, otherwise if there is a transient failure, then the error is logged.

func (*TopicInfoGatherer) Run

func (t *TopicInfoGatherer) Run()

Run the background go routine that collects topic info from each topic. NOTE: this should not be run in maintenance mode. WARNING: Do not call this method more than once per process!

func (*TopicInfoGatherer) Shutdown

func (t *TopicInfoGatherer) Shutdown() error

Shutdown the topic info gatherer; blocks until the topic info has completed. WARNING: Do not call this method more than once per process!