kafka-topics.sh --bootstrap-server localhost:9092 \
  --describe --topic orders

# Output shows ISR for each partition:
# Topic: orders  Partition: 0  Leader: 1  Replicas: 1,2  Isr: 1,2
# Topic: orders  Partition: 1  Leader: 2  Replicas: 2,3  Isr: 2,3
# If broker 2 lags, ISR for partition 1 becomes just: 3

# Kafka stores partitions like this on disk:
/kafka-logs/orders-0/
  00000000000000000000.log        # messages from offset 0
  00000000000000000000.index      # offset index
  00000000000000000000.timeindex  # time index
  00000000000001048576.log        # next segment (started at offset 1048576)
  00000000000001048576.index

# Dump a segment to see contents:
kafka-dump-log.sh \
  --files /kafka-logs/orders-0/00000000000000000000.log \
  --print-data-log

# Set retention to 3 days for a topic
kafka-configs.sh --bootstrap-server localhost:9092 \
  --entity-type topics --entity-name orders \
  --alter --add-config retention.ms=259200000

# Set retention by size (keep only 5 GB per partition)
kafka-configs.sh --bootstrap-server localhost:9092 \
  --entity-type topics --entity-name orders \
  --alter --add-config retention.bytes=5368709120

# Create a compacted topic
kafka-topics.sh --bootstrap-server localhost:9092 \
  --create --topic user-profiles \
  --partitions 3 \
  --replication-factor 2 \
  --config cleanup.policy=compact \
  --config min.compaction.lag.ms=60000  # keep for at least 1 min before compaction

# Delete a key using a tombstone (null value)
# Producer sends: key="user1", value=null → user1 gets deleted

from kafka import KafkaProducer
import json

producer = KafkaProducer(
    bootstrap_servers='localhost:9092',
    key_serializer=lambda k: k.encode('utf-8'),
    value_serializer=lambda v: json.dumps(v).encode('utf-8'),
    acks='all',             # wait for all ISR replicas to ack
    enable_idempotence=True  # exactly-once at producer level
)

# key="user_123" → always goes to same partition → order guaranteed
producer.send(
    'orders',
    key='user_123',
    value={'order_id': 'A001', 'amount': 99.99, 'status': 'placed'}
)

producer.flush()  # wait for all pending messages to be sent
producer.close()

from kafka import KafkaProducer

# Idempotent producer — safest configuration
producer = KafkaProducer(
    bootstrap_servers='localhost:9092',
    acks='all',                   # required for idempotence
    enable_idempotence=True,        # enables PID + sequence number tracking
    max_in_flight_requests_per_connection=5,  # max 5 unacked batches
    retries=2147483647,             # retry virtually forever
    compression_type='snappy'       # compress for throughput
)

# Even if broker crashes mid-send, this message appears exactly once
future = producer.send('payments', key=b'pay_001', value=b'{"amount":500}')

# Check send result
try:
    record_metadata = future.get(timeout=10)
    print(f"Sent to partition {record_metadata.partition}, offset {record_metadata.offset}")
except Exception as e:
    print(f"Failed to send: {e}")

from kafka import KafkaConsumer

consumer = KafkaConsumer(
    'orders',
    bootstrap_servers='localhost:9092',
    group_id='order-processor',
    auto_offset_reset='earliest',    # start from beginning if no committed offset
    enable_auto_commit=False,         # manual commit for reliability
    value_deserializer=lambda m: m.decode('utf-8')
)

for message in consumer:
    print(f"Partition: {message.partition}, Offset: {message.offset}, Value: {message.value}")

    # Process the message (save to DB, etc.)
    process_order(message.value)

    # Commit AFTER successful processing
    consumer.commit()  # commitSync()

from kafka import KafkaConsumer

# Use cooperative (incremental) rebalance for lower disruption
consumer = KafkaConsumer(
    'orders',
    bootstrap_servers='localhost:9092',
    group_id='order-processor',
    partition_assignment_strategy=['cooperative-sticky'],
    session_timeout_ms=45000,    # how long before consumer considered dead
    heartbeat_interval_ms=3000    # heartbeat frequency to coordinator
)

# Spark Structured Streaming handles rebalances automatically
# by re-fetching Kafka partition assignments each micro-batch

from pyspark.sql import SparkSession
from pyspark.sql.functions import from_json, col
from pyspark.sql.types import StructType, StructField, StringType, DoubleType

spark = SparkSession.builder.appName("KafkaJSON").getOrCreate()

# Define expected schema
order_schema = StructType([
    StructField("order_id", StringType()),
    StructField("user_id", StringType()),
    StructField("amount", DoubleType()),
    StructField("status", StringType())
])

# Read from Kafka — value is binary bytes
raw_df = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "localhost:9092") \
    .option("subscribe", "orders") \
    .load()

