# Create topic with specific configurations
# --bootstrap-server: comma-separated list of broker addresses for initial connection
# --topic: name of the topic (kebab-case recommended)
# --partitions: number of partitions (determines max parallelism)
# --replication-factor: number of replicas per partition (3 for production)
# --config: topic-level overrides for server properties

kafka-topics.sh --create \
  --bootstrap-server kafka-broker-0:9092,kafka-broker-1:9092,kafka-broker-2:9092 \
  --topic order-events \
  --partitions 12 \
  --replication-factor 3 \
  --config min.insync.replicas=2 \
  --config max.message.bytes=10485760 \
  --config retention.ms=604800000 \
  --config cleanup.policy=delete \
  --config compression.type=lz4 \
  --config segment.bytes=1073741824 \
  --config flush.messages=10000 \
  --config index.interval.bytes=4096 \
  --config unclean.leader.election.enable=false

# min.insync.replicas=2: minimum replicas that must acknowledge a write before it's acknowledged to producer
# max.message.bytes=10485760: maximum message size (10MB) per batch
# retention.ms=604800000: retain data for 7 days (604800000ms)
# cleanup.policy=delete: delete segments after retention expires (alternative: compact)
# compression.type=lz4: compression algorithm for batch encoding (none, gzip, snappy, lz4, zstd)
# segment.bytes=1073741824: roll log segment after 1GB (1073741824 bytes)
# flush.messages=10000: flush to disk after every 10000 messages (durability vs performance)
# index.interval.bytes=4096: add index entry every 4096 bytes for offset lookup
# unclean.leader.election.enable=false: prevent out-of-sync replica from becoming leader

#!/bin/bash
# Monitor Kafka cluster health and partition distribution
# Uses kafka CLI tools for comprehensive cluster visibility

BOOTSTRAP_SERVER="kafka-broker-0:9092"

echo "=== KAFKA CLUSTER STATUS ==="
echo "Timestamp: $(date)"
echo ""

# List all brokers and their API versions
# Shows available APIs and supported versions for client compatibility
echo "--- Active Brokers ---"
kafka-broker-api-versions.sh --bootstrap-server $BOOTSTRAP_SERVER | head -20

echo ""
echo "--- Topic Partition Distribution ---"
# Describe shows partition leaders, ISRs, and replica placement
# Use --topic to filter to specific topic
kafka-topics.sh --describe --bootstrap-server $BOOTSTRAP_SERVER --topic order-events

echo ""
echo "--- Consumer Group Lag ---"
# Lag = log_end_offset - committed_offset per partition
# High lag indicates consumer cannot keep up with production rate
kafka-consumer-groups.sh --bootstrap-server $BOOTSTRAP_SERVER \
  --group order-processing-service \
  --describe

echo ""
echo "--- Cluster Config ---"
# List all broker-level configurations
kafka-configs.sh --bootstrap-server $BOOTSTRAP_SERVER \
  --entity-type brokers \
  --entity-default \
  --describe

echo ""
echo "--- Under-replicated Partitions ---"
# Under-replicated = ISR size < replication factor
# Indicates broker failure or network partition
kafka-topics.sh --describe --bootstrap-server $BOOTSTRAP_SERVER \
  | grep -i "under-replicated"

echo ""
echo "--- Active Controller ---"
# Exactly one broker should be the controller
kafka-metadata.sh --snapshot /var/kafka/metadata/metadata-1.log \
  --cluster-id <cluster-id> --command-config admin.properties 2>/dev/null || \
  echo "Check controller via JMX: kafka.controller:type=KafkaController,name=ActiveControllerCount"

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;
import java.util.Properties;
import java.util.concurrent.Future;

public class KafkaProducerExample {
    
    public static void main(String[] args) {
        Properties props = new Properties();
        
        // Core configurations
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, 
            "kafka-broker-0:9092,kafka-broker-1:9092,kafka-broker-2:9092");
        // Comma-separated broker list for initial metadata fetch;
        // client discovers all brokers from metadata response
        
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, 
            StringSerializer.class.getName());
        // Serializer for message keys; determines partition routing
        // Options: StringSerializer, LongSerializer, ByteArraySerializer
        
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, 
            StringSerializer.class.getName());
        // Serializer for message values; independent of key serializer
        
        // Reliability configurations
        props.put(ProducerConfig.ACKS_CONFIG, "all");
        // acks=0: no acknowledgment (fastest, no durability)
        // acks=1: leader acknowledgment only (moderate durability)
        // acks=all: all ISR replicas acknowledge (highest durability)
        
        props.put(ProducerConfig.RETRIES_CONFIG, 10);
        // Number of retries on transient failures (e.g., LEADER_NOT_AVAILABLE)
        // With idempotence enabled, retries are unlimited by default
        
        props.put(ProducerConfig.RETRY_BACKOFF_MS_CONFIG, 100);
        // Delay between retries to avoid overwhelming broker
        // Exponential backoff applied on consecutive failures
        
        props.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 5);
        // Max unacknowledged requests per broker connection
        // >1 enables pipelining but may cause reordering without idempotence
        // With idempotence, ordering is preserved even with value > 1
        
        // Idempotent producer for exactly-once semantics
        props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
        // Prevents duplicate messages during retries
        // Assigns Producer ID (PID) and per-partition sequence numbers
        // Automatically sets acks=all, retries=Integer.MAX_VALUE
        
        props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "order-producer-1");
        // Unique identifier for transactional producer
        // Enables atomic writes across multiple partitions/topics
        // Must be unique per producer instance (fencing via epoch)
        
        // Performance configurations
        props.put(ProducerConfig.BATCH_SIZE_CONFIG, 65536);
        // Maximum batch size in bytes (64KB)
        // Batch is sent when full OR linger.ms expires, whichever first
        // Larger batches improve throughput, increase latency
        
        props.put(ProducerConfig.LINGER_MS_CONFIG, 10);
        // Time to wait before sending batch even if not full
        // 0 = send immediately (lowest latency, lowest throughput)
        // 10-50ms = good balance for most workloads
        
        props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);
        // Total memory for buffering records (32MB)
        // Records waiting to be sent to broker
        // If buffer is full, send() blocks until space is available
        
        props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "lz4");
        // Compression algorithm applied to batches
        // none: no compression (fastest, largest payload)
        // gzip: best compression ratio, high CPU usage
        // snappy: moderate compression, low CPU
        // lz4: good compression, fast decompression (recommended)
        // zstd: best compression ratio, moderate CPU (Kafka 2.1+)
        
        // Monitoring
        props.put(ProducerConfig.METRICS_NUM_SAMPLES_CONFIG, 10);
        // Number of samples to keep for metric calculations
        
        props.put(ProducerConfig.METRICS_SAMPLE_WINDOW_MS_CONFIG, 30000);
        // Time window for metric aggregation (30 seconds)
        
        KafkaProducer<String, String> producer = new KafkaProducer<>(props);
        
        // Initialize transactions (required if transactional.id is set)
        // Acquires Producer ID and epoch from Transaction Coordinator
        producer.initTransactions();
        
        try {
            // Begin transaction for atomic multi-partition writes
            // All sends until commitTransaction() are atomic
            producer.beginTransaction();
            
            for (int i = 0; i < 1000; i++) {
                String key = "order-" + (i % 100);
                // Key determines partition: hash(key) % numPartitions
                // Same key always routes to same partition (ordering guarantee)
                
                String value = "{\"orderId\": \"" + i + "\", \"amount\": " + 
                    (Math.random() * 1000) + "}";
                
                ProducerRecord<String, String> record = 
                    new ProducerRecord<>("order-events", key, value);
                // Topic, key, value; partition determined by key hash
                // Optional: specify partition directly as 3rd argument
                
                record.headers().add("correlation-id", 
                    java.util.UUID.randomUUID().toString().getBytes());
                // Headers carry metadata without affecting partition routing
                // Useful for tracing, audit, and cross-system correlation
                
                record.headers().add("source-system", "order-service".getBytes());
                // Identify the originating service for observability
                
                Future<RecordMetadata> future = producer.send(record, 
                    (metadata, exception) -> {
                        if (exception != null) {
                            System.err.println("Failed to send record: " + 
                                exception.getMessage());
                        } else {
                            System.out.printf("Sent record to partition %d, " +
                                "offset %d, timestamp %d%n",
                                metadata.partition(), metadata.offset(),
                                metadata.timestamp());
                        }
                    });
                // Async send with callback; callback runs on sender thread
                // For sync send, call future.get() (blocks, lower throughput)
            }
            
            // Commit transaction atomically
            // Writes TxnMarker to all affected partitions
            // Marks transaction as committed in __transaction_state
            producer.commitTransaction();
            
        } catch (Exception e) {
            // Abort rolls back all writes in this transaction
            // Writes become invisible to read_committed consumers
            producer.abortTransaction();
            throw e;
        } finally {
            // Close releases resources, flushes pending sends
            producer.close();
        }
    }
}

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;
import java.time.Duration;
import java.util.*;
import java.util.concurrent.atomic.AtomicBoolean;

public class KafkaConsumerExample {
    
    private static final AtomicBoolean running = new AtomicBoolean(true);
    
    public static void main(String[] args) {
        Properties props = new Properties();
        
        // Core configurations
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, 
            "kafka-broker-0:9092,kafka-broker-1:9092,kafka-broker-2:9092");
        // Initial broker list; consumer discovers full cluster from metadata
        
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "order-processing-service");
        // Consumer group identifier
        // Partitions are assigned among consumers with same group.id
        
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, 
            StringDeserializer.class.getName());
        // Deserializer for message keys; must match producer's key serializer
        
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, 
            StringDeserializer.class.getName());
        // Deserializer for message values; must match producer's value serializer
        
        // Consumer behavior
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
        // false = manual offset commit (recommended for exactly-once)
        // true = auto-commit every auto.commit.interval.ms (default 5000ms)
        // Auto-commit may lose messages if processing fails after commit
        
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        // earliest: read from beginning if no committed offset exists
        // latest: read from end (new messages only)
        // none: throw exception if no committed offset
        
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 500);
        // Maximum records returned per poll() call
        // Lower values reduce poll loop latency
        // Higher values improve throughput
        
        props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 300000);
        // Maximum time between poll() calls before consumer is evicted
        // If processing takes longer, consumer is considered dead
        // Increase for slow processing; decrease for fast detection
        
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 45000);
        // Maximum time without heartbeat before consumer is considered dead
        // Detected by group coordinator; triggers rebalance
        
        props.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, 15000);
        // Heartbeat sent to coordinator periodically
        // Should be <= session.timeout.ms / 3 for reliable detection
        
        // Isolation level for transactional reads
        props.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");
        // read_committed: only see records from committed transactions
        // read_uncommitted: see all records including uncommitted
        // Required for exactly-once semantics end-to-end
        
        // Partition assignment strategy
        props.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG, 
            "org.apache.kafka.clients.consumer.CooperativeStickyAssignor");
        // CooperativeStickyAssignor: incremental rebalancing (recommended)
        //   Only revokes partitions that need to move
        // RangeAssignor: assigns contiguous partition ranges (default)
        // RoundRobinAssignor: distributes partitions evenly
        
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        
        // Subscribe to topics with rebalance listener
        consumer.subscribe(Arrays.asList("order-events", "payment-events"), 
            new ConsumerRebalanceListener() {
                @Override
                public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
                    // Called before partitions are revoked (rebalance start)
                    // Commit pending offsets to avoid reprocessing
                    System.out.println("Partitions revoked: " + partitions);
                    consumer.commitSync(Duration.ofSeconds(30));
                }
                
                @Override
                public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
                    // Called after new partitions are assigned
                    // Seek to specific offsets if needed
                    System.out.println("Partitions assigned: " + partitions);
                }
            });
        
        try {
            while (running.get()) {
                // poll() fetches records and maintains group membership
                // Timeout controls how long to wait if no records available
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
                
                for (ConsumerRecord<String, String> record : records) {
                    processRecord(record);
                }
                
                // Commit offsets after successful processing
                // commitSync: blocks until successful or timeout
                // commitAsync: non-blocking, may fail silently
                if (!records.isEmpty()) {
                    consumer.commitSync(Duration.ofSeconds(30));
                }
            }
        } finally {
            // Close releases consumer from group, triggers rebalance
            // May want to delay close for graceful shutdown
            consumer.close();
        }
    }
    
    private static void processRecord(ConsumerRecord<String, String> record) {
        System.out.printf("Topic: %s, Partition: %d, Offset: %d, Key: %s, Value: %s%n",
            record.topic(), record.partition(), record.offset(), 
            record.key(), record.value());
        // Access record.timestamp() for event-time processing
        // Access record.headers() for message metadata
    }
}

from kafka import KafkaProducer
import json
import time

# Initialize producer with configuration
producer = KafkaProducer(
    bootstrap_servers=['kafka-broker-0:9092', 'kafka-broker-1:9092'],
    # Comma-separated broker addresses for metadata discovery
    
    key_serializer=lambda k: k.encode('utf-8'),
    # Serialize key to bytes; determines partition via hash
    
    value_serializer=lambda v: json.dumps(v).encode('utf-8'),
    # Serialize value to JSON bytes
    
    acks='all',
    # Wait for all ISR replicas to acknowledge
    # 'none' = 0 (no ack), 'one' = 1 (leader only), 'all' = all ISR
    
    retries=5,
    # Retry transient failures up to 5 times
    
    retry_backoff_ms=100,
    # 100ms delay between retries
    
    batch_size=65536,
    # 64KB batch size for network efficiency
    
    linger_ms=10,
    # Wait up to 10ms to fill batch
    
    compression_type='lz4',
    # Compress batches with lz4 (good balance of speed/ratio)
    
    max_in_flight_requests_per_connection=5,
    # Allow 5 concurrent requests per broker (pipelining)
    
    enable_idempotence=True,
    # Prevent duplicate messages during retries
    
    buffer_memory=33554432,
    # 32MB buffer for pending records
    
    request_timeout_ms=30000,
    # 30s timeout for broker responses
)

# Send messages with callback
for i in range(1000):
    key = f"order-{i % 100}"
    value = {
        "orderId": str(i),
        "amount": round(1000.0, 2),
        "timestamp": int(time.time() * 1000)
    }
    
    future = producer.send(
        topic='order-events',
        key=key,
        value=value,
        headers=[
            ("correlation-id", str(i).encode('utf-8')),
            ("source", b"python-producer")
        ]
    )
    
    # Add callback for async result handling
    future.add_callback(lambda metadata: print(
        f"Sent to partition {metadata.partition()}, "
        f"offset {metadata.offset()}, "
        f"timestamp {metadata.timestamp}"
    ))
    
    future.add_errback(lambda exc: print(f"Error: {exc}"))

# Flush ensures all pending records are sent
producer.flush()
producer.close()

from kafka import KafkaConsumer
import json

# Initialize consumer with configuration
consumer = KafkaConsumer(
    'order-events',
    # Topic(s) to subscribe to; can be list for multiple topics
    
    bootstrap_servers=['kafka-broker-0:9092', 'kafka-broker-1:9092'],
    # Initial broker list
    
    group_id='order-processing-service',
    # Consumer group for partition assignment
    
    key_deserializer=lambda k: k.decode('utf-8') if k else None,
    # Deserialize key from bytes
    
    value_deserializer=lambda v: json.loads(v.decode('utf-8')) if v else None,
    # Deserialize value from JSON bytes
    
    auto_offset_reset='earliest',
    # 'earliest': read from beginning if no offset
    # 'latest': read from end if no offset
    # 'none': raise exception if no offset
    
    enable_auto_commit=False,
    # Disable auto-commit for manual offset management
    
    isolation_level='read_committed',
    # Only read committed transactional records
    
    max_poll_records=500,
    # Max records per poll() call
    
    max_poll_interval_ms=300000,
    # Max time between polls before eviction (300s)
    
    session_timeout_ms=45000,
    # Consumer failure detection timeout
    
    heartbeat_interval_ms=15000,
    # Heartbeat frequency to coordinator
    
    fetch_min_bytes=1,
    # Minimum bytes to fetch (batch optimization)
    
    fetch_max_wait_ms=500,
    # Max wait if fetch_min_bytes not reached
)

# Subscribe with rebalance listener
def on_partitions_revoked(partitions):
    print(f"Partitions revoked: {partitions}")
    consumer.commitSync()

def on_partitions_assigned(partitions):
    print(f"Partitions assigned: {partitions}")

consumer.subscribe(
    topics=['order-events', 'payment-events'],
    listener=ConsumerRebalanceListener(
        on_partitions_revoked=on_partitions_revoked,
        on_partitions_assigned=on_partitions_assigned
    )
)

# Process records
try:
    while True:
        records = consumer.poll(timeout_ms=100)
        
        for tp, records_list in records.items():
            for record in records_list:
                print(f"Topic: {record.topic}, "
                      f"Partition: {record.partition}, "
                      f"Offset: {record.offset}, "
                      f"Key: {record.key}, "
                      f"Value: {record.value}")
                
                # Process record here
                # After processing, commit offset
                consumer.commitSync({
                    tp: OffsetAndMetadata(record.offset + 1, "processed")
                })
                
except KeyboardInterrupt:
    consumer.close()