# Step 1: Generate CA, broker keystore, client keystore
# CA key and certificate
openssl req -new -x509 -keyout ca.key -out ca.crt -days 365 \
  -subj "/CN=Kafka-CA/O=Company"

# Broker: generate key, CSR, sign with CA
keytool -keystore broker.keystore.jks -alias broker -validity 365 \
  -keyalg RSA -genkey -dname "CN=kafka-broker-1,OU=Backend,O=Company"

keytool -keystore broker.keystore.jks -alias broker -certreq \
  -file broker.csr

openssl x509 -req -CA ca.crt -CAkey ca.key -in broker.csr \
  -out broker-signed.crt -days 365 -CAcreateserial

keytool -keystore broker.keystore.jks -alias CARoot -import -file ca.crt
keytool -keystore broker.keystore.jks -alias broker -import -file broker-signed.crt

# Client: same pattern with client.keystore.jks

# server.properties — Kafka broker with mTLS enabled
listeners=SSL://0.0.0.0:9093
advertised.listeners=SSL://kafka-broker-1.internal:9093
ssl.keystore.location=/etc/kafka/certs/broker.keystore.jks
ssl.keystore.password=${KEYSTORE_PASSWORD}
ssl.truststore.location=/etc/kafka/certs/broker.truststore.jks
ssl.truststore.password=${TRUSTSTORE_PASSWORD}
ssl.client.auth=required         # mutual: client MUST present a certificate
ssl.endpoint.identification.algorithm=HTTPS  # broker hostname verification

// kafkajs client with mTLS
import { readFileSync } from 'fs'

const kafka = new Kafka({
  clientId: 'payment-service',
  brokers:  ['kafka-broker-1.internal:9093'],
  ssl: {
    rejectUnauthorized: true,
    ca:   [readFileSync('/etc/certs/ca.crt')],
    cert:  readFileSync('/etc/certs/client.crt'),
    key:   readFileSync('/etc/certs/client.key'),
  }
})

// kafkajs SASL/SCRAM-SHA-512
const kafka = new Kafka({
  clientId: 'analytics-service',
  brokers:  ['kafka-broker-1.internal:9092'],
  sasl: {
    mechanism: 'scram-sha-512',
    username:  process.env.KAFKA_USERNAME!,
    password:  process.env.KAFKA_PASSWORD!,
  }
})
// Create user: kafka-configs.sh --bootstrap-server ... --alter --add-config
//   'SCRAM-SHA-512=[iterations=8192,password=xxx]' --entity-type users --entity-name analytics-service

# Grant analytics-service read access to payments topic only
kafka-acls.sh --bootstrap-server kafka:9092 \
  --add --allow-principal "User:analytics-service" \
  --operation Read \
  --topic payments.charged

# Grant payment-service write access to payments topic
kafka-acls.sh --bootstrap-server kafka:9092 \
  --add --allow-principal "User:payment-service" \
  --operation Write \
  --topic payments.charged

# Deny all other principals by default (set allow.everyone.if.no.acl.found=false)

// ✅ MSK IAM: no username, no password, no certificate — IAM role IS the identity
// The ECS task role or EC2 instance profile provides credentials automatically

import { generateAuthToken } from 'aws-msk-iam-sasl-signer-js'

const kafka = new Kafka({
  clientId: 'payment-service',
  brokers:  [process.env.MSK_BOOTSTRAP!],
  ssl:      true,
  sasl: {
    mechanism: 'oauthbearer',
    oauthBearerProvider: async () => {
      const { token, expiryTime } = await generateAuthToken({ region: 'us-east-1' })
      return { value: token }
    }
  }
})
// MSK rotates the token automatically via the IAM SDK
// No password to store, no certificate to rotate, no secret to leak
// Audit: every connection attempt appears in CloudTrail as an IAM event

# mm2.properties — MirrorMaker 2 active-passive replication
clusters=us-east-1, eu-west-1

us-east-1.bootstrap.servers=kafka-primary:9092
eu-west-1.bootstrap.servers=kafka-replica:9092

us-east-1->eu-west-1.enabled=true
us-east-1->eu-west-1.topics=payments.charged,orders.created,inventory.events
us-east-1->eu-west-1.groups=fulfillment-service,analytics-service,notification-service

# Offset translation: MirrorMaker 2 maps us-east-1 offsets to eu-west-1 equivalents
# Stored in: eu-west-1.mm2-offset-syncs.us-east-1.internal topic
us-east-1->eu-west-1.emit.checkpoints.enabled=true
us-east-1->eu-west-1.sync.group.offsets.enabled=true
us-east-1->eu-west-1.sync.group.offsets.interval.seconds=60

# DR RUNBOOK: Promoting eu-west-1 passive cluster to active
# Estimated time to recovery: ~15 minutes with pre-staged consumers

# Step 1: Verify replication lag (should be < 30s in steady state)
kafka-consumer-groups.sh --bootstrap-server eu-west-1-kafka:9092 \
  --describe --group us-east-1.fulfillment-service

# Step 2: Translate consumer offsets from us-east-1 → eu-west-1
# MirrorMaker 2 stores translated offsets in:
# eu-west-1.mm2-offset-syncs.us-east-1.internal
# Use MirrorMaker 2 RemoteClusterUtils to get translated offsets:
kafka-mirror-maker-offsets.sh \
  --source-cluster us-east-1 \
  --consumer-group fulfillment-service \
  --bootstrap-server eu-west-1-kafka:9092

# Step 3: Set consumer group offsets on eu-west-1 to translated positions
kafka-consumer-groups.sh --bootstrap-server eu-west-1-kafka:9092 \
  --group fulfillment-service \
  --reset-offsets --from-file translated-offsets.csv \
  --topic us-east-1.payments.charged --execute

# Step 4: Update producer endpoints to eu-west-1 (DNS failover or config change)
# Step 5: Start consumer groups on eu-west-1 — they resume from translated offsets
# Step 6: Monitor for duplicate processing (at-least-once: ~60s of lag may reprocess)