Takhin Kafka Platform Development

Expert guidance for developing Takhin, a Kafka-compatible streaming platform rewritten in Go.

Project Overview

Takhin is a Kafka-compatible streaming platform with two main components:

This is a backend-focused monorepo in Go with frontend React code at `projects/console/frontend/`.

Architecture & Component Structure

Core Services Layout

```

backend/pkg/

├── kafka/ # Kafka protocol implementation

│ ├── server/ # TCP server handling Kafka connections

│ ├── handler/ # Request handlers (produce, fetch, metadata)

│ └── protocol/ # Binary protocol encoding/decoding

├── storage/ # Storage layer

│ ├── topic/ # Topic and partition management

│ └── log/ # Log segment storage

├── console/ # REST API server (Chi router)

├── coordinator/ # Consumer group coordination

├── raft/ # Raft consensus (no ZooKeeper)

├── config/ # Koanf-based configuration (YAML + env)

├── logger/ # Structured logging (slog)

└── metrics/ # Prometheus metrics

```

Component Interaction Pattern

1. **Kafka Handler** receives binary protocol requests → decodes using `pkg/kafka/protocol`

2. **Handler** validates and routes to backend (implements `Backend` interface)

3. **Backend** interacts with `topic.Manager` for storage operations

4. **Topic Manager** manages partitions as `log.Log` instances

5. **Console Server** provides HTTP REST API wrapping similar operations

Reference: `backend/pkg/kafka/handler/handler.go` shows this pattern - handlers instantiate with `config.Config` and `topic.Manager`, use protocol decoders/encoders for wire format.

Configuration System

Uses Koanf with layered approach:

1. Load from `configs/takhin.yaml`

2. Override with `TAKHIN_` prefixed env vars (e.g., `TAKHIN_SERVER_PORT`)

3. Validate and set defaults in `backend/pkg/config/config.go`

Example: `TAKHIN_STORAGE_DATA_DIR=/data` overrides `storage.data.dir` in YAML.

Development Workflows

Building & Running

Use Taskfile (not Makefile):

```bash

task backend:build # Builds to build/takhin

task backend:run # Runs with configs/takhin.yaml

task backend:test # Run all tests with race detector

task backend:test:unit # Run only unit tests

task backend:lint # golangci-lint with timeout

Manual runs for debugging

cd backend

go run ./cmd/takhin -config configs/takhin.yaml

go run ./cmd/console -data-dir /tmp/data -api-addr :8080

```

Testing Patterns

Follow these standards:

Example test structure:

```go

func TestSomething(t *testing.T) {

cfg := &config.Config{Storage: config.StorageConfig{DataDir: t.TempDir()}}

mgr := topic.NewManager(cfg.Storage.DataDir, 1024*1024)

handler := handler.New(cfg, mgr)

// ... test logic

}

```

Swagger API Documentation

Console API uses Swag for OpenAPI docs:

Project Conventions

Error Handling

Logging

Protocol Implementation

Kafka binary protocol is complex:

1. Check API version in request header

2. Use protocol encoder/decoder functions in `pkg/kafka/protocol/`

3. Return proper error codes (see `protocol.ErrorCode*` constants)

4. Write tests with actual binary data when possible

Example: `backend/pkg/kafka/handler/alter_configs_test.go` shows full encode → handle → decode cycle.

Authentication (Console API)

Common Implementation Tasks

Adding a New Kafka API Handler

1. Add request/response structs to `pkg/kafka/protocol/`

2. Implement encoder/decoder functions

3. Add handler method to `pkg/kafka/handler/handler.go`

4. Register in `Handle()` switch statement

5. Write table-driven tests in `*_test.go`

Adding Console REST Endpoint

1. Add handler method to `pkg/console/server.go`

2. Add Swag comment above handler (see existing examples)

3. Register route in `setupRoutes()` method

4. Return JSON using `respondJSON()` helper

5. Update Swagger docs with `swag init`

Configuration Changes

1. Add field to struct in `pkg/config/config.go`

2. Add to YAML schema with koanf tags

3. Update `setDefaults()` and `validate()` functions

4. Add to `configs/takhin.yaml` with comments

Storage Operations

Code Quality Standards

Go Conventions (Critical)

Git Workflow

- Examples: `feat(kafka): add metadata v9 support`, `fix(console): auth middleware bypass`

Documentation Location

All docs in `docs/` directory:

Known Gotchas

References

Instructions for AI Agent

When working on Takhin:

1. **Understand the context**: Check which component you're working in (kafka/, storage/, console/, etc.) and review the relevant architecture in `docs/architecture/`

2. **Follow the patterns**: Use existing code as reference. Handler tests follow table-driven patterns, console endpoints use Swag annotations, storage operations go through `topic.Manager`

3. **Maintain quality**: Run `task backend:lint` and `task backend:test` before proposing changes. Ensure 80%+ test coverage on new code.

4. **Configuration changes**: Update both the struct in `config.go` and the YAML example in `configs/takhin.yaml`. Document env var overrides.

5. **Protocol work**: When implementing Kafka APIs, always check the protocol version, use the encoder/decoder functions, and write tests with binary data.

6. **Documentation**: Update relevant docs in `docs/` when adding features or changing architecture.

7. **Git commits**: Use Conventional Commits format: `<type>(<scope>): <subject>`