kaf

README

Kaf

Kafka CLI inspired by kubectl & docker

Install

Install from source:

go get github.com/infinimesh/kaf/cmd/kaf

Install binary:

curl https://raw.githubusercontent.com/infinimesh/kaf/master/godownloader.sh | BINDIR=$HOME/bin bash

Configuration

See the examples folder

Documentation

Index

• Variables
• func ParseConfluentCloudConfig(path string) (username, password, broker string, err error)
• func TryFindCcloudConfigFile() (string, error)
• type Cluster
• type Config
  • func ReadConfig() (c Config, err error)
  • func (c *Config) ActiveCluster() *Cluster
  • func (c *Config) SetCurrentCluster(name string) error
  • func (c *Config) Write() error
• type PacketDecoder
  • func NewDecoder(raw []byte) PacketDecoder
• type SASL
• type SubscriptionInfo
  • func (s *SubscriptionInfo) Decode(pd PacketDecoder) (err error)
• type TLS
• type TaskID

Variables

var ErrInsufficientData = errors.New("kafka: insufficient data to decode packet, more bytes expected")

ErrInsufficientData is returned when decoding and the packet is truncated. This can be expected when requesting messages, since as an optimization the server is allowed to return a partial message at the end of the message set.

Functions

func ParseConfluentCloudConfig

func ParseConfluentCloudConfig(path string) (username, password, broker string, err error)

TODO return []string for brokers

func TryFindCcloudConfigFile

func TryFindCcloudConfigFile() (string, error)

Types

type Cluster

type Cluster struct {
	Name              string
	Brokers           []string `yaml:"brokers"`
	SASL              *SASL    `yaml:"SASL"`
	TLS               *TLS     `yaml:"TLS"`
	SecurityProtocol  string   `yaml:"security-protocol"`
	SchemaRegistryURL string   `yaml:"schema-registry-url"`
}

type Config

type Config struct {
	CurrentCluster string     `yaml:"current-cluster"`
	Clusters       []*Cluster `yaml:"clusters"`
}

func ReadConfig

func ReadConfig() (c Config, err error)

func (*Config) ActiveCluster

func (c *Config) ActiveCluster() *Cluster

func (*Config) SetCurrentCluster

func (c *Config) SetCurrentCluster(name string) error

func (*Config) Write

func (c *Config) Write() error

type PacketDecoder

type PacketDecoder interface {
	// contains filtered or unexported methods
}

func NewDecoder

func NewDecoder(raw []byte) PacketDecoder

type SASL

type SASL struct {
	Mechanism string
	Username  string
	Password  string
}

type SubscriptionInfo

type SubscriptionInfo struct {
	Version      int32
	UUID         []byte // 16-byte UUID
	ProcessID    string
	PrevTasks    []TaskID
	StandbyTasks []TaskID
	UserEndpoint string
}

func (*SubscriptionInfo) Decode

func (s *SubscriptionInfo) Decode(pd PacketDecoder) (err error)

Support version 1+2

type TLS

type TLS struct {
	Cafile   string
	Insecure bool
}

type TaskID

type TaskID struct {
	TopicGroupID int32
	Partition    int32
}