# Deployment Guide

## Prerequisites

- Linux with systemd (Ubuntu 20.04+, CentOS 7+, Debian 10+)
- Network access to Kafka cluster and Lake instance
- Go 1.23+ (for building from source) or pre-built binary from [releases](https://github.com/tidbcloud/lake-ingest-kafka/releases)

## Build

```bash
git clone https://github.com/tidbcloud/lake-ingest-kafka.git
cd bend-ingest-kafka
CGO_ENABLED=1 go build -o bend-ingest-kafka .
```

## Install

```bash
# Create installation directory
sudo mkdir -p /opt/bend-ingest-kafka

# Copy binary
sudo cp bend-ingest-kafka /opt/bend-ingest-kafka/
sudo chmod +x /opt/bend-ingest-kafka/bend-ingest-kafka

# Copy and edit config
sudo cp config/conf.json /opt/bend-ingest-kafka/config.json
sudo vi /opt/bend-ingest-kafka/config.json

# Create dedicated service user
sudo useradd -r -s /sbin/nologin -d /opt/bend-ingest-kafka lake
sudo chown -R lake:lake /opt/bend-ingest-kafka
```

## Configuration

Edit `/opt/bend-ingest-kafka/config.json`:

```json
{
  "kafkaBootstrapServers": "kafka1:9092,kafka2:9092,kafka3:9092",
  "kafkaTopic": "your-topic",
  "KafkaConsumerGroup": "your-consumer-group",
  "isSASL": true,
  "saslUser": "kafka-user",
  "saslPassword": "kafka-password",
  "lakeDSN": "http://user:password@lake-host:8002",
  "lakeTable": "db.target_table",
  "isJsonTransform": true,
  "batchSize": 5000,
  "batchMaxInterval": 30,
  "dataFormat": "json",
  "workers": 2,
  "metricsPort": 2112,
  "enableRebalanceOptimization": true,
  "partitionAssignmentStrategy": "range,roundrobin",
  "topicMetadataRefreshIntervalMs": 60000
}
```

Key parameters to tune for production:
- `batchSize`: larger batches = higher throughput, higher latency (recommended: 5000-10000)
- `batchMaxInterval`: max seconds to wait before flushing an incomplete batch
- `workers`: number of parallel consumers (match Kafka partition count for best throughput)

### Rebalance Optimization

Kafka consumer group rebalancing remains enabled whenever the consumer uses a
topic subscription. By default, bend-ingest-kafka optimizes that behavior to
detect added partitions faster, apply an explicit assignment strategy, and log
assignment and revocation events.

| Parameter | Default | Description |
|-----------|---------|-------------|
| `enableRebalanceOptimization` | `true` | Configure faster metadata refresh, the assignor list, and rebalance logging. Disabling it restores librdkafka defaults but does not disable consumer group rebalancing. |
| `partitionAssignmentStrategy` | `range,roundrobin` | Ordered assignor list matching librdkafka's legacy defaults for rolling-upgrade compatibility. To use incremental cooperative rebalancing, set it to `cooperative-sticky` only after every consumer in the group supports that assignor. |
| `topicMetadataRefreshIntervalMs` | `60000` | How often (in ms) the consumer refreshes topic metadata to detect changes such as newly added partitions. A lower value detects partition expansion faster. |

These can also be set via CLI flags: `-enable-rebalance-optimization`,
`-partition-assignment-strategy`, and `-topic-metadata-refresh-interval-ms`.

When partitions are added to a topic (for example, scaling from 36 to 144),
the running consumers detect the new partitions within
`topicMetadataRefreshIntervalMs` and rejoin the group to take on the new
assignment. Partition assignment and revocation events are written to the logs
(look for `kafka_rebalance`).

## systemd Service

### Install Service

```bash
sudo cp deploy/bend-ingest-kafka.service /etc/systemd/system/
sudo systemctl daemon-reload
sudo systemctl enable bend-ingest-kafka
sudo systemctl start bend-ingest-kafka
```

### Service Management

```bash
# Start
sudo systemctl start bend-ingest-kafka

# Stop (graceful — waits for current batch to finish)
sudo systemctl stop bend-ingest-kafka

# Restart
sudo systemctl restart bend-ingest-kafka

# Check status
sudo systemctl status bend-ingest-kafka

# Reload config (requires restart)
sudo systemctl restart bend-ingest-kafka
```

### Logs

```bash
# Follow logs in real-time
journalctl -u bend-ingest-kafka -f

# View recent logs
journalctl -u bend-ingest-kafka --since "30 minutes ago"

# View logs from last boot
journalctl -u bend-ingest-kafka -b

# Export logs to file
journalctl -u bend-ingest-kafka --since today > /tmp/ingest-kafka.log
```

## Auto-restart & Failure Handling

The systemd service is configured with:

| Setting | Value | Description |
|---------|-------|-------------|
| `Restart` | `on-failure` | Restarts automatically on non-zero exit |
| `RestartSec` | `5` | Wait 5 seconds between restarts |
| `StartLimitBurst` | `5` | Max 5 restarts per interval |
| `StartLimitIntervalSec` | `60` | Interval for burst limit (1 minute) |
| `TimeoutStopSec` | `30` | Allows 30s for graceful shutdown |
| `KillSignal` | `SIGTERM` | Triggers graceful shutdown |

If the service crashes more than 5 times in 60 seconds, systemd stops trying. To reset:

```bash
sudo systemctl reset-failed bend-ingest-kafka
sudo systemctl start bend-ingest-kafka
```

## Monitoring

### Prometheus Metrics

The service exposes metrics at `http://<host>:2112/metrics`.

Add to your Prometheus `scrape_configs`:

```yaml
scrape_configs:
  - job_name: 'bend-ingest-kafka'
    static_configs:
      - targets: ['your-host:2112']
```

### Health Check

A simple health check script for external monitoring:

```bash
#!/bin/bash
# Check if process is running
systemctl is-active --quiet bend-ingest-kafka || exit 1

# Check if metrics endpoint responds
curl -sf http://localhost:2112/metrics > /dev/null || exit 1

exit 0
```

### Alerting Recommendations

| Alert | Condition | Severity |
|-------|-----------|----------|
| Service down | `systemctl is-active` != active | Critical |
| High error rate | `rate(bend_ingest_kafka_ingest_errors_total[5m]) > 0.1` | Warning |
| Ingest stalled | `rate(bend_ingest_kafka_ingest_rows_total[5m]) == 0` | Critical |
| High upload latency | `histogram_quantile(0.95, bend_ingest_kafka_upload_stage_duration_seconds) > 10` | Warning |
| High copy latency | `histogram_quantile(0.95, bend_ingest_kafka_copy_into_duration_seconds) > 10` | Warning |

## Upgrade

```bash
# Build or download new version
CGO_ENABLED=1 go build -o bend-ingest-kafka .

# Replace binary and restart
sudo cp bend-ingest-kafka /opt/bend-ingest-kafka/
sudo systemctl restart bend-ingest-kafka

# Verify
sudo systemctl status bend-ingest-kafka
journalctl -u bend-ingest-kafka --since "1 minute ago"
```

## Troubleshooting

### Service won't start

```bash
# Check logs for error details
journalctl -u bend-ingest-kafka -n 50 --no-pager

# Common issues:
# - Config file not found: check path in ExecStart
# - Permission denied: check file ownership (lake:lake)
# - Port conflict: check metricsPort not in use
```

### Kafka connection issues

```bash
# Test connectivity from the host
nc -z kafka-host 9092 && echo "OK" || echo "UNREACHABLE"

# Check for SASL/TLS misconfiguration in logs
journalctl -u bend-ingest-kafka | grep -i "sasl\|tls\|auth"
```

### Lake connection issues

```bash
# Test Lake health
curl -u user:password http://lake-host:8000/v1/health

# Check DSN format in config
# Correct: "http://user:password@host:8002"
```

### Consumer lag growing

If Kafka consumer lag keeps increasing:
1. Increase `workers` to match partition count
2. Increase `batchSize` to reduce per-batch overhead
3. Check Lake performance (upload/copy latency in metrics)
4. Consider scaling horizontally with multiple instances (each with unique consumer group or partitions)

### New partitions not being consumed

If partitions were added to the topic but the consumer is not reading from them:
1. Confirm `enableRebalanceOptimization` is `true` (the default).
2. Check the logs for `kafka_rebalance` entries; a rebalance should occur within `topicMetadataRefreshIntervalMs` after partitions are added.
3. Lower `topicMetadataRefreshIntervalMs` if faster detection is needed.
4. Increase `workers` so there are enough consumer instances to cover the new partition count.