run_query.sh

#!/usr/bin/bash 

# Configurable variables
ENV_DIR="environment"


LOG_DIR="/tmp/kraft-combined-logs"  # KRaft storage
BROKER_LOGS="$ENV_DIR/kafka_2.13-4.1.0/logs/server.log"

CURRENT_DIR="$(pwd)"

CONFIG="config.json"

#RUN_TOPIC=$(jq -r '.run_topic' "$CONFIG")
RUN_TOPIC=$1

# System config
FPS=$(jq -r '.system.input_fps' "$CONFIG")
WINDOW_SIZE=$(jq -r '.system.window_size' "$CONFIG")
SKIP_AMOUNT=$(jq -r '.system.skip_amount' "$CONFIG")

# LLM config
LLM=$(jq -r '.LLM' "$CONFIG")

#Topic Config
KAFKA_TOPIC=$(jq -r ".topics.$RUN_TOPIC.kafka_topic" "$CONFIG")
VIDEO_SENDER=$(jq -r ".topics.$RUN_TOPIC.video_sender" "$CONFIG")

QUERY_NAME="stolen_red_cars_naive"

#SAVE_PATH="results/$LLM/$RUN_TOPIC/$WINDOW_SIZE-$SKIP_AMOUNT/$(date +'%Y_%m_%d_%H_%M_%S')"
SAVE_PATH="results/$LLM/$2/$RUN_TOPIC/$WINDOW_SIZE-$SKIP_AMOUNT"
mkdir -p "$SAVE_PATH"

echo "+---------- CONFIG ----------+"
echo
echo "Run Topic: $RUN_TOPIC"
echo
echo "Input FPS: $FPS"
echo "Window Size: $WINDOW_SIZE"
echo "Skip Amount: $SKIP_AMOUNT"
echo
echo "LLM: $LLM"
echo
echo "Kafka Topic: $KAFKA_TOPIC"
echo "Video Sender: $VIDEO_SENDER"
echo "Result Folder: $SAVE_PATH"
echo
echo "+-------- CONFIG END --------+"

echo

# Set JAVA_HOME variable
export JAVA_HOME="$PWD/$ENV_DIR/jdk-17.0.9"
export PATH="$JAVA_HOME/bin:$PATH"
echo "Set JAVA_HOME to $PWD/$ENV_DIR/jdk-17.0.9" 
echo 

# Clean old Kafka logs
if [ -d "$LOG_DIR" ]; then
    echo "Cleaning old Kafka log."
    rm -rf "$LOG_DIR"
fi

# Clean previous broker log
if [ -f "$BROKER_LOGS" ]; then
    echo "Cleaning old Kafka broker log."
    rm -f "$BROKER_LOGS"
fi

# Format Kafka storage
UUID=$("$ENV_DIR/kafka_2.13-4.1.0/bin/kafka-storage.sh" random-uuid)
"$ENV_DIR/kafka_2.13-4.1.0/bin/kafka-storage.sh" format --standalone -t "$UUID" -c "$ENV_DIR/kafka_2.13-4.1.0/config/server.properties"


# Start Kafka in daemon mode
echo
echo "Starting Kafka broker."
"$ENV_DIR/kafka_2.13-4.1.0/bin/kafka-server-start.sh" -daemon "$ENV_DIR/kafka_2.13-4.1.0/config/server.properties"

# Wait for broker to be ready
echo "Waiting for Kafka broker to be ready."

MAX_WAIT=60
WAITED=0
until "$ENV_DIR/kafka_2.13-4.1.0/bin/kafka-topics.sh" --list --bootstrap-server localhost:9092 >/dev/null 2>&1; do
    sleep 2
    WAITED=$((WAITED+2))
    if [ $WAITED -ge $MAX_WAIT ]; then
        echo "Kafka broker did not start in time. Check logs at $BROKER_LOGS"
        exit 1
    fi
done
echo "Kafka broker is ready."

# Create topic with retries
echo "Creating topic '$KAFKA_TOPIC'."
"$ENV_DIR/kafka_2.13-4.1.0/bin/kafka-topics.sh" \
        --create --topic "$KAFKA_TOPIC" \
        --bootstrap-server localhost:9092 \
        --partitions 1 --replication-factor 1 --if-not-exists


# Send video frames to topic
echo 
echo "Sending frames to topic $KAFKA_TOPIC."
python "$VIDEO_SENDER" --topic "$KAFKA_TOPIC" --fps "$FPS"

# Wait to ensure messages are produced
sleep 5

# Read queries and their job paths
declare -A QUERIES

while IFS="=" read -r key value; do
    QUERIES["$key"]="$value"
done < <(
    jq -r --arg run_topic "$RUN_TOPIC" '.topics[$run_topic].queries | to_entries[] | "\(.key)=\(.value)"' "$CONFIG"

)

echo
echo "Stopping running Flink clusters."
"$ENV_DIR/flink-1.20.3/bin/stop-cluster.sh"

echo 
echo "+---------- QUERIES ----------+"

for QUERY in "${!QUERIES[@]}"; do
    echo
    echo "Processing query: $QUERY - ${QUERIES[$QUERY]}."

    # Start Flink cluster
    echo
    "$ENV_DIR/flink-1.20.3/bin/start-cluster.sh"

    sleep 30

    nohup "$ENV_DIR/flink-1.20.3/bin/flink" run -d \
        -py "$CURRENT_DIR/${QUERIES[$QUERY]}" \
        --pyFiles llm_call.py,helper_functions.py,yolo11n.pt,config.json

    echo "Waiting for the Flink job to be submitted."

    JOB_ID=""

    while [ -z "$JOB_ID" ]; do
        JOB_ID=$("$ENV_DIR/flink-1.20.3/bin/flink" list | awk -F' : ' -v job="$QUERY" '$3 ~ ("^" job " \\(") { print $2 }')
        if [ -z "$JOB_ID" ]; then
            echo "No jobs yet, waiting."
            sleep 5
        fi
    done

    echo "Job submitted with Job ID: $JOB_ID"

    python receive_answers.py --query_name "$QUERY" --save_path "$SAVE_PATH"

    sleep 10

    "$ENV_DIR/flink-1.20.3/bin/flink" cancel $JOB_ID

    # Stop Flink cluster
    echo "Stopping Flink cluster."
    "$ENV_DIR/flink-1.20.3/bin/stop-cluster.sh"
done

echo 
echo "+-------- QUERIES END --------+" 


# Stop Kafka broker
echo
echo "Stopping Kafka broker."
"$ENV_DIR/kafka_2.13-4.1.0/bin/kafka-server-stop.sh"

echo
echo "Done."