#!/bin/bash
#===============================================================================
# install-kafka-kraft.sh
# Deploy Confluent Kafka Stack on Kubernetes (KRaft mode - No Zookeeper)
# Components: Kafka 3-node cluster, Schema Registry, Control Center
#===============================================================================
set -euo pipefail
NAMESPACE="kafka"
GREEN='\033[0;32m'
NC='\033[0m'
info() { echo -e "${GREEN}[INFO]${NC} $*"; }
#===============================================================================
# Step 1: Create Namespace
#===============================================================================
info "Creating namespace '${NAMESPACE}'..."
kubectl create namespace ${NAMESPACE} --dry-run=client -o yaml | kubectl apply -f -
#===============================================================================
# Step 2: Generate Cluster ID
#===============================================================================
CLUSTER_ID=$(docker run --rm confluentinc/cp-kafka:7.6.0 kafka-storage random-uuid 2>/dev/null || cat /proc/sys/kernel/random/uuid | base64 | head -c 22)
info "Generated Kafka Cluster ID: ${CLUSTER_ID}"
#===============================================================================
# Step 3: Deploy Kafka 3-node cluster in KRaft Mode
#===============================================================================
info "Deploying Kafka 3-node cluster (KRaft mode - no Zookeeper)..."
cat > /tmp/kafka.yaml <<OUTER
---
apiVersion: v1
kind: Service
metadata:
  name: kafka
  labels:
    app: kafka
spec:
  ports:
    - port: 9092
      name: internal
    - port: 29092
      name: external
    - port: 9093
      name: controller
  clusterIP: None
  selector:
    app: kafka
---
apiVersion: v1
kind: Service
metadata:
  name: kafka-external
  labels:
    app: kafka
spec:
  type: LoadBalancer
  ports:
    - port: 9092
      targetPort: 9092
      name: external
  selector:
    app: kafka
---
apiVersion: apps/v1
kind: StatefulSet
metadata:
  name: kafka
spec:
  serviceName: kafka
  replicas: 3
  podManagementPolicy: Parallel
  selector:
    matchLabels:
      app: kafka
  template:
    metadata:
      labels:
        app: kafka
    spec:
      securityContext:
        fsGroup: 1000
      initContainers:
        - name: fix-permissions
          image: busybox:1.35
          command: ["sh", "-c", "rm -rf /var/lib/kafka/data/lost+found && chown -R 1000:1000 /var/lib/kafka/data"]
          volumeMounts:
            - name: kafka-data
              mountPath: /var/lib/kafka/data
      containers:
        - name: kafka
          image: confluentinc/cp-kafka:7.6.0
          ports:
            - containerPort: 9092
            - containerPort: 29092
            - containerPort: 9093
          env:
            - name: CLUSTER_ID
              value: "${CLUSTER_ID}"
            - name: KAFKA_PROCESS_ROLES
              value: "broker,controller"
            - name: KAFKA_CONTROLLER_QUORUM_VOTERS
              value: "0@kafka-0.kafka:9093,1@kafka-1.kafka:9093,2@kafka-2.kafka:9093"
            - name: KAFKA_LISTENER_SECURITY_PROTOCOL_MAP
              value: "PLAINTEXT:PLAINTEXT,CONTROLLER:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT"
            - name: KAFKA_CONTROLLER_LISTENER_NAMES
              value: "CONTROLLER"
            - name: KAFKA_INTER_BROKER_LISTENER_NAME
              value: "PLAINTEXT"
            - name: KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR
              value: "3"
            - name: KAFKA_TRANSACTION_STATE_LOG_MIN_ISR
              value: "2"
            - name: KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR
              value: "3"
            - name: KAFKA_DEFAULT_REPLICATION_FACTOR
              value: "3"
            - name: KAFKA_MIN_INSYNC_REPLICAS
              value: "2"
            - name: KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS
              value: "0"
            - name: KAFKA_AUTO_CREATE_TOPICS_ENABLE
              value: "true"
            - name: KAFKA_LOG_RETENTION_HOURS
              value: "168"
            - name: KAFKA_LOG_DIRS
              value: "/var/lib/kafka/data"
          command:
            - /bin/bash
            - -c
            - |
              # Derive NODE_ID from hostname (kafka-0 -> 0, kafka-1 -> 1, kafka-2 -> 2)
              export KAFKA_NODE_ID=\${HOSTNAME##*-}
              export KAFKA_LISTENERS="PLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093,PLAINTEXT_HOST://0.0.0.0:29092"
              export KAFKA_ADVERTISED_LISTENERS="PLAINTEXT://\${HOSTNAME}.kafka:9092,PLAINTEXT_HOST://\${HOSTNAME}.kafka:29092"
              exec /etc/confluent/docker/run
          resources:
            requests:
              memory: "1Gi"
              cpu: "500m"
            limits:
              memory: "2Gi"
              cpu: "1000m"
          volumeMounts:
            - name: kafka-data
              mountPath: /var/lib/kafka/data
  volumeClaimTemplates:
    - metadata:
        name: kafka-data
      spec:
        accessModes: ["ReadWriteOnce"]
        resources:
          requests:
            storage: 5Gi
OUTER
kubectl apply -n ${NAMESPACE} -f /tmp/kafka.yaml
info "Waiting for Kafka to be ready..."
kubectl rollout status statefulset/kafka -n ${NAMESPACE} --timeout=300s
#===============================================================================
# Step 4: Deploy Schema Registry
#===============================================================================
info "Deploying Schema Registry..."
cat > /tmp/schema-registry.yaml <<'EOF'
---
apiVersion: v1
kind: Service
metadata:
  name: schema-registry
  labels:
    app: schema-registry
spec:
  ports:
    - port: 8081
      name: http
  selector:
    app: schema-registry
---
apiVersion: apps/v1
kind: Deployment
metadata:
  name: schema-registry
spec:
  replicas: 2
  selector:
    matchLabels:
      app: schema-registry
  template:
    metadata:
      labels:
        app: schema-registry
    spec:
      enableServiceLinks: false
      containers:
        - name: schema-registry
          image: confluentinc/cp-schema-registry:7.6.0
          ports:
            - containerPort: 8081
          env:
            - name: PORT
              value: "8081"
            - name: SCHEMA_REGISTRY_HOST_NAME
              value: "schema-registry"
            - name: SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS
              value: "kafka-0.kafka:9092,kafka-1.kafka:9092,kafka-2.kafka:9092"
            - name: SCHEMA_REGISTRY_LISTENERS
              value: "http://0.0.0.0:8081"
            - name: SCHEMA_REGISTRY_KAFKASTORE_TOPIC
              value: "_schemas"
            - name: SCHEMA_REGISTRY_DEBUG
              value: "true"
          resources:
            requests:
              memory: "512Mi"
              cpu: "250m"
            limits:
              memory: "1Gi"
              cpu: "500m"
          readinessProbe:
            httpGet:
              path: /
              port: 8081
            initialDelaySeconds: 60
            periodSeconds: 10
            failureThreshold: 10
          livenessProbe:
            httpGet:
              path: /
              port: 8081
            initialDelaySeconds: 120
            periodSeconds: 15
            failureThreshold: 10
EOF
kubectl apply -n ${NAMESPACE} -f /tmp/schema-registry.yaml
info "Waiting for Schema Registry to be ready..."
kubectl rollout status deployment/schema-registry -n ${NAMESPACE} --timeout=120s
#===============================================================================
# Step 5: Deploy Control Center
#===============================================================================
info "Deploying Control Center..."
cat > /tmp/control-center.yaml <<'EOF'
---
apiVersion: v1
kind: Service
metadata:
  name: control-center
  labels:
    app: control-center
spec:
  type: LoadBalancer
  ports:
    - port: 9021
      targetPort: 9021
      name: http
  selector:
    app: control-center
---
apiVersion: apps/v1
kind: Deployment
metadata:
  name: control-center
spec:
  replicas: 1
  selector:
    matchLabels:
      app: control-center
  template:
    metadata:
      labels:
        app: control-center
    spec:
      enableServiceLinks: false
      containers:
        - name: control-center
          image: confluentinc/cp-enterprise-control-center:7.6.0
          ports:
            - containerPort: 9021
          env:
            - name: CONTROL_CENTER_BOOTSTRAP_SERVERS
              value: "kafka-0.kafka:9092,kafka-1.kafka:9092,kafka-2.kafka:9092"
            - name: CONTROL_CENTER_SCHEMA_REGISTRY_URL
              value: "http://schema-registry:8081"
            - name: CONTROL_CENTER_REPLICATION_FACTOR
              value: "3"
            - name: CONTROL_CENTER_INTERNAL_TOPICS_PARTITIONS
              value: "3"
            - name: CONTROL_CENTER_MONITORING_INTERCEPTOR_TOPIC_PARTITIONS
              value: "3"
            - name: CONFLUENT_METRICS_TOPIC_REPLICATION
              value: "3"
            - name: PORT
              value: "9021"
          resources:
            requests:
              memory: "2Gi"
              cpu: "1000m"
            limits:
              memory: "4Gi"
              cpu: "2000m"
          readinessProbe:
            httpGet:
              path: /
              port: 9021
            initialDelaySeconds: 60
            periodSeconds: 10
            failureThreshold: 10
          livenessProbe:
            httpGet:
              path: /
              port: 9021
            initialDelaySeconds: 120
            periodSeconds: 15
            failureThreshold: 10
EOF
kubectl apply -n ${NAMESPACE} -f /tmp/control-center.yaml
info "Waiting for Control Center to be ready..."
kubectl rollout status deployment/control-center -n ${NAMESPACE} --timeout=300s
#===============================================================================
# Step 6: Cleanup temp files
#===============================================================================
rm -f /tmp/kafka.yaml /tmp/schema-registry.yaml /tmp/control-center.yaml
#===============================================================================
# Step 7: Summary
#===============================================================================
info "============================================"
info "Kafka Stack (KRaft) Deployed Successfully!"
info "============================================"
echo ""
echo "  Mode:             KRaft (No Zookeeper)"
echo "  Cluster ID:       ${CLUSTER_ID}"
echo "  Replicas:         3"
echo "  Namespace:        ${NAMESPACE}"
echo "  Kafka Brokers:    kafka-0.kafka:9092, kafka-1.kafka:9092, kafka-2.kafka:9092"
echo "  Schema Registry:  schema-registry.${NAMESPACE}:8081"
echo ""
info "Get external IPs:"
echo "  kubectl get svc -n ${NAMESPACE}"
echo ""
info "Check all pods:"
echo "  kubectl get pods -n ${NAMESPACE}"
echo ""
info "Test Kafka:"
echo "  kubectl exec -n ${NAMESPACE} kafka-0 -- kafka-topics --bootstrap-server localhost:9092 --list"
echo "  kubectl exec -n ${NAMESPACE} kafka-0 -- kafka-topics --bootstrap-server localhost:9092 --create --topic test --partitions 3 --replication-factor 3"
echo "  kubectl exec -n ${NAMESPACE} kafka-0 -- kafka-metadata --snapshot /var/lib/kafka/data/__cluster_metadata-0/00000000000000000000.log --cluster-id ${CLUSTER_ID}"