brew install --cask docker

# Update package index
sudo apt-get update

# Install required packages
sudo apt-get install -y \
    apt-transport-https \
    ca-certificates \
    curl \
    gnupg \
    lsb-release

# Add Docker's official GPG key
curl -fsSL https://download.docker.com/linux/ubuntu/gpg | sudo gpg --dearmor -o /usr/share/keyrings/docker-archive-keyring.gpg

# Set up the stable repository
echo \
  "deb [arch=amd64 signed-by=/usr/share/keyrings/docker-archive-keyring.gpg] https://download.docker.com/linux/ubuntu \
  $(lsb_release -cs) stable" | sudo tee /etc/apt/sources.list.d/docker.list > /dev/null

# Install Docker Engine
sudo apt-get update
sudo apt-get install -y docker-ce docker-ce-cli containerd.io

# Add your user to the docker group
sudo usermod -aG docker $USER

docker --version
docker run hello-world

brew install kubectl

# Download the latest release
curl -LO "https://dl.k8s.io/release/$(curl -L -s https://dl.k8s.io/release/stable.txt)/bin/linux/amd64/kubectl"

# Make it executable
chmod +x kubectl

# Move to PATH
sudo mv kubectl /usr/local/bin/

choco install kubernetes-cli

kubectl version --client

# For Linux
curl -Lo ./kind https://kind.sigs.k8s.io/dl/v0.30.0/kind-linux-amd64
chmod +x ./kind
sudo mv ./kind /usr/local/bin/kind

# For macOS
curl -Lo ./kind https://kind.sigs.k8s.io/dl/v0.30.0/kind-darwin-amd64
chmod +x ./kind
sudo mv ./kind /usr/local/bin/kind

# For Windows (in PowerShell)
curl.exe -Lo kind-windows-amd64.exe https://kind.sigs.k8s.io/dl/v0.30.0/kind-windows-amd64
Move-Item .\kind-windows-amd64.exe c:\some-dir-in-your-PATH\kind.exe

brew install kind

kind version

brew install helm

# Download and install
curl https://raw.githubusercontent.com/helm/helm/main/scripts/get-helm-3 | bash

choco install kubernetes-helm

helm version

brew install git

sudo apt-get install -y git

git --version

mkdir -p ~/kafka-k8s-tutorial
cd ~/kafka-k8s-tutorial

# Export variables for the session
export TUTORIAL_DIR=~/kafka-k8s-tutorial
export KAFKA_NAMESPACE=kafka
export ZOOKEEPER_NAMESPACE=zookeeper
export MONITORING_NAMESPACE=default

# Make them persistent (add to your shell profile)
echo "export TUTORIAL_DIR=~/kafka-k8s-tutorial" >> ~/.bashrc
echo "export KAFKA_NAMESPACE=kafka" >> ~/.bashrc
echo "export ZOOKEEPER_NAMESPACE=zookeeper" >> ~/.bashrc
echo "export MONITORING_NAMESPACE=default" >> ~/.bashrc

# Reload your shell or source the file
source ~/.bashrc

# Check Docker system info
docker system info

# Check available resources
docker system df

cd $TUTORIAL_DIR
git clone https://github.com/amuraru/k8s-kafka-the-hard-way.git
cd k8s-kafka-the-hard-way

# Check Docker
echo "Docker version:"
docker --version
echo ""

# Check kubectl
echo "kubectl version:"
kubectl version --client
echo ""

# Check kind
echo "kind version:"
kind version
echo ""

# Check Helm
echo "Helm version:"
helm version
echo ""

# Check Git
echo "Git version:"
git --version
echo ""

# Check working directory
echo "Working directory:"
ls -la $TUTORIAL_DIR

# Add your user to the docker group
sudo usermod -aG docker $USER

# Log out and log back in, or run:
newgrp docker

# Check if kubectl is in your PATH
which kubectl

# If not found, ensure /usr/local/bin is in your PATH
echo $PATH

# Add to PATH if needed
export PATH=$PATH:/usr/local/bin

# Check Docker is running
docker ps

# Check available disk space
df -h

# Check Docker resources in Docker Desktop settings

cd $TUTORIAL_DIR

# Create kind configuration directory
mkdir -p ~/.kind

# Create the cluster configuration
cat > ~/.kind/kind-config.yaml <<EOF
kind: Cluster
apiVersion: kind.x-k8s.io/v1alpha4
nodes:
  - role: control-plane
    kubeadmConfigPatches:
    - |
      kind: InitConfiguration
      nodeRegistration:
        kubeletExtraArgs:
          node-labels: "ingress-ready=true"
  - role: worker
  - role: worker
  - role: worker
  - role: worker
  - role: worker
  - role: worker
containerdConfigPatches:
- |-
  [plugins."io.containerd.grpc.v1.cri".containerd]
    snapshotter = "overlayfs"
  [plugins."io.containerd.grpc.v1.cri".registry.mirrors."localhost:5000"]
    endpoint = ["http://localhost:5000"]
EOF

# Create the cluster (this may take 5-10 minutes)
kind create cluster \
  --name kafka \
  --config ~/.kind/kind-config.yaml \
  --image kindest/node:v1.33.4

# Wait for cluster to be ready
echo "Waiting for cluster to be ready..."
kubectl wait --for=condition=Ready nodes --all --timeout=300s

# Check cluster info
kubectl cluster-info --context kind-kafka

# List all nodes
kubectl get nodes -o wide

# Check node status
kubectl get nodes --show-labels

# Label all worker nodes with region
kubectl label nodes \
  kafka-worker \
  kafka-worker2 \
  kafka-worker3 \
  kafka-worker4 \
  kafka-worker5 \
  kafka-worker6 \
  topology.kubernetes.io/region=region1

# AZ1: kafka-worker and kafka-worker2
kubectl label nodes kafka-worker kafka-worker2 \
  topology.kubernetes.io/zone=az1

# AZ2: kafka-worker3 and kafka-worker4
kubectl label nodes kafka-worker3 kafka-worker4 \
  topology.kubernetes.io/zone=az2

# AZ3: kafka-worker5 and kafka-worker6
kubectl label nodes kafka-worker5 kafka-worker6 \
  topology.kubernetes.io/zone=az3

# Display nodes with region and zone labels
kubectl get nodes \
  --label-columns=topology.kubernetes.io/region,topology.kubernetes.io/zone

# Show detailed node information
kubectl describe nodes | grep -E "Name:|topology.kubernetes.io"

# Set the current context to the kind cluster
kubectl config use-context kind-kafka

# Verify current context
kubectl config current-context

# Test cluster access
kubectl get namespaces

# Check node resources
kubectl top nodes 2>/dev/null || echo "Metrics server not yet available"

# Check cluster capacity
kubectl describe nodes | grep -A 5 "Capacity:"

# Check storage classes
kubectl get storageclass

# Check default namespace
kubectl get all

# Delete the failed cluster
kind delete cluster --name kafka

# Check Docker resources
docker system df
docker system prune -f

# Retry cluster creation
kind create cluster --name kafka --config ~/.kind/kind-config.yaml --image kindest/node:v1.33.4

# Check node status
kubectl describe nodes

# Check system pods
kubectl get pods -n kube-system

# Check kubelet logs (from Docker)
docker logs kafka-worker

# List available contexts
kubectl config get-contexts

# Set the correct context
kubectl config use-context kind-kafka

# Verify
kubectl config current-context

# Delete the cluster
kind delete cluster --name kafka

# Verify deletion
kind get clusters

# Remove configuration
rm ~/.kind/kind-config.yaml

# Install cert-manager CRDs
kubectl create --validate=false -f https://github.com/cert-manager/cert-manager/releases/download/v1.18.2/cert-manager.crds.yaml

# Create namespace for cert-manager
kubectl create namespace cert-manager

# Add cert-manager Helm repository
helm repo add cert-manager https://charts.jetstack.io
helm repo update

# Install cert-manager
helm install cert-manager cert-manager/cert-manager \
  --namespace cert-manager \
  --version v1.18.2

# Wait for cert-manager to be ready
kubectl wait --for=condition=Available deployment --all -n cert-manager --timeout=300s

# Check cert-manager pods
kubectl get pods -n cert-manager

# Check cert-manager services
kubectl get svc -n cert-manager

# Verify cert-manager is working
kubectl get certificates -A

# Create namespace for ZooKeeper
kubectl create namespace zookeeper

# Install ZooKeeper CRDs
kubectl create -f https://raw.githubusercontent.com/adobe/zookeeper-operator/master/config/crd/bases/zookeeper.pravega.io_zookeeperclusters.yaml

# Clone the ZooKeeper operator repository
cd $TUTORIAL_DIR
rm -rf /tmp/zookeeper-operator
git clone --single-branch --branch master https://github.com/adobe/zookeeper-operator /tmp/zookeeper-operator
cd /tmp/zookeeper-operator

# Install ZooKeeper operator
helm template zookeeper-operator \
  --namespace=zookeeper \
  --set crd.create=false \
  --set image.repository='adobe/zookeeper-operator' \
  --set image.tag='0.2.15-adobe-20250914' \
  ./charts/zookeeper-operator | kubectl create -n zookeeper -f -

# Wait for operator to be ready
kubectl wait --for=condition=Available deployment --all -n zookeeper --timeout=300s

# Create ZooKeeper cluster
kubectl create --namespace zookeeper -f - <<EOF
apiVersion: zookeeper.pravega.io/v1beta1
kind: ZookeeperCluster
metadata:
  name: zk
  namespace: zookeeper
spec:
  replicas: 3
  image:
    repository: adobe/zookeeper
    tag: 3.8.4-0.2.15-adobe-20250914
    pullPolicy: IfNotPresent
  config:
    initLimit: 10
    tickTime: 2000
    syncLimit: 5
  probes:
    livenessProbe:
      initialDelaySeconds: 41
  persistence:
    reclaimPolicy: Delete
    spec:
      accessModes:
      - ReadWriteOnce
      resources:
        requests:
          storage: 20Gi
EOF

# Check ZooKeeper cluster status
kubectl get zookeepercluster -n zookeeper -o wide

# Watch ZooKeeper cluster creation
kubectl get pods -n zookeeper -w
# Press Ctrl+C to stop watching when all pods are running

# Check ZooKeeper services
kubectl get svc -n zookeeper

# Verify ZooKeeper cluster is ready
kubectl wait --for=condition=Ready pod --all -n zookeeper --timeout=600s

# Install Prometheus Operator CRDs
kubectl create -f https://raw.githubusercontent.com/prometheus-operator/prometheus-operator/master/example/prometheus-operator-crd/monitoring.coreos.com_alertmanagers.yaml
kubectl create -f https://raw.githubusercontent.com/prometheus-operator/prometheus-operator/master/example/prometheus-operator-crd/monitoring.coreos.com_alertmanagerconfigs.yaml
kubectl create -f https://raw.githubusercontent.com/prometheus-operator/prometheus-operator/master/example/prometheus-operator-crd/monitoring.coreos.com_prometheuses.yaml
kubectl create -f https://raw.githubusercontent.com/prometheus-operator/prometheus-operator/master/example/prometheus-operator-crd/monitoring.coreos.com_prometheusrules.yaml
kubectl create -f https://raw.githubusercontent.com/prometheus-operator/prometheus-operator/master/example/prometheus-operator-crd/monitoring.coreos.com_servicemonitors.yaml
kubectl create -f https://raw.githubusercontent.com/prometheus-operator/prometheus-operator/master/example/prometheus-operator-crd/monitoring.coreos.com_podmonitors.yaml
kubectl create -f https://raw.githubusercontent.com/prometheus-operator/prometheus-operator/master/example/prometheus-operator-crd/monitoring.coreos.com_thanosrulers.yaml
kubectl create -f https://raw.githubusercontent.com/prometheus-operator/prometheus-operator/master/example/prometheus-operator-crd/monitoring.coreos.com_probes.yaml

# Add Prometheus community Helm repository
helm repo add prometheus-community https://prometheus-community.github.io/helm-charts
helm repo update

# Install kube-prometheus-stack (includes Prometheus, Grafana, and AlertManager)
helm install monitoring \
  --namespace=default \
  prometheus-community/kube-prometheus-stack \
  --set prometheusOperator.createCustomResource=false

# Wait for monitoring stack to be ready
kubectl wait --for=condition=Available deployment --all -n default --timeout=600s

# Check monitoring pods
kubectl get pods -l release=monitoring

# Check monitoring services
kubectl get svc -l release=monitoring

# Check Prometheus targets (optional)
kubectl get prometheus -o wide

# Get Grafana admin password
kubectl get secret --namespace default monitoring-grafana \
  -o jsonpath="{.data.admin-password}" | base64 --decode
echo ""

# Prometheus (in a separate terminal)
kubectl --namespace default port-forward svc/monitoring-kube-prometheus-prometheus 9090 &

# Grafana (in a separate terminal)
kubectl --namespace default port-forward svc/monitoring-grafana 3000:80 &

# AlertManager (in a separate terminal)
kubectl --namespace default port-forward svc/monitoring-kube-prometheus-alertmanager 9093 &

echo "=== cert-manager ==="
kubectl get pods -n cert-manager

echo -e "\n=== ZooKeeper ==="
kubectl get pods -n zookeeper
kubectl get zookeepercluster -n zookeeper

echo -e "\n=== Monitoring ==="
kubectl get pods -l release=monitoring

echo -e "\n=== All Namespaces ==="
kubectl get namespaces

# Check cert-manager logs
kubectl logs -n cert-manager deployment/cert-manager

# Check webhook connectivity
kubectl get validatingwebhookconfigurations

# Check ZooKeeper operator logs
kubectl logs -n zookeeper deployment/zookeeper-operator

# Check ZooKeeper cluster events
kubectl describe zookeepercluster zk -n zookeeper

# Check Prometheus operator logs
kubectl logs -l app.kubernetes.io/name=prometheus-operator

# Check Prometheus configuration
kubectl get prometheus -o yaml

# Create namespace for Kafka
kubectl create namespace kafka

# Verify namespace creation
kubectl get namespaces | grep kafka

# Install KafkaCluster CRD
kubectl apply --server-side -f https://raw.githubusercontent.com/adobe/koperator/refs/heads/master/config/base/crds/kafka.banzaicloud.io_kafkaclusters.yaml

# Install KafkaTopic CRD
kubectl apply --server-side -f https://raw.githubusercontent.com/adobe/koperator/refs/heads/master/config/base/crds/kafka.banzaicloud.io_kafkatopics.yaml

# Install KafkaUser CRD
kubectl apply --server-side -f https://raw.githubusercontent.com/adobe/koperator/refs/heads/master/config/base/crds/kafka.banzaicloud.io_kafkausers.yaml

# Install CruiseControlOperation CRD
kubectl apply --server-side -f https://raw.githubusercontent.com/adobe/koperator/refs/heads/master/config/base/crds/kafka.banzaicloud.io_cruisecontroloperations.yaml

# Check that all CRDs are installed
kubectl get crd | grep kafka.banzaicloud.io

# Get detailed information about the CRDs
kubectl get crd kafkaclusters.kafka.banzaicloud.io -o yaml | head -20

# Install Koperator using OCI Helm chart
helm install kafka-operator oci://ghcr.io/adobe/helm-charts/kafka-operator \
  --namespace=kafka \
  --set webhook.enabled=false \
  --version 0.28.0-adobe-20250923

# Wait for the operator to be ready
kubectl wait --for=condition=Available deployment --all -n kafka --timeout=300s

# Check Koperator pods
kubectl get pods -n kafka

# Check pod details
kubectl describe pods -n kafka

# Check operator logs
kubectl logs -l app.kubernetes.io/instance=kafka-operator -c manager -n kafka

# Check services in kafka namespace
kubectl get svc -n kafka

# Check operator deployment
kubectl get deployment -n kafka

# Check if the operator is watching for KafkaCluster resources
kubectl get kafkaclusters -n kafka

# Check operator configuration
kubectl get deployment kafka-operator-operator -n kafka -o yaml | grep -A 10 -B 10 image:

# Check RBAC resources
kubectl get serviceaccount -n kafka
kubectl get clusterrole | grep kafka-operator
kubectl get rolebinding -n kafka

# Check operator deployment configuration
kubectl get deployment kafka-operator-operator -n kafka -o yaml

# Check operator environment variables
kubectl get deployment kafka-operator-operator -n kafka -o jsonpath='{.spec.template.spec.containers[0].env}' | jq .

# Check pod events
kubectl describe pod -l app.kubernetes.io/instance=kafka-operator -n kafka

# Check operator logs
kubectl logs -l app.kubernetes.io/instance=kafka-operator -c manager -n kafka --previous

# Check resource constraints
kubectl top pod -n kafka

# Reinstall CRDs
kubectl delete crd kafkaclusters.kafka.banzaicloud.io
kubectl apply --server-side -f https://raw.githubusercontent.com/adobe/koperator/refs/heads/master/config/base/crds/kafka.banzaicloud.io_kafkaclusters.yaml

# Check CRD status
kubectl get crd kafkaclusters.kafka.banzaicloud.io -o yaml

# Check Helm release status
helm list -n kafka

# Uninstall and reinstall
helm uninstall kafka-operator -n kafka
helm install kafka-operator oci://ghcr.io/adobe/helm-charts/kafka-operator \
  --namespace=kafka \
  --set webhook.enabled=false \
  --version 0.28.0-adobe-20250923

echo "=== Namespace ==="
kubectl get namespace kafka

echo -e "\n=== CRDs ==="
kubectl get crd | grep kafka.banzaicloud.io

