kafka

package

v0.0.0-...-5cd769b Latest Latest Go to latest Published: Mar 27, 2024 License: MIT Imports: 16 Imported by: 4

Details

Valid go.mod file

The Go module system was introduced in Go 1.11 and is the official dependency management solution for Go.
Redistributable license

Redistributable licenses place minimal restrictions on how software can be used, modified, and redistributed.
Tagged version

Modules with tagged versions give importers more predictable builds.
Stable version

When a project reaches major version v1 it is considered stable.
Learn more about best practices

Repository

github.com/uw-labs/substrate

Links

Open Source Insights

Documentation ¶

Overview ¶

Package kafka provides kafka support for substrate

Usage ¶

This package support two methods of use. The first is to directly use this package. See the function documentation for more details.

The second method is to use the suburl package. See https://godoc.org/github.com/uw-labs/substrate/suburl for more information.

Using suburl ¶

The url structure is kafka://host:port/topic/

The following url parameters are available:

broker - Specifies additional broker addresses in the form host%3Aport (where %3A is a url encoded ':')
version - Specifies the version of the broker

Additionally, for sources, the following url parameters are available

offset           - The initial offset. Valid values are `newest` and `oldest`.
consumer-group   - The consumer group id
metadata-refresh - How frequently to refresh the cluster metadata. E.g., '10s' '2m'

Additionally, for sinks, the following url parameters are available

debug               - Boolean indicating if debug logs should be written.
max-message-bytes   - The maximum size in bytes for the produced messages.

Index ¶

Constants
func NewAsyncMessageSink(config AsyncMessageSinkConfig) (substrate.AsyncMessageSink, error)
func NewAsyncMessageSource(c AsyncMessageSourceConfig) (substrate.AsyncMessageSource, error)
type AsyncMessageSinkConfig
type AsyncMessageSourceConfig

Constants ¶

View Source

const (
	// OffsetOldest indicates the oldest appropriate message available on the broker.
	OffsetOldest int64 = sarama.OffsetOldest
	// OffsetNewest indicates the next appropriate message available on the broker.
	OffsetNewest int64 = sarama.OffsetNewest
)

Variables ¶

This section is empty.

Functions ¶

func NewAsyncMessageSink ¶

func NewAsyncMessageSink(config AsyncMessageSinkConfig) (substrate.AsyncMessageSink, error)

func NewAsyncMessageSource ¶

func NewAsyncMessageSource(c AsyncMessageSourceConfig) (substrate.AsyncMessageSource, error)

Types ¶

type AsyncMessageSinkConfig ¶

type AsyncMessageSinkConfig struct {
	Brokers         []string
	Topic           string
	MaxMessageBytes int
	KeyFunc         func(substrate.Message) []byte
	Version         string

	Debug bool
}

type AsyncMessageSourceConfig ¶

type AsyncMessageSourceConfig struct {
	ConsumerGroup            string
	Topic                    string
	Brokers                  []string
	Offset                   int64
	MetadataRefreshFrequency time.Duration
	OffsetsRetention         time.Duration
	SessionTimeout           time.Duration
	RebalanceTimeout         time.Duration
	Version                  string

	Debug bool
}

AsyncMessageSource represents a kafka message source and implements the substrate.AsyncMessageSource interface.

Source Files ¶

View all Source files

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL