Amazon MSK: Managed Kafka Streaming — Production Patterns That Don't Fail at 3 AM

// io.thecodeforge — DevOps tutorial

# Create a topic with safe defaults for production
kafka-topics.sh --create \
  --bootstrap-server $MSK_BROKER_STRING \
  --replication-factor 3 \
  --partitions 6 \
  --config min.insync.replicas=2 \
  --config retention.ms=604800000 \
  --config retention.bytes=53687091200 \
  --config cleanup.policy=delete \
  --topic orders

// io.thecodeforge — DevOps tutorial

import org.apache.kafka.clients.producer.*;
import java.util.Properties;

public class OrderProducer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", System.getenv("MSK_BROKER_STRING"));
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        // High throughput settings
        props.put("batch.size", 65536);       // 64 KB batches
        props.put("linger.ms", 10);            // Wait up to 10ms for more messages
        props.put("compression.type", "snappy");
        props.put("acks", "1");               // Leader ack only — faster, safe with min.insync.replicas=2
        props.put("max.in.flight.requests.per.connection", 5);
        props.put("retries", 3);
        props.put("enable.idempotence", true); // Prevent duplicates on retry

        KafkaProducer<String, String> producer = new KafkaProducer<>(props);
        // Send orders
        for (int i = 0; i < 1000; i++) {
            producer.send(new ProducerRecord<>("orders", "order-" + i, "{\"id\":" + i + "}"));
        }
        producer.close();
    }
}

// io.thecodeforge — DevOps tutorial

import org.apache.kafka.clients.consumer.*;
import java.time.Duration;
import java.util.Properties;
import java.util.List;

public class OrderConsumer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", System.getenv("MSK_BROKER_STRING"));
        props.put("group.id", "order-processors");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        // Stable consumer settings
        props.put("session.timeout.ms", 45000);
        props.put("heartbeat.interval.ms", 15000);
        props.put("max.poll.interval.ms", 300000);  // 5 min to process a batch
        props.put("max.poll.records", 500);
        props.put("partition.assignment.strategy", "org.apache.kafka.clients.consumer.CooperativeStickyAssignor");
        props.put("enable.auto.commit", false);  // Manual commit for control

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(List.of("orders"));

        try {
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
                for (ConsumerRecord<String, String> record : records) {
                    processOrder(record.value());
                }
                consumer.commitSync();  // Commit after successful batch
            }
        } finally {
            consumer.close();
        }
    }

    private static void processOrder(String orderJson) {
        // Simulate processing
        System.out.println("Processing: " + orderJson);
    }
}

// io.thecodeforge — DevOps tutorial

# Create CloudWatch alarm for broker disk usage
aws cloudwatch put-metric-alarm \
  --alarm-name MSK-HighDiskUsage \
  --alarm-description "Alert when broker disk usage exceeds 80%" \
  --metric-name KafkaDataLogsDiskUsed \
  --namespace AWS/Kafka \
  --statistic Maximum \
  --period 300 \
  --threshold 80 \
  --comparison-operator GreaterThanThreshold \
  --dimensions Name=Cluster Name,Value=my-msk-cluster \
  --evaluation-periods 2 \
  --alarm-actions arn:aws:sns:us-east-1:123456789012:MyTopic

// io.thecodeforge — DevOps tutorial

# Update MSK cluster to increase number of brokers
aws kafka update-broker-count \
  --cluster-arn arn:aws:kafka:us-east-1:123456789012:cluster/my-cluster/abcd1234 \
  --target-number-of-broker-nodes 6

# Wait for update to complete (check status)
aws kafka describe-cluster \
  --cluster-arn arn:aws:kafka:us-east-1:123456789012:cluster/my-cluster/abcd1234 \
  --query 'ClusterInfo.State'