echo -e "\n=== Operator Pod ==="
kubectl get pods -n kafka

echo -e "\n=== Operator Logs (last 10 lines) ==="
kubectl logs -l app.kubernetes.io/instance=kafka-operator -c manager -n kafka --tail=10

echo -e "\n=== Ready for Kafka Cluster Deployment ==="
kubectl get kafkaclusters -n kafka

cd $TUTORIAL_DIR

# Create the KafkaCluster resource
kubectl create -n kafka -f - <<EOF
apiVersion: kafka.banzaicloud.io/v1beta1
kind: KafkaCluster
metadata:
  name: kafka
  namespace: kafka
spec:
  headlessServiceEnabled: true
  zkAddresses:
    - "zk-client.zookeeper:2181"
  rackAwareness:
    labels:
      - "topology.kubernetes.io/zone"
  brokerConfigGroups:
    default:
      brokerAnnotations:
        prometheus.io/scrape: "true"
        prometheus.io/port: "9020"
      storageConfigs:
        - mountPath: "/kafka-logs"
          pvcSpec:
            accessModes:
              - ReadWriteOnce
            resources:
              requests:
                storage: 10Gi
      serviceAccountName: "default"
      resourceRequirements:
        limits:
          cpu: "2"
          memory: "4Gi"
        requests:
          cpu: "1"
          memory: "2Gi"
      jvmPerformanceOpts: "-server -XX:+UseG1GC -XX:MaxGCPauseMillis=20 -XX:InitiatingHeapOccupancyPercent=35 -XX:+ExplicitGCInvokesConcurrent -Djava.awt.headless=true -Dsun.awt.fontpath=/usr/share/fonts/TTF"
      config:
        "auto.create.topics.enable": "true"
        "cruise.control.metrics.topic.auto.create": "true"
        "cruise.control.metrics.topic.num.partitions": "1"
        "cruise.control.metrics.topic.replication.factor": "2"
        "default.replication.factor": "2"
        "min.insync.replicas": "1"
        "num.partitions": "3"
        "offsets.topic.replication.factor": "2"
        "replica.lag.time.max.ms": "30000"
        "transaction.state.log.replication.factor": "2"
        "transaction.state.log.min.isr": "1"
  brokers:
    - id: 101
      brokerConfigGroup: "default"
      nodePortExternalIP:
        externalIP: "127.0.0.1"
    - id: 102
      brokerConfigGroup: "default"
      nodePortExternalIP:
        externalIP: "127.0.0.1"
    - id: 201
      brokerConfigGroup: "default"
      nodePortExternalIP:
        externalIP: "127.0.0.1"
    - id: 202
      brokerConfigGroup: "default"
      nodePortExternalIP:
        externalIP: "127.0.0.1"
    - id: 301
      brokerConfigGroup: "default"
      nodePortExternalIP:
        externalIP: "127.0.0.1"
    - id: 302
      brokerConfigGroup: "default"
      nodePortExternalIP:
        externalIP: "127.0.0.1"
  rollingUpgradeConfig:
    failureThreshold: 1
  cruiseControlConfig:
    cruiseControlEndpoint: "kafka-cruisecontrol-svc.kafka:8090"
    config: |
      # Copyright 2017 LinkedIn Corp. Licensed under the BSD 2-Clause License (the "License").
      # Sample Cruise Control configuration file.
      
      # Configuration for the metadata client.
      # =======================================
      
      # The maximum interval in milliseconds between two metadata refreshes.
      metadata.max.age.ms=300000
      
      # Client id for the Cruise Control. It is used for the metadata client.
      client.id=kafka-cruise-control
      
      # The size of TCP send buffer for Kafka network client.
      send.buffer.bytes=131072
      
      # The size of TCP receive buffer for Kafka network client.
      receive.buffer.bytes=131072
      
      # The time to wait for response from a server.
      request.timeout.ms=30000
      
      # Configurations for the load monitor
      # ===================================
      
      # The number of metric fetcher thread to fetch metrics for the Kafka cluster
      num.metric.fetchers=1
      
      # The metric sampler class
      metric.sampler.class=com.linkedin.kafka.cruisecontrol.monitor.sampling.CruiseControlMetricsReporterSampler
      
      # Configurations for CruiseControlMetricsReporter
      cruise.control.metrics.reporter.interval.ms=10000
      cruise.control.metrics.reporter.kubernetes.mode=true
      
      # The sample store class name
      sample.store.class=com.linkedin.kafka.cruisecontrol.monitor.sampling.KafkaSampleStore
      
      # The config for the Kafka sample store to save the partition metric samples
      partition.metric.sample.store.topic=__CruiseControlMetrics
      
      # The config for the Kafka sample store to save the model training samples
      broker.metric.sample.store.topic=__CruiseControlModelTrainingSamples
      
      # The replication factor of Kafka metric sample store topic
      sample.store.topic.replication.factor=2
      
      # The config for the number of Kafka sample store consumer threads
      num.sample.loading.threads=8
      
      # The partition assignor class for the metric samplers
      metric.sampler.partition.assignor.class=com.linkedin.kafka.cruisecontrol.monitor.sampling.DefaultMetricSamplerPartitionAssignor
      
      # The metric sampling interval in milliseconds
      metric.sampling.interval.ms=120000
      
      # The partition metrics window size in milliseconds
      partition.metrics.window.ms=300000
      
      # The number of partition metric windows to keep in memory
      num.partition.metrics.windows=1
      
      # The minimum partition metric samples required for a partition in each window
      min.samples.per.partition.metrics.window=1
      
      # The broker metrics window size in milliseconds
      broker.metrics.window.ms=300000
      
      # The number of broker metric windows to keep in memory
      num.broker.metrics.windows=20
      
      # The minimum broker metric samples required for a broker in each window
      min.samples.per.broker.metrics.window=1
      
      # The configuration for the BrokerCapacityConfigFileResolver (supports JBOD and non-JBOD broker capacities)
      capacity.config.file=config/capacity.json
      
      # Configurations for the analyzer
      # ===============================
      
      # The list of goals to optimize the Kafka cluster for with pre-computed proposals
      default.goals=com.linkedin.kafka.cruisecontrol.analyzer.goals.RackAwareGoal,com.linkedin.kafka.cruisecontrol.analyzer.goals.ReplicaCapacityGoal,com.linkedin.kafka.cruisecontrol.analyzer.goals.DiskCapacityGoal,com.linkedin.kafka.cruisecontrol.analyzer.goals.NetworkInboundCapacityGoal,com.linkedin.kafka.cruisecontrol.analyzer.goals.NetworkOutboundCapacityGoal,com.linkedin.kafka.cruisecontrol.analyzer.goals.CpuCapacityGoal,com.linkedin.kafka.cruisecontrol.analyzer.goals.ReplicaDistributionGoal,com.linkedin.kafka.cruisecontrol.analyzer.goals.PotentialNwOutGoal,com.linkedin.kafka.cruisecontrol.analyzer.goals.DiskUsageDistributionGoal,com.linkedin.kafka.cruisecontrol.analyzer.goals.NetworkInboundUsageDistributionGoal,com.linkedin.kafka.cruisecontrol.analyzer.goals.NetworkOutboundUsageDistributionGoal,com.linkedin.kafka.cruisecontrol.analyzer.goals.CpuUsageDistributionGoal,com.linkedin.kafka.cruisecontrol.analyzer.goals.TopicReplicaDistributionGoal,com.linkedin.kafka.cruisecontrol.analyzer.goals.LeaderReplicaDistributionGoal,com.linkedin.kafka.cruisecontrol.analyzer.goals.LeaderBytesInDistributionGoal
      
      # The list of supported goals
      goals=com.linkedin.kafka.cruisecontrol.analyzer.goals.RackAwareGoal,com.linkedin.kafka.cruisecontrol.analyzer.goals.ReplicaCapacityGoal,com.linkedin.kafka.cruisecontrol.analyzer.goals.DiskCapacityGoal,com.linkedin.kafka.cruisecontrol.analyzer.goals.NetworkInboundCapacityGoal,com.linkedin.kafka.cruisecontrol.analyzer.goals.NetworkOutboundCapacityGoal,com.linkedin.kafka.cruisecontrol.analyzer.goals.CpuCapacityGoal,com.linkedin.kafka.cruisecontrol.analyzer.goals.ReplicaDistributionGoal,com.linkedin.kafka.cruisecontrol.analyzer.goals.PotentialNwOutGoal,com.linkedin.kafka.cruisecontrol.analyzer.goals.DiskUsageDistributionGoal,com.linkedin.kafka.cruisecontrol.analyzer.goals.NetworkInboundUsageDistributionGoal,com.linkedin.kafka.cruisecontrol.analyzer.goals.NetworkOutboundUsageDistributionGoal,com.linkedin.kafka.cruisecontrol.analyzer.goals.CpuUsageDistributionGoal,com.linkedin.kafka.cruisecontrol.analyzer.goals.TopicReplicaDistributionGoal,com.linkedin.kafka.cruisecontrol.analyzer.goals.LeaderReplicaDistributionGoal,com.linkedin.kafka.cruisecontrol.analyzer.goals.LeaderBytesInDistributionGoal,com.linkedin.kafka.cruisecontrol.analyzer.goals.PreferredLeaderElectionGoal
      
      # The list of supported hard goals
      hard.goals=com.linkedin.kafka.cruisecontrol.analyzer.goals.RackAwareGoal,com.linkedin.kafka.cruisecontrol.analyzer.goals.ReplicaCapacityGoal,com.linkedin.kafka.cruisecontrol.analyzer.goals.DiskCapacityGoal,com.linkedin.kafka.cruisecontrol.analyzer.goals.NetworkInboundCapacityGoal,com.linkedin.kafka.cruisecontrol.analyzer.goals.NetworkOutboundCapacityGoal,com.linkedin.kafka.cruisecontrol.analyzer.goals.CpuCapacityGoal
      
      # The minimum percentage of well monitored partitions out of all the partitions
      min.monitored.partition.percentage=0.95
      
      # The balance threshold for CPU
      cpu.balance.threshold=1.1
      
      # The balance threshold for disk
      disk.balance.threshold=1.1
      
      # The balance threshold for network inbound utilization
      network.inbound.balance.threshold=1.1
      
      # The balance threshold for network outbound utilization
      network.outbound.balance.threshold=1.1
      
      # The balance threshold for the replica count
      replica.count.balance.threshold=1.1
      
      # The capacity threshold for CPU in percentage
      cpu.capacity.threshold=0.8
      
      # The capacity threshold for disk in percentage
      disk.capacity.threshold=0.8
      
      # The capacity threshold for network inbound utilization in percentage
      network.inbound.capacity.threshold=0.8
      
      # The capacity threshold for network outbound utilization in percentage
      network.outbound.capacity.threshold=0.8
      
      # The threshold for the number of replicas per broker
      replica.capacity.threshold=1000
      
      # The weight adjustment in the optimization algorithm
      cpu.low.utilization.threshold=0.0
      disk.low.utilization.threshold=0.0
      network.inbound.low.utilization.threshold=0.0
      network.outbound.low.utilization.threshold=0.0
      
      # The metric anomaly percentile upper threshold
      metric.anomaly.percentile.upper.threshold=90.0
      
      # The metric anomaly percentile lower threshold
      metric.anomaly.percentile.lower.threshold=10.0
      
      # How often should the cached proposal be expired and recalculated if necessary
      proposal.expiration.ms=60000
      
      # The maximum number of replicas that can reside on a broker at any given time.
      max.replicas.per.broker=10000
      
      # The number of threads to use for proposal candidate precomputing.
      num.proposal.precompute.threads=1
      
      # the topics that should be excluded from the partition movement.
      #topics.excluded.from.partition.movement
      
      # Configurations for the executor
      # ===============================
      
      # The max number of partitions to move in/out on a given broker at a given time.
      num.concurrent.partition.movements.per.broker=10
      
      # The interval between two execution progress checks.
      execution.progress.check.interval.ms=10000
      
      # Configurations for anomaly detector
      # ===================================
      
      # The goal violation notifier class
      anomaly.notifier.class=com.linkedin.kafka.cruisecontrol.detector.notifier.SelfHealingNotifier
      
      # The metric anomaly finder class
      metric.anomaly.finder.class=com.linkedin.kafka.cruisecontrol.detector.KafkaMetricAnomalyFinder
      
      # The anomaly detection interval
      anomaly.detection.interval.ms=10000
      
      # The goal violation to detect.
      anomaly.detection.goals=com.linkedin.kafka.cruisecontrol.analyzer.goals.RackAwareGoal,com.linkedin.kafka.cruisecontrol.analyzer.goals.ReplicaCapacityGoal,com.linkedin.kafka.cruisecontrol.analyzer.goals.DiskCapacityGoal,com.linkedin.kafka.cruisecontrol.analyzer.goals.NetworkInboundCapacityGoal,com.linkedin.kafka.cruisecontrol.analyzer.goals.NetworkOutboundCapacityGoal,com.linkedin.kafka.cruisecontrol.analyzer.goals.CpuCapacityGoal
      
      # The interested metrics for metric anomaly analyzer.
      metric.anomaly.analyzer.metrics=BROKER_PRODUCE_LOCAL_TIME_MS_MAX,BROKER_PRODUCE_LOCAL_TIME_MS_MEAN,BROKER_CONSUMER_FETCH_LOCAL_TIME_MS_MAX,BROKER_CONSUMER_FETCH_LOCAL_TIME_MS_MEAN,BROKER_FOLLOWER_FETCH_LOCAL_TIME_MS_MAX,BROKER_FOLLOWER_FETCH_LOCAL_TIME_MS_MEAN,BROKER_LOG_FLUSH_TIME_MS_MAX,BROKER_LOG_FLUSH_TIME_MS_MEAN
      
      # The zk path to store the anomaly detector state. This is to avoid duplicate anomaly detection due to controller failure.
      anomaly.detection.state.path=/CruiseControlAnomalyDetector/AnomalyDetectorState
      
      # Enable self healing for all anomaly detectors, unless the particular anomaly detector is explicitly disabled
      self.healing.enabled=true
      
      # Enable self healing for broker failure detector
      self.healing.broker.failure.enabled=true
      
      # Enable self healing for goal violation detector
      self.healing.goal.violation.enabled=true
      
      # Enable self healing for metric anomaly detector
      self.healing.metric.anomaly.enabled=true
      
      # configurations for the webserver
      # ================================
      
      # HTTP listen port for the Cruise Control
      webserver.http.port=8090
      
      # HTTP listen address for the Cruise Control
      webserver.http.address=0.0.0.0
      
      # Whether CORS support is enabled for API or not
      webserver.http.cors.enabled=false
      
      # Value for Access-Control-Allow-Origin
      webserver.http.cors.origin=http://localhost:8090/
      
      # Value for Access-Control-Request-Method
      webserver.http.cors.allowmethods=OPTIONS,GET,POST
      
      # Headers that should be exposed to the Browser (Webapp)
      # This is a special header that is used by the
      # User Tasks subsystem and should be explicitly
      # Enabled when CORS mode is used as part of the
      # Admin Interface
      webserver.http.cors.exposeheaders=User-Task-ID
      
      # REST API default prefix (dont forget the ending /*)
      webserver.api.urlprefix=/kafkacruisecontrol/*
      
      # Location where the Cruise Control frontend is deployed
      webserver.ui.diskpath=./cruise-control-ui/dist/
      
      # URL path prefix for UI
      webserver.ui.urlprefix=/*
      
      # Time After which request is converted to Async
      webserver.request.maxBlockTimeMs=10000
      
      # Default Session Expiry Period
      webserver.session.maxExpiryTimeMs=60000
      
      # Session cookie path
      webserver.session.path=/
      
      # Server Access Logs
      webserver.accesslog.enabled=true
      
      # Location of HTTP Request Logs
      webserver.accesslog.path=access.log
      
      # HTTP Request Log retention days
      webserver.accesslog.retention.days=14
EOF

# Watch KafkaCluster status
kubectl get kafkacluster kafka -n kafka -w -o wide
# Press Ctrl+C to stop watching when cluster is ready

# Check broker pods
kubectl get pods -n kafka -l kafka_cr=kafka

# Check all resources in kafka namespace
kubectl get all -n kafka

# Create ServiceMonitor for Kafka metrics
kubectl apply -n kafka -f - <<EOF
apiVersion: monitoring.coreos.com/v1
kind: ServiceMonitor
metadata:
  name: kafka-servicemonitor
  namespace: kafka
  labels:
    app: kafka
    release: monitoring
spec:
  selector:
    matchLabels:
      app: kafka
  endpoints:
  - port: metrics
    interval: 30s
    path: /metrics
---
apiVersion: monitoring.coreos.com/v1
kind: ServiceMonitor
metadata:
  name: kafka-jmx-servicemonitor
  namespace: kafka
  labels:
    app: kafka-jmx
    release: monitoring
spec:
  selector:
    matchLabels:
      app: kafka
  endpoints:
  - port: jmx-metrics
    interval: 30s
    path: /metrics
EOF

# Create PrometheusRule for Kafka alerting
kubectl apply -n kafka -f - <<EOF
apiVersion: monitoring.coreos.com/v1
kind: PrometheusRule
metadata:
  name: kafka-alerts
  namespace: kafka
  labels:
    app: kafka
    release: monitoring
spec:
  groups:
  - name: kafka.rules
    rules:
    - alert: KafkaOfflinePartitions
      expr: kafka_controller_kafkacontroller_offlinepartitionscount > 0
      for: 5m
      labels:
        severity: critical
      annotations:
        summary: "Kafka has offline partitions"
        description: "Kafka cluster {{ \$labels.instance }} has {{ \$value }} offline partitions"
    
    - alert: KafkaUnderReplicatedPartitions
      expr: kafka_server_replicamanager_underreplicatedpartitions > 0
      for: 5m
      labels:
        severity: warning
      annotations:
        summary: "Kafka has under-replicated partitions"
        description: "Kafka cluster {{ \$labels.instance }} has {{ \$value }} under-replicated partitions"
    
    - alert: KafkaHighProducerRequestRate
      expr: rate(kafka_network_requestmetrics_requests_total{request="Produce"}[5m]) > 1000
      for: 10m
      labels:
        severity: warning
        command: "upScale"
      annotations:
        summary: "High Kafka producer request rate"
        description: "Kafka producer request rate is {{ \$value }} requests/sec"
    
    - alert: KafkaLowProducerRequestRate
      expr: rate(kafka_network_requestmetrics_requests_total{request="Produce"}[5m]) < 100
      for: 30m
      labels:
        severity: info
        command: "downScale"
      annotations:
        summary: "Low Kafka producer request rate"
        description: "Kafka producer request rate is {{ \$value }} requests/sec"
EOF

# Apply the complete Kafka Looking Glass dashboard directly
kubectl apply -n default \
  -f https://raw.githubusercontent.com/amuraru/k8s-kafka-operator/master/grafana-dashboard.yaml

# Describe the KafkaCluster
kubectl describe kafkacluster kafka -n kafka

# Check broker distribution across zones
kubectl get pods -n kafka -l kafka_cr=kafka -o wide

# Check persistent volumes
kubectl get pv,pvc -n kafka

# Port forward to Cruise Control (in a separate terminal)
kubectl port-forward -n kafka svc/kafka-cruisecontrol-svc 8090:8090 &

# Check Cruise Control status (optional)
curl -s http://localhost:8090/kafkacruisecontrol/v1/state | jq .

# Check if Prometheus is scraping Kafka metrics
kubectl port-forward -n default svc/monitoring-kube-prometheus-prometheus 9090 &

# Visit http://localhost:9090 and search for kafka_ metrics

# Get Grafana admin password
kubectl get secret --namespace default monitoring-grafana \
  -o jsonpath="{.data.admin-password}" | base64 --decode
echo ""

# Port forward to Grafana
kubectl port-forward -n default svc/monitoring-grafana 3000:80 &

# List topics using kubectl run
kubectl run kafka-topics --rm -i --tty=true \
  --image=ghcr.io/adobe/koperator/kafka:2.13-3.9.1 \
  --restart=Never \
  -- /opt/kafka/bin/kafka-topics.sh \
  --bootstrap-server kafka-headless:29092 \
  --list

# Get cluster metadata
kubectl run kafka-metadata --rm -i --tty=true \
  --image=ghcr.io/adobe/koperator/kafka:2.13-3.9.1 \
  --restart=Never \
  -- /opt/kafka/bin/kafka-broker-api-versions.sh \
  --bootstrap-server kafka-headless:29092

# Create a test topic with 12 partitions distributed across brokers
kubectl run kafka-topics --rm -i --tty=true \
  --image=ghcr.io/adobe/koperator/kafka:2.13-3.9.1 \
  --restart=Never \
  -- /opt/kafka/bin/kafka-topics.sh \
  --bootstrap-server kafka-headless:29092 \
  --topic perf_topic \
  --replica-assignment 101:201:301,102:202:302,101:201:301,102:202:302,101:201:301,102:202:302,101:201:301,102:202:302,101:201:301,102:202:302,101:201:301,102:202:302 \
  --create

# Describe the topic to verify configuration
kubectl run kafka-topics --rm -i --tty=true \
  --image=ghcr.io/adobe/koperator/kafka:2.13-3.9.1 \
  --restart=Never \
  -- /opt/kafka/bin/kafka-topics.sh \
  --bootstrap-server kafka-headless:29092 \
  --topic perf_topic \
  --describe

# Set custom retention period (12 minutes for testing)
kubectl run kafka-configs --rm -i --tty=true \
  --image=ghcr.io/adobe/koperator/kafka:2.13-3.9.1 \
  --restart=Never \
  -- /opt/kafka/bin/kafka-configs.sh \
  --zookeeper zk-client.zookeeper:2181/kafka \
  --alter --entity-name perf_topic \
  --entity-type topics \
  --add-config retention.ms=720000

# Start a simple producer (run in one terminal)
kubectl run kafka-producer \
  --image=ghcr.io/adobe/koperator/kafka:2.13-3.9.1 \
  --restart=Never \
  -- /opt/kafka/bin/kafka-console-producer.sh \
  --bootstrap-server kafka-headless:29092 \
  --topic perf_topic

# Start a consumer (run in another terminal)
kubectl run kafka-consumer \
  --image=ghcr.io/adobe/koperator/kafka:2.13-3.9.1 \
  --restart=Never \
  -- /opt/kafka/bin/kafka-console-consumer.sh \
  --bootstrap-server kafka-headless:29092 \
  --topic perf_topic \
  --from-beginning

# Clean up the test pods
kubectl delete pod kafka-producer --ignore-not-found=true
kubectl delete pod kafka-consumer --ignore-not-found=true

# Start producer performance test
kubectl run kafka-producer-perf \
  --image=ghcr.io/adobe/koperator/kafka:2.13-3.9.1 \
  --restart=Never \
  -- /opt/kafka/bin/kafka-producer-perf-test.sh \
  --producer-props bootstrap.servers=kafka-headless:29092 acks=all \
  --topic perf_topic \
  --record-size 1000 \
  --throughput 29000 \
  --num-records 2110000

# Start consumer performance test
kubectl run kafka-consumer-perf \
  --image=ghcr.io/adobe/koperator/kafka:2.13-3.9.1 \
  --restart=Never \
  -- /opt/kafka/bin/kafka-consumer-perf-test.sh \
  --broker-list kafka-headless:29092 \
  --group perf-consume \
  --messages 10000000 \
  --topic perf_topic \
  --show-detailed-stats \
  --from-latest \
  --timeout 100000

# Port forward to Prometheus (if not already done)
kubectl port-forward -n default svc/monitoring-kube-prometheus-prometheus 9090 &

# Check if Kafka metrics are being collected
curl -s "http://localhost:9090/api/v1/query?query=kafka_server_brokertopicmetrics_messagesin_total" | jq .

# Port forward to Grafana (if not already done)
kubectl port-forward -n default svc/monitoring-grafana 3000:80 &

# Get Grafana admin password
kubectl get secret --namespace default monitoring-grafana \
  -o jsonpath="{.data.admin-password}" | base64 --decode
echo ""

# Port forward to AlertManager
kubectl port-forward -n default svc/monitoring-kube-prometheus-alertmanager 9093 &

# Check broker pod distribution
kubectl get pods -n kafka -l kafka_cr=kafka -o wide \
  --sort-by='.spec.nodeName'

# Check node labels
kubectl get nodes \
  --label-columns=topology.kubernetes.io/zone \
  -l topology.kubernetes.io/zone

# Check if rack awareness is working by examining topic partition distribution
kubectl run kafka-topics --rm -i --tty=true \
  --image=ghcr.io/adobe/koperator/kafka:2.13-3.9.1 \
  --restart=Never \
  -- /opt/kafka/bin/kafka-topics.sh \
  --bootstrap-server kafka-headless:29092 \
  --topic perf_topic \
  --describe

# Create topic using KafkaTopic CRD
kubectl apply -n kafka -f - <<EOF
apiVersion: kafka.banzaicloud.io/v1alpha1
kind: KafkaTopic
metadata:
  name: test-topic-crd
  namespace: kafka
spec:
  clusterRef:
    name: kafka
  name: test-topic-crd
  partitions: 6
  replicationFactor: 2
  config:
    "retention.ms": "604800000"
    "cleanup.policy": "delete"
EOF

# Check KafkaTopic resource
kubectl get kafkatopic -n kafka

# Verify topic exists in Kafka
kubectl run kafka-topics --rm -i --tty=true \
  --image=ghcr.io/adobe/koperator/kafka:2.13-3.9.1 \
  --restart=Never \
  -- /opt/kafka/bin/kafka-topics.sh \
  --bootstrap-server kafka-headless:29092 \
  --list | grep test-topic-crd

# Create multiple consumers in the same group
for i in {1..3}; do
  kubectl run kafka-consumer-group-$i \
    --image=ghcr.io/adobe/koperator/kafka:2.13-3.9.1 \
    --restart=Never \
    -- /opt/kafka/bin/kafka-console-consumer.sh \
    --bootstrap-server kafka-headless:29092 \
    --topic perf_topic \
    --group test-group &
done

# Check consumer group status
kubectl run kafka-consumer-groups --rm -i --tty=true \
  --image=ghcr.io/adobe/koperator/kafka:2.13-3.9.1 \
  --restart=Never \
  -- /opt/kafka/bin/kafka-consumer-groups.sh \
  --bootstrap-server kafka-headless:29092 \
  --group test-group \
  --describe

# Clean up performance test pods
kubectl delete pod kafka-producer-perf --ignore-not-found=true
kubectl delete pod kafka-consumer-perf --ignore-not-found=true

# Clean up consumer group pods
for i in {1..3}; do
  kubectl delete pod kafka-consumer-group-$i --ignore-not-found=true
done

# Optionally delete test topics
kubectl delete kafkatopic test-topic-crd -n kafka

# Check broker connectivity
kubectl run kafka-test --rm -i --tty=true \
  --image=ghcr.io/adobe/koperator/kafka:2.13-3.9.1 \
  --restart=Never \
  -- /bin/bash

# Inside the pod, test connectivity
telnet kafka-headless 29092

# Check broker resource usage
kubectl top pods -n kafka

# Check broker logs
kubectl logs -n kafka kafka-101-xyz123

# Check JVM metrics
kubectl exec -n kafka kafka-101-xyz123 -- jps -v

# Check ServiceMonitor
kubectl get servicemonitor -n kafka

# Check Prometheus targets
curl -s "http://localhost:9090/api/v1/targets" | jq '.data.activeTargets[] | select(.labels.job | contains("kafka"))'

# Verify cluster is healthy
kubectl get kafkacluster kafka -n kafka
kubectl get pods -n kafka -l kafka_cr=kafka

# Create a test topic with data
kubectl run kafka-producer-dr --rm -i --tty=true \
  --image=ghcr.io/adobe/koperator/kafka:2.13-3.9.1 \
  --restart=Never \
  -- /opt/kafka/bin/kafka-console-producer.sh \
  --bootstrap-server kafka-headless:29092 \
  --topic disaster-recovery-test <<EOF
message-1-before-disaster
message-2-before-disaster
message-3-before-disaster
EOF

echo "=== Initial Kafka Cluster State ==="

# Get broker pods
echo "Kafka Broker Pods:"
kubectl get pods -l kafka_cr=kafka -n kafka -o wide

# Get persistent volumes
echo -e "\nPersistent Volumes:"
kubectl get pv | grep kafka

# Get persistent volume claims
echo -e "\nPersistent Volume Claims:"
kubectl get pvc -n kafka | grep kafka

# Get broker services
echo -e "\nKafka Services:"
kubectl get svc -n kafka | grep kafka

# Save state to file for comparison
kubectl get pods -l kafka_cr=kafka -n kafka -o yaml > /tmp/kafka-pods-initial.yaml
kubectl get pvc -n kafka -o yaml > /tmp/kafka-pvc-initial.yaml

# Get a broker pod name
BROKER_POD=$(kubectl get pods -n kafka -l kafka_cr=kafka -o jsonpath='{.items[0].metadata.name}')
echo "Testing JVM crash on pod: $BROKER_POD"

# Kill the Java process (PID 1 in the container)
kubectl exec -n kafka $BROKER_POD -- kill 1

# Monitor pod restart
kubectl get pods -n kafka -l kafka_cr=kafka -w
# Press Ctrl+C after observing the restart

# Check if pod restarted
kubectl get pods -n kafka -l kafka_cr=kafka

# Verify the same PVC is reused
kubectl describe pod -n kafka $BROKER_POD | grep -A 5 "Volumes:"

# Test data persistence
kubectl run kafka-consumer-dr --rm -i --tty=true \
  --image=ghcr.io/adobe/koperator/kafka:2.13-3.9.1 \
  --restart=Never \
  -- /opt/kafka/bin/kafka-console-consumer.sh \
  --bootstrap-server kafka-headless:29092 \
  --topic disaster-recovery-test \
  --from-beginning \
  --timeout-ms 10000

# Get another broker pod
BROKER_POD_2=$(kubectl get pods -n kafka -l kafka_cr=kafka -o jsonpath='{.items[1].metadata.name}')
echo "Testing pod deletion on: $BROKER_POD_2"

# Record the PVC before deletion
kubectl get pod -n kafka $BROKER_POD_2 -o yaml | grep -A 10 "volumes:" > /tmp/pvc-before-deletion.yaml

# Delete the pod
kubectl delete pod -n kafka $BROKER_POD_2

# Monitor recreation
kubectl get pods -n kafka -l kafka_cr=kafka -w
# Press Ctrl+C after new pod is running

# Check new pod is running
kubectl get pods -n kafka -l kafka_cr=kafka

# Verify PVC reattachment
NEW_BROKER_POD=$(kubectl get pods -n kafka -l kafka_cr=kafka | grep $BROKER_POD_2 | awk '{print $1}')
kubectl get pod -n kafka $NEW_BROKER_POD -o yaml | grep -A 10 "volumes:" > /tmp/pvc-after-deletion.yaml

# Compare PVC usage
echo "PVC comparison:"
diff /tmp/pvc-before-deletion.yaml /tmp/pvc-after-deletion.yaml || echo "PVCs are identical - Good!"

# Test cluster functionality
kubectl run kafka-test-after-deletion --rm -i --tty=true \
  --image=ghcr.io/adobe/koperator/kafka:2.13-3.9.1 \
  --restart=Never \
  -- /opt/kafka/bin/kafka-topics.sh \
  --bootstrap-server kafka-headless:29092 \
  --list

# Get a worker node with Kafka pods
NODE_WITH_KAFKA=$(kubectl get pods -n kafka -l kafka_cr=kafka -o wide | grep kafka | head -1 | awk '{print $7}')
echo "Simulating failure on node: $NODE_WITH_KAFKA"

# Get pods on this node before cordoning
echo "Pods on node before cordoning:"
kubectl get pods -n kafka -l kafka_cr=kafka -o wide | grep $NODE_WITH_KAFKA

# Cordon the node (prevent new pods)
kubectl cordon $NODE_WITH_KAFKA

# Drain the node (evict existing pods)
kubectl drain $NODE_WITH_KAFKA --ignore-daemonsets --delete-emptydir-data --force

# Watch pods being rescheduled
kubectl get pods -n kafka -l kafka_cr=kafka -o wide -w
# Press Ctrl+C after pods are rescheduled

# Verify pods moved to other nodes
echo "Pods after node drain:"
kubectl get pods -n kafka -l kafka_cr=kafka -o wide | grep -v $NODE_WITH_KAFKA

# Uncordon the node
kubectl uncordon $NODE_WITH_KAFKA

# Verify node is ready
kubectl get nodes | grep $NODE_WITH_KAFKA

echo "=== Persistent Volume Analysis ==="

# List all Kafka PVCs
kubectl get pvc -n kafka | grep kafka

# Check PV reclaim policy
kubectl get pv | grep kafka | head -3

# Verify PVC-PV binding
for pvc in $(kubectl get pvc -n kafka -o jsonpath='{.items[*].metadata.name}' | grep kafka); do
  echo "PVC: $pvc"
  kubectl get pvc -n kafka $pvc -o jsonpath='{.spec.volumeName}'
  echo ""
done

# Create test data
kubectl run kafka-persistence-test --rm -i --tty=true \
  --image=ghcr.io/adobe/koperator/kafka:2.13-3.9.1 \
  --restart=Never \
  -- /opt/kafka/bin/kafka-console-producer.sh \
  --bootstrap-server kafka-headless:29092 \
  --topic persistence-test <<EOF
persistence-message-1
persistence-message-2
persistence-message-3
EOF

# Delete multiple broker pods simultaneously
kubectl delete pods -n kafka -l kafka_cr=kafka --grace-period=0 --force

# Wait for recreation
kubectl wait --for=condition=Ready pod -l kafka_cr=kafka -n kafka --timeout=300s

# Verify data survived
kubectl run kafka-persistence-verify --rm -i --tty=true \
  --image=ghcr.io/adobe/koperator/kafka:2.13-3.9.1 \
  --restart=Never \
  -- /opt/kafka/bin/kafka-console-consumer.sh \
  --bootstrap-server kafka-headless:29092 \
  --topic persistence-test \
  --from-beginning \
  --timeout-ms 10000

# Create a network policy that isolates one broker
kubectl apply -n kafka -f - <<EOF
apiVersion: networking.k8s.io/v1
kind: NetworkPolicy
metadata:
  name: isolate-broker
  namespace: kafka
spec:
  podSelector:
    matchLabels:
      brokerId: "101"
  policyTypes:
  - Ingress
  - Egress
  ingress: []
  egress: []
EOF

# Check cluster state during network partition
kubectl run kafka-network-test --rm -i --tty=true \
  --image=ghcr.io/adobe/koperator/kafka:2.13-3.9.1 \
  --restart=Never \
  -- /opt/kafka/bin/kafka-topics.sh \
  --bootstrap-server kafka-headless:29092 \
  --topic persistence-test \
  --describe

# Check under-replicated partitions
kubectl logs -n kafka kafka-101 | grep -i "under.replicated" | tail -5

# Remove the network policy
kubectl delete networkpolicy isolate-broker -n kafka

# Verify cluster recovery
sleep 30
kubectl run kafka-recovery-test --rm -i --tty=true \
  --image=ghcr.io/adobe/koperator/kafka:2.13-3.9.1 \
  --restart=Never \
  -- /opt/kafka/bin/kafka-topics.sh \
  --bootstrap-server kafka-headless:29092 \
  --topic persistence-test \
  --describe

# Check current ZooKeeper state
kubectl get pods -n zookeeper

# Scale down ZooKeeper to 1 replica (simulating failure)
kubectl patch zookeepercluster zk -n zookeeper --type='merge' -p='{"spec":{"replicas":1}}'

# Monitor Kafka behavior
kubectl logs -n kafka kafka-101 | grep -i zookeeper | tail -10

# Try to create a topic (should fail or be delayed)
timeout 30 kubectl run kafka-zk-test --rm -i --tty=true \
  --image=ghcr.io/adobe/koperator/kafka:2.13-3.9.1 \
  --restart=Never \
  -- /opt/kafka/bin/kafka-topics.sh \
  --bootstrap-server kafka-headless:29092 \
  --topic zk-failure-test \
  --create --partitions 3 --replication-factor 2 || echo "Topic creation failed as expected"

# Test existing topic access (should still work)
kubectl run kafka-existing-test --rm -i --tty=true \
  --image=ghcr.io/adobe/koperator/kafka:2.13-3.9.1 \
  --restart=Never \
  -- /opt/kafka/bin/kafka-console-producer.sh \
  --bootstrap-server kafka-headless:29092 \
  --topic persistence-test <<EOF
message-during-zk-failure
EOF

# Scale ZooKeeper back to 3 replicas
kubectl patch zookeepercluster zk -n zookeeper --type='merge' -p='{"spec":{"replicas":3}}'

# Wait for ZooKeeper recovery
kubectl wait --for=condition=Ready pod -l app=zookeeper -n zookeeper --timeout=300s

# Verify Kafka functionality restored
kubectl run kafka-zk-recovery --rm -i --tty=true \
  --image=ghcr.io/adobe/koperator/kafka:2.13-3.9.1 \
  --restart=Never \
  -- /opt/kafka/bin/kafka-topics.sh \
  --bootstrap-server kafka-headless:29092 \
  --topic zk-recovery-test \
  --create --partitions 3 --replication-factor 2

echo "=== Disaster Recovery Test Summary ==="
echo "Test Date: $(date)"
echo ""

echo "1. Broker JVM Crash Test: PASSED"
echo "   - Pod restarted automatically"
echo "   - PVC reused successfully"
echo "   - Data preserved"
echo ""

echo "2. Broker Pod Deletion Test: PASSED"
echo "   - New pod created automatically"
echo "   - PVC reattached successfully"
echo "   - Cluster remained functional"
echo ""

echo "3. Node Failure Simulation: PASSED"
echo "   - Pods rescheduled to healthy nodes"
echo "   - PVCs reattached successfully"
echo "   - No data loss"
echo ""

echo "4. Persistent Volume Validation: PASSED"
echo "   - Data survived multiple pod deletions"
echo "   - PVC-PV bindings maintained"
echo "   - Storage reclaim policy working"
echo ""

echo "5. Network Partition Test: PASSED"
echo "   - Cluster detected partition"
echo "   - Maintained availability"
echo "   - Recovered after partition resolution"
echo ""

echo "6. ZooKeeper Failure Test: PASSED"
echo "   - Existing functionality maintained during ZK degradation"
echo "   - Full functionality restored after ZK recovery"
echo ""

# Final cluster health check
echo "=== Final Cluster Health Check ==="
kubectl get kafkacluster kafka -n kafka
kubectl get pods -n kafka -l kafka_cr=kafka
kubectl get pvc -n kafka | grep kafka

# Clean up test topics
kubectl run kafka-cleanup --rm -i --tty=true \
  --image=ghcr.io/adobe/koperator/kafka:2.13-3.9.1 \
  --restart=Never \
  -- /opt/kafka/bin/kafka-topics.sh \
  --bootstrap-server kafka-headless:29092 \
  --delete --topic disaster-recovery-test

kubectl run kafka-cleanup-2 --rm -i --tty=true \
  --image=ghcr.io/adobe/koperator/kafka:2.13-3.9.1 \
  --restart=Never \
  -- /opt/kafka/bin/kafka-topics.sh \
  --bootstrap-server kafka-headless:29092 \
  --delete --topic persistence-test

# Remove temporary files
rm -f /tmp/kafka-pods-initial.yaml /tmp/kafka-pvc-initial.yaml
rm -f /tmp/pvc-before-deletion.yaml /tmp/pvc-after-deletion.yaml

# Set namespace context for convenience
kubectl config set-context --current --namespace=kafka

# Quick cluster health check
echo "=== Cluster Health Overview ==="
kubectl get kafkacluster kafka -o wide
kubectl get pods -l kafka_cr=kafka
kubectl get svc | grep kafka
kubectl get pvc | grep kafka

# Comprehensive cluster diagnostics
function kafka_diagnostics() {
    echo "=== Kafka Cluster Diagnostics ==="
    echo "Timestamp: $(date)"
    echo ""
    
    echo "1. KafkaCluster Resource:"
    kubectl describe kafkacluster kafka
    echo ""
    
    echo "2. Broker Pods:"
    kubectl get pods -l kafka_cr=kafka -o wide
    echo ""
    
    echo "3. Pod Events:"
    kubectl get events --sort-by=.metadata.creationTimestamp | grep kafka | tail -10
    echo ""
    
    echo "4. Persistent Volumes:"
    kubectl get pv | grep kafka
    echo ""
    
    echo "5. Services:"
    kubectl get svc | grep kafka
    echo ""
    
    echo "6. Operator Status:"
    kubectl get pods -l app.kubernetes.io/instance=kafka-operator
    echo ""
}

# Run diagnostics
kafka_diagnostics

# Check operator pod status
kubectl get pods -l app.kubernetes.io/instance=kafka-operator

# Check operator logs
kubectl logs -l app.kubernetes.io/instance=kafka-operator -c manager --tail=50

# Check operator events
kubectl describe pod -l app.kubernetes.io/instance=kafka-operator

# 1. Restart operator deployment
kubectl rollout restart deployment kafka-operator-operator

# 2. Check RBAC permissions
kubectl auth can-i create kafkaclusters --as=system:serviceaccount:kafka:kafka-operator-operator

# 3. Reinstall operator
helm uninstall kafka-operator -n kafka
helm install kafka-operator oci://ghcr.io/adobe/helm-charts/kafka-operator \
  --namespace=kafka \
  --set webhook.enabled=false \
  --version 0.28.0-adobe-20250923

# Check broker pod status
kubectl get pods -l kafka_cr=kafka -o wide

# Check specific broker logs
BROKER_POD=$(kubectl get pods -l kafka_cr=kafka -o jsonpath='{.items[0].metadata.name}')
kubectl logs $BROKER_POD --tail=100

# Check pod events
kubectl describe pod $BROKER_POD

# 1. Check resource constraints
kubectl describe nodes | grep -A 5 "Allocated resources"

# 2. Check storage class
kubectl get storageclass

# 3. Check ZooKeeper connectivity
kubectl run zk-test --rm -i --tty=true \
  --image=busybox \
  --restart=Never \
  -- telnet zk-client.zookeeper 2181

# 4. Force broker recreation
kubectl delete pod $BROKER_POD

# Test DNS resolution
kubectl run dns-test --rm -i --tty=true \
  --image=busybox \
  --restart=Never \
  -- nslookup kafka-headless.kafka.svc.cluster.local

# Test port connectivity
kubectl run port-test --rm -i --tty=true \
  --image=busybox \
  --restart=Never \
  -- telnet kafka-headless 29092

# Check service endpoints
kubectl get endpoints kafka-headless

# 1. Verify service configuration
kubectl get svc kafka-headless -o yaml

# 2. Check network policies
kubectl get networkpolicy -A

# 3. Restart CoreDNS (if DNS issues)
kubectl rollout restart deployment coredns -n kube-system

# Check external services
kubectl get svc | grep LoadBalancer

# Check ingress configuration
kubectl get ingress -A

# Test external connectivity
kubectl run external-test --rm -i --tty=true \
  --image=ghcr.io/adobe/koperator/kafka:2.13-3.9.1 \
  --restart=Never \
  -- /opt/kafka/bin/kafka-broker-api-versions.sh \
  --bootstrap-server <EXTERNAL_IP>:<EXTERNAL_PORT>

# Check broker resource usage
kubectl top pods -n kafka

# Check JVM metrics
kubectl exec -n kafka $BROKER_POD -- jstat -gc 1

# Check disk I/O
kubectl exec -n kafka $BROKER_POD -- iostat -x 1 5

# Check network metrics
kubectl exec -n kafka $BROKER_POD -- ss -tuln

# 1. Increase broker resources
kubectl patch kafkacluster kafka --type='merge' -p='
{
  "spec": {
    "brokerConfigGroups": {
      "default": {
        "resourceRequirements": {
          "requests": {
            "cpu": "2",
            "memory": "4Gi"
          },
          "limits": {
            "cpu": "4",
            "memory": "8Gi"
          }
        }
      }
    }
  }
}'

# 2. Optimize JVM settings
kubectl patch kafkacluster kafka --type='merge' -p='
{
  "spec": {
    "brokerConfigGroups": {
      "default": {
        "jvmPerformanceOpts": "-server -XX:+UseG1GC -XX:MaxGCPauseMillis=20 -XX:InitiatingHeapOccupancyPercent=35 -Xms4g -Xmx4g"
      }
    }
  }
}'

# Check disk usage in broker pods
for pod in $(kubectl get pods -l kafka_cr=kafka -o jsonpath='{.items[*].metadata.name}'); do
  echo "=== $pod ==="
  kubectl exec $pod -- df -h /kafka-logs
done

# Check PVC usage
kubectl get pvc | grep kafka

# 1. Increase PVC size (if storage class supports expansion)
kubectl patch pvc kafka-101-storage-0 -p='{"spec":{"resources":{"requests":{"storage":"20Gi"}}}}'

# 2. Configure log retention
kubectl patch kafkacluster kafka --type='merge' -p='
{
  "spec": {
    "brokerConfigGroups": {
      "default": {
        "config": {
          "log.retention.hours": "168",
          "log.segment.bytes": "1073741824",
          "log.retention.check.interval.ms": "300000"
        }
      }
    }
  }
}'

# Check ServiceMonitor
kubectl get servicemonitor -n kafka

# Check Prometheus targets
kubectl port-forward -n default svc/monitoring-kube-prometheus-prometheus 9090 &
curl -s "http://localhost:9090/api/v1/targets" | jq '.data.activeTargets[] | select(.labels.job | contains("kafka"))'

# Check metrics endpoints
kubectl exec -n kafka $BROKER_POD -- curl -s localhost:9020/metrics | head -10

# 1. Recreate ServiceMonitor
kubectl delete servicemonitor kafka-servicemonitor -n kafka
kubectl apply -n kafka -f - <<EOF
apiVersion: monitoring.coreos.com/v1
kind: ServiceMonitor
metadata:
  name: kafka-servicemonitor
  namespace: kafka
  labels:
    app: kafka
    release: monitoring
spec:
  selector:
    matchLabels:
      app: kafka
  endpoints:
  - port: metrics
    interval: 30s
    path: /metrics
EOF

# 2. Check Prometheus configuration
kubectl get prometheus -o yaml | grep -A 10 serviceMonitorSelector

# Check Grafana pod
kubectl get pods -l app.kubernetes.io/name=grafana

# Check dashboard ConfigMap
kubectl get configmap -l grafana_dashboard=1

# Check Grafana logs
kubectl logs -l app.kubernetes.io/name=grafana

# 1. Restart Grafana
kubectl rollout restart deployment monitoring-grafana

# 2. Recreate dashboard ConfigMap
kubectl delete configmap kafka-looking-glass-dashboard
# Then recreate using the configuration from the deployment section

# Check operator logs for errors
kubectl logs -l app.kubernetes.io/instance=kafka-operator -c manager --tail=100

# Check KafkaCluster status
kubectl describe kafkacluster kafka

# Check operator events
kubectl get events --field-selector involvedObject.kind=KafkaCluster

# 1. Restart operator
kubectl rollout restart deployment kafka-operator-operator

# 2. Check CRD versions
kubectl get crd kafkaclusters.kafka.banzaicloud.io -o yaml | grep version

# 3. Force reconciliation
kubectl annotate kafkacluster kafka kubectl.kubernetes.io/restartedAt="$(date +%Y-%m-%dT%H:%M:%S%z)"

# Check disk usage
kubectl exec -n kafka $BROKER_POD -- df -h

# Clean up old log segments
kubectl exec -n kafka $BROKER_POD -- find /kafka-logs -name "*.log" -mtime +7 -delete

# Increase PVC size or configure retention

# Check if broker is listening
kubectl exec -n kafka $BROKER_POD -- netstat -tuln | grep 9092

# Check broker configuration
kubectl exec -n kafka $BROKER_POD -- cat /opt/kafka/config/server.properties | grep listeners

# Restart broker if needed
kubectl delete pod $BROKER_POD

# Check ZooKeeper status
kubectl get pods -n zookeeper

# Test ZooKeeper connectivity
kubectl run zk-test --rm -i --tty=true \
  --image=busybox \
  --restart=Never \
  -- telnet zk-client.zookeeper 2181

# Check ZooKeeper logs
kubectl logs -n zookeeper zk-0

# Function to start all monitoring port-forwards
function start_monitoring() {
    echo "Starting monitoring port-forwards..."
    
    # Prometheus
    kubectl port-forward -n default svc/monitoring-kube-prometheus-prometheus 9090 &
    echo "Prometheus: http://localhost:9090"
    
    # Grafana
    kubectl port-forward -n default svc/monitoring-grafana 3000:80 &
    echo "Grafana: http://localhost:3000"
    echo "Grafana password: $(kubectl get secret --namespace default monitoring-grafana -o jsonpath="{.data.admin-password}" | base64 --decode)"
    
    # AlertManager
    kubectl port-forward -n default svc/monitoring-kube-prometheus-alertmanager 9093 &
    echo "AlertManager: http://localhost:9093"
    
    # Cruise Control
    kubectl port-forward -n kafka svc/kafka-cruisecontrol-svc 8090:8090 &
    echo "Cruise Control: http://localhost:8090"
    
    echo "All monitoring tools are now accessible!"
}

# Run the function
start_monitoring

# Function to stop all port-forwards
function stop_monitoring() {
    echo "Stopping all port-forwards..."
    pkill -f "kubectl port-forward"
    echo "All port-forwards stopped."
}

# WARNING: This will delete all data!
function emergency_reset() {
    echo "WARNING: This will delete all Kafka data!"
    read -p "Are you sure? (yes/no): " confirm
    
    if [ "$confirm" = "yes" ]; then
        # Delete KafkaCluster
        kubectl delete kafkacluster kafka -n kafka
        
        # Delete all Kafka pods
        kubectl delete pods -l kafka_cr=kafka -n kafka --force --grace-period=0
        
        # Delete PVCs (this deletes data!)
        kubectl delete pvc -l app=kafka -n kafka
        
        # Recreate cluster
        echo "Recreate your KafkaCluster resource to start fresh"
    else
        echo "Reset cancelled"
    fi
}

# Backup ZooKeeper data
kubectl exec -n zookeeper zk-0 -- tar czf /tmp/zk-backup.tar.gz /data

# Copy backup locally
kubectl cp zookeeper/zk-0:/tmp/zk-backup.tar.gz ./zk-backup-$(date +%Y%m%d).tar.gz

# Backup Kafka topic metadata
kubectl run kafka-backup --rm -i --tty=true \
  --image=ghcr.io/adobe/koperator/kafka:2.13-3.9.1 \
  --restart=Never \
  -- /opt/kafka/bin/kafka-topics.sh \
  --bootstrap-server kafka-headless:29092 \
  --list > kafka-topics-backup-$(date +%Y%m%d).txt

# Generate support bundle
function collect_support_info() {
    local output_dir="kafka-support-$(date +%Y%m%d-%H%M%S)"
    mkdir -p $output_dir
    
    # Cluster information
    kubectl cluster-info > $output_dir/cluster-info.txt
    kubectl get nodes -o wide > $output_dir/nodes.txt
    
    # Kafka resources
    kubectl get kafkacluster kafka -n kafka -o yaml > $output_dir/kafkacluster.yaml
    kubectl get pods -n kafka -o wide > $output_dir/kafka-pods.txt
    kubectl get svc -n kafka > $output_dir/kafka-services.txt
    kubectl get pvc -n kafka > $output_dir/kafka-pvcs.txt
    
    # Logs
    kubectl logs -l app.kubernetes.io/instance=kafka-operator -c manager --tail=1000 > $output_dir/operator-logs.txt
    
    # Events
    kubectl get events -n kafka --sort-by=.metadata.creationTimestamp > $output_dir/kafka-events.txt
    
    # Create archive
    tar czf $output_dir.tar.gz $output_dir
    rm -rf $output_dir
    
    echo "Support bundle created: $output_dir.tar.gz"
}

# Run support collection
collect_support_info