WIP: add cluster mode using Raft and Serf

2025-02-23 18:13:12 -06:00
parent 6a14cfaa08
commit b5b0eb3396
27 changed files with 1664 additions and 233 deletions
@@ -11,7 +11,7 @@ I am running `go1.23.4` for my local testing.
 1. Start the server locally on port 9092:

    ```go
-    go run main.go
+    go run main.go --bootstrap --node-id 1

    Server is listening on port 9092...
    ```
@@ -73,6 +73,20 @@ I am running `go1.23.4` for my local testing.

    ```

+# Starting a MonKafka Cluster
+Cluster mode operates on a distributed state using the Raft protocol, with membership management monitored by HashiCorp Serf. This combinations is used by the great Consul and Jocko.
+
+
+```shell
+# The first node bootstraps (using `--bootstrap`) the Raft cluster. This needs to be done only once.
+go run main.go --bootstrap --node-id 1 --serf-addr 127.0.0.1:3331
+
+# Subsequent nodes join the cluster via Serf by specifying an existing Serf endpoint with `--serf-join`
+go run main.go --node-id 2 --broker-port 9093 --raft-addr localhost:2222 --serf-addr 127.0.0.1:3332 --serf-join "127.0.0.1:3331"
+
+go run main.go --node-id 3 --broker-port 9094 --raft-addr localhost:2223 --serf-addr 127.0.0.1:3333 --serf-join "127.0.0.1:3331"
+```
+
 # Running tests
 Simple end-to-end tests have been implemented to verify topic creation and basic producer and consumer functionality.

@@ -88,6 +102,7 @@ go test -v  ./...
 - [X] Multiple log segments per partition
 - [X] Parse requests properly
 - [X] Compression
+- [ ] Cluster Mode / multiple brokers
 - [ ] Improve Consumer offsets management
 - [ ] SSL / ACls 
 - [ ] Configurability 
@@ -95,7 +110,6 @@ go test -v  ./...
 - [ ] Better concurrency
 - [ ] Error handling (e.g. topic not found)
 - [ ] Transactions
- [ ] Distributed system / multiple brokers


 # Credits
@@ -1,99 +0,0 @@
-package broker
-
-import (
-	"fmt"
-	"io"
-	"net"
-	"os"
-
-	log "github.com/CefBoud/monkafka/logging"
-
-	"github.com/CefBoud/monkafka/protocol"
-	"github.com/CefBoud/monkafka/serde"
-	"github.com/CefBoud/monkafka/storage"
-	"github.com/CefBoud/monkafka/types"
-)
-
-// Broker represents a Kafka broker instance
-type Broker struct {
-	Config         types.Configuration
-	ShutDownSignal chan bool
-}
-
-// NewBroker creates a new Broker instance with the provided configuration
-func NewBroker(config types.Configuration) *Broker {
-	return &Broker{Config: config, ShutDownSignal: make(chan bool)}
-}
-
-// Startup initializes the broker, starts the storage, loads group metadata,
-// and listens for incoming connections
-func (b Broker) Startup() {
-	storage.Startup(b.Config, b.ShutDownSignal)
-	protocol.LoadGroupMetadataState()
-
-	// Set up a TCP listener on the specified port
-	listener, err := net.Listen("tcp", fmt.Sprintf(":%d", b.Config.BrokerPort))
-	if err != nil {
-		log.Error("Error starting server: %v\n", err)
-		os.Exit(1)
-	}
-	defer listener.Close()
-
-	log.Info("Server is listening on port %d...\n", b.Config.BrokerPort)
-
-	// Accept and handle incoming connections
-	for {
-		conn, err := listener.Accept()
-		if err != nil {
-			log.Error("Error accepting connection: %v\n", err)
-			continue
-		}
-		go b.HandleConnection(conn)
-	}
-}
-
-// HandleConnection processes incoming requests from a client connection
-func (b Broker) HandleConnection(conn net.Conn) {
-	defer conn.Close()
-	connectionAddr := conn.RemoteAddr().String()
-	log.Debug("Connection established with %s\n", connectionAddr)
-
-	for {
-		// First, we read the length, then allocate a byte slice based on it.
-		// ReadFull (not Read) is used to ensure the entire request is read. Partial data would result in parsing errors
-		lengthBuffer := make([]byte, 4)
-		_, err := io.ReadFull(conn, lengthBuffer)
-		if err != nil {
-			log.Info("failed to read request's length. Error: %v ", err)
-			return
-		}
-		length := serde.Encoding.Uint32(lengthBuffer)
-		buffer := make([]byte, length+4)
-		copy(buffer, lengthBuffer)
-		_, err = io.ReadFull(conn, buffer[4:])
-		if err != nil {
-			if err.Error() != "EOF" {
-				log.Error("Error reading from connection: %v\n", err)
-			}
-			break
-		}
-		req := serde.ParseHeader(buffer, connectionAddr)
-
-		log.Debug("Received RequestAPIKey: %v | RequestAPIVersion: %v | CorrelationID: %v | Length: %v \n\n", protocol.APIDispatcher[req.RequestAPIKey].Name, req.RequestAPIVersion, req.CorrelationID, length)
-		response := protocol.APIDispatcher[req.RequestAPIKey].Handler(req)
-
-		_, err = conn.Write(response)
-		if err != nil {
-			log.Error("Error writing to connection: %v\n", err)
-			break
-		}
-	}
-	log.Debug("Connection with %s closed.\n", connectionAddr)
-}
-
-// Shutdown gracefully shuts down the broker and its components
-func (b Broker) Shutdown() {
-	log.Info("Broker Shutdown...")
-	close(b.ShutDownSignal)
-	storage.Shutdown()
-}
@@ -3,8 +3,36 @@ module github.com/CefBoud/monkafka
 go 1.23.4

 require (
+	github.com/armon/go-metrics v0.4.1 // indirect
+	github.com/boltdb/bolt v1.3.1 // indirect
 	github.com/eapache/go-xerial-snappy v0.0.0-20230731223053-c322873962e3 // indirect
+	github.com/fatih/color v1.13.0 // indirect
 	github.com/golang/snappy v0.0.4 // indirect
+	github.com/google/btree v1.1.2 // indirect
+	github.com/hashicorp/errwrap v1.1.0 // indirect
+	github.com/hashicorp/go-hclog v1.6.2 // indirect
+	github.com/hashicorp/go-immutable-radix v1.3.1 // indirect
+	github.com/hashicorp/go-metrics v0.5.4 // indirect
+	github.com/hashicorp/go-msgpack v0.5.5 // indirect
+	github.com/hashicorp/go-msgpack/v2 v2.1.2 // indirect
+	github.com/hashicorp/go-multierror v1.1.1 // indirect
+	github.com/hashicorp/go-sockaddr v1.0.5 // indirect
+	github.com/hashicorp/golang-lru v1.0.2 // indirect
+	github.com/hashicorp/memberlist v0.5.2 // indirect
+	github.com/hashicorp/raft v1.7.2 // indirect
+	github.com/hashicorp/raft-boltdb v0.0.0-20250113192317-e8660f88bcc9 // indirect
+	github.com/hashicorp/serf v0.10.2 // indirect
+	github.com/inconshreveable/mousetrap v1.1.0 // indirect
 	github.com/klauspost/compress v1.17.11 // indirect
+	github.com/mattn/go-colorable v0.1.12 // indirect
+	github.com/mattn/go-isatty v0.0.14 // indirect
+	github.com/miekg/dns v1.1.56 // indirect
 	github.com/pierrec/lz4/v4 v4.1.22 // indirect
+	github.com/sean-/seed v0.0.0-20170313163322-e2103e2c3529 // indirect
+	github.com/spf13/cobra v1.8.1 // indirect
+	github.com/spf13/pflag v1.0.6 // indirect
+	golang.org/x/mod v0.13.0 // indirect
+	golang.org/x/net v0.17.0 // indirect
+	golang.org/x/sys v0.15.0 // indirect
+	golang.org/x/tools v0.14.0 // indirect
 )
@@ -1,8 +1,230 @@
+cloud.google.com/go v0.34.0/go.mod h1:aQUYkXzVsufM+DwF1aE+0xfcU+56JwCaLick0ClmMTw=
+github.com/DataDog/datadog-go v2.2.0+incompatible/go.mod h1:LButxg5PwREeZtORoXG3tL4fMGNddJ+vMq1mwgfaqoQ=
+github.com/DataDog/datadog-go v3.2.0+incompatible/go.mod h1:LButxg5PwREeZtORoXG3tL4fMGNddJ+vMq1mwgfaqoQ=
+github.com/alecthomas/template v0.0.0-20160405071501-a0175ee3bccc/go.mod h1:LOuyumcjzFXgccqObfd/Ljyb9UuFJ6TxHnclSeseNhc=
+github.com/alecthomas/template v0.0.0-20190718012654-fb15b899a751/go.mod h1:LOuyumcjzFXgccqObfd/Ljyb9UuFJ6TxHnclSeseNhc=
+github.com/alecthomas/units v0.0.0-20151022065526-2efee857e7cf/go.mod h1:ybxpYRFXyAe+OPACYpWeL0wqObRcbAqCMya13uyzqw0=
+github.com/alecthomas/units v0.0.0-20190717042225-c3de453c63f4/go.mod h1:ybxpYRFXyAe+OPACYpWeL0wqObRcbAqCMya13uyzqw0=
+github.com/alecthomas/units v0.0.0-20190924025748-f65c72e2690d/go.mod h1:rBZYJk541a8SKzHPHnH3zbiI+7dagKZ0cgpgrD7Fyho=
+github.com/armon/go-metrics v0.0.0-20190430140413-ec5e00d3c878/go.mod h1:3AMJUQhVx52RsWOnlkpikZr01T/yAVN2gn0861vByNg=
+github.com/armon/go-metrics v0.4.1 h1:hR91U9KYmb6bLBYLQjyM+3j+rcd/UhE+G78SFnF8gJA=
+github.com/armon/go-metrics v0.4.1/go.mod h1:E6amYzXo6aW1tqzoZGT755KkbgrJsSdpwZ+3JqfkOG4=
+github.com/beorn7/perks v0.0.0-20180321164747-3a771d992973/go.mod h1:Dwedo/Wpr24TaqPxmxbtue+5NUziq4I4S80YR8gNf3Q=
+github.com/beorn7/perks v1.0.0/go.mod h1:KWe93zE9D1o94FZ5RNwFwVgaQK1VOXiVxmqh+CedLV8=
+github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw=
+github.com/boltdb/bolt v1.3.1 h1:JQmyP4ZBrce+ZQu0dY660FMfatumYDLun9hBCUVIkF4=
+github.com/boltdb/bolt v1.3.1/go.mod h1:clJnj/oiGkjum5o1McbSZDSLxVThjynRyGBgiAx27Ps=
+github.com/cespare/xxhash/v2 v2.1.1/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
+github.com/circonus-labs/circonus-gometrics v2.3.1+incompatible/go.mod h1:nmEj6Dob7S7YxXgwXpfOuvO54S+tGdZdw9fuRZt25Ag=
+github.com/circonus-labs/circonusllhist v0.1.3/go.mod h1:kMXHVDlOchFAehlya5ePtbp5jckzBHf4XRpQvBOLI+I=
+github.com/cpuguy83/go-md2man/v2 v2.0.4/go.mod h1:tgQtvFlXSQOSOSIRvRPT7W67SCa46tRHOmNcaadrF8o=
+github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/eapache/go-xerial-snappy v0.0.0-20230731223053-c322873962e3 h1:Oy0F4ALJ04o5Qqpdz8XLIpNA3WM/iSIXqxtqo7UGVws=
 github.com/eapache/go-xerial-snappy v0.0.0-20230731223053-c322873962e3/go.mod h1:YvSRo5mw33fLEx1+DlK6L2VV43tJt5Eyel9n9XBcR+0=
+github.com/fatih/color v1.13.0 h1:8LOYc1KYPPmyKMuN8QV2DNRWNbLo6LZ0iLs8+mlH53w=
+github.com/fatih/color v1.13.0/go.mod h1:kLAiJbzzSOZDVNGyDpeOxJ47H46qBXwg5ILebYFFOfk=
+github.com/go-kit/kit v0.8.0/go.mod h1:xBxKIO96dXMWWy0MnWVtmwkA9/13aqxPnvrjFYMA2as=
+github.com/go-kit/kit v0.9.0/go.mod h1:xBxKIO96dXMWWy0MnWVtmwkA9/13aqxPnvrjFYMA2as=
+github.com/go-kit/log v0.1.0/go.mod h1:zbhenjAZHb184qTLMA9ZjW7ThYL0H2mk7Q6pNt4vbaY=
+github.com/go-logfmt/logfmt v0.3.0/go.mod h1:Qt1PoO58o5twSAckw1HlFXLmHsOX5/0LbT9GBnD5lWE=
+github.com/go-logfmt/logfmt v0.4.0/go.mod h1:3RMwSq7FuexP4Kalkev3ejPJsZTpXXBr9+V4qmtdjCk=
+github.com/go-logfmt/logfmt v0.5.0/go.mod h1:wCYkCAKZfumFQihp8CzCvQ3paCTfi41vtzG1KdI/P7A=
+github.com/go-stack/stack v1.8.0/go.mod h1:v0f6uXyyMGvRgIKkXu+yp6POWl0qKG85gN/melR3HDY=
+github.com/gogo/protobuf v1.1.1/go.mod h1:r8qH/GZQm5c6nD/R0oafs1akxWv10x8SbQlK7atdtwQ=
+github.com/golang/protobuf v1.2.0/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
+github.com/golang/protobuf v1.3.1/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
+github.com/golang/protobuf v1.3.2/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
+github.com/golang/protobuf v1.4.0-rc.1/go.mod h1:ceaxUfeHdC40wWswd/P6IGgMaK3YpKi5j83Wpe3EHw8=
+github.com/golang/protobuf v1.4.0-rc.1.0.20200221234624-67d41d38c208/go.mod h1:xKAWHe0F5eneWXFV3EuXVDTCmh+JuBKY0li0aMyXATA=
+github.com/golang/protobuf v1.4.0-rc.2/go.mod h1:LlEzMj4AhA7rCAGe4KMBDvJI+AwstrUpVNzEA03Pprs=
+github.com/golang/protobuf v1.4.0-rc.4.0.20200313231945-b860323f09d0/go.mod h1:WU3c8KckQ9AFe+yFwt9sWVRKCVIyN9cPHBJSNnbL67w=
+github.com/golang/protobuf v1.4.0/go.mod h1:jodUvKwWbYaEsadDk5Fwe5c77LiNKVO9IDvqG2KuDX0=
+github.com/golang/protobuf v1.4.2/go.mod h1:oDoupMAO8OvCJWAcko0GGGIgR6R6ocIYbsSw735rRwI=
+github.com/golang/protobuf v1.4.3/go.mod h1:oDoupMAO8OvCJWAcko0GGGIgR6R6ocIYbsSw735rRwI=
 github.com/golang/snappy v0.0.4 h1:yAGX7huGHXlcLOEtBnF4w7FQwA26wojNCwOYAEhLjQM=
 github.com/golang/snappy v0.0.4/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q=
+github.com/google/btree v1.1.2 h1:xf4v41cLI2Z6FxbKm+8Bu+m8ifhj15JuZ9sa0jZCMUU=
+github.com/google/btree v1.1.2/go.mod h1:qOPhT0dTNdNzV6Z/lhRX0YXUafgPLFUh+gZMl761Gm4=
+github.com/google/go-cmp v0.3.0/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMywk6iLU=
+github.com/google/go-cmp v0.3.1/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMywk6iLU=
+github.com/google/go-cmp v0.4.0/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
+github.com/google/go-cmp v0.5.4/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
+github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
+github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg=
+github.com/hashicorp/errwrap v1.0.0/go.mod h1:YH+1FKiLXxHSkmPseP+kNlulaMuP3n2brvKWEqk/Jc4=
+github.com/hashicorp/errwrap v1.1.0 h1:OxrOeh75EUXMY8TBjag2fzXGZ40LB6IKw45YeGUDY2I=
+github.com/hashicorp/errwrap v1.1.0/go.mod h1:YH+1FKiLXxHSkmPseP+kNlulaMuP3n2brvKWEqk/Jc4=
+github.com/hashicorp/go-cleanhttp v0.5.0/go.mod h1:JpRdi6/HCYpAwUzNwuwqhbovhLtngrth3wmdIIUrZ80=
+github.com/hashicorp/go-hclog v0.9.1/go.mod h1:5CU+agLiy3J7N7QjHK5d05KxGsuXiQLrjA0H7acj2lQ=
+github.com/hashicorp/go-hclog v1.6.2 h1:NOtoftovWkDheyUM/8JW3QMiXyxJK3uHRK7wV04nD2I=
+github.com/hashicorp/go-hclog v1.6.2/go.mod h1:W4Qnvbt70Wk/zYJryRzDRU/4r0kIg0PVHBcfoyhpF5M=
+github.com/hashicorp/go-immutable-radix v1.0.0 h1:AKDB1HM5PWEA7i4nhcpwOrO2byshxBjXVn/J/3+z5/0=
+github.com/hashicorp/go-immutable-radix v1.0.0/go.mod h1:0y9vanUI8NX6FsYoO3zeMjhV/C5i9g4Q3DwcSNZ4P60=
+github.com/hashicorp/go-immutable-radix v1.3.1 h1:DKHmCUm2hRBK510BaiZlwvpD40f8bJFeZnpfm2KLowc=
+github.com/hashicorp/go-immutable-radix v1.3.1/go.mod h1:0y9vanUI8NX6FsYoO3zeMjhV/C5i9g4Q3DwcSNZ4P60=
+github.com/hashicorp/go-metrics v0.5.4 h1:8mmPiIJkTPPEbAiV97IxdAGNdRdaWwVap1BU6elejKY=
+github.com/hashicorp/go-metrics v0.5.4/go.mod h1:CG5yz4NZ/AI/aQt9Ucm/vdBnbh7fvmv4lxZ350i+QQI=
+github.com/hashicorp/go-msgpack v0.5.5 h1:i9R9JSrqIz0QVLz3sz+i3YJdT7TTSLcfLLzJi9aZTuI=
+github.com/hashicorp/go-msgpack v0.5.5/go.mod h1:ahLV/dePpqEmjfWmKiqvPkv/twdG7iPBM1vqhUKIvfM=
+github.com/hashicorp/go-msgpack/v2 v2.1.2 h1:4Ee8FTp834e+ewB71RDrQ0VKpyFdrKOjvYtnQ/ltVj0=
+github.com/hashicorp/go-msgpack/v2 v2.1.2/go.mod h1:upybraOAblm4S7rx0+jeNy+CWWhzywQsSRV5033mMu4=
+github.com/hashicorp/go-multierror v1.1.1 h1:H5DkEtf6CXdFp0N0Em5UCwQpXMWke8IA0+lD48awMYo=
+github.com/hashicorp/go-multierror v1.1.1/go.mod h1:iw975J/qwKPdAO1clOe2L8331t/9/fmwbPZ6JB6eMoM=
+github.com/hashicorp/go-retryablehttp v0.5.3/go.mod h1:9B5zBasrRhHXnJnui7y6sL7es7NDiJgTc6Er0maI1Xs=
+github.com/hashicorp/go-sockaddr v1.0.5 h1:dvk7TIXCZpmfOlM+9mlcrWmWjw/wlKT+VDq2wMvfPJU=
+github.com/hashicorp/go-sockaddr v1.0.5/go.mod h1:uoUUmtwU7n9Dv3O4SNLeFvg0SxQ3lyjsj6+CCykpaxI=
+github.com/hashicorp/go-uuid v1.0.0/go.mod h1:6SBZvOh/SIDV7/2o3Jml5SYk/TvGqwFJ/bN7x4byOro=
+github.com/hashicorp/golang-lru v0.5.0 h1:CL2msUPvZTLb5O648aiLNJw3hnBxN2+1Jq8rCOH9wdo=
+github.com/hashicorp/golang-lru v0.5.0/go.mod h1:/m3WP610KZHVQ1SGc6re/UDhFvYD7pJ4Ao+sR/qLZy8=
+github.com/hashicorp/golang-lru v1.0.2 h1:dV3g9Z/unq5DpblPpw+Oqcv4dU/1omnb4Ok8iPY6p1c=
+github.com/hashicorp/golang-lru v1.0.2/go.mod h1:iADmTwqILo4mZ8BN3D2Q6+9jd8WM5uGBxy+E8yxSoD4=
+github.com/hashicorp/memberlist v0.5.2 h1:rJoNPWZ0juJBgqn48gjy59K5H4rNgvUoM1kUD7bXiuI=
+github.com/hashicorp/memberlist v0.5.2/go.mod h1:Ri9p/tRShbjYnpNf4FFPXG7wxEGY4Nrcn6E7jrVa//4=
+github.com/hashicorp/raft v1.1.0/go.mod h1:4Ak7FSPnuvmb0GV6vgIAJ4vYT4bek9bb6Q+7HVbyzqM=
+github.com/hashicorp/raft v1.7.2 h1:pyvxhfJ4R8VIAlHKvLoKQWElZspsCVT6YWuxVxsPAgc=
+github.com/hashicorp/raft v1.7.2/go.mod h1:DfvCGFxpAUPE0L4Uc8JLlTPtc3GzSbdH0MTJCLgnmJQ=
+github.com/hashicorp/raft-boltdb v0.0.0-20250113192317-e8660f88bcc9 h1:DtRY4x+oreq0BTrrfF66XeCg6DPJuR2AL4Ejeipau/A=
+github.com/hashicorp/raft-boltdb v0.0.0-20250113192317-e8660f88bcc9/go.mod h1:FLQZr+lEOtW/5JZQCqRihQOrmyqWRqpJ+pP1gjb8XTE=
+github.com/hashicorp/serf v0.10.2 h1:m5IORhuNSjaxeljg5DeQVDlQyVkhRIjJDimbkCa8aAc=
+github.com/hashicorp/serf v0.10.2/go.mod h1:T1CmSGfSeGfnfNy/w0odXQUR1rfECGd2Qdsp84DjOiY=
+github.com/inconshreveable/mousetrap v1.1.0 h1:wN+x4NVGpMsO7ErUn/mUI3vEoE6Jt13X2s0bqwp9tc8=
+github.com/inconshreveable/mousetrap v1.1.0/go.mod h1:vpF70FUmC8bwa3OWnCshd2FqLfsEA9PFc4w1p2J65bw=
+github.com/jpillora/backoff v1.0.0/go.mod h1:J/6gKK9jxlEcS3zixgDgUAsiuZ7yrSoa/FX5e0EB2j4=
+github.com/json-iterator/go v1.1.6/go.mod h1:+SdeFBvtyEkXs7REEP0seUULqWtbJapLOCVDaaPEHmU=
+github.com/json-iterator/go v1.1.9/go.mod h1:KdQUCv79m/52Kvf8AW2vK1V8akMuk1QjK/uOdHXbAo4=
+github.com/json-iterator/go v1.1.10/go.mod h1:KdQUCv79m/52Kvf8AW2vK1V8akMuk1QjK/uOdHXbAo4=
+github.com/json-iterator/go v1.1.11/go.mod h1:KdQUCv79m/52Kvf8AW2vK1V8akMuk1QjK/uOdHXbAo4=
+github.com/julienschmidt/httprouter v1.2.0/go.mod h1:SYymIcj16QtmaHHD7aYtjjsJG7VTCxuUUipMqKk8s4w=
+github.com/julienschmidt/httprouter v1.3.0/go.mod h1:JR6WtHb+2LUe8TCKY3cZOxFyyO8IZAc4RVcycCCAKdM=
 github.com/klauspost/compress v1.17.11 h1:In6xLpyWOi1+C7tXUUWv2ot1QvBjxevKAaI6IXrJmUc=
 github.com/klauspost/compress v1.17.11/go.mod h1:pMDklpSncoRMuLFrf1W9Ss9KT+0rH90U12bZKk7uwG0=
+github.com/konsorten/go-windows-terminal-sequences v1.0.1/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ=
+github.com/konsorten/go-windows-terminal-sequences v1.0.3/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ=
+github.com/kr/logfmt v0.0.0-20140226030751-b84e30acd515/go.mod h1:+0opPa2QZZtGFBFZlji/RkVcI2GknAs/DXo4wKdlNEc=
+github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo=
+github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
+github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
+github.com/mattn/go-colorable v0.1.9/go.mod h1:u6P/XSegPjTcexA+o6vUJrdnUu04hMope9wVRipJSqc=
+github.com/mattn/go-colorable v0.1.12 h1:jF+Du6AlPIjs2BiUiQlKOX0rt3SujHxPnksPKZbaA40=
+github.com/mattn/go-colorable v0.1.12/go.mod h1:u5H1YNBxpqRaxsYJYSkiCWKzEfiAb1Gb520KVy5xxl4=
+github.com/mattn/go-isatty v0.0.12/go.mod h1:cbi8OIDigv2wuxKPP5vlRcQ1OAZbq2CE4Kysco4FUpU=
+github.com/mattn/go-isatty v0.0.14 h1:yVuAays6BHfxijgZPzw+3Zlu5yQgKGP2/hcQbHb7S9Y=
+github.com/mattn/go-isatty v0.0.14/go.mod h1:7GGIvUiUoEMVVmxf/4nioHXj79iQHKdU27kJ6hsGG94=
+github.com/matttproud/golang_protobuf_extensions v1.0.1/go.mod h1:D8He9yQNgCq6Z5Ld7szi9bcBfOoFv/3dc6xSMkL2PC0=
+github.com/miekg/dns v1.1.56 h1:5imZaSeoRNvpM9SzWNhEcP9QliKiz20/dA2QabIGVnE=
+github.com/miekg/dns v1.1.56/go.mod h1:cRm6Oo2C8TY9ZS/TqsSrseAcncm74lfK5G+ikN2SWWY=
+github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q=
+github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q=
+github.com/modern-go/reflect2 v0.0.0-20180701023420-4b7aa43c6742/go.mod h1:bx2lNnkwVCuqBIxFjflWJWanXIb3RllmbCylyMrvgv0=
+github.com/modern-go/reflect2 v1.0.1/go.mod h1:bx2lNnkwVCuqBIxFjflWJWanXIb3RllmbCylyMrvgv0=
+github.com/mwitkow/go-conntrack v0.0.0-20161129095857-cc309e4a2223/go.mod h1:qRWi+5nqEBWmkhHvq77mSJWrCKwh8bxhgT7d/eI7P4U=
+github.com/mwitkow/go-conntrack v0.0.0-20190716064945-2f068394615f/go.mod h1:qRWi+5nqEBWmkhHvq77mSJWrCKwh8bxhgT7d/eI7P4U=
+github.com/pascaldekloe/goe v0.1.0/go.mod h1:lzWF7FIEvWOWxwDKqyGYQf6ZUaNfKdP144TG7ZOy1lc=
 github.com/pierrec/lz4/v4 v4.1.22 h1:cKFw6uJDK+/gfw5BcDL0JL5aBsAFdsIT18eRtLj7VIU=
 github.com/pierrec/lz4/v4 v4.1.22/go.mod h1:gZWDp/Ze/IJXGXf23ltt2EXimqmTUXEy0GFuRQyBid4=
+github.com/pkg/errors v0.8.0/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
+github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
+github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
+github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/prometheus/client_golang v0.9.1/go.mod h1:7SWBe2y4D6OKWSNQJUaRYU/AaXPKyh/dDVn+NZz0KFw=
+github.com/prometheus/client_golang v0.9.2/go.mod h1:OsXs2jCmiKlQ1lTBmv21f2mNfw4xf/QclQDMrYNZzcM=
+github.com/prometheus/client_golang v1.0.0/go.mod h1:db9x61etRT2tGnBNRi70OPL5FsnadC4Ky3P0J6CfImo=
+github.com/prometheus/client_golang v1.4.0/go.mod h1:e9GMxYsXl05ICDXkRhurwBS4Q3OK1iX/F2sw+iXX5zU=
+github.com/prometheus/client_golang v1.7.1/go.mod h1:PY5Wy2awLA44sXw4AOSfFBetzPP4j5+D6mVACh+pe2M=
+github.com/prometheus/client_golang v1.11.1/go.mod h1:Z6t4BnS23TR94PD6BsDNk8yVqroYurpAkEiz0P2BEV0=
+github.com/prometheus/client_model v0.0.0-20180712105110-5c3871d89910/go.mod h1:MbSGuTsp3dbXC40dX6PRTWyKYBIrTGTE9sqQNg2J8bo=
+github.com/prometheus/client_model v0.0.0-20190129233127-fd36f4220a90/go.mod h1:xMI15A0UPsDsEKsMN9yxemIoYk6Tm2C1GtYGdfGttqA=
+github.com/prometheus/client_model v0.2.0/go.mod h1:xMI15A0UPsDsEKsMN9yxemIoYk6Tm2C1GtYGdfGttqA=
+github.com/prometheus/common v0.0.0-20181126121408-4724e9255275/go.mod h1:daVV7qP5qjZbuso7PdcryaAu0sAZbrN9i7WWcTMWvro=
+github.com/prometheus/common v0.4.1/go.mod h1:TNfzLD0ON7rHzMJeJkieUDPYmFC7Snx/y86RQel1bk4=
+github.com/prometheus/common v0.9.1/go.mod h1:yhUN8i9wzaXS3w1O07YhxHEBxD+W35wd8bs7vj7HSQ4=
+github.com/prometheus/common v0.10.0/go.mod h1:Tlit/dnDKsSWFlCLTWaA1cyBgKHSMdTB80sz/V91rCo=
+github.com/prometheus/common v0.26.0/go.mod h1:M7rCNAaPfAosfx8veZJCuw84e35h3Cfd9VFqTh1DIvc=
+github.com/prometheus/procfs v0.0.0-20181005140218-185b4288413d/go.mod h1:c3At6R/oaqEKCNdg8wHV1ftS6bRYblBhIjjI8uT2IGk=
+github.com/prometheus/procfs v0.0.0-20181204211112-1dc9a6cbc91a/go.mod h1:c3At6R/oaqEKCNdg8wHV1ftS6bRYblBhIjjI8uT2IGk=
+github.com/prometheus/procfs v0.0.2/go.mod h1:TjEm7ze935MbeOT/UhFTIMYKhuLP4wbCsTZCD3I8kEA=
+github.com/prometheus/procfs v0.0.8/go.mod h1:7Qr8sr6344vo1JqZ6HhLceV9o3AJ1Ff+GxbHq6oeK9A=
+github.com/prometheus/procfs v0.1.3/go.mod h1:lV6e/gmhEcM9IjHGsFOCxxuZ+z1YqCvr4OA4YeYWdaU=
+github.com/prometheus/procfs v0.6.0/go.mod h1:cz+aTbrPOrUb4q7XlbU9ygM+/jj0fzG6c1xBZuNvfVA=
+github.com/russross/blackfriday/v2 v2.1.0/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM=
+github.com/sean-/seed v0.0.0-20170313163322-e2103e2c3529 h1:nn5Wsu0esKSJiIVhscUtVbo7ada43DJhG55ua/hjS5I=
+github.com/sean-/seed v0.0.0-20170313163322-e2103e2c3529/go.mod h1:DxrIzT+xaE7yg65j358z/aeFdxmN0P9QXhEzd20vsDc=
+github.com/sirupsen/logrus v1.2.0/go.mod h1:LxeOpSwHxABJmUn/MG1IvRgCAasNZTLOkJPxbbu5VWo=
+github.com/sirupsen/logrus v1.4.2/go.mod h1:tLMulIdttU9McNUspp0xgXVQah82FyeX6MwdIuYE2rE=
+github.com/sirupsen/logrus v1.6.0/go.mod h1:7uNnSEd1DgxDLC74fIahvMZmmYsHGZGEOFrfsX/uA88=
+github.com/spf13/cobra v1.8.1 h1:e5/vxKd/rZsfSJMUX1agtjeTDf+qv1/JdBF8gg5k9ZM=
+github.com/spf13/cobra v1.8.1/go.mod h1:wHxEcudfqmLYa8iTfL+OuZPbBZkmvliBWKIezN3kD9Y=
+github.com/spf13/pflag v1.0.5/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An2Bg=
+github.com/spf13/pflag v1.0.6 h1:jFzHGLGAlb3ruxLB8MhbI6A8+AQX/2eW4qeyNZXNp2o=
+github.com/spf13/pflag v1.0.6/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An2Bg=
+github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
+github.com/stretchr/objx v0.1.1/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
+github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs=
+github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
+github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4=
+github.com/stretchr/testify v1.7.2/go.mod h1:R6va5+xMeoiuVRoj+gSkQ7d3FALtqAAGI1FQKckRals=
+github.com/tv42/httpunix v0.0.0-20150427012821-b75d8614f926/go.mod h1:9ESjWnEqriFuLhtthL60Sar/7RFoluCcXsuvEwTV5KM=
+golang.org/x/crypto v0.0.0-20180904163835-0709b304e793/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4=
+golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
+golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
+golang.org/x/mod v0.13.0 h1:I/DsJXRlw/8l/0c24sM9yb0T4z9liZTduXvdAWYiysY=
+golang.org/x/mod v0.13.0/go.mod h1:hTbmBsO62+eylJbnUtE2MGJUyE7QWk4xUqPFrRgJ+7c=
+golang.org/x/net v0.0.0-20180724234803-3673e40ba225/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
+golang.org/x/net v0.0.0-20181114220301-adae6a3d119a/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
+golang.org/x/net v0.0.0-20181201002055-351d144fa1fc/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
+golang.org/x/net v0.0.0-20190108225652-1e06a53dbb7e/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
+golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
+golang.org/x/net v0.0.0-20190613194153-d28f0bde5980/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
+golang.org/x/net v0.0.0-20200625001655-4c5254603344/go.mod h1:/O7V0waA8r7cgGh81Ro3o1hOxt32SMVPicZroKQ2sZA=
+golang.org/x/net v0.17.0 h1:pVaXccu2ozPjCXewfr1S7xza/zcXTity9cCdXQYSjIM=
+golang.org/x/net v0.17.0/go.mod h1:NxSsAGuq816PNPmqtQdLE42eU2Fs7NoRIZrHJAlaCOE=
+golang.org/x/oauth2 v0.0.0-20190226205417-e64efc72b421/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw=
+golang.org/x/sync v0.0.0-20181108010431-42b317875d0f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sync v0.0.0-20181221193216-37e7f081c4d4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sync v0.0.0-20201207232520-09787c993a3a/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sys v0.0.0-20180905080454-ebe1bf3edb33/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
+golang.org/x/sys v0.0.0-20181116152217-5ac8a444bdc5/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
+golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
+golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20190422165155-953cdadca894/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20200106162015-b016eb3dc98e/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20200116001909-b77594299b42/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20200122134326-e047566fdf82/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20200223170610-d5e6a3e2c0ae/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20200323222414-85ca7c5b95cd/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20200615200032-f1bc736245b1/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20200625212154-ddb9806d33ae/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20210124154548-22da62e12c0c/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20210603081109-ebe580a85c40/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.0.0-20210630005230-0f9fa26af87c/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.0.0-20210927094055-39ccf1dd6fa6/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.0.0-20220503163025-988cb79eb6c6/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.13.0 h1:Af8nKPmuFypiUBjVoU9V20FiaFXOcuZI21p0ycVYYGE=
+golang.org/x/sys v0.13.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.15.0 h1:h48lPFYpsTvQJZF4EKyI4aLHaev3CxivZmv7yZig9pc=
+golang.org/x/sys v0.15.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
+golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
+golang.org/x/text v0.3.2/go.mod h1:bEr9sfX3Q8Zfm5fL9x+3itogRgK3+ptLWKqgva+5dAk=
+golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
+golang.org/x/tools v0.14.0 h1:jvNa2pY0M4r62jkRQ6RwEZZyPcymeL9XZMLBbV7U2nc=
+golang.org/x/tools v0.14.0/go.mod h1:uYBEerGOWcJyEORxN+Ek8+TT266gXkNlHdJBwexUsBg=
+golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
+google.golang.org/appengine v1.4.0/go.mod h1:xpcJRLb0r/rnEns0DIKYYv+WjYCduHsrkT7/EB5XEv4=
+google.golang.org/protobuf v0.0.0-20200109180630-ec00e32a8dfd/go.mod h1:DFci5gLYBciE7Vtevhsrf46CRTquxDuWsQurQQe4oz8=
+google.golang.org/protobuf v0.0.0-20200221191635-4d8936d0db64/go.mod h1:kwYJMbMJ01Woi6D6+Kah6886xMZcty6N08ah7+eCXa0=
+google.golang.org/protobuf v0.0.0-20200228230310-ab0ca4ff8a60/go.mod h1:cfTl7dwQJ+fmap5saPgwCLgHXTUD7jkjRqWcaiX5VyM=
+google.golang.org/protobuf v1.20.1-0.20200309200217-e05f789c0967/go.mod h1:A+miEFZTKqfCUM6K7xSMQL9OKL/b6hQv+e19PK+JZNE=
+google.golang.org/protobuf v1.21.0/go.mod h1:47Nbq4nVaFHyn7ilMalzfO3qCViNmqZ2kzikPIcrTAo=
+google.golang.org/protobuf v1.23.0/go.mod h1:EGpADcykh3NcUnDUJcl1+ZksZNG86OlYog2l/sGQquU=
+google.golang.org/protobuf v1.26.0-rc.1/go.mod h1:jlhhOSvTdKEhbULTjvd4ARK9grFBp09yW+WbY/TyQbw=
+gopkg.in/alecthomas/kingpin.v2 v2.2.6/go.mod h1:FMv+mEhP44yOT+4EoQTLFTRgOQ1FBLkstjWtayDeSgw=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/yaml.v2 v2.2.1/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
+gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
+gopkg.in/yaml.v2 v2.2.4/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
+gopkg.in/yaml.v2 v2.2.5/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
+gopkg.in/yaml.v2 v2.3.0/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
+gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
@@ -3,39 +3,60 @@ package main
 import (
 	"os"
 	"os/signal"
-	"path/filepath"
 	"syscall"

-	"github.com/CefBoud/monkafka/broker"
 	log "github.com/CefBoud/monkafka/logging"
+	"github.com/CefBoud/monkafka/protocol"
 	"github.com/CefBoud/monkafka/types"
+	"github.com/hashicorp/serf/serf"
+	"github.com/spf13/cobra"
 )

 var config = types.Configuration{
-	LogDir:                      filepath.Join(os.TempDir(), "MonKafka"),
+	LogDir:                      os.TempDir(),
 	BrokerHost:                  "localhost",
 	BrokerPort:                  9092,
+	SerfConfig:                  serf.DefaultConfig(),
 	FlushIntervalMs:             5000,
 	LogRetentionCheckIntervalMs: 1000 * 30,          // 30 sec  //5 * 60 * 1000, // 5 min
 	LogRetentionMs:              3 * 60 * 60 * 1000, // 3h //604800000 (7 days)
 	LogSegmentSizeBytes:         104857600 * 5,      // 500 MiB
 	LogSegmentMs:                1800000,            // 30 min
-
 }

 func main() {
-	// TODO: config from args / env
-	broker := broker.NewBroker(config)
+	var rootCmd = &cobra.Command{
+		Use:   "broker",
+		Short: "Kafka broker with Raft protocol support",
+		Run: func(cmd *cobra.Command, args []string) {
+			broker := protocol.NewBroker(&config)
+			log.SetLogLevel(log.DEBUG)

-	log.SetLogLevel(log.DEBUG)
-	signalChannel := make(chan os.Signal, 1)
-	signal.Notify(signalChannel, syscall.SIGINT, syscall.SIGTERM)
-	go func() {
-		sig := <-signalChannel
-		log.Info("\nReceived signal: %s. Shutting down...\n", sig)
-		broker.Shutdown()
-		os.Exit(0)
-	}()
+			// Handle termination signals (e.g., Ctrl+C)
+			signalChannel := make(chan os.Signal, 1)
+			signal.Notify(signalChannel, syscall.SIGINT, syscall.SIGTERM)
+			go func() {
+				sig := <-signalChannel
+				log.Info("Received signal: %s. Shutting down...", sig)
+				broker.Shutdown()
+				os.Exit(0)
+			}()

-	broker.Startup()
+			// Start the broker
+			broker.Startup()
+		},
+	}
+	// Add flags using Cobra
+	rootCmd.Flags().BoolVar(&config.Bootstrap, "bootstrap", false, "Indicates if this broker should bootstrap the first broker in the cluster")
+	rootCmd.Flags().IntVar(&config.NodeID, "node-id", -1, "Unique node ID for this broker instance")
+	rootCmd.MarkFlagRequired("node-id")
+	rootCmd.Flags().StringVar(&config.RaftAddress, "raft-addr", "localhost:2221", "The address where Raft will bind")
+	rootCmd.Flags().StringVar(&config.SerfAddress, "serf-addr", "127.0.0.1:3331", "The address where Serf will bind")
+	rootCmd.Flags().StringVar(&config.SerfJoinAddress, "serf-join", "", "The Serf Join address")
+	rootCmd.Flags().IntVar(&config.BrokerPort, "broker-port", 9092, "Port where the broker will listen for client connections")
+
+	// Execute the root command
+	if err := rootCmd.Execute(); err != nil {
+		log.Panic("Failed to execute root command %v", err)
+	}
 }
@@ -20,7 +20,7 @@ type APIVersionsResponse struct {
 }

 // APIVersion (Api key = 18)
-func getAPIVersionResponse(req types.Request) []byte {
+func (b *Broker) getAPIVersionResponse(req types.Request) []byte {
 	response := APIVersionsResponse{
 		APIKeys: []APIKey{
 			{APIKey: produceKey, MinVersion: 0, MaxVersion: 11},
@@ -36,6 +36,7 @@ func getAPIVersionResponse(req types.Request) []byte {
 			{APIKey: apiVersionKey, MinVersion: 0, MaxVersion: 4},
 			{APIKey: createTopicKey, MinVersion: 0, MaxVersion: 7},
 			{APIKey: initProducerIDKey, MinVersion: 0, MaxVersion: 5},
+			{APIKey: describeConfigsKey, MinVersion: 0, MaxVersion: 4},
 		},
 	}

@@ -0,0 +1,404 @@
+package protocol
+
+import (
+	"fmt"
+	"io"
+	"net"
+	"os"
+	"path"
+	"path/filepath"
+	"strconv"
+	"strings"
+	"time"
+
+	"github.com/CefBoud/monkafka/logging"
+	log "github.com/CefBoud/monkafka/logging"
+	"github.com/CefBoud/monkafka/raft"
+	"github.com/CefBoud/monkafka/serde"
+	"github.com/CefBoud/monkafka/storage"
+	"github.com/CefBoud/monkafka/types"
+	"github.com/CefBoud/monkafka/utils"
+
+	hraft "github.com/hashicorp/raft"
+	raftboltdb "github.com/hashicorp/raft-boltdb"
+	"github.com/hashicorp/serf/serf"
+)
+
+const (
+	// serfEventChSize is the size of the buffered channel to get Serf
+	// events. If this is exhausted we will block Serf and Memberlist.
+	serfEventChSize = 2048
+)
+
+// Broker represents a Kafka broker instance
+type Broker struct {
+	Config         *types.Configuration
+	ShutDownSignal chan bool
+	Serf           *serf.Serf  // Serf cluster maintained inside the DC
+	Raft           *hraft.Raft // Raft cluster maintained inside the DC
+	FSM            *raft.FSM
+
+	RaftNotifyCh <-chan bool // raftNotifyCh ensures that we get reliable leader transition notifications from the Raft layer.
+
+	SerfEventCh chan serf.Event  // eventCh is used to receive events from the serf cluster
+	ReconcileCh chan serf.Member //  used to pass events from the serf handler into the leader manager to update the strong state
+}
+
+// NewBroker creates a new Broker instance with the provided configuration
+func NewBroker(config *types.Configuration) *Broker {
+	return &Broker{
+		Config:         config,
+		ShutDownSignal: make(chan bool),
+		RaftNotifyCh:   make(<-chan bool),
+		SerfEventCh:    make(chan serf.Event, serfEventChSize),
+		ReconcileCh:    make(chan serf.Member),
+	}
+}
+
+// Startup initializes the broker, starts the storage, loads group metadata,
+// and listens for incoming connections
+func (b *Broker) Startup() {
+	var err error
+	b.Config.LogDir = filepath.Join(b.Config.LogDir, fmt.Sprintf("MonKafka-%v", b.Config.NodeID))
+	b.FSM = &raft.FSM{NodeID: uint32(b.Config.NodeID), Topics: make(map[string]types.Topic), Nodes: make(map[uint32]types.Node)}
+
+	err = b.SetupRaft()
+	if err != nil {
+		log.Panic("Raft Setup failed: %v", err)
+	}
+
+	err = b.SetupSerf()
+	if err != nil {
+		log.Panic("Serf Setup failed: %v", err)
+	}
+
+	go b.handleSerfEvent()
+	go b.monitorLeadership()
+	go b.printClusteringInfo()
+
+	storage.Startup(b.Config, b.ShutDownSignal)
+	LoadGroupMetadataState()
+
+	// Set up a TCP listener on the specified port
+	listener, err := net.Listen("tcp", fmt.Sprintf(":%d", b.Config.BrokerPort))
+	if err != nil {
+		log.Error("Error starting server: %v\n", err)
+		os.Exit(1)
+	}
+	defer listener.Close()
+
+	log.Info("Server is listening on port %d...\n", b.Config.BrokerPort)
+
+	// Accept and handle incoming connections
+	for {
+		conn, err := listener.Accept()
+		if err != nil {
+			log.Error("Error accepting connection: %v\n", err)
+			continue
+		}
+		go b.HandleConnection(conn)
+	}
+}
+
+// HandleConnection processes incoming requests from a client connection
+func (b *Broker) HandleConnection(conn net.Conn) {
+	defer conn.Close()
+	connectionAddr := conn.RemoteAddr().String()
+	log.Debug("Connection established with %s\n", connectionAddr)
+
+	for {
+		// First, we read the length, then allocate a byte slice based on it.
+		// ReadFull (not Read) is used to ensure the entire request is read. Partial data would result in parsing errors
+		lengthBuffer := make([]byte, 4)
+		_, err := io.ReadFull(conn, lengthBuffer)
+		if err != nil {
+			log.Info("failed to read request's length. Error: %v ", err)
+			return
+		}
+		length := serde.Encoding.Uint32(lengthBuffer)
+		buffer := make([]byte, length+4)
+		copy(buffer, lengthBuffer)
+		_, err = io.ReadFull(conn, buffer[4:])
+		if err != nil {
+			if err.Error() != "EOF" {
+				log.Error("Error reading from connection: %v\n", err)
+			}
+			break
+		}
+		req := serde.ParseHeader(buffer, connectionAddr)
+		apiKeyHandler := b.APIDispatcher(req.RequestAPIKey)
+		log.Debug("Received RequestAPIKey: %v | RequestAPIVersion: %v | CorrelationID: %v | Length: %v \n\n", apiKeyHandler.Name, req.RequestAPIVersion, req.CorrelationID, length)
+		response := apiKeyHandler.Handler(req)
+
+		_, err = conn.Write(response)
+		if err != nil {
+			log.Error("Error writing to connection: %v\n", err)
+			break
+		}
+	}
+	log.Debug("Connection with %s closed.\n", connectionAddr)
+}
+
+// Shutdown gracefully shuts down the broker and its components
+func (b *Broker) Shutdown() {
+	// close ShutDownSignal so any goroutine waiting on it will run
+	close(b.ShutDownSignal)
+	log.Info("Broker Shut down...")
+	log.Info("Shutting down Serf ...")
+
+	log.Info("Waiting a bit after Serf leaving to allow other servers to be notified")
+	if b.IsController() { // raft leader
+		raftServers, err := b.getRaftServers()
+		if err != nil {
+			log.Error("failed to get raft server %v", err)
+		} else {
+			if len(raftServers) > 2 {
+				log.Info("Node is raft leader and there are >2 raft servers, calling LeadershipTransfer")
+				future := b.Raft.RemoveServer(hraft.ServerID(b.Config.RaftID), 0, 0)
+				if err := future.Error(); err != nil {
+					log.Error("failed to remove self from raft cluster %v", err)
+				}
+			}
+		}
+	}
+
+	if err := b.Serf.Leave(); err != nil {
+		log.Error("Serf leave failed: %s", err)
+	}
+	LeaveDrainTime := 5 * time.Second
+
+	time.Sleep(LeaveDrainTime)
+	storage.Shutdown()
+
+	if b.Serf != nil {
+		b.Serf.Shutdown()
+	}
+
+	if b.Raft != nil {
+		future := b.Raft.Shutdown()
+		if err := future.Error(); err != nil {
+			log.Warn("error shutting down raft:  %v", err)
+		}
+	}
+}
+
+func (b *Broker) getRaftServers() ([]hraft.Server, error) {
+	configFuture := b.Raft.GetConfiguration()
+	if err := configFuture.Error(); err != nil {
+		return nil, fmt.Errorf("getRaftServer: can't get raft configuration error: %s", err)
+	}
+	return configFuture.Configuration().Servers, nil
+}
+
+// AppendRaftEntry add a new entry to the raft log
+func (b *Broker) AppendRaftEntry(entryType raft.CommandType, entry any) (any, error) {
+	bytes, err := raft.EncodeLogEntry(entryType, entry)
+	if err != nil {
+		return nil, err
+	}
+	future := b.Raft.Apply(bytes, 10*time.Second)
+	if err := future.Error(); err != nil {
+		return nil, err
+	}
+	log.Info("added entry to raft: %+v", entry)
+	return future.Response(), nil
+}
+
+// IsController return if the broker is the cluster's controller which is also the raft leader
+func (b *Broker) IsController() bool {
+	return b.Raft.State() == hraft.Leader
+}
+
+// SetupRaft inits Raft for the broker
+func (b *Broker) SetupRaft() error {
+	raftAddress := b.Config.RaftAddress // fmt.Sprintf("localhost:%v", b.Config.RaftPort)
+	dir := path.Join(b.Config.LogDir, "raft"+b.Config.RaftID)
+	err := os.MkdirAll(dir, os.ModePerm)
+	if err != nil {
+		return fmt.Errorf("could not create data directory: %s", err)
+	}
+
+	store, err := raftboltdb.NewBoltStore(path.Join(dir, "bolt"))
+	if err != nil {
+		return fmt.Errorf("could not create bolt store: %s", err)
+	}
+
+	snapshots, err := hraft.NewFileSnapshotStore(path.Join(dir, "snapshot"), 2, os.Stderr)
+	if err != nil {
+		return fmt.Errorf("could not create snapshot store: %s", err)
+	}
+
+	tcpAddr, err := net.ResolveTCPAddr("tcp", raftAddress)
+	if err != nil {
+		return fmt.Errorf("could not resolve address: %s", err)
+	}
+
+	transport, err := hraft.NewTCPTransport(raftAddress, tcpAddr, 10, time.Second*10, os.Stderr)
+	if err != nil {
+		return fmt.Errorf("could not create tcp transport: %s", err)
+	}
+
+	raftCfg := hraft.DefaultConfig()
+	raftCfg.LogLevel = "INFO"
+
+	if b.Config.RaftID == "" {
+		b.Config.RaftID = fmt.Sprintf("raft-broker-%d", b.Config.NodeID)
+	}
+	nodeID := b.Config.RaftID
+	raftCfg.LocalID = hraft.ServerID(nodeID)
+
+	// Set up a channel for reliable leader notifications.
+	raftNotifyCh := make(chan bool, 1)
+	raftCfg.NotifyCh = raftNotifyCh
+	b.RaftNotifyCh = raftNotifyCh
+
+	b.Raft, err = hraft.NewRaft(raftCfg, b.FSM, store, store, snapshots, transport)
+	if err != nil {
+		return fmt.Errorf("could not create raft instance: %s", err)
+	}
+
+	if b.Config.Bootstrap {
+		logging.Info("bootstrapping raft with nodeID %v ....", nodeID)
+		hasState, err := hraft.HasExistingState(store, store, snapshots)
+		if err != nil {
+			return err
+		}
+		logging.Info("bootstrapping hasState %v", hasState)
+		if !hasState {
+			future := b.Raft.BootstrapCluster(hraft.Configuration{
+				Servers: []hraft.Server{
+					{
+						ID:      hraft.ServerID(nodeID),
+						Address: transport.LocalAddr(),
+					},
+				},
+			})
+			if err := future.Error(); err != nil {
+				logging.Error(" bootstrap cluster error: %s", err)
+			}
+		}
+	}
+	return nil
+}
+
+// SetupSerf to setup the serf agent and maybe join a serf cluster
+func (b *Broker) SetupSerf() error {
+	var err error
+	conf := b.Config.SerfConfig
+	conf.Init()
+	conf.NodeName = b.Config.RaftID
+	bindIP, bindPort, err := net.SplitHostPort(b.Config.SerfAddress)
+	if err != nil {
+		return err
+	}
+	log.Debug("SetupSerf: bindIP=%v bindPort=%v", bindIP, bindPort)
+	conf.MemberlistConfig.BindAddr = bindIP
+	conf.MemberlistConfig.BindPort, err = strconv.Atoi(bindPort)
+	if err != nil {
+		return err
+	}
+	conf.Tags["role"] = "broker"
+	conf.Tags["ID"] = strconv.Itoa(b.Config.NodeID)
+	conf.Tags["broker_addr"] = fmt.Sprintf("%s:%d", b.Config.BrokerHost, b.Config.BrokerPort)
+	conf.Tags["raft_server_id"] = b.Config.RaftID
+	conf.Tags["raft_addr"] = b.Config.RaftAddress
+	conf.Tags["serf_addr"] = b.Config.SerfAddress
+
+	conf.EventCh = b.SerfEventCh
+	conf.SnapshotPath = filepath.Join(b.Config.LogDir, "serf-snapshot")
+
+	if err = utils.EnsurePath(conf.SnapshotPath, false); err != nil {
+		return fmt.Errorf("could not serf SnapshotPath dir: %s", err)
+	}
+
+	b.Serf, err = serf.Create(conf)
+
+	if len(b.Config.SerfJoinAddress) > 0 {
+		existingSerfNodes := strings.Split(b.Config.SerfJoinAddress, ",")
+		log.Info("joining serf nodes: %v", existingSerfNodes)
+		n, err := b.Serf.Join(existingSerfNodes, true)
+		if err != nil {
+			log.Error("Couldn't join cluster, starting own: %v\n", err)
+		} else {
+			log.Info("Serf join: successfully contacted %v node. Members: %v", n, b.Serf.Members())
+		}
+	}
+	return err
+}
+
+func (b *Broker) handleSerfEvent() {
+	for {
+		log.Info("handleSerfEvent .... %v", b.Serf.Memberlist().Members())
+		select {
+		case e := <-b.SerfEventCh:
+			log.Info("serf EventType: %v", e.EventType())
+			switch e.EventType() {
+			case serf.EventMemberJoin:
+				b.handleSerfMemberJoin(e.(serf.MemberEvent))
+			case serf.EventMemberFailed:
+				// a node is moved from `fail` to `reap` (completely ousted from the cluster) after `reconnect_timeout` (defaults to 24h)
+				log.Error("handleSerfEvent... EventMemberFailed ")
+			case serf.EventMemberReap, serf.EventMemberLeave:
+				b.handleSerfMemberLeft(e.(serf.MemberEvent))
+			}
+		case <-b.ShutDownSignal:
+
+			return
+		}
+	}
+}
+
+// GetClusterNodes returns the raft cluster nodes each representing a broker
+func (b *Broker) GetClusterNodes() ([]*types.Node, error) {
+	configFuture := b.Raft.GetConfiguration()
+	if err := configFuture.Error(); err != nil {
+		log.Error("handleSerfMemberLeft: can't get raft configuration error: %s", err)
+		return nil, err
+	}
+	nodes := make(map[string]*types.Node)
+
+	for _, server := range configFuture.Configuration().Servers {
+		nodes[string(server.ID)] = &types.Node{}
+	}
+
+	_, leaderID := b.Raft.LeaderWithID()
+	for _, m := range b.Serf.Members() {
+		raftServerID := m.Tags["raft_server_id"]
+		n, ok := nodes[raftServerID]
+		if ok {
+			// log.Debug("serf node tags %v", m.Tags)
+			id, err := strconv.Atoi(m.Tags["ID"])
+			if err != nil {
+				log.Error("GetClusterNodes: unable to convert serf ID to uint32: %v", err)
+				continue
+			}
+			n.NodeID = uint32(id)
+			host, port, err := net.SplitHostPort(m.Tags["broker_addr"])
+			if err != nil {
+				log.Error("GetClusterNodes: unable to parse broker_addr: %v", err)
+				continue
+			}
+			portInt, _ := strconv.Atoi(port)
+			n.Host, n.Port = host, uint32(portInt)
+			n.IsController = string(leaderID) == raftServerID
+		}
+	}
+	var res []*types.Node
+	for _, n := range nodes {
+		res = append(res, n)
+	}
+	return res, nil
+}
+
+func (b *Broker) printClusteringInfo() {
+	ticker := time.NewTicker(30000 * time.Millisecond)
+	defer ticker.Stop() // Ensure the ticker is stopped when done
+	for {
+		select {
+		case <-ticker.C:
+			leaderAddr, leaderID := b.Raft.LeaderWithID()
+			log.Debug("Serf members:  %v", b.Serf.Members())
+			log.Debug("Raft LeaderAddr: [%v] - leaderID [%v]", leaderAddr, leaderID)
+		}
+	}
+}
@@ -1,10 +1,12 @@
 package protocol

 import (
+	"fmt"
+	"math/rand"
+
 	log "github.com/CefBoud/monkafka/logging"
+	"github.com/CefBoud/monkafka/raft"
 	"github.com/CefBoud/monkafka/serde"
-	"github.com/CefBoud/monkafka/state"
-	"github.com/CefBoud/monkafka/storage"
 	"github.com/CefBoud/monkafka/types"
 )

@@ -63,11 +65,12 @@ type CreateTopicsResponseConfig struct {
 }

 // CreateTopics	(Api key = 19)
-func getCreateTopicResponse(req types.Request) []byte {
+func (b *Broker) getCreateTopicResponse(req types.Request) []byte {
 	decoder := serde.NewDecoder(req.Body)
 	createTopicsRequest := decoder.Decode(&CreateTopicsRequest{}).(*CreateTopicsRequest)
 	log.Debug("CreateTopicsRequest %+v", createTopicsRequest)
 	response := CreateTopicsResponse{}
+
 	for _, topic := range createTopicsRequest.Topics {

 		if int32(topic.NumPartitions) == -1 {
@@ -80,20 +83,63 @@ func getCreateTopicResponse(req types.Request) []byte {
 			ReplicationFactor: topic.ReplicationFactor,
 			Configs:           []CreateTopicsResponseConfig{}, // TODO handle conf
 		}
-		if state.TopicExists(topic.Name) {
-			topicResponse.ErrorCode = uint16(ErrTopicAlreadyExists.Code)
-			topicResponse.ErrorMessage = ErrTopicAlreadyExists.Message

+		if !b.IsController() {
+			log.Error("getCreateTopicResponse: not controller")
+			topicResponse.ErrorCode = uint16(ErrNotController.Code)
+			topicResponse.ErrorMessage = ErrNotController.Message
 		} else {
-			err := storage.CreateTopic(topic.Name, topic.NumPartitions)
-			if err != nil {
-				log.Error("Error creating topic %v", err)
+			if b.FSM.TopicExists(topic.Name) {
+				topicResponse.ErrorCode = uint16(ErrTopicAlreadyExists.Code)
+				topicResponse.ErrorMessage = ErrTopicAlreadyExists.Message
+
+			} else {
+				configs := make(map[string]string)
+				for _, c := range topic.Configs {
+					configs[c.Name] = c.Value
+				}
+				err := b.CreateTopicPartitions(topic.Name, topic.NumPartitions, configs)
+				// resp, err := b.AppendRaftEntry(raft.AddTopic, raft.Topic{Name: topic.Name})
+				if err != nil {
+					log.Error("Error CreateTopicPartitions %v", err)
+				}
 			}
 		}
 		response.Topics = append(response.Topics, topicResponse)

 	}
-
+	log.Debug("CreateTopicResponse %+v", response)
 	encoder := serde.NewEncoder()
 	return encoder.EncodeResponseBytes(req, response)
 }
+
+// CreateTopicPartitions creates a new topic with its partition by appending them to the raft log.
+func (b *Broker) CreateTopicPartitions(name string, numPartitions uint32, configs map[string]string) error {
+	resp, err := b.AppendRaftEntry(raft.AddTopic, types.Topic{Name: name, Configs: configs})
+	if err != nil {
+		log.Error("Error append topic to raft log %v", err)
+	}
+	log.Debug("raft AddTopic entry for %v with configs [%+v]. Result: %v ", name, configs, resp)
+
+	nodes, err := b.GetClusterNodes()
+	if err != nil || len(nodes) < 1 {
+		return fmt.Errorf("CreateTopicPartitions GetClusterNodes error: %v", err)
+	}
+
+	for i := uint32(0); i < numPartitions; i++ {
+		// pick a leader randomly
+		leaderID := nodes[rand.Intn(len(nodes))].NodeID
+		partition := types.PartitionState{
+			Topic:          name,
+			PartitionIndex: i,
+			LeaderID:       leaderID,
+		}
+		log.Debug("adding partition %+v", partition)
+		resp, err = b.AppendRaftEntry(raft.AddPartition, partition)
+		if err != nil {
+			return fmt.Errorf("Error appending partition to raft log %v", err)
+		}
+		log.Debug("raft AddPartition entry for %v. Result: %v ", name, resp)
+	}
+	return nil
+}
@@ -0,0 +1,110 @@
+package protocol
+
+import (
+	"strconv"
+
+	log "github.com/CefBoud/monkafka/logging"
+	"github.com/CefBoud/monkafka/serde"
+	"github.com/CefBoud/monkafka/types"
+)
+
+// ResourceType represents the kafka resource type (topic, group, etc.)
+type ResourceType int8
+
+// https://github.com/apache/kafka/blob/c6335c2ae86913954d940036917b7556e9ac0460/clients/src/main/java/org/apache/kafka/common/resource/ResourceType.java#L31
+const (
+	ResourceTypeUnknown         ResourceType = 0
+	ResourceTypeAny             ResourceType = 1
+	ResourceTypeTopic           ResourceType = 2
+	ResourceTypeGroup           ResourceType = 3
+	ResourceTypeBroker          ResourceType = 4
+	ResourceTypeCluster         ResourceType = 4
+	ResourceTypeTransactionalID ResourceType = 5
+	ResourceTypeDelegationToken ResourceType = 6
+)
+
+// DescribeConfigsRequest struct represents the request for DescribeConfigs with version 4.
+type DescribeConfigsRequest struct {
+	Resources            []DescribeConfigsResource
+	IncludeSynonyms      bool
+	IncludeDocumentation bool
+}
+
+// DescribeConfigsResource struct represents the resource inside DescribeConfigsRequest.
+type DescribeConfigsResource struct {
+	ResourceType      uint8
+	ResourceName      string   `kafka:"CompactString"`
+	ConfigurationKeys []string `kafka:"CompactString"`
+}
+
+// DescribeConfigsResponse struct represents the response for DescribeConfigs with version 4.
+type DescribeConfigsResponse struct {
+	ThrottleTimeMs uint32
+	Results        []DescribeConfigsResponseResult
+}
+
+// DescribeConfigsResponseResult struct represents the result inside DescribeConfigsResponse.
+type DescribeConfigsResponseResult struct {
+	ErrorCode    uint16
+	ErrorMessage string `kafka:"CompactString"`
+	ResourceType uint8
+	ResourceName string `kafka:"CompactString"`
+	Configs      []DescribeConfigsResponseConfig
+}
+
+// DescribeConfigsResponseConfig struct represents the configuration details within a result.
+type DescribeConfigsResponseConfig struct {
+	Name          string
+	Value         string `kafka:"CompactNullableString"`
+	ReadOnly      bool
+	ConfigSource  uint8
+	IsSensitive   bool
+	Synonyms      []DescribeConfigsResponseSynonym
+	ConfigType    uint
+	Documentation string `kafka:"CompactNullableString"`
+}
+
+// DescribeConfigsResponseSynonym struct represents synonym details within a config.
+type DescribeConfigsResponseSynonym struct {
+	Name   string `kafka:"CompactString"`
+	Value  string `kafka:"CompactNullableString"`
+	Source uint
+}
+
+// DescribeConfigs	(Api key = 32)
+func (b *Broker) getDescribeConfigsResponse(req types.Request) []byte {
+	decoder := serde.NewDecoder(req.Body)
+	describeConfigsRequest := decoder.Decode(&DescribeConfigsRequest{}).(*DescribeConfigsRequest)
+	log.Debug("DescribeConfigsRequest %+v", describeConfigsRequest)
+	response := DescribeConfigsResponse{}
+
+	for _, resourceConfReq := range describeConfigsRequest.Resources {
+		resourceResp := DescribeConfigsResponseResult{
+			ResourceType: resourceConfReq.ResourceType,
+			ResourceName: resourceConfReq.ResourceName,
+		}
+		// TODO: handle other resource types
+		if ResourceType(resourceConfReq.ResourceType) == ResourceTypeTopic {
+			topic, ok := b.FSM.GetTopic(resourceConfReq.ResourceName)
+			if !ok {
+				resourceResp.ErrorCode = uint16(ErrUnknownTopicOrPartition.Code)
+				resourceResp.ErrorMessage = ErrUnknownTopicOrPartition.Message
+			} else {
+				resourceResp.Configs = append(resourceResp.Configs, DescribeConfigsResponseConfig{
+					Name:  "partitions",
+					Value: strconv.Itoa(len(topic.Partitions)),
+				})
+				for key, value := range topic.Configs {
+					resourceResp.Configs = append(resourceResp.Configs, DescribeConfigsResponseConfig{
+						Name:  key,
+						Value: value,
+					})
+				}
+			}
+			response.Results = append(response.Results, resourceResp)
+		}
+	}
+	log.Debug("DescribeConfigsResponse %+v", response)
+	encoder := serde.NewEncoder()
+	return encoder.EncodeResponseBytes(req, response)
+}
@@ -0,0 +1,65 @@
+package protocol
+
+import "github.com/CefBoud/monkafka/types"
+
+// https://kafka.apache.org/protocol#protocol_api_keys
+var produceKey = uint16(0)
+var fetchKey = uint16(1)
+var listOffsetsKey = uint16(2)
+var metadataKey = uint16(3)
+var offsetCommitKey = uint16(8)
+var offsetFetchKey = uint16(9)
+var findCoordinatorKey = uint16(10)
+var joinGroupKey = uint16(11)
+var heartbeatKey = uint16(12)
+var syncGroupKey = uint16(14)
+var apiVersionKey = uint16(18)
+var createTopicKey = uint16(19)
+var initProducerIDKey = uint16(22)
+var describeConfigsKey = uint16(32)
+
+// APIKeyHandler represents a kafka api key with its handler
+type APIKeyHandler struct {
+	Name    string
+	Handler func(req types.Request) []byte
+}
+
+// APIDispatcher maps the Request key to its handler
+func (b *Broker) APIDispatcher(requestAPIKey uint16) APIKeyHandler {
+	switch requestAPIKey {
+	case produceKey:
+		return APIKeyHandler{Name: "Produce", Handler: b.getProduceResponse}
+	case fetchKey:
+		return APIKeyHandler{Name: "Fetch", Handler: b.getFetchResponse}
+	case listOffsetsKey:
+		return APIKeyHandler{Name: "ListOffsets", Handler: b.getListOffsetsResponse}
+	case metadataKey:
+		return APIKeyHandler{Name: "Metadata", Handler: b.getMetadataResponse}
+	case offsetCommitKey:
+		return APIKeyHandler{Name: "OffsetCommit", Handler: b.getOffsetCommitResponse}
+	case offsetFetchKey:
+		return APIKeyHandler{Name: "OffsetFetch", Handler: b.getOffsetFetchResponse}
+	case findCoordinatorKey:
+		return APIKeyHandler{Name: "FindCoordinator", Handler: b.getFindCoordinatorResponse}
+	case joinGroupKey:
+		return APIKeyHandler{Name: "JoinGroup", Handler: b.getJoinGroupResponse}
+	case heartbeatKey:
+		return APIKeyHandler{Name: "Heartbeat", Handler: b.getHeartbeatResponse}
+	case syncGroupKey:
+		return APIKeyHandler{Name: "SyncGroup", Handler: b.getSyncGroupResponse}
+	case apiVersionKey:
+		return APIKeyHandler{Name: "APIVersion", Handler: b.getAPIVersionResponse}
+	case createTopicKey:
+		return APIKeyHandler{Name: "CreateTopic", Handler: b.getCreateTopicResponse}
+	case initProducerIDKey:
+		return APIKeyHandler{Name: "InitProducerID", Handler: b.getInitProducerIDResponse}
+	case describeConfigsKey:
+		return APIKeyHandler{Name: "DescribeConfigs", Handler: b.getDescribeConfigsResponse}
+	default:
+		return APIKeyHandler{}
+	}
+
+}
+
+// ConsumerOffsetsTopic is the topic where consumers committed offsets are saved
+var ConsumerOffsetsTopic = "__consumer-offsets"
@@ -53,6 +53,7 @@ var (
 	ErrInvalidReplicationFactor     = Error{Code: 38, Message: "Replication factor is below 1 or larger than the number of available brokers.", IsRetriable: false}
 	ErrInvalidReplicaAssignment     = Error{Code: 39, Message: "Replica assignment is invalid.", IsRetriable: false}
 	ErrInvalidConfig                = Error{Code: 40, Message: "Configuration is invalid.", IsRetriable: false}
+	ErrNotController                = Error{Code: 41, Message: "This is not the correct controller for this cluster.", IsRetriable: true}
 )

 // ErrorMap associates error codes with corresponding Error structs
@@ -78,7 +78,7 @@ type AbortedTransaction struct {
 	FirstOffset uint64
 }

-func getFetchResponse(req types.Request) []byte {
+func (b *Broker) getFetchResponse(req types.Request) []byte {
 	decoder := serde.NewDecoder(req.Body)
 	fetchRequest := decoder.Decode(&FetchRequest{}).(*FetchRequest)
 	log.Debug("fetchRequest %+v", fetchRequest)
@@ -88,6 +88,16 @@ func getFetchResponse(req types.Request) []byte {
 	for _, tp := range fetchRequest.Topics {
 		fetchTopicResponse := FetchTopicResponse{TopicName: tp.Name}
 		for _, p := range tp.Partitions {
+			partition, exists := b.FSM.GetPartition(tp.Name, p.PartitionIndex)
+			if !exists {
+				response.ErrorCode = uint16(ErrReplicaNotAvailable.Code)
+				goto FINISH
+			}
+			if partition.LeaderID != b.FSM.NodeID {
+				log.Error("Fetch: Not leader for partition %v-%v", partition.Topic, partition.PartitionIndex)
+				response.ErrorCode = uint16(ErrNotLeaderOrFollower.Code)
+				goto FINISH
+			}
 			recordBytes, err := storage.GetRecordBatch(p.FetchOffset, tp.Name, p.PartitionIndex)
 			// log.Debug("getFetchResponse GetRecord recordBytes %v", recordBytes)
 			numTotalRecordBytes += len(recordBytes)
@@ -97,12 +107,12 @@ func getFetchResponse(req types.Request) []byte {

 			fetchTopicResponse.Partitions = append(fetchTopicResponse.Partitions,
 				FetchPartitionResponse{
-					PartitionIndex:       p.PartitionIndex,
-					HighWatermark:        state.GetPartition(tp.Name, p.PartitionIndex).EndOffset(), //uint64(MinusOne),
-					LastStableOffset:     uint64(MinusOne),
-					LogStartOffset:       state.GetPartition(tp.Name, p.PartitionIndex).StartOffset(),
-					PreferredReadReplica: 1,
-					Records:              recordBytes,
+					PartitionIndex:   p.PartitionIndex,
+					HighWatermark:    state.GetPartition(tp.Name, p.PartitionIndex).EndOffset(), //uint64(MinusOne),
+					LastStableOffset: uint64(MinusOne),
+					LogStartOffset:   state.GetPartition(tp.Name, p.PartitionIndex).StartOffset(),
+					// PreferredReadReplica: 1,
+					Records: recordBytes,
 				})
 		}
 		response.Responses = append(response.Responses, fetchTopicResponse)
@@ -111,6 +121,7 @@ func getFetchResponse(req types.Request) []byte {
 		log.Info("There is no data available for this fetch request, waiting for a bit ..")
 		time.Sleep(300 * time.Millisecond) // TODO get this from consumer settings
 	}
+FINISH:
 	encoder := serde.NewEncoder()
 	return encoder.EncodeResponseBytes(req, response)
 }
@@ -0,0 +1,126 @@
+package protocol
+
+import (
+	log "github.com/CefBoud/monkafka/logging"
+	"github.com/hashicorp/raft"
+	"github.com/hashicorp/serf/serf"
+)
+
+func (b *Broker) handleSerfMemberJoin(event serf.MemberEvent) error {
+	_, leaderID := b.Raft.LeaderWithID()
+	noLeader := leaderID == ""
+	if noLeader {
+		if b.Config.Bootstrap {
+			log.Info("handleSerfMemberJoin: there is no leader, current node will bootstrap")
+			// if bootstrapping, but no yet controller, we give raft state some time to transition to leader
+			// if !b.IsController() {
+			// 	log.Info("Waiting for node to bootstrap raft and become leader")
+			// 	timeToWaitForLeaderShip := time.Now().Add(5 * time.Second)
+			// 	leader := false
+			// 	for !leader && time.Now().Before(timeToWaitForLeaderShip) {
+			// 		time.Sleep(50 * time.Millisecond)
+			// 		leader = b.IsController()
+			// 	}
+			// 	log.Info("finished waiting. Is Node leader ? %v", leader)
+
+			// }
+		} else {
+			log.Info("handleSerfMemberJoin: there is no leader, current node WILL NOT bootstrap")
+			return nil
+		}
+	} else {
+		if !b.IsController() {
+			log.Info("handleSerfMemberJoin: node is not the leader, ignoring join event")
+			return nil
+		}
+	}
+
+	newMembers := make(map[string]serf.Member)
+	for _, m := range event.Members {
+		if m.Tags["role"] != "broker" {
+			log.Info("handleSerfMemberJoin: new member [%v - %v] is not a broker", m.Name, m.Addr)
+			continue
+		}
+		newMembers[m.Tags["raft_addr"]] = m
+	}
+
+	raftServers, err := b.getRaftServers()
+	if err != nil {
+		return err
+	}
+	for _, server := range raftServers {
+		for raftAddr := range newMembers {
+			if server.Address == raft.ServerAddress(raftAddr) {
+				log.Info("handleSerfMemberJoin: member [%v] already in raft cluster", raftAddr)
+				delete(newMembers, raftAddr)
+				if len(newMembers) == 0 {
+					return nil
+				}
+			}
+		}
+	}
+	for raftAddr, m := range newMembers {
+		log.Info("handleSerfMemberJoin: adding voter to the raft cluster with addr %s", raftAddr)
+		err := b.Raft.AddVoter(raft.ServerID(m.Tags["raft_server_id"]), raft.ServerAddress(m.Tags["raft_addr"]), 0, 0).Error()
+		if err != nil {
+			log.Error("Failed to add follower: %s", err)
+			return err
+		}
+	}
+	return nil
+}
+
+func (b *Broker) handleSerfMemberLeft(event serf.MemberEvent) error {
+	log.Debug("Inside handleSerfMemberLeft")
+	_, leaderID := b.Raft.LeaderWithID()
+	if leaderID == "" {
+		log.Info("handleSerfMemberLeft: there is no leader. Nothing to do.")
+		return nil
+	} else if !b.IsController() {
+		log.Info("handleSerfMemberLeft: node is not the leader, ignoring left/reap/failed event")
+		return nil
+	}
+
+	eventMembers := make(map[string]serf.Member)
+	for _, m := range event.Members {
+		if m.Tags["role"] != "broker" {
+			log.Info("handleSerfMemberLeft: member [%v - %v] is not a broker", m.Name, m.Addr)
+			continue
+		}
+		eventMembers[m.Tags["raft_addr"]] = m
+	}
+
+	raftServers, err := b.getRaftServers()
+	if err != nil {
+		return err
+	}
+	for _, server := range raftServers {
+		for raftAddr := range eventMembers {
+			if server.Address == raft.ServerAddress(raftAddr) {
+				log.Info("handleSerfMemberLeft: removing member [%v] from raft cluster", raftAddr)
+
+				future := b.Raft.RemoveServer(server.ID, 0, 0)
+				if err := future.Error(); err != nil {
+					log.Error("handleSerfMemberLeft: remove server [%v] from raft error: %s", server.Address, err)
+					return err
+				}
+			}
+		}
+	}
+	return nil
+}
+
+func (b *Broker) monitorLeadership() {
+
+	// var weAreLeaderCh chan struct{}
+	// var leaderLoop sync.WaitGroup
+	for {
+		log.Info("monitorLeadership loop")
+		select {
+		case isLeader := <-b.RaftNotifyCh:
+			servers, err := b.getRaftServers()
+			log.Info("monitorLeadership isLeader: %v  | servers: %+v %v", isLeader, servers, err)
+
+		}
+	}
+}
@@ -3,8 +3,6 @@ package protocol
 import (
 	log "github.com/CefBoud/monkafka/logging"
 	"github.com/CefBoud/monkafka/serde"
-	"github.com/CefBoud/monkafka/state"
-	"github.com/CefBoud/monkafka/storage"
 	"github.com/CefBoud/monkafka/types"
 )

@@ -61,55 +59,83 @@ type MetadataResponse struct {
 	Topics         []MetadataResponseTopic
 }

-func getMetadataResponse(req types.Request) []byte {
+func (b *Broker) getMetadataResponse(req types.Request) []byte {
 	decoder := serde.NewDecoder(req.Body)
+	metadataRequest := decoder.Decode(&MetadataRequest{}).(*MetadataRequest)
+	log.Debug("metadataRequest %+v", metadataRequest)
+	// log.Debug("metadataRequest isBroker Controller %+v", b.IsController())
+	topics := make([]MetadataResponseTopic, len(metadataRequest.Topics))
+
+	brokers := []MetadataResponseBroker{}
+
+	nodes, err := b.GetClusterNodes()
+	if err != nil {
+		log.Error("Error get cluster nodes. %v", err)
+	}
+
+	controllerID := uint32(MinusOne)
+	for i, n := range nodes {
+		log.Debug("Metadata Node %v : %+v", i, n)
+		brokers = append(brokers, MetadataResponseBroker{
+			NodeID: n.NodeID,
+			Host:   n.Host,
+			Port:   n.Port,
+		})
+		if n.IsController {
+			controllerID = n.NodeID
+		}
+	}
+
+	foundController := controllerID != uint32(MinusOne)
+
 	// https://github.com/apache/kafka/blob/430892654bcf45d644e66b532d83aab0f569cb7d/clients/src/main/resources/common/message/MetadataRequest.json#L26-L27
 	// An empty array indicates "request metadata for no topics," and a null array is used to
 	// indicate "request metadata for all topics."
-	// TODO: we are only handling empty and non empty array case, null array (all topics) is not handled
-	metadataRequest := decoder.Decode(&MetadataRequest{}).(*MetadataRequest)
-	log.Debug("metadataRequest %+v", metadataRequest)
-
-	topics := make([]MetadataResponseTopic, len(metadataRequest.Topics))
-
+	// TODO: we are not handling the null array case ..
+	if len(metadataRequest.Topics) == 0 {
+		for t := range b.FSM.Topics {
+			metadataRequest.Topics = append(metadataRequest.Topics, MetadataRequestTopic{Name: t})
+		}
+	}
 	for _, topic := range metadataRequest.Topics {
-		topic := MetadataResponseTopic{ErrorCode: 0, Name: topic.Name, TopicID: topic.TopicID, IsInternal: false}
+		topic := MetadataResponseTopic{Name: topic.Name, TopicID: topic.TopicID, IsInternal: false}

-		if state.TopicExists(topic.Name) {
-			for partitionIndex := range state.TopicStateInstance[types.TopicName(topic.Name)] {
-				topic.Partitions = append(topic.Partitions, MetadataResponsePartition{
-					PartitionIndex: uint32(partitionIndex),
-					LeaderID:       1,
-					ReplicaNodes:   []uint32{1},
-					IsrNodes:       []uint32{1}})
-			}
-		} else {
-			if metadataRequest.AllowAutoTopicCreation {
-				err := storage.CreateTopic(topic.Name, 1)
-				if err != nil {
-					log.Error("Error creating topic. %v", err)
+		if foundController {
+			log.Debug("fsm.Topics: %+v", b.FSM.Topics)
+			if b.FSM.TopicExists(topic.Name) {
+				for partitionIndex, partition := range b.FSM.Topics[topic.Name].Partitions {
+					topic.Partitions = append(topic.Partitions, MetadataResponsePartition{
+						PartitionIndex: uint32(partitionIndex),
+						LeaderID:       partition.LeaderID,
+						ReplicaNodes:   []uint32{partition.LeaderID},  // TODO fix this
+						IsrNodes:       []uint32{partition.LeaderID}}) // TODO fix this
 				}
 			} else {
-				topic.ErrorCode = uint16(ErrUnknownTopicOrPartition.Code)
+				if metadataRequest.AllowAutoTopicCreation {
+					err := b.CreateTopicPartitions(topic.Name, 1, nil)
+					if err != nil {
+						log.Error("Error creating topic. %v", err)
+					}
+				} else {
+					topic.ErrorCode = uint16(ErrUnknownTopicOrPartition.Code)
+				}
 			}
+		} else {
+			topic.ErrorCode = uint16(ErrLeaderNotAvailable.Code)
 		}

 		topics = append(topics, topic)
 	}
+
 	response := MetadataResponse{
 		ThrottleTimeMs: 0,
-		Brokers: []MetadataResponseBroker{
-			{
-				NodeID: 1,
-				Host:   state.Config.BrokerHost,
-				Port:   state.Config.BrokerPort,
-				Rack:   ""},
-		},
-		ClusterID:    ClusterID,
-		ControllerID: 1,
-		Topics:       topics,
+		Brokers:        brokers,
+		ClusterID:      ClusterID,
+		ControllerID:   controllerID,
+		Topics:         topics,
 	}

+	log.Debug("MetadataResponse %+v", response)
 	encoder := serde.NewEncoder()
 	return encoder.EncodeResponseBytes(req, response)
 }
@@ -58,7 +58,7 @@ type RecordError struct {
 	BatchIndexErrorMessage string // compact_nullable
 }

-func getProduceResponse(req types.Request) []byte {
+func (b *Broker) getProduceResponse(req types.Request) []byte {
 	decoder := serde.NewDecoder(req.Body)
 	produceRequest := decoder.Decode(&ProduceRequest{}).(*ProduceRequest)
 	log.Debug("ProduceRequest %+v", produceRequest)
@@ -1,41 +0,0 @@
-package protocol
-
-import "github.com/CefBoud/monkafka/types"
-
-// https://kafka.apache.org/protocol#protocol_api_keys
-var produceKey = uint16(0)
-var fetchKey = uint16(1)
-var listOffsetsKey = uint16(2)
-var metadataKey = uint16(3)
-var offsetCommitKey = uint16(8)
-var offsetFetchKey = uint16(9)
-var findCoordinatorKey = uint16(10)
-var joinGroupKey = uint16(11)
-var heartbeatKey = uint16(12)
-var syncGroupKey = uint16(14)
-var apiVersionKey = uint16(18)
-var createTopicKey = uint16(19)
-var initProducerIDKey = uint16(22)
-
-// APIDispatcher maps the Request key to its handler
-var APIDispatcher = map[uint16]struct {
-	Name    string
-	Handler func(req types.Request) []byte
-}{
-	produceKey:         {Name: "Produce", Handler: getProduceResponse},
-	fetchKey:           {Name: "Fetch", Handler: getFetchResponse},
-	listOffsetsKey:     {Name: "ListOffsets", Handler: getListOffsetsResponse},
-	metadataKey:        {Name: "Metadata", Handler: getMetadataResponse},
-	offsetCommitKey:    {Name: "OffsetCommit", Handler: getOffsetCommitResponse},
-	offsetFetchKey:     {Name: "OffsetFetch", Handler: getOffsetFetchResponse},
-	findCoordinatorKey: {Name: "FindCoordinator", Handler: getFindCoordinatorResponse},
-	joinGroupKey:       {Name: "JoinGroup", Handler: getJoinGroupResponse},
-	heartbeatKey:       {Name: "Heartbeat", Handler: getHeartbeatResponse},
-	syncGroupKey:       {Name: "SyncGroup", Handler: getSyncGroupResponse},
-	apiVersionKey:      {Name: "APIVersion", Handler: getAPIVersionResponse},
-	createTopicKey:     {Name: "CreateTopic", Handler: getCreateTopicResponse},
-	initProducerIDKey:  {Name: "InitProducerID", Handler: getInitProducerIDResponse},
-}
-
-// ConsumerOffsetsTopic is the topic where consumers committed offsets are saved
-var ConsumerOffsetsTopic = "__consumer-offsets"
@@ -21,7 +21,7 @@ var MinusOne int = -1
 const DefaultNumPartition = 1

 // InitProducerID (Api key = 22)
-func getInitProducerIDResponse(req types.Request) []byte {
+func (b *Broker) getInitProducerIDResponse(req types.Request) []byte {
 	decoder := serde.NewDecoder(req.Body)
 	initProducerIDRequest := decoder.Decode(&InitProducerIDRequest{}).(*InitProducerIDRequest)
 	log.Debug(" initProducerIDRequest %+v", initProducerIDRequest)
@@ -42,21 +42,21 @@ func getInitProducerIDResponse(req types.Request) []byte {
 	return encoder.EncodeResponseBytes(req, response)
 }

-func getFindCoordinatorResponse(req types.Request) []byte {
+func (b *Broker) getFindCoordinatorResponse(req types.Request) []byte {
 	// TODO: get requested coordinator keys
 	response := FindCoordinatorResponse{
 		Coordinators: []FindCoordinatorResponseCoordinator{{
 			Key:    "dummy", //"console-consumer-22229",
 			NodeID: 1,
 			Host:   state.Config.BrokerHost,
-			Port:   state.Config.BrokerPort,
+			Port:   uint32(state.Config.BrokerPort),
 		}},
 	}
 	encoder := serde.NewEncoder()
 	return encoder.EncodeResponseBytes(req, response)
 }

-func getJoinGroupResponse(req types.Request) []byte {
+func (b *Broker) getJoinGroupResponse(req types.Request) []byte {
 	decoder := serde.NewDecoder(req.Body)
 	joinGroupRequest := decoder.Decode(&JoinGroupRequest{}).(*JoinGroupRequest)
 	log.Debug("joinGroupRequest %+v", joinGroupRequest)
@@ -78,13 +78,13 @@ func getJoinGroupResponse(req types.Request) []byte {
 	return encoder.EncodeResponseBytes(req, response)
 }

-func getHeartbeatResponse(req types.Request) []byte {
+func (b *Broker) getHeartbeatResponse(req types.Request) []byte {
 	response := HeartbeatResponse{}
 	encoder := serde.NewEncoder()
 	return encoder.EncodeResponseBytes(req, response)
 }

-func getSyncGroupResponse(req types.Request) []byte {
+func (b *Broker) getSyncGroupResponse(req types.Request) []byte {
 	decoder := serde.NewDecoder(req.Body)
 	syncGroupRequest := decoder.Decode(&SyncGroupRequest{}).(*SyncGroupRequest)

@@ -99,7 +99,7 @@ func getSyncGroupResponse(req types.Request) []byte {
 	return encoder.EncodeResponseBytes(req, response)
 }

-func getOffsetFetchResponse(req types.Request) []byte {
+func (b *Broker) getOffsetFetchResponse(req types.Request) []byte {
 	decoder := serde.NewDecoder(req.Body)
 	offsetFetchRequest := decoder.Decode(&OffsetFetchRequest{}).(*OffsetFetchRequest)
 	log.Debug("offsetFetchRequest %+v", offsetFetchRequest)
@@ -126,7 +126,7 @@ func getOffsetFetchResponse(req types.Request) []byte {
 	return encoder.EncodeResponseBytes(req, response)
 }

-func getOffsetCommitResponse(req types.Request) []byte {
+func (b *Broker) getOffsetCommitResponse(req types.Request) []byte {

 	decoder := serde.NewDecoder(req.Body)
 	offsetCommitRequest := decoder.Decode(&OffsetCommitRequest{}).(*OffsetCommitRequest)
@@ -150,7 +150,7 @@ func getOffsetCommitResponse(req types.Request) []byte {
 	return encoder.EncodeResponseBytes(req, response)
 }

-func getListOffsetsResponse(req types.Request) []byte {
+func (b *Broker) getListOffsetsResponse(req types.Request) []byte {
 	decoder := serde.NewDecoder(req.Body)
 	listOffsetsRequest := decoder.Decode(&ListOffsetsRequest{}).(*ListOffsetsRequest)
 	log.Debug("listOffsetsRequest %+v", listOffsetsRequest)
@@ -0,0 +1,73 @@
+package raft
+
+import (
+	"encoding/json"
+	"fmt"
+
+	log "github.com/CefBoud/monkafka/logging"
+	"github.com/CefBoud/monkafka/types"
+)
+
+// CommandType is a raft log command type
+type CommandType int
+
+// Commands types that can be applied to the raft log to change the state machine
+const (
+	AddNode CommandType = iota
+	RemoveNode
+	AddTopic
+	RemoveTopic
+	AddPartition
+	RemovePartition
+)
+
+// Command represents a command type with its payload
+type Command struct {
+	Kind    CommandType
+	Payload json.RawMessage
+}
+
+// ApplyCommand to append a command to the raft log
+func (kf *FSM) ApplyCommand(cmd Command) error {
+	log.Info("Inside ApplyCommand %v", cmd.Kind)
+	switch cmd.Kind {
+	case AddNode:
+	case RemoveNode:
+	case AddTopic:
+		var topic types.Topic
+		err := json.Unmarshal(cmd.Payload, &topic)
+		if err != nil {
+			return fmt.Errorf("could not parse topic: %s", err)
+		}
+		log.Debug("Raft ApplyCommand AddTopic: %+v", topic)
+		// kf.DB.Store(sp.Key, sp.Value)
+		kf.StoreTopic(topic)
+
+	case AddPartition:
+		var partition types.PartitionState
+		err := json.Unmarshal(cmd.Payload, &partition)
+		if err != nil {
+			return fmt.Errorf("could not parse partition command: %s", err)
+		}
+		log.Debug("Raft ApplyCommand AddPartition: %+v", partition)
+		err = kf.StorePartition(partition)
+		if err != nil {
+			return fmt.Errorf("error applying partition %+v command: %s", partition, err)
+		}
+	default:
+		return fmt.Errorf("unknown command type: %#v", cmd.Kind)
+	}
+	return nil
+}
+
+// EncodeLogEntry converts a raft log entry into bytes
+// TODO: use protobuf or some better encoding
+func EncodeLogEntry(entryType CommandType, entry any) (res []byte, err error) {
+	cmd := Command{Kind: entryType}
+	cmd.Payload, err = json.Marshal(entry)
+	if err != nil {
+		return
+	}
+	res, err = json.Marshal(cmd)
+	return
+}
@@ -0,0 +1,59 @@
+package raft
+
+import (
+	"encoding/json"
+	"fmt"
+
+	"io"
+
+	"github.com/CefBoud/monkafka/logging"
+	"github.com/hashicorp/raft"
+)
+
+// Apply applies a `raft.Log` to the FSM
+func (kf *FSM) Apply(log *raft.Log) any {
+	logging.Info("Inside KV APPLY %v", log.Type)
+	switch log.Type {
+	case raft.LogCommand:
+		var cmd Command
+		err := json.Unmarshal(log.Data, &cmd)
+		logging.Info("case raft.LogCommand Error: %v", err)
+		if err != nil {
+			return fmt.Errorf("could not parse payload: %s", err)
+		}
+		kf.ApplyCommand(cmd)
+	default:
+		return fmt.Errorf("unknown raft log type: %#v", log.Type)
+	}
+
+	return nil
+}
+
+type snapshotNoop struct{}
+
+func (sn snapshotNoop) Persist(_ raft.SnapshotSink) error { return nil }
+func (sn snapshotNoop) Release()                          {}
+
+// Snapshot snapshots the FSM into a struct that implements the raft.FSMSnapshot interface
+func (kf *FSM) Snapshot() (raft.FSMSnapshot, error) {
+	return snapshotNoop{}, nil
+}
+
+// Restore is used to restore an FSM from a snapshot
+func (kf *FSM) Restore(rc io.ReadCloser) error {
+	// deleting first isn't really necessary since there's no exposed DELETE operation anyway.
+	// so any changes over time will just get naturally overwritten
+	decoder := json.NewDecoder(rc)
+	logging.Info("Inside RESTORE")
+	for decoder.More() {
+		var cmd Command
+		err := decoder.Decode(&cmd)
+		fmt.Printf("inside restore, parsed command %+v \n", cmd)
+		if err != nil {
+			return fmt.Errorf("could not decode entry during restore: %s", err)
+		}
+		kf.ApplyCommand(cmd)
+	}
+
+	return rc.Close()
+}
@@ -0,0 +1,86 @@
+package raft
+
+import (
+	"fmt"
+	"sync"
+
+	"github.com/CefBoud/monkafka/logging"
+	"github.com/CefBoud/monkafka/storage"
+	"github.com/CefBoud/monkafka/types"
+)
+
+// FSM is the finite-state-machine of the raft log
+type FSM struct {
+	NodeID uint32
+	Nodes  map[uint32]types.Node
+	Topics map[string]types.Topic
+	sync.RWMutex
+}
+
+// StoreNode stores a node (broker) in the FSM
+func (fsm *FSM) StoreNode(node types.Node) {
+	fsm.Lock()
+	defer fsm.Unlock()
+	fsm.Nodes[node.NodeID] = node
+}
+
+// StoreTopic stores a topic in the FSM
+func (fsm *FSM) StoreTopic(topic types.Topic) {
+	fsm.Lock()
+	defer fsm.Unlock()
+	if _, ok := fsm.Topics[topic.Name]; !ok {
+		fsm.Topics[topic.Name] = types.Topic{Name: topic.Name, Partitions: make(map[uint32]types.PartitionState), Configs: topic.Configs}
+	}
+}
+
+// StorePartition stores a partition in the FSM
+func (fsm *FSM) StorePartition(partition types.PartitionState) error {
+	fsm.Lock()
+	defer fsm.Unlock()
+	if _, ok := fsm.Topics[partition.Topic]; !ok {
+		return fmt.Errorf("topic %v doesn't exist in raft FSM", partition.Topic)
+	}
+	fsm.Topics[partition.Topic].Partitions[partition.PartitionIndex] = partition
+
+	logging.Info("StorePartition partition.LeaderID %v, fsm.NodeID %v", partition.LeaderID, fsm.NodeID)
+	if partition.LeaderID == fsm.NodeID { //|| slices.Contains(partition.ReplicaNodes,  fsm.NodeID)
+		return storage.EnsurePartition(partition.Topic, partition.PartitionIndex)
+	}
+	return nil
+}
+
+// GetNode retrieves a node (broker) from the FSM
+func (fsm *FSM) GetNode(nodeID uint32) (types.Node, bool) {
+	fsm.RLock()
+	defer fsm.RUnlock()
+	node, exists := fsm.Nodes[nodeID]
+	return node, exists
+}
+
+// GetTopic retrieves a topic from the FSM
+func (fsm *FSM) GetTopic(topicName string) (types.Topic, bool) {
+	fsm.RLock()
+	defer fsm.RUnlock()
+	topic, exists := fsm.Topics[topicName]
+	return topic, exists
+}
+
+// GetPartition retrieves a partition from the FSM
+func (fsm *FSM) GetPartition(topicName string, partitionIndex uint32) (types.PartitionState, bool) {
+	fsm.RLock()
+	defer fsm.RUnlock()
+	topic, topicExists := fsm.Topics[topicName]
+	if !topicExists {
+		return types.PartitionState{}, false
+	}
+	partition, partitionExists := topic.Partitions[partitionIndex]
+	return partition, partitionExists
+}
+
+// TopicExists checks if topicName exists in the FSM
+func (fsm *FSM) TopicExists(topicName string) bool {
+	fsm.RLock()
+	defer fsm.RUnlock()
+	_, exists := fsm.Topics[topicName]
+	return exists
+}
@@ -5,7 +5,7 @@ import (
 )

 // Config holds the global configuration.
-var Config types.Configuration
+var Config *types.Configuration

 // TopicStateInstance holds the state of topics.
 var TopicStateInstance types.TopicsState = make(types.TopicsState)
@@ -1,8 +1,10 @@
 package storage

 import (
+	"errors"
 	"fmt"
 	"io"
+	"io/fs"
 	"os"
 	"path/filepath"
 	"strconv"
@@ -234,6 +236,38 @@ func CreateTopic(name string, numPartitions uint32) error {
 	return nil
 }

+// EnsurePartition creates a partition if it doesn't exist
+func EnsurePartition(topicName string, partitionIndex uint32) error {
+	partitionDir := GetPartitionDir(topicName, partitionIndex)
+	_, err := os.Stat(partitionDir)
+	if err == nil {
+		return nil
+	}
+	if errors.Is(err, fs.ErrNotExist) {
+		err := os.MkdirAll(partitionDir, 0750)
+		if err != nil {
+			log.Error("error creating partition directory: %v", err)
+			return err
+		}
+		if !state.TopicExists(topicName) {
+			state.TopicStateInstance[types.TopicName(topicName)] = make(map[types.PartitionIndex]*types.Partition)
+		}
+		partition := &types.Partition{
+			TopicName: topicName,
+			Index:     partitionIndex,
+		}
+		segment, err := NewSegment(partition)
+		if err != nil {
+			log.Error("Error creating segment: %v", err)
+			return err
+		}
+		partition.Segments = append(partition.Segments, segment)
+		state.TopicStateInstance[types.TopicName(topicName)][types.PartitionIndex(partitionIndex)] = partition
+		return nil
+	}
+	return err
+}
+
 // GetPartitionDir returns the directory path for the log of a specific partition.
 func GetPartitionDir(topic string, partition uint32) string {
 	return filepath.Join(state.Config.LogDir, topic+"-"+strconv.Itoa(int(partition)))
@@ -267,7 +301,7 @@ func FlushDataToDisk() {
 }

 // Startup initializes log management system
-func Startup(Config types.Configuration, shutdown chan bool) {
+func Startup(Config *types.Configuration, shutdown chan bool) {
 	state.Config = Config
 	_, err := LoadTopicsState()
 	if err != nil {
@@ -0,0 +1,187 @@
+package cluster
+
+import (
+	"fmt"
+	"log"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/CefBoud/monkafka/logging"
+	broker "github.com/CefBoud/monkafka/protocol"
+	"github.com/CefBoud/monkafka/types"
+	"github.com/hashicorp/serf/serf"
+)
+
+var LogDir = filepath.Join(os.TempDir(), "MonKafkaTestCluster")
+
+var TestConfig = types.Configuration{
+	LogDir:                      LogDir,
+	BrokerHost:                  "localhost",
+	BrokerPort:                  19091,
+	FlushIntervalMs:             5000,
+	NodeID:                      1,
+	Bootstrap:                   true,
+	RaftAddress:                 "localhost:12221",
+	SerfAddress:                 "127.0.0.1:13331",
+	SerfConfig:                  serf.DefaultConfig(),
+	LogRetentionCheckIntervalMs: 1000 * 30,          // 30 sec  //5 * 60 * 1000, // 5 min
+	LogRetentionMs:              3 * 60 * 60 * 1000, // 3h //604800000 (7 days)
+	LogSegmentSizeBytes:         104857600 * 5,      // 500 MiB
+	LogSegmentMs:                1800000,            // 30 min
+
+}
+
+var TestConfig2 = types.Configuration{
+	LogDir:                      LogDir,
+	BrokerHost:                  "localhost",
+	BrokerPort:                  19092,
+	FlushIntervalMs:             5000,
+	NodeID:                      2,
+	Bootstrap:                   false,
+	RaftAddress:                 "localhost:12222",
+	SerfAddress:                 "127.0.0.1:13332",
+	SerfJoinAddress:             "127.0.0.1:13331",
+	SerfConfig:                  serf.DefaultConfig(),
+	LogRetentionCheckIntervalMs: 1000 * 30,          // 30 sec  //5 * 60 * 1000, // 5 min
+	LogRetentionMs:              3 * 60 * 60 * 1000, // 3h //604800000 (7 days)
+	LogSegmentSizeBytes:         104857600 * 5,      // 500 MiB
+	LogSegmentMs:                1800000,            // 30 min
+
+}
+
+var TestConfig3 = types.Configuration{
+	LogDir:                      LogDir,
+	BrokerHost:                  "localhost",
+	BrokerPort:                  19093,
+	FlushIntervalMs:             5000,
+	NodeID:                      3,
+	Bootstrap:                   false,
+	RaftAddress:                 "localhost:12223",
+	SerfAddress:                 "127.0.0.1:13333",
+	SerfJoinAddress:             "127.0.0.1:13331",
+	SerfConfig:                  serf.DefaultConfig(),
+	LogRetentionCheckIntervalMs: 1000 * 30,          // 30 sec  //5 * 60 * 1000, // 5 min
+	LogRetentionMs:              3 * 60 * 60 * 1000, // 3h //604800000 (7 days)
+	LogSegmentSizeBytes:         104857600 * 5,      // 500 MiB
+	LogSegmentMs:                1800000,            // 30 min
+
+}
+
+var BootstrapServers = fmt.Sprintf("%s:%d", TestConfig.BrokerHost, TestConfig.BrokerPort)
+var HomeDir, _ = os.UserHomeDir()
+var KafkaBinDir = HomeDir + "/kafka_2.13-3.9.0/bin/" // assumes kafka is in HomeDir
+
+var topicName = "cluster-test-topic"
+
+func TestMain(m *testing.M) {
+	// Initialization logic
+	logging.SetLogLevel(logging.INFO)
+	log.Println("Setup: Initializing resources")
+	os.RemoveAll(TestConfig.LogDir)
+	v, exists := os.LookupEnv("KAFKA_BIN_DIR")
+	if exists {
+		KafkaBinDir = v
+	}
+	if _, err := os.Stat(KafkaBinDir); err != nil {
+		log.Printf("Ensure Kafka bin dir exists. %v", err)
+		os.Exit(1)
+	}
+	broker1 := broker.NewBroker(&TestConfig)
+	go broker1.Startup()
+	time.Sleep(5 * time.Second)
+	broker2 := broker.NewBroker(&TestConfig2)
+	go broker2.Startup()
+	broker3 := broker.NewBroker(&TestConfig3)
+	go broker3.Startup()
+	// wait for the cluster to settle
+	time.Sleep(3 * time.Second)
+	// Run the tests
+	exitCode := m.Run()
+
+	// Teardown
+	log.Println("Teardown: Cleaning up resources")
+	// broker1.Shutdown()
+	// broker2.Shutdown()
+	// broker3.Shutdown()
+
+	os.Exit(exitCode)
+}
+
+func TestTopicCreation(t *testing.T) {
+	cmd := exec.Command(
+		filepath.Join(KafkaBinDir, "kafka-topics.sh"),
+		"--bootstrap-server", BootstrapServers,
+		"--create",
+		"--topic", topicName,
+		"--partitions", "10",
+	)
+	output, err := cmd.CombinedOutput()
+
+	if err != nil {
+		t.Error(err.Error())
+	}
+	if !strings.Contains(string(output), fmt.Sprintf("Created topic %s.", topicName)) {
+		t.Errorf("Expected output to contain 'Created topic %s.'. Output: %v", topicName, string(output))
+	}
+
+	cmd = exec.Command(
+		filepath.Join(KafkaBinDir, "kafka-topics.sh"),
+		"--bootstrap-server", BootstrapServers,
+		"--describe",
+		"--topic", topicName,
+	)
+	output, err = cmd.CombinedOutput()
+
+	if err != nil {
+		logging.Error("Describe failed .. %v", string(output))
+		t.Error(err.Error())
+	}
+
+	// each node id should lead at least one partition
+	nodeIds := []int{TestConfig.NodeID, TestConfig2.NodeID, TestConfig3.NodeID}
+
+	for _, nodeID := range nodeIds {
+		expected := fmt.Sprintf("Leader: %d", nodeID)
+		if !strings.Contains(string(output), expected) {
+			t.Errorf("Expected output to contain '%s'. Output: %v", expected, string(output))
+		}
+	}
+
+}
+
+func TestProducerAndConsumer(t *testing.T) {
+	nbRecords := "1000000"
+	cmd := exec.Command(
+		filepath.Join(KafkaBinDir, "kafka-producer-perf-test.sh"),
+		"--topic", topicName,
+		"--num-records", nbRecords,
+		"--record-size", "500",
+		"--throughput", "100000",
+		"--producer-props", "acks=1", "batch.size=16384", "linger.ms=5", fmt.Sprintf("bootstrap.servers=%s", BootstrapServers),
+	)
+	_, err := cmd.Output()
+
+	if err != nil {
+		t.Error(err.Error())
+	}
+
+	consumerCmd := exec.Command(
+		filepath.Join(KafkaBinDir, "/kafka-console-consumer.sh"),
+		"--bootstrap-server", BootstrapServers,
+		"--topic", topicName,
+		"--max-messages", nbRecords,
+		"--from-beginning",
+	)
+	output, err := consumerCmd.CombinedOutput()
+
+	if err != nil {
+		t.Error(err.Error())
+	}
+	if !strings.Contains(string(output), fmt.Sprintf("Processed a total of %s messages", nbRecords)) {
+		t.Errorf("Expected consumer output to contain 'Processed a total of %s messages'. Output: %v", nbRecords, string(output))
+	}
+}
@@ -10,17 +10,25 @@ import (
 	"strings"
 	"testing"

-	"github.com/CefBoud/monkafka/broker"
 	"github.com/CefBoud/monkafka/compress"
+	"github.com/CefBoud/monkafka/logging"
+	broker "github.com/CefBoud/monkafka/protocol"
 	"github.com/CefBoud/monkafka/storage"
 	"github.com/CefBoud/monkafka/types"
+	"github.com/hashicorp/serf/serf"
 )

 var TestConfig = types.Configuration{
-	LogDir:                      filepath.Join(os.TempDir(), "MonKafkaTest"),
-	BrokerHost:                  "localhost",
-	BrokerPort:                  19092,
-	FlushIntervalMs:             5000,
+	LogDir:          filepath.Join(os.TempDir(), "MonKafkaTest"),
+	BrokerHost:      "localhost",
+	BrokerPort:      19090,
+	FlushIntervalMs: 5000,
+	NodeID:          1,
+
+	Bootstrap:                   true,
+	RaftAddress:                 "localhost:12220",
+	SerfAddress:                 "127.0.0.1:13330",
+	SerfConfig:                  serf.DefaultConfig(),
 	LogRetentionCheckIntervalMs: 1000 * 30,          // 30 sec  //5 * 60 * 1000, // 5 min
 	LogRetentionMs:              3 * 60 * 60 * 1000, // 3h //604800000 (7 days)
 	LogSegmentSizeBytes:         104857600 * 5,      // 500 MiB
@@ -34,6 +42,7 @@ var KafkaBinDir = HomeDir + "/kafka_2.13-3.9.0/bin/" // assumes kafka is in Home

 func TestMain(m *testing.M) {
 	// Initialization logic
+	logging.SetLogLevel(logging.DEBUG)
 	log.Println("Setup: Initializing resources")
 	os.RemoveAll(TestConfig.LogDir)
 	v, exists := os.LookupEnv("KAFKA_BIN_DIR")
@@ -44,7 +53,8 @@ func TestMain(m *testing.M) {
 		log.Printf("Ensure Kafka bin dir exists. %v", err)
 		os.Exit(1)
 	}
-	broker := broker.NewBroker(TestConfig)
+	broker := broker.NewBroker(&TestConfig)
+
 	go broker.Startup()

 	// Run the tests
@@ -53,9 +63,6 @@ func TestMain(m *testing.M) {
 	// Teardown logic
 	log.Println("Teardown: Cleaning up resources")
 	broker.Shutdown()
-	// log.Printf("deleting LogDir %v", TestConfig.LogDir)
-	// os.RemoveAll(TestConfig.LogDir)
-
 	os.Exit(exitCode)
 }

@@ -1,10 +1,19 @@
 package types

+import "github.com/hashicorp/serf/serf"
+
 // Configuration represents the broker configuration and settings
 type Configuration struct {
+	Bootstrap                   bool // Bootstrap is used to bring up the first broker. It is required so that it can elect a leader without any other nodes
+	NodeID                      int
+	RaftID                      string
+	RaftAddress                 string
+	SerfAddress                 string
+	SerfJoinAddress             string
+	SerfConfig                  *serf.Config
 	LogDir                      string
 	BrokerHost                  string `kafka:"CompactString"`
-	BrokerPort                  uint32
+	BrokerPort                  int
 	FlushIntervalMs             uint64 // flush.ms:time in ms that a message in any topic is kept in memory before flushed to disk.
 	LogRetentionCheckIntervalMs uint64 // check interval to manage log retention
 	LogRetentionMs              uint64 // The number of milliseconds to keep a log file before deleting it
@@ -0,0 +1,29 @@
+package types
+
+// Topic in the state
+type Topic struct {
+	TopicID    [16]byte
+	Name       string
+	Partitions map[uint32]PartitionState
+	Configs    map[string]string
+}
+
+// Node represents a broker
+type Node struct {
+	NodeID       uint32
+	Host         string
+	Port         uint32
+	Rack         string
+	IsController bool
+}
+
+// PartitionState represents a partition in the state
+type PartitionState struct {
+	Topic           string // Topic Name
+	PartitionIndex  uint32
+	LeaderID        uint32
+	LeaderEpoch     uint32
+	ReplicaNodes    []uint32
+	IsrNodes        []uint32
+	OfflineReplicas []uint32
+}
@@ -1,8 +1,20 @@
 package utils

-import "time"
+import (
+	"os"
+	"path/filepath"
+	"time"
+)

 // NowAsUnixMilli returns current time in ms
 func NowAsUnixMilli() uint64 {
 	return uint64(time.Now().UnixNano() / 1e6)
 }
+
+// EnsurePath is used to make sure a path exists
+func EnsurePath(path string, dir bool) error {
+	if !dir {
+		path = filepath.Dir(path)
+	}
+	return os.MkdirAll(path, 0755)
+}