README
¶
JOCKO
Kafka/distributed commit log service in Go.
Goals of this project:
- Implement Kafka in Go
- Protocol compatible with Kafka so Kafka clients and services work with Jocko
- Make operating simpler
- Distribute a single binary
- Use Serf for discovery, Raft for consensus (and remove the need to run ZooKeeper)
- Smarter configuration settings
- Able to use percentages of disk space for retention policies rather than only bytes and time kept
- Handling size configs when you change the number of partitions or add topics
- Learn a lot and have fun
TODO
- Producing
- Fetching
- Partition consensus and distribution
- Protocol
- Produce
- Fetch
- Metadata
- Create Topics
- Delete Topics
- Consumer group
- Discovery
- Replication [current task]
Reading
- How Jocko's built-in service discovery and consensus works
- How Jocko's (and Kafka's) storage internals work
License
MIT
- travisjeffery.com
- GitHub @travisjeffery
- Twitter @travisjeffery
- Medium @travisjeffery
Documentation
¶
Index ¶
- type Broker
- type ClusterMember
- type CommitLog
- type MemberStatus
- type Partition
- func (p *Partition) Append(ms []byte) (int64, error)
- func (p *Partition) Delete() error
- func (p *Partition) HighWatermark() int64
- func (r *Partition) IsFollowing(id int32) bool
- func (r *Partition) IsLeader(id int32) bool
- func (r *Partition) IsOpen() bool
- func (p *Partition) LeaderID() int32
- func (p *Partition) LowWatermark() int64
- func (p *Partition) NewReader(offset int64, maxBytes int32) (io.Reader, error)
- func (p *Partition) Read(b []byte) (int, error)
- func (r *Partition) String() string
- func (p *Partition) TruncateTo(offset int64) error
- func (p *Partition) Write(b []byte) (int, error)
- type Proxy
- type Raft
- type RaftCmdType
- type RaftCommand
- type Serf
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Broker ¶
type Broker interface {
ID() int32
IsController() bool
CreateTopic(topic string, partitions int32) error
StartReplica(*Partition) error
DeleteTopic(topic string) error
Partition(topic string, id int32) (*Partition, error)
ClusterMember(brokerID int32) *ClusterMember
BecomeLeader(topic string, id int32, command *protocol.PartitionState) error
BecomeFollower(topic string, id int32, command *protocol.PartitionState) error
Join(addr ...string) (int, error)
Cluster() []*ClusterMember
TopicPartitions(topic string) ([]*Partition, error)
IsLeaderOfPartition(topic string, id int32, leaderID int32) bool
}
Broker is the interface that wraps the Broker's methods.
type ClusterMember ¶
type ClusterMember struct {
ID int32 `json:"id"`
Port int `json:"port"`
IP string `json:"addr"`
SerfPort int `json:"-"`
RaftPort int `json:"-"`
Status MemberStatus `json:"-"`
// contains filtered or unexported fields
}
ClusterMember is used as a wrapper around a broker's info and a connection to it.
func (*ClusterMember) Addr ¶
func (b *ClusterMember) Addr() *net.TCPAddr
Addr is used to get the address of the member.
type CommitLog ¶
type CommitLog interface {
DeleteAll() error
NewReader(offset int64, maxBytes int32) (io.Reader, error)
TruncateTo(int64) error
NewestOffset() int64
OldestOffset() int64
Append([]byte) (int64, error)
}
CommitLog is the interface that wraps the commit log's methods and is used to manage a partition's data.
type MemberStatus ¶
type MemberStatus int
MemberStatus is the state that a member is in.
const ( StatusNone MemberStatus = iota StatusAlive StatusLeaving StatusLeft StatusFailed StatusReap )
Different possible states of serf member
type Partition ¶
type Partition struct {
Topic string `json:"topic"`
ID int32 `json:"id"`
Replicas []int32 `json:"replicas"`
ISR []int32 `json:"isr"`
Leader int32 `json:"leader"`
PreferredLeader int32 `json:"preferred_leader"`
LeaderAndISRVersionInZK int32 `json:"-"`
CommitLog CommitLog `json:"-"`
Conn io.ReadWriter `json:"-"`
}
Partition is the unit of storage in Jocko.
func NewPartition ¶
NewPartition is used to create a new partition.
func (*Partition) HighWatermark ¶
HighWatermark is used to get the newest offset of the partition.
func (*Partition) IsFollowing ¶
IsFollowing is used to check if the given broker ID's should follow/replicate the leader.
func (*Partition) IsLeader ¶
IsLeader is used to check if the given broker ID's the partition's leader.
func (*Partition) IsOpen ¶
IsOpen is used to check whether the partition's commit log has been initialized.
func (*Partition) LowWatermark ¶
LowWatermark is used to oldest offset of the partition.
func (*Partition) NewReader ¶
NewReader is used to create a reader at the given offset and will read up to maxBytes.
func (*Partition) Read ¶
Write is used to directly read the given bytes from the partition's leader.
func (*Partition) TruncateTo ¶
TruncateTo is used to truncate the partition's logs before the given offset.
type Proxy ¶
type Proxy interface {
FetchMessages(clientID string, fetchRequest *protocol.FetchRequest) (*protocol.FetchResponses, error)
CreateTopic(clientID string, createRequest *protocol.CreateTopicRequest) (*protocol.CreateTopicsResponse, error)
}
Proxy is the interface that wraps Proxy methods for forwarding requests to an existing Jocko server and returning server response to caller
type Raft ¶
type Raft interface {
Bootstrap(peers []*ClusterMember, commandCh chan<- RaftCommand, leaderCh chan<- bool) (err error)
Apply(cmd RaftCommand) error
IsLeader() bool
LeaderID() string
WaitForBarrier() error
AddPeer(addr string) error
RemovePeer(addr string) error
Shutdown() error
Addr() string
}
Raft is the interface that wraps Raft's methods and is used to manage consensus for the Jocko cluster.
type RaftCmdType ¶
type RaftCmdType int
type RaftCommand ¶
type RaftCommand struct {
Cmd RaftCmdType `json:"type"`
Data *json.RawMessage `json:"data"`
}
type Serf ¶
type Serf interface {
Bootstrap(node *ClusterMember, reconcileCh chan<- *ClusterMember) error
Cluster() []*ClusterMember
Member(memberID int32) *ClusterMember
Join(addrs ...string) (int, error)
Shutdown() error
Addr() string
}
Serf is the interface that wraps Serf methods and is used to manage the cluster membership for Jocko nodes.
Source Files
Directories