Milena

README

Milena

kafka probe tool

While using kafka as a prodcuer, it is a problem to know what if prodcuer pushs right message to it (At least kafka-manager doesn't show topic contents); Or when producer wants to know whether consumers can pull right msg from kafka , producers may has to write another test-consumer.

I wrote this toy tool just for proving to testers that my producer program works very well !

Once Milena started , she can watch your every new messgae of kafka topic confed in conf file and write it to a log file. And she also supports duration for offsets which means U can always restart Milena and don't need to worry about reconsume history messages.

download

go get github.com/JodeZer/Milena

compile

make build

config

config is yml file like this:

datadir : data

servers:
- name: cluster1
  brokers:
  - "192.168.1.x:9092"
  - "192.168.1.x:9092"
  - "192.168.1.x:9092"
  topics:
  - name: "xxx"
    partitions:
    - partition: 0
      start: 7
  - name: "xxxxxx"

run

the default config file is ${curdir}/Milena.yml

cd bin && ./Milena -f ${conf}.yml

stop

in the start cmd dir

./Milena -s stop

results

topic contents will append to a file which is name by

${datadir}/${clustername}/${topicName}.log

conetnt will like this:

ts=>[${timestamp}] p:${parition} o:${offset} =>${value}

there is a metadata dir in ${datadir}/${clustername}, never delete it unless lt is broken.It stored offsets already consumed.

thanks to

• sarama
• leveldb
• configor
• logrus (though my own fork actually)

TODO

• optimize log append engine(cur can't be called engine, it's just a working shit)
• testify and optimize stop mechanism
• gen more cmd to help
  • reload signal -s
  • repair command
• add travis.ci to help
• more powerful makefile
• fix shit code(always)

Documentation

Index

• type Config
  • func NewConfig(fileName string) *Config
• type Instance
  • func NewInsatnce(conf *Config) (*Instance, error)
  • func (i *Instance) Start()
  • func (i *Instance) Stop()

Types

type Config

type Config struct {
	//config file path
	FileName string

	//data files dir
	DataDir string

	//log level
	LogLevel string

	//lockDir
	LockFile string `yaml:"-"`

	//listen servers
	Servers []struct {
		//kafka cluster name
		Name string

		// storage consumer metadata format: {clusterName}/{topic}.meta
		MetaDataDir string `yaml:"-"`

		// Server address
		Brokers []string

		// listen all topics
		ListenAll bool

		// listen topic
		Topics []topicSetting
	}
}

Config Entity

func NewConfig

func NewConfig(fileName string) *Config

NewConfig

type Instance

type Instance struct {
	// contains filtered or unexported fields
}

Milena Instance

func NewInsatnce

func NewInsatnce(conf *Config) (*Instance, error)

NewInstances

func (*Instance) Start

func (i *Instance) Start()

StartInstance

func (*Instance) Stop

func (i *Instance) Stop()

StopInstance