# Cast value bytes to string, then parse JSON
orders_df = raw_df \
    .selectExpr("CAST(key AS STRING) as key", "CAST(value AS STRING) as json_str") \
    .withColumn("data", from_json(col("json_str"), order_schema)) \
    .select("key", "data.*")

orders_df.printSchema()

from pyspark.sql.avro.functions import from_avro

avro_schema = """
{
  "type": "record",
  "name": "Order",
  "fields": [
    {"name": "order_id", "type": "string"},
    {"name": "amount",   "type": "double"},
    {"name": "status",   "type": "string"}
  ]
}
"""

avro_df = raw_df \
    .select(from_avro(col("value"), avro_schema).alias("data")) \
    .select("data.*")

# Using spark-avro with Confluent Schema Registry
# Requires: spark-avro + confluent platform kafka packages

from pyspark.sql.avro.functions import from_avro
import requests

# Fetch schema from registry
schema_registry_url = "http://schema-registry:8081"
subject = "orders-value"
response = requests.get(f"{schema_registry_url}/subjects/{subject}/versions/latest")
avro_schema = response.json()["schema"]

# Strip the 5-byte Confluent wire format header, then parse Avro
# (magic byte + 4-byte schema ID prefix)
parsed_df = raw_df.select(
    from_avro(
        expr("substring(value, 6)"),  # skip 5-byte header
        avro_schema
    ).alias("data")
).select("data.*")

# Spark + Kafka + Delta = end-to-end exactly-once
# Spark checkpoints offsets + Delta's idempotent writes = no duplicates

query = orders_df \
    .writeStream \
    .format("delta") \
    .option("checkpointLocation", "/mnt/checkpoints/orders") \
    .outputMode("append") \
    .start("/mnt/delta/orders")

# If Spark restarts, it:
# 1. Reads last committed Kafka offset from checkpoint
# 2. Re-reads only new messages
# 3. Writes to Delta with same batch ID (Delta deduplicates)
# → Exactly-once end-to-end

from pyspark.sql.functions import when, col, lit, to_json, struct

def process_batch(batch_df, batch_id):
    # Classify records
    classified = batch_df.withColumn(
        "is_valid",
        col("amount").isNotNull() & (col("amount") > 0)
    )

    good_records = classified.filter(col("is_valid"))
    bad_records  = classified.filter(~col("is_valid"))

    # Write good records to Delta
    good_records.drop("is_valid") \
        .write.format("delta") \
        .mode("append") \
        .save("/mnt/delta/orders")

    # Write bad records to Kafka DLQ topic
    dlq_df = bad_records \
        .withColumn("error_reason", lit("invalid_amount")) \
        .withColumn("batch_id", lit(batch_id)) \
        .withColumn("value", to_json(struct("*"))) \
        .select("value")

    dlq_df.write \
        .format("kafka") \
        .option("kafka.bootstrap.servers", "localhost:9092") \
        .option("topic", "orders.DLT") \
        .save()

query = orders_df.writeStream \
    .foreachBatch(process_batch) \
    .option("checkpointLocation", "/mnt/checkpoints/orders") \
    .start()

from pyspark.sql import SparkSession
from pyspark.sql.functions import from_json, col, expr
from pyspark.sql.types import StructType, StructField, StringType, DoubleType

spark = SparkSession.builder \
    .appName("KafkaToSpark") \
    .config("spark.jars.packages",
           "org.apache.spark:spark-sql-kafka-0-10_2.12:3.5.0") \
    .getOrCreate()

# ── Step 1: Read raw bytes from Kafka ──
raw_df = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "localhost:9092") \
    .option("subscribe", "orders") \
    .option("startingOffsets", "latest") \
    .option("maxOffsetsPerTrigger", 10000) \
    .option("failOnDataLoss", "false") \
    .load()

# ── Step 2: Define schema for JSON payload ──
schema = StructType([
    StructField("order_id", StringType()),
    StructField("user_id",  StringType()),
    StructField("amount",   DoubleType()),
    StructField("product",  StringType())
])

# ── Step 3: Parse JSON from the value column ──
orders_df = raw_df \
    .select(
        expr("CAST(key AS STRING) AS key"),
        expr("CAST(value AS STRING) AS raw_json"),
        col("topic"),
        col("partition"),
        col("offset"),
        col("timestamp")
    ) \
    .withColumn("data", from_json(col("raw_json"), schema)) \
    .select("key", "data.*", "topic", "partition", "offset", "timestamp")

# ── Step 4: Write to console for testing ──
query = orders_df.writeStream \
    .format("console") \
    .outputMode("append") \
    .option("truncate", False) \
    .start()

query.awaitTermination()

# Batch read: read a specific range of offsets
batch_df = spark.read \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "localhost:9092") \
    .option("subscribe", "orders") \
    .option("startingOffsets", """{"orders":{"0":0,"1":0,"2":0}}""") \
    .option("endingOffsets",   """{"orders":{"0":100,"1":100,"2":100}}""") \
    .load()

print(f"Records read: {batch_df.count()}")

from pyspark.sql.functions import to_json, struct, col

# Enrich orders — add a "priority" field
enriched_df = orders_df \
    .withColumn("priority",
        when(col("amount") > 1000, "HIGH")
        .when(col("amount") > 100, "MEDIUM")
        .otherwise("LOW")
    )

# Prepare for Kafka: must have 'key' and 'value' columns
kafka_output_df = enriched_df \
    .withColumn("key", col("order_id")) \
    .withColumn("value", to_json(struct(
        "order_id", "user_id", "amount", "priority"
    ))) \
    .select("key", "value")

# Write enriched stream back to Kafka
query = kafka_output_df.writeStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "localhost:9092") \
    .option("topic", "orders-enriched") \
    .option("checkpointLocation", "/mnt/checkpoints/orders-enriched") \
    .outputMode("append") \
    .start()

from pyspark.sql.functions import when, col

# Route HIGH priority orders to "orders-high" topic, others to "orders-standard"
routed_df = enriched_df \
    .withColumn("key", col("order_id")) \
    .withColumn("value", to_json(struct("*"))) \
    .withColumn("topic",           # dynamic topic column!
        when(col("priority") == "HIGH", "orders-high")
        .otherwise("orders-standard")
    ) \
    .select("key", "value", "topic")  # include 'topic' column for routing

query = routed_df.writeStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "localhost:9092") \
    .option("checkpointLocation", "/mnt/checkpoints/routed") \
    .start()  # no .option("topic") needed — uses per-row 'topic' column

# Option 1: Start from latest (skip all existing messages)
df = spark.readStream.format("kafka") \
    .option("kafka.bootstrap.servers", "localhost:9092") \
    .option("subscribe", "orders") \
    .option("startingOffsets", "latest") \
    .load()

# Option 2: Start from beginning (read all historical data)
df = spark.readStream.format("kafka") \
    .option("startingOffsets", "earliest") \
    .load()

# Option 3: Start from specific offsets (per partition)
specific_offsets = """
{
  "orders": {
    "0": 1500,
    "1": 2000,
    "2": 800
  }
}
"""
df = spark.readStream.format("kafka") \
    .option("startingOffsets", specific_offsets) \
    .load()

# Option 4: Start from timestamp (Kafka 0.10+)
# Start from messages produced after a specific time
timestamp_ms = 1700000000000  # epoch milliseconds
df = spark.readStream.format("kafka") \
    .option("startingOffsetsByTimestamp",
           f'{{"orders":{{"0":{timestamp_ms},"1":{timestamp_ms}}}}}') \
    .load()

df = spark.readStream.format("kafka") \
    .option("kafka.bootstrap.servers", "localhost:9092") \
    .option("subscribe", "orders") \
    .option("failOnDataLoss", "false") \  # don't fail if Kafka deleted old offsets
    .option("maxOffsetsPerTrigger", 50000) \  # limit records per batch
    .option("kafka.session.timeout.ms", "30000") \
    .load()

# Subscribe to multiple topics explicitly
multi_df = spark.readStream.format("kafka") \
    .option("kafka.bootstrap.servers", "localhost:9092") \
    .option("subscribe", "orders,payments,refunds") \  # comma-separated
    .option("startingOffsets", "latest") \
    .load()

# Use the 'topic' column to route records
from pyspark.sql.functions import col

orders_only   = multi_df.filter(col("topic") == "orders")
payments_only = multi_df.filter(col("topic") == "payments")
refunds_only  = multi_df.filter(col("topic") == "refunds")

# Subscribe to all topics matching pattern (e.g. orders, orders-eu, orders-us)
pattern_df = spark.readStream.format("kafka") \
    .option("kafka.bootstrap.servers", "localhost:9092") \
    .option("subscribePattern", "orders.*") \  # Java regex
    .option("startingOffsets", "latest") \
    .load()

# New topic "orders-apac" created → auto-included next trigger

# Handling different schemas per topic using a when() approach
from pyspark.sql.functions import when, from_json, col

orders_schema  = StructType([StructField("order_id", StringType()), StructField("amount", DoubleType())])
payments_schema= StructType([StructField("pay_id",   StringType()), StructField("amount", DoubleType())])

value_str = pattern_df.selectExpr("topic", "CAST(value AS STRING) AS val")

parsed = value_str \
    .withColumn("order_data",
        when(col("topic") == "orders", from_json(col("val"), orders_schema))
    ) \
    .withColumn("pay_data",
        when(col("topic") == "payments", from_json(col("val"), payments_schema))
    )