Why Go
- Case Studies
  
  Common problems companies solve with Go
- Use Cases
  
  Stories about how and why companies use Go
- Security Policy
  
  How Go can help keep you secure by default
Learn
Docs
- Effective Go
  
  Tips for writing clear, performant, and idiomatic Go code
- Go User Manual
  
  A complete introduction to building software with Go
- Standard library
  
  Reference documentation for Go's standard library
- Release Notes
  
  Learn what's new in each Go release
Packages
Community
- Recorded Talks
  
  Videos from prior events
- Meetups
  
  Meet other local Go developers
- Conferences
  
  Learn and network with Go developers from around the world
- Go blog
  
  The Go project's official blog.
- Go project
  
  Get help and stay informed from Go
- Get connected

kafka

package module

Go to main page

Versions in this module

v0

v0.9.1

Apr 7, 2022

Changes in this version

+ const Plain

+ const SHA256

+ const SHA512

+ var CompressionCodecs = map[string]compress.Codec

+ var ReaderBytes = stats.New("kafka.reader.message.bytes", stats.Counter, stats.Data)

+ var ReaderDialTime = stats.New("kafka.reader.dial.seconds", stats.Trend, stats.Time)

+ var ReaderDials = stats.New("kafka.reader.dial.count", stats.Counter)

+ var ReaderErrors = stats.New("kafka.reader.error.count", stats.Counter)

+ var ReaderFetchBytes = stats.New("kafka.reader.fetch.bytes", stats.Counter, stats.Data)

+ var ReaderFetchSize = stats.New("kafka.reader.fetch.size", stats.Counter)

+ var ReaderFetches = stats.New("kafka.reader.fetches.count", stats.Counter)

+ var ReaderLag = stats.New("kafka.reader.lag", stats.Gauge)

+ var ReaderMaxBytes = stats.New("kafka.reader.fetch_bytes.max", stats.Gauge)

+ var ReaderMaxWait = stats.New("kafka.reader.fetch_wait.max", stats.Gauge, stats.Time)

+ var ReaderMessages = stats.New("kafka.reader.message.count", stats.Counter)

+ var ReaderMinBytes = stats.New("kafka.reader.fetch_bytes.min", stats.Gauge)

+ var ReaderOffset = stats.New("kafka.reader.offset", stats.Gauge)

+ var ReaderQueueCapacity = stats.New("kafka.reader.queue.capacity", stats.Gauge)

+ var ReaderQueueLength = stats.New("kafka.reader.queue.length", stats.Gauge)

+ var ReaderReadTime = stats.New("kafka.reader.read.seconds", stats.Trend, stats.Time)

+ var ReaderRebalances = stats.New("kafka.reader.rebalance.count", stats.Counter)

+ var ReaderTimeouts = stats.New("kafka.reader.timeouts.count", stats.Counter)

+ var ReaderWaitTime = stats.New("kafka.reader.wait.seconds", stats.Trend, stats.Time)

+ var WriterAsync = stats.New("kafka.writer.async", stats.Rate)

+ var WriterBatchBytes = stats.New("kafka.writer.batch.bytes", stats.Counter, stats.Data)

+ var WriterBatchSize = stats.New("kafka.writer.batch.size", stats.Counter)

+ var WriterBatchTimeout = stats.New("kafka.writer.batch.timeout", stats.Gauge, stats.Time)

+ var WriterBytes = stats.New("kafka.writer.message.bytes", stats.Counter, stats.Data)

+ var WriterDialTime = stats.New("kafka.writer.dial.seconds", stats.Trend, stats.Time)

+ var WriterDials = stats.New("kafka.writer.dial.count", stats.Counter)

+ var WriterErrors = stats.New("kafka.writer.error.count", stats.Counter)

+ var WriterMaxAttempts = stats.New("kafka.writer.attempts.max", stats.Gauge)

+ var WriterMaxBatchSize = stats.New("kafka.writer.batch.max", stats.Gauge)

+ var WriterMessages = stats.New("kafka.writer.message.count", stats.Counter)

+ var WriterQueueCapacity = stats.New("kafka.writer.queue.capacity", stats.Gauge)

+ var WriterQueueLength = stats.New("kafka.writer.queue.length", stats.Gauge)

+ var WriterReadTimeout = stats.New("kafka.writer.read.timeout", stats.Gauge, stats.Time)

+ var WriterRebalanceInterval = stats.New("kafka.writer.rebalance.interval", stats.Gauge, stats.Time)

+ var WriterRebalances = stats.New("kafka.writer.rebalance.count", stats.Counter)

+ var WriterRequiredAcks = stats.New("kafka.writer.acks.required", stats.Gauge)

+ var WriterRetries = stats.New("kafka.writer.retries.count", stats.Counter)

+ var WriterWaitTime = stats.New("kafka.writer.wait.seconds", stats.Trend, stats.Time)

+ var WriterWriteTime = stats.New("kafka.writer.write.seconds", stats.Trend, stats.Time)

+ var WriterWriteTimeout = stats.New("kafka.writer.write.timeout", stats.Gauge, stats.Time)

+ var WriterWrites = stats.New("kafka.writer.write.count", stats.Counter)

+ func ConsumeInternal(ctx context.Context, reader *kafkago.Reader, limit int64, ...) []map[string]interface

+ func FromAvro(message []byte, schema string) interface

+ func ProduceInternal(ctx context.Context, writer *kafkago.Writer, messages []map[string]interface{}, ...) error

+ func ReportError(err error, msg string)

+ func ReportReaderStats(ctx context.Context, currentStats kafkago.ReaderStats) error

+ func ReportWriterStats(ctx context.Context, currentStats kafkago.WriterStats) error

+ func SerializeAvro(configuration Configuration, topic string, data interface{}, keyOrValue string, ...) ([]byte, error)

+ func SerializeByteArray(configuration Configuration, topic string, data interface{}, keyOrValue string, ...) ([]byte, error)

+ func SerializeString(configuration Configuration, topic string, data interface{}, keyOrValue string, ...) ([]byte, error)

+ func ToAvro(value string, schema string) []byte

+ type BasicAuth struct

+ CredentialsSource string

+ UserInfo string

+ type Configuration struct

+ Consumer ConsumerConfiguration

+ Producer ProducerConfiguration

+ SchemaRegistry SchemaRegistryConfiguration

+ type ConsumerConfiguration struct

+ KeyDeserializer string

+ ValueDeserializer string

+ type Credentials struct

+ Algorithm string

+ Password string

+ Username string

+ type Kafka struct

+ func (*Kafka) Consume(ctx context.Context, reader *kafkago.Reader, limit int64, keySchema string, ...) []map[string]interface{}

+ func (*Kafka) ConsumeWithConfiguration(ctx context.Context, reader *kafkago.Reader, limit int64, ...) []map[string]interface{}

+ func (*Kafka) CreateTopic(address, topic string, partitions, replicationFactor int, compression string) error

+ func (*Kafka) ListTopics(address string) ([]string, error)

+ func (*Kafka) Produce(ctx context.Context, writer *kafkago.Writer, messages []map[string]interface{}, ...) error

+ func (*Kafka) ProduceWithConfiguration(ctx context.Context, writer *kafkago.Writer, messages []map[string]interface{}, ...) error

+ func (*Kafka) Reader(brokers []string, topic string, partition int, groupID string, offset int64, ...) *kafkago.Reader

+ func (*Kafka) Writer(brokers []string, topic string, auth string, compression string) *kafkago.Writer

+ type ProducerConfiguration struct

+ KeySerializer string

+ ValueSerializer string

+ type SchemaRegistryConfiguration struct

+ BasicAuth BasicAuth

+ Url string

+ type Serializer func(configuration Configuration, topic string, data interface{}, keyOrValue string, ...) ([]byte, error)

+ func GetSerializer(serializer string) Serializer

v0.0.2

Jul 26, 2020

v0.0.1

Jul 24, 2020

JackTT - Gopher 🇻🇳