# Configure Kafka Clients for MSK

## Producer Configuration

| Setting | Recommended Value | Why |
|---|---|---|
| `linger.ms` | 5 ms minimum; 25 ms for most use cases | NEVER use 0. A value of 0 sends one request per message, saturating broker request handlers. Even low-latency use cases benefit from 5 ms. |
| `batch.size` | 65536 (64 KB) or 131072 (128 KB) | Larger batches reduce request count and broker CPU. Default 16384 is often too small. |
| `buffer.memory` | 67108864 (64 MB) | Increase when using larger batch sizes to avoid `BufferExhaustedException`. |
| `compression.type` | `lz4` or `zstd` | Reduces network bandwidth and storage. `lz4` for low latency; `zstd` for best compression ratio. |
| `acks` | `all` | Required for durability with MSK default `min.insync.replicas=2` and `RF=3`. Ensures all in-sync replicas acknowledge. Combined with `min.insync.replicas=2`, writes succeed as long as at least 2 of 3 replicas are in the ISR. |
| `retries` | 2147483647 (Integer.MAX_VALUE) | Allow unlimited retries. Use `delivery.timeout.ms` to bound total time. Failure to retry breaks Kafka's high availability during broker failover. |
| `delivery.timeout.ms` | 60000 minimum; 120000 (default) or higher | Upper bound for total send time including retries. Must be ≥ `request.timeout.ms` + `linger.ms`. AWS recommends a minimum of 60 seconds. With RF=3 and `min.insync.replicas=2`, producers only stall during leader election (seconds), so the 2-min default covers most cases. Increase if you observe `TimeoutException` during maintenance. |
| `request.timeout.ms` | 10000 (10 seconds) or higher | Max wait time for a single request before retry. |
| `retry.backoff.ms` | 200 minimum | Prevents retry storms during broker failover. |
| `send.buffer.bytes` | -1 (OS default) | Let the OS manage TCP buffers, especially on high-latency networks. |

## Consumer Configuration

| Setting | Recommended Value | Why |
|---|---|---|
| `session.timeout.ms` | 45000-60000 | Controls how long the broker waits without a heartbeat before evicting the consumer. Higher values tolerate GC pauses and network blips but delay failure detection. When using static membership (`group.instance.id`), must also exceed expected restart time. |
| `heartbeat.interval.ms` | 10000-15000 | Should be less than 1/3 of `session.timeout.ms`. Controls how quickly the group coordinator detects consumer failures. |
| `max.poll.interval.ms` | Based on processing time | If message processing takes > 5 minutes, increase this. Default 300000 (5 min). If exceeded, consumer is evicted from the group. |
| `max.poll.records` | Tune to processing capacity | Reduce if processing is slow to avoid exceeding `max.poll.interval.ms`. |
| `partition.assignment.strategy` | `CooperativeStickyAssignor` | Enables incremental rebalances instead of stop-the-world. **Migration requires two rolling restarts**: first deploy with `RangeAssignor,CooperativeStickyAssignor`, then remove `RangeAssignor`. Mixing eager and cooperative protocols causes `InconsistentGroupProtocolException`. |
| `group.instance.id` | Unique per consumer (e.g., hostname, pod-id) | Enables static group membership. Prevents unnecessary rebalances on short consumer restarts. |
| `auto.offset.reset` | `latest` for new consumer groups | Avoids reprocessing the entire topic on first start, which can overload the cluster. |
| `auto.commit.interval.ms` | 5000 minimum | Prevents excessive commit requests that add broker load. |
| `fetch.min.bytes` | 1024-131072 (1 KB-128 KB) | Reduces number of fetch requests. 1 KB for low-latency use cases; 32-128 KB for throughput-oriented workloads. |
| `fetch.max.wait.ms` | 1000 | How long to wait if `fetch.min.bytes` is not met. |
| `client.rack` | AZ ID (e.g., `use1-az1`) | Enables nearest-replica reads to reduce cross-AZ network costs. |
| `isolation.level` | `read_uncommitted` (default) | SHOULD NOT use `read_committed` when reading from tiered storage unless actively using transactions. |
| `receive.buffer.bytes` | -1 (OS default) | Let OS manage TCP buffers on high-latency networks. |

## Connection Management

- Create Kafka clients (producer, consumer, admin) once per application lifecycle — use singleton pattern. For AWS Lambda, create the client in global/init scope, NOT inside the handler function.
- Add random jitter (random sleep) before creating clients to avoid connection storms during deployments
- Add a shutdown hook with a random sleep before closing clients on SIGTERM — this prevents all clients from disconnecting simultaneously during rolling deployments. The random sleep should fit within the window before SIGKILL occurs.
- Ensure your deployment mechanism does not restart all producers/consumers at once — deploy in smaller batches
- Set `reconnect.backoff.ms = 1000` to handle connection retries gracefully
- Monitor `connection-count`, `connection-creation-rate`, `connection-close-rate` client metrics — these should be stable. High connection creation/termination rates cause unnecessary broker load.

## IAM Authentication

MSK IAM auth client configuration:

```
security.protocol=SASL_SSL
sasl.mechanism=AWS_MSK_IAM
sasl.jaas.config=software.amazon.msk.auth.iam.IAMLoginModule required;
sasl.client.callback.handler.class=software.amazon.msk.auth.iam.IAMClientCallbackHandler
```

**Constraints:**

- Maximum 3000 TCP connections per broker with IAM. This limit is adjustable via `listener.name.client_iam.max.connections` dynamic config.
- Maximum 100 new IAM connections per second per broker (M5/M7g); 4 per second on T3. This rate limit is not customer-adjustable.

## SASL/SCRAM Authentication

```
security.protocol=SASL_SSL
sasl.mechanism=SCRAM-SHA-512
sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required \
  username="<username>" password="<password>";
```

Store credentials in AWS Secrets Manager. Associate the secret with the MSK cluster. This config format is required for the Kafka CLI (`kafka-console-producer.sh`, etc.). In application code, retrieve credentials from Secrets Manager at runtime and inject into the JAAS config programmatically — do not store passwords in source-controlled config files.

## TLS (mTLS) Authentication

```
security.protocol=SSL
ssl.truststore.location=/path/to/truststore.jks
ssl.truststore.password=<password>
ssl.keystore.location=/path/to/keystore.jks
ssl.keystore.password=<password>
ssl.key.password=<password>
```

This config format is required for the Kafka CLI. In application code, load keystore/truststore passwords from Secrets Manager or SSM Parameter Store (SecureString) at startup — do not commit passwords to source-controlled config files.

If you don't have an existing CA, [AWS Private CA](https://docs.aws.amazon.com/msk/latest/developerguide/msk-authentication.html) can issue and rotate client certificates for MSK mTLS.

# MSK Maintenance Operations

## How MSK Maintenance Works

### Standard brokers

During patching and version upgrades, MSK performs **rolling broker restarts** — one broker at a time. The cluster enters `MAINTENANCE` state. You can still produce and consume data, but you cannot perform MSK API update operations until the cluster returns to `ACTIVE`. These operations appear as `SECURITY_PATCHING` in the `DescribeClusterOperation` API.

**Expected client impact**: Transient disconnect errors and brief p99 latency spikes (high milliseconds, up to ~2 seconds) lasting up to 2 minutes per broker restart as clients reconnect to new leaders. With the default RF=3 and proper client configuration (retries, `delivery.timeout.ms >= 60000`, `acks=all`), this does NOT cause data loss or prolonged unavailability — retries transparently reconnect to the new leader within seconds. **Topics with RF=1 become completely unavailable while their broker restarts** — there is no replica to fail over to, so producers receive errors and consumers stall for the full restart duration (5-15 min). See [configure-clients.md](configure-clients.md) and the "Consumer Resilience During Maintenance" section below.

**Expected metric impact**: `UnderReplicatedPartitions` increases temporarily (partitions on the offline broker stop replicating). After restart, the broker catches up on missed messages — you may see increased volume throughput and CPU usage during catch-up.

### Express brokers

Express brokers have **no maintenance windows**. MSK updates Express broker software on an ongoing basis in a **time-distributed manner** — occasional singular broker reboots spread across the month. The cluster stays `ACTIVE` during all maintenance. These operations appear as `BROKER_UPDATE` in the `DescribeClusterOperation` API.

**Why Express patching is less disruptive**:

- No cluster-wide maintenance window to plan around
- Throughput quotas prevent overloading during broker restarts
- Fixed RF=3 guarantees all topics survive a single broker restart (no RF=1 or RF=2 topics)
- Faster catch-up after restart than Standard brokers
- No advance notification needed

**Client contract still applies**: Clients must still handle leadership failover. Configure producers with retries, `delivery.timeout.ms >= 60000`, and `acks=all`; configure consumers with `session.timeout.ms = 45000-60000`. See [configure-clients.md](configure-clients.md) and the "Consumer Resilience During Maintenance" section below.

## What Happens During a Rolling Restart

When a broker restarts during maintenance:

1. **Broker goes offline**: The broker's metrics disappear from CloudWatch for several minutes.
2. **Leadership transfer**: Partition leadership moves from the restarting broker to other in-sync replicas. `LeaderCount` shifts across brokers.
3. **UnderReplicatedPartitions (URP) spikes** (Standard only — Express does not emit URP): While the broker is down, its partitions are under-replicated. This is expected and temporary.
4. **ActiveControllerCount may change**: If the controller broker is restarted, a new controller is elected.
5. **Consumer group rebalances**: If consumers were connected to the restarting broker, the session timeout triggers a rebalance.
6. **Broker restarts and catches up**: The broker comes back online, loads logs, replicates missed data, and rejoins ISR. URP decreases as replicas catch up.
7. **MSK moves to the next broker**: MSK waits for the broker to fully catch up before restarting the next one.

**Typical timeline per broker**: 5-15 minutes depending on data volume and partition count. Log loading progress can be tracked via the JMX metrics `remainingLogsToRecover` and `remainingSegmentsToRecover` (available through Prometheus/JMX monitoring, not via CloudWatch).

**Speeding up log recovery**: By default, Kafka uses a single thread per log directory for log recovery after an unclean shutdown. With thousands of partitions, recovery can take hours. Set `num.recovery.threads.per.data.dir` to the number of CPU cores to parallelize recovery. This is a broker-side configuration — update via `aws kafka update-cluster-configuration`.

## What NOT To Do During Maintenance

- **NEVER restart additional brokers** — MSK is already performing a rolling restart. Manual restarts compound the problem.
- **NEVER reassign partitions during URP** — Reassignment adds replication load on already-stressed brokers.
- **NEVER lower `min.insync.replicas`** — This weakens durability guarantees. The `NotEnoughReplicasException` during maintenance is transient.
- **NEVER escalate as a cluster-level issue** if URP is decreasing and only one broker has a metrics gap.

## Impact of Scaling Operations (Standard)

Scaling operations on Standard clusters trigger rolling restarts or add replication load. Plan these during low-traffic periods and ensure the cluster has headroom.

### Broker size updates

Updating the broker size (e.g., kafka.m5.large → kafka.m5.xlarge) triggers a **rolling restart** — MSK takes brokers offline one at a time and temporarily reassigns partition leadership to other brokers. This is the same process as a maintenance rolling restart. A size update typically takes **10-15 minutes per broker**. During this time:

- `UnderReplicatedPartitions` will spike per broker, same as during patching
- Remaining brokers absorb extra leadership and replication load
- Ensure CPU is under 60% before initiating a size change

### Adding brokers and reassigning partitions

After adding brokers to expand a Standard cluster, existing partitions are NOT automatically redistributed. You must manually reassign partitions using `kafka-reassign-partitions.sh`. This creates replication load as data is copied from existing brokers to new ones.

**Constraints:**

- Limit to **10 partitions per reassignment call** for safe operations on Standard clusters
- Do NOT reassign partitions when CPU utilization is above **70%** — replication adds significant CPU and network load that can cascade
- Do NOT reassign partitions while `UnderReplicatedPartitions` > 0
- Consider using [Cruise Control](https://docs.aws.amazon.com/msk/latest/developerguide/cruise-control.html) for continuous, automated partition rebalancing

### Storage expansion

Expanding EBS storage does NOT trigger a rolling restart — it happens online. However, the volume enters an **optimizing** state that can take up to 24 hours, and a second expansion cannot be performed for at least 6 hours. See [manage-storage.md](manage-storage.md) for details.

## Impact of Scaling Operations (Express)

Express scaling is simpler than Standard, but broker size changes still involve rolling restarts.

### Broker size updates

Updating the Express broker size also triggers a **rolling restart**, same as Standard. MSK takes brokers offline one at a time. However, the cluster stays **ACTIVE** (not MAINTENANCE) throughout. Key differences from Standard:

- Express does **not** emit `UnderReplicatedPartitions` — you cannot use URP to track restart progress. Monitor `ProduceThrottleTime`, `FetchThrottleTime`, and consumer lag instead.
- Ensure CPU (CpuUser + CpuSystem) is under 60% before initiating a size change, same as Standard.

### Adding brokers and partition redistribution

When you add brokers to an Express cluster:

- If **Intelligent Rebalancing is enabled** (default): Partitions are automatically redistributed to new brokers. No manual action needed. You cannot use `kafka-reassign-partitions.sh` while Intelligent Rebalancing is active.
- If **Intelligent Rebalancing is disabled**: You must manually reassign partitions using `kafka-reassign-partitions.sh`. Limit to **20 partitions per reassignment call** (vs 10 for Standard).

### Storage

Express storage is fully managed — there is no expansion operation, no cooldown period, and no provisioning required. Storage scales automatically with data retained. However, you should still monitor `StorageUsed` and per-topic ingress to catch runaway growth that impacts cost. See [manage-storage.md](manage-storage.md) for investigation steps.

## Consumer Resilience During Maintenance

Configure consumers to survive broker restarts gracefully. See [configure-clients.md](configure-clients.md) for full settings.

**Key settings for maintenance resilience:**

| Setting | Recommended | Why |
|---|---|---|
| `session.timeout.ms` | 45000-60000 | Must exceed time for broker restart + consumer reconnection. Default 10000 is too short. |
| `heartbeat.interval.ms` | 10000-15000 | Should be < 1/3 of `session.timeout.ms`. |
| `partition.assignment.strategy` | `CooperativeStickyAssignor` | Incremental rebalances instead of stop-the-world. Only moved partitions are reassigned. |
| `group.instance.id` | Unique per consumer | Enables static membership. Consumer can rejoin after brief disconnect without triggering full rebalance. |
| `group.initial.rebalance.delay.ms` | Match average deployment time | **Broker-side config** (set via `aws kafka update-cluster-configuration`, not consumer properties). Prevents cascading rebalances during rolling deployments. |

**Producer settings for maintenance:**

| Setting | Recommended | Why |
|---|---|---|
| `retries` | Integer.MAX_VALUE | Allows retrying through broker restart. |
| `delivery.timeout.ms` | 60000 minimum; 120000 (2 minutes) or higher | Bounds total retry time. AWS recommends a minimum of 60 seconds. Must be ≥ `request.timeout.ms` + `linger.ms`. With RF=3 and `min.insync.replicas=2`, producers only stall during leader election (seconds, not minutes). The 2-min default covers this. Increase if you observe `TimeoutException` during maintenance. |
| `acks` | `all` | With `min.insync.replicas=2` (MSK default), writes succeed as long as 2 of 3 replicas are available. One broker offline is tolerated. |

## Preparing for Maintenance Windows (Standard)

1. **Ensure CPU < 60%**: During maintenance, remaining brokers handle extra leadership and replication. If CPU is already near 60%, the added load during maintenance may cause cascading issues.
2. **Ensure storage headroom**: Brokers that take over leadership temporarily handle more writes.
3. **Use 3-AZ clusters with RF=3 and min.insync.replicas=2**: This tolerates one broker offline.
4. **Distribute connection strings across AZs**: Client bootstrap servers SHOULD include at least one broker from each AZ.
5. **Test consumer resilience**: Simulate broker failure by rebooting a broker via the MSK API: `aws kafka reboot-broker --cluster-arn <arn> --broker-ids <id>`.

## Kafka Version Upgrades

Version upgrades trigger rolling restarts. The process:

1. MSK updates brokers one at a time to the new version.
2. Between each broker restart, MSK waits for the broker to fully catch up.
3. The cluster enters `UPDATING` state during the upgrade.

**Constraints:**

- You MUST check that partition counts per broker are within the limits for the target version before upgrading (see [size-and-choose-cluster.md](size-and-choose-cluster.md)).
- Upgrades are forward-only — you cannot downgrade Kafka versions.
- Express brokers support Kafka versions 3.6, 3.8, and 3.9.

## Monitoring During Maintenance

Watch these metrics to track maintenance progress:

| Metric | What to Look For |
|---|---|
| `UnderReplicatedPartitions` (Standard only) | Should spike when a broker restarts, then decrease as it catches up. If URP stays elevated for > 30 min after a broker comes back, investigate. Express does not emit this metric. |
| `ActiveControllerCount` | Should always be 1. Brief fluctuation during controller broker restart is normal. |
| `CpuUser` on remaining brokers | Should increase temporarily as they absorb extra leadership. If > 80%, cluster is undersized for maintenance. |
| `BytesInPerSec` per broker | Should redistribute when a broker goes offline and rebalance when it returns. |
| `LeaderCount` per broker | Should shift during restart and rebalance afterward via `auto.leader.rebalance.enable=true` (MSK default). |

# Manage MSK Storage

## Standard Brokers: EBS Storage Management

### Monitor disk usage

Monitor `KafkaDataLogsDiskUsed` (DEFAULT level, dimensions: Cluster Name, Broker ID). This metric shows the percentage of disk used for data logs per broker. Set a CloudWatch alarm at 85%.

MSK also sends proactive storage alerts at 60% and 80% usage via the MSK console, Health Dashboard, EventBridge, and account email.

### Expand EBS storage

You can increase but NEVER decrease EBS volume size. Maximum: 16,384 GiB (16 TiB) per broker.

```
aws kafka update-broker-storage \
  --cluster-arn <cluster-arn> \
  --current-version <cluster-version> \
  --target-broker-ebs-volume-info '[{"KafkaBrokerNodeId": "All", "VolumeSizeGB": <target-size>}]'
```

**Constraints:**

- You MUST get the current cluster version first: `aws kafka describe-cluster-v2 --cluster-arn <arn>` — the version string looks like `KTVPDKIKX0DER`, not a simple integer.
- Storage expansion has a cool-down period of minimum 6 hours. A second expansion attempt during cool-down fails with "storage is optimizing."
- Optimization after expansion can take up to 24 hours proportional to storage size.

### Enable auto-scaling

Auto-scaling automatically expands storage when utilization reaches a threshold. Configure via the MSK console or Application Auto Scaling API.

Policy parameters:

- **Storage Utilization Target**: Recommended 50-60%. Range: 10-80%.
- **Maximum Storage Capacity**: Up to 16 TiB per broker.

Auto-scaling increases storage by the larger of 10 GiB or 10% of current storage. A scaling action can occur only once every 6 hours.

**Long-term alternative**: If recurring storage management is a pain point, consider migrating to Express brokers — storage is fully managed, pay-as-you-go, and requires no provisioning or monitoring.

### Identify high-growth topics

To find which topics consume the most storage, check per-topic `BytesInPerSec` (PER_TOPIC_PER_BROKER level) and multiply by retention period:

`Estimated storage per topic = SUM(BytesInPerSec across all brokers for the topic) × retention_seconds × ReplicationFactor`

Use this to identify topics that need retention adjustment. Retention changes require app-owner approval — reducing retention deletes data permanently.

### Provision storage throughput (Standard only)

For broker sizes `kafka.m5.4xlarge` or larger (or `kafka.m7g.2xlarge` or larger), you can provision storage throughput above the default of 250 MiB/s (for volumes 10 GiB+). Check the [MSK storage throughput documentation](https://docs.aws.amazon.com/msk/latest/developerguide/msk-provision-throughput-management.html) for current max provisioned throughput per broker size.

When enabling provisioned throughput, also increase `num.replica.fetchers` (default 2) to match the instance size — e.g., 4 for m5.4xl, 8 for m5.8xl.

## Express Brokers: Managed Storage

Express brokers have fully managed, pay-as-you-go storage. You do NOT provision or manage EBS volumes.

### How Express storage works

Express storage is elastic and virtually unlimited. You do not provision volumes or manage capacity. However, storage is pay-as-you-go based on total data retained, so runaway growth directly impacts cost.

### Monitor Express storage

Monitor total cluster storage with the `StorageUsed` metric (DEFAULT level, dimension: Cluster Name). There are no per-broker disk usage metrics for Express because storage is not tied to individual brokers.

Set a CloudWatch alarm on `StorageUsed` based on expected retention:

`Expected StorageUsed ≈ SUM(BytesInPerSec across all topics) × retention_seconds × 3`

The `× 3` accounts for Express's fixed replication factor. If `StorageUsed` significantly exceeds this estimate, investigate per-topic growth.

### Identify and resolve runaway storage growth

Even though Express storage scales automatically, you should actively monitor for unexpected growth to control costs:

1. **Check per-topic ingress**: Use `BytesInPerSec` at PER_TOPIC_PER_BROKER level to identify which topics are driving the most data volume.
2. **Check topic-level retention overrides**: A topic with `retention.ms` set higher than the cluster default will retain more data. Audit topic configs with:

   ```
   kafka-configs.sh --bootstrap-server <bootstrap> --describe --entity-type topics --entity-name <topic>
   ```

3. **Check for compacted topics**: Topics with `cleanup.policy=compact` retain data indefinitely based on key cardinality, not time. High-cardinality compacted topics can grow without bound.
4. **Check for topic proliferation**: A growing number of topics (each with RF=3) compounds storage. Monitor `GlobalTopicCount` at the cluster level.
5. **Reduce retention**: Lowering `retention.ms` on high-volume topics is the most direct way to reduce stored data. Coordinate with app owners before changing — reducing retention deletes data permanently.

### Storage costs on Express

Express storage is fully managed. Unlike Standard where you explicitly manage EBS and optionally enable tiered storage, Express storage requires no configuration. Storage costs are based on total data retained — reducing retention or cleaning up unused topics is the primary lever for cost control.

## Standard Brokers: Tiered Storage

Standard brokers can optionally enable tiered storage to extend retention beyond EBS capacity.

### How tiered storage works on Standard

1. Closed log segments are copied from primary (EBS) storage to tiered (S3) storage automatically.
2. Active segments are NOT eligible for tiering — segment size (`segment.bytes`, default 128 MiB for tiered clusters) or segment roll time (`segment.ms`) controls when segments close.
3. `local.retention.ms` controls how long data stays on primary storage after being copied to tiered storage. Default `-2` means use `retention.ms` (data stays on both local and tiered until retention expires).
4. `retention.ms` controls total retention (local + tiered). Minimum 3 days for tiered topics.

### Example retention scenario

With `retention.ms = 5 days` and `local.retention.ms = 12 hours`:

- Data stays on fast primary storage for 12 hours
- Data remains in tiered storage for the full 5 days
- Consumers reading data older than 12 hours fetch from tiered storage with slightly higher initial latency

### Tiered storage constraints

- Compacted topics (`cleanup.policy=compact`) are NOT supported with tiered storage
- When explicitly set to a positive value, `local.retention.ms` MUST be less than `retention.ms`. The default `-2` is a sentinel meaning "use `retention.ms`" and is always valid.
- Minimum log segment size: 48 MiB; minimum segment roll time: 10 minutes
- Once disabled for a topic, tiered storage CANNOT be re-enabled on that topic
- Supported on Kafka versions 3.6.0+ and 2.8.2.tiered
- Not supported on `kafka.t3.small` instances
- Clients SHOULD NOT use `read_committed` isolation level when reading from tiered storage unless actively using transactions

### Enable tiered storage on a topic

```
kafka-configs.sh --bootstrap-server <bootstrap> --alter --entity-type topics --entity-name <topic> \
  --add-config 'remote.storage.enable=true,local.retention.ms=43200000,retention.ms=604800000'
```

# Monitor and Alarm MSK Clusters

## Monitoring Levels

MSK Provisioned clusters support 4 monitoring levels. Each level includes all metrics from the previous level.

| Level | Cost | Dimensions | What It Adds |
|---|---|---|---|
| DEFAULT | Free | Cluster Name; Cluster Name + Broker ID; Cluster Name + Consumer Group + Topic | Core metrics: CPU, memory, disk, bytes in/out, connection count, URP (Standard), partition count, consumer lag, TrafficShaping (Standard) |
| PER_BROKER | Paid | Cluster Name + Broker ID | Request handler/network idle %, EBS volume metrics (Standard), traffic shaping detailed (Standard), connection rates, produce/fetch latency, tiered storage metrics (Standard) |
| PER_TOPIC_PER_BROKER | Paid | Cluster Name + Broker ID + Topic | Per-topic message rates, conversion rates, tiered storage per topic (Standard) |
| PER_TOPIC_PER_PARTITION | Paid | Consumer Group + Topic + Partition | Per-partition consumer lag (OffsetLag, EstimatedTimeLag) |

**Recommendation**: Use PER_BROKER as the minimum for production clusters. Use PER_TOPIC_PER_BROKER if you need per-topic throughput visibility. Use PER_TOPIC_PER_PARTITION only if you need partition-level consumer lag granularity.

**Key metrics NOT available on Express** (do not alarm on these for Express clusters): `UnderReplicatedPartitions`, `OfflinePartitionsCount`, `KafkaDataLogsDiskUsed`, `HeapMemoryAfterGC`, `BurstBalance`, `VolumeQueueLength`, `BwInAllowanceExceeded`, `BwOutAllowanceExceeded`. Express uses `ProduceThrottleTime`, `FetchThrottleTime`, and `StorageUsed` instead.

For the full metric list per broker type and monitoring level, search AWS docs for `"MSK CloudWatch metrics Standard brokers"` or `"MSK CloudWatch metrics Express brokers"`.

Update monitoring level:

```
# Get the cluster's current revision version first (this is an opaque revision string, NOT a Kafka version number):
# aws kafka describe-cluster-v2 --cluster-arn <cluster-arn> --query 'ClusterInfo.CurrentVersion' --output text

aws kafka update-monitoring --cluster-arn <cluster-arn> --current-version <cluster-current-version> \
  --enhanced-monitoring PER_BROKER
```

Follow [SNS security best practices](https://docs.aws.amazon.com/sns/latest/dg/sns-security-best-practices.html) when wiring alarm actions to an SNS topic (e.g. `--alarm-actions` for `aws cloudwatch put-metric-alarm` calls).

## Recommended Alarms

### Critical alarms (set these for every cluster)

| Metric | Condition | Action | Source |
|---|---|---|---|
| `CpuUser + CpuSystem` | Average > 60% for 5 min | Scale broker size or add brokers. Check client batch config first. | MSK Best Practices |
| `KafkaDataLogsDiskUsed` (Standard) | Max > 85% per broker | Expand EBS storage. Review topic retention. Enable auto-scaling. | MSK Best Practices |
| `UnderReplicatedPartitions` (Standard) | Sum > 0 for 15 min | Check if maintenance is in progress. If not, investigate broker health. | MSK Best Practices |
| `UnderMinIsrPartitionCount` (Standard) | Average ≥ 1 for 1 min | Partitions have fallen below `min.insync.replicas`. Produce calls with `acks=all` will fail with `NotEnoughReplicas`. Differs from URP: URP fires when any replica is behind; UnderMinIsr fires only when ISR drops to a level that breaks writes. Confirm broker health and replication factor ≥ 3 across topics. | MSK Best Practices |
| `OfflinePartitionsCount` (Standard) | Sum > 0 | Immediate investigation — partitions unavailable for produce/consume. | MSK Best Practices |
| `ActiveControllerCount` | Maximum < 1 for 5 min | No active controller. Brief dip during controller election on broker restart is normal. | MSK Best Practices |

### Performance alarms (most require PER_BROKER level)

| Metric | Condition | Action |
|---|---|---|
| `RequestHandlerAvgIdlePercent` | Average < 0.3 (30%) for 5 min | Request threads saturated. Check client batch sizes before scaling. |
| `NetworkProcessorAvgIdlePercent` | Average < 0.3 for 5 min | Network threads saturated. May indicate connection storms or high TLS overhead. |
| `BwInAllowanceExceeded` or `BwOutAllowanceExceeded` (Standard) | Sum > 0 for 5 min | Traffic shaping active. Check per-broker traffic distribution. Consider larger broker. |
| `HeapMemoryAfterGC` (Standard, DEFAULT level) | Average > 60% for 5 min, 3 consecutive periods | Memory pressure after GC. Check connection count, consumer groups, partition count. Reduce `transactional.id.expiration.ms` from 604800000 (7 days) to 86400000 (1 day) to lower per-transaction memory footprint. Review high-cardinality consumer groups and excessive partition counts. |
| `CPUCreditBalance` (Standard, t3 brokers only) | Average ≤ 100 for 5 min, 3 consecutive periods | t3 brokers earn CPU credits up to a max of 576. When the balance hits 0, CPU is capped at the 20% baseline and broker performance degrades sharply. Upgrade from t3 to m7g brokers — t3 is appropriate only for dev/test, never production sustained load. |
| `(Sum(VolumeReadBytes) + Sum(VolumeWriteBytes)) / (5 * 60 * 1024 * 1024)` (Standard) | ≥ 80% of available volume throughput in MiB/s for 5 min, 3 consecutive periods | Underlying EBS volume read+write activity is approaching the volume's throughput limit. EBS throughput ceilings are invisible in Kafka metrics — this is the alarm that surfaces them. Use a metric math expression: sum `VolumeReadBytes` + `VolumeWriteBytes` per broker over a 5-min period, divide by `5*60` for per-second, then by `1024*1024` for MiB/s. Threshold = 0.8 × the volume's provisioned throughput. Remediation: provision higher EBS storage throughput, switch to gp3 with explicit throughput, or scale to a larger broker. See [MSK provisioned throughput management](https://docs.aws.amazon.com/msk/latest/developerguide/msk-provision-throughput-management.html). |
| `IAMTooManyConnections` | Sum > 0 for 5 min | Clients exceeding the 100 new IAM connections/sec per broker limit (4/sec on t3). Check for connection storms from deployments, missing singleton patterns, or Lambda cold starts creating new clients per invocation. Add random jitter before client creation. See [configure-clients.md](configure-clients.md) connection management guidance. |
| `ConnectionCreationRate` (IAM) | Sum ≥ 80 per minute for 1 min, 3 consecutive periods | Proactive alarm — fires before `IAMTooManyConnections` throttling kicks in. The 100/sec limit is fixed and cannot be raised. Use **Sum** statistic, NOT Average: a single broker reports `ConnectionCreationRate` as multiple datapoints across its network processors, so summing at the broker level gives the true total. Same remediation as `IAMTooManyConnections`. |
| `ClientConnectionCount` per broker | Sum > 80% of the configured limit for 5 min, 3 consecutive periods | **For IAM**: the 3000 default cap is enforced — at 3000 new connections are rejected. Adjustable via `listener.name.client_iam.max.connections` (Kafka dynamic config, applied with `kafka-configs.sh --bootstrap-server <bootstrap> --command-config client.properties --alter ...`, where `client.properties` carries the IAM SASL settings — `security.protocol=SASL_SSL`, `sasl.mechanism=AWS_MSK_IAM`, `sasl.jaas.config=software.amazon.msk.auth.iam.IAMLoginModule required;`, `sasl.client.callback.handler.class=software.amazon.msk.auth.iam.IAMClientCallbackHandler` — and is NOT an AWS CLI command). **For SASL/SCRAM and mTLS**: MSK does NOT enforce a connection cap, but high counts still consume network threads, file descriptors, and heap, and can cascade into `RequestHandlerAvgIdlePercent` saturation and produce/fetch latency. Alarm at the same 2400 (80% of 3000) baseline for non-IAM unless you have measured a different ceiling for the workload. Use the `Client Authentication` dimension to scope the alarm to the relevant listener. Use **Sum** statistic per minute — a single broker reports this metric as multiple datapoints across network processors. Common causes: creating new producer/consumer instances per request instead of reusing one per application, connection leaks from missing `close()` calls, or too many microservices connecting to the same cluster. See [configure-clients.md](configure-clients.md) connection management guidance. |

### Consumer lag alarms (DEFAULT level)

| Metric | Condition | Action |
|---|---|---|
| `MaxOffsetLag` | Maximum > threshold (app-specific, e.g., 10000 offsets) for 15 min | Per-partition worst case. Catches a single hot partition falling behind even when the topic-wide total looks fine. Use this when partition-level SLAs matter. Check consumer group health. |
| `SumOffsetLag` | Average ≥ threshold (app-specific) for 5 min, 3 consecutive periods | Aggregated lag across all partitions in a topic for a consumer group. Catches whole-topic backlog growth. Use this when total backlog or end-to-end latency matters more than any single partition. For partition-level detail, use the `Offset` metric (PER_TOPIC_PER_PARTITION level) or `kafka-consumer-groups.sh --describe`. |
| `EstimatedMaxTimeLag` | > threshold (app-specific) | Lag expressed as time. Set threshold based on SLA. |

### Per-broker capacity alarms

| Metric | Applies to | Condition | Action |
|---|---|---|---|
| `PartitionCount` per broker | Standard and Express | > recommended limit for broker size (see [size-and-choose-cluster.md](size-and-choose-cluster.md)) | Approaching partition hard limit. Scale to larger broker size or add brokers to redistribute. |
| `BytesInPerSec` per broker | Express only | > ~80% of sustained ingress limit for broker size | Approaching per-broker ingress quota. Scale to larger broker size or add brokers. |
| `BytesOutPerSec` per broker | Express only | > ~80% of sustained egress limit for broker size | Approaching per-broker egress quota. Scale or reduce consumer groups. |

Standard brokers do not have MSK-enforced per-broker throughput quotas — throughput is bounded by EBS volume type and EC2 network bandwidth instead. For Standard throughput alarming, use `BwInAllowanceExceeded` and `BwOutAllowanceExceeded` in the performance alarms section above.

### Express-specific alarms

| Metric | Condition | Action |
|---|---|---|
| `ProduceThrottleTime` | Average > 0 for 5 min | Per-broker ingress quota exceeded. Scale to larger Express broker or add brokers. |
| `FetchThrottleTime` | Average > 0 for 5 min | Per-broker egress quota exceeded. Scale or reduce consumer groups. |
| `StorageUsed` | See below — anomaly detection by default; static threshold when retention is known | Express storage is fully managed but billed per byte-hour, so unbounded growth = unbounded cost. ALWAYS create a storage alarm — never skip it. |

**`StorageUsed` alarm — choose ONE of the two patterns:**

1. **Default (retention unknown, or you want to catch unusual growth):** anomaly-detection alarm on `StorageUsed`. CloudWatch builds a per-cluster baseline and alerts when usage falls outside the predicted band — no static threshold needed and it adapts as the workload changes. Use `LessThanLowerOrGreaterThanUpperThreshold` with `ANOMALY_DETECTION_BAND(m1, 2)`. This is the recommended default for new clusters where you don't yet know the steady-state size.

   ```
   aws cloudwatch put-metric-alarm \
     --alarm-name MSK-<cluster>-StorageUsed-Anomaly \
     --metrics '[
       {"Id":"m1","MetricStat":{"Metric":{"Namespace":"AWS/Kafka","MetricName":"StorageUsed","Dimensions":[{"Name":"Cluster Name","Value":"<cluster>"}]},"Period":300,"Stat":"Maximum"},"ReturnData":true},
       {"Id":"ad1","Expression":"ANOMALY_DETECTION_BAND(m1, 2)","Label":"StorageUsed (expected)"}
     ]' \
     --threshold-metric-id ad1 \
     --comparison-operator LessThanLowerOrGreaterThanUpperThreshold \
     --evaluation-periods 3 --datapoints-to-alarm 3 \
     --treat-missing-data notBreaching
   ```

2. **Static threshold (retention is known):** `Maximum > SUM(BytesInPerSec) × retention_seconds × 3` for 15 min (RF=3 on Express). Use this once the cluster has been running long enough to know steady-state ingress and retention is locked.

## CloudWatch Dimension Reference

All MSK metrics use namespace `AWS/Kafka`.

| Dimension | Format | Example |
|---|---|---|
| Cluster Name | String (cluster name, not ARN) | `my-msk-cluster` |
| Broker ID | Integer string | `1`, `2`, `3` |
| Topic | String | `my-topic` |
| Consumer Group | ASCII string (non-ASCII drops metrics) | `my-consumer-group` |
| Partition | Integer string | `0`, `1`, `2` |
| Client Authentication | String | `SSL`, `SASL_SCRAM`, `IAM` |

**Important**: Consumer group names MUST use ASCII characters only. Non-ASCII characters cause consumer lag metrics to be silently dropped from CloudWatch.

## Dashboard Construction

Create a CloudWatch dashboard with these sections:

1. **Cluster Health**: `ActiveControllerCount`, `OfflinePartitionsCount` (Standard), `UnderReplicatedPartitions` (Standard); for Express add `ProduceThrottleTime` and `FetchThrottleTime`
2. **CPU**: `CpuUser` per broker (line chart, one series per broker to spot AZ skew)
3. **Throughput**: `BytesInPerSec` and `BytesOutPerSec` per broker
4. **Storage**: `KafkaDataLogsDiskUsed` per broker (Standard); `StorageUsed` cluster-level (Express)
5. **Consumer Lag**: `MaxOffsetLag` per consumer group/topic
6. **Request Performance**: `RequestHandlerAvgIdlePercent`, `ProduceTotalTimeMsMean` per broker

Use metric math for composite CPU alarm: `m1 + m2` where m1 is defined as `CpuUser` and m2 is defined as `CpuSystem` for the target broker.

# Size and Choose an MSK Cluster

## For any sizing question, run [`scripts/msk_sizing.py`](../scripts/msk_sizing.py)

If the user is asking how many brokers, what instance size, what cost, etc., you **MUST** run [`scripts/msk_sizing.py`](../scripts/msk_sizing.py) before answering. Do not size by hand. Do not estimate from memory. Run the script from the skill directory with `python3`.

The script is the source of truth for broker counts and costs in this skill — it models broker capacity (EBS, NIC, partitions, PST, storage), AZ rounding, 1-AZ-down headroom, fan-out, Tiered Storage detection, EBS headroom, cross-AZ producer replication, optional cross-AZ consumer fetch, Express storage and data-in charges, and PST cost. It enumerates every Standard and Express instance size and returns a "Recommended pick per class" section naming the lowest-cost option per class within the broker quota.

Required workflow for any sizing answer:

1. Translate the user's inputs (avg/peak ingress, avg/peak egress, partitions, retention, primary retention, RF, PST, rack affinity) into the script's flags. If a value is missing, ask before guessing.
2. Run the script with `--explain`. Always pass `--retention-hours` and `--primary-retention-hours` — set them equal to disable Tiered Storage; set retention > primary to enable it.
3. Read the "Recommended pick per class" section. Quote those instance types, broker counts, bottlenecks, and total monthly costs back to the user verbatim — do not round, re-derive, or substitute your own numbers.
4. Use the `--explain` per-instance breakdown only to explain *why* the script picked what it did, never to override the recommendation.

You may suggest a larger size than the recommended pick only when (a) the user explicitly asks for one, or (b) the workload exceeds the broker quota and a quota increase is impractical. In both cases, name the recommended pick first and the alternative second.

```
python scripts/msk_sizing.py \
  --avg-data-in-mbs 100  --peak-data-in-mbs 500 \
  --avg-data-out-mbs 200 --peak-data-out-mbs 1000 \
  --num-partitions 1000 \
  --retention-hours 720  --primary-retention-hours 24 \
  --explain
```

Flag reference:

- `--explain` — print per-constraint analysis (which dimension is the bottleneck and by how much) and per-cost-factor breakdown (brokers, storage, Tiered Storage, Provisioned ST, Express data-in, cross-AZ) for every instance and the recommended pick per class. **Always include this flag.**
- `--broker-quota N` — change the per-cluster broker quota used to pick a "recommended" instance per class. Default is 60 (the MSK Provisioned default soft quota). The script picks the cheapest instance per class whose broker count fits within the quota.
- `--use-max-partitions` — size against the hard partition cap instead of the recommended cap (use only when the user accepts the operational risk).
- `--pst-per-broker-mbs` — apply a Provisioned Storage Throughput limit (4xlarge+ Standard only). Pass when the user mentions PST, gp3 provisioned throughput, or EBS write IO bottleneck.
- `--utilization-standard` / `--utilization-express` — override the headroom factor (defaults: 0.50 / 0.75). Do not change unless the user explicitly asks.
- `--no-rack-affined-consumers` — include consumer fetch traffic in the cross-AZ cost (use when consumers fetch from any leader rather than local-AZ replicas). Affects cost only, not broker count.

The narrative steps below explain what the script computes and when each constraint dominates. Use them to interpret `--explain` output, **never as a substitute for running the script**.

## Standard vs Express: Decision Framework

| Factor | Standard | Express |
|---|---|---|
| Storage | Customer-managed EBS (1 GiB - 16 TiB per broker) | Fully managed, pay-as-you-go, unlimited |
| Throughput per broker | Depends on instance type + EBS volume type + provisioned throughput | Defined per broker size (up to 500 MiB/s ingress); up to 3x more throughput per broker than equivalent Standard instance sizes |
| Maintenance windows | Yes — cluster enters MAINTENANCE state during patching | No maintenance windows — stays ACTIVE |
| Scaling speed | Hours for partition rebalancing | Up to 20x faster scaling |
| Storage management | Manual or auto-scaling EBS; optional tiered storage | No storage management required |
| Replication factor | Configurable (default 3) | Fixed at 3 |
| min.insync.replicas | Configurable (default 2) | Fixed at 2 |
| unclean.leader.election | Configurable (default true for non-tiered) | Fixed at false |
| Instance families | kafka.t3, kafka.m5, kafka.m7g | express.m7g only |
| Availability zones | 2 or 3 AZs | 3 AZs only |

**Choose Express for most workloads.** Express provides fully managed storage, no maintenance windows, faster scaling, up to 3x more throughput per broker, and managed best-practice defaults — making it the right choice for the majority of use cases.

**Choose Standard only when:** You need fine-grained control over broker configuration (e.g., custom `min.insync.replicas`, `unclean.leader.election`, replication factor other than 3) or you require a 2-AZ deployment.

## Sizing Standard Clusters

### Step 1: Determine throughput requirement

Calculate total ingress (write) throughput across all topics. Account for replication: total broker write IO = ingress × RF. For RF=3 and 100 MiB/s ingress, total write IO = 300 MiB/s across the cluster.

### Step 2: Choose instance type

Check the [MSK Best Practices for Standard brokers](https://docs.aws.amazon.com/msk/latest/developerguide/bestpractices.html) for current partition limits per Standard broker size. Larger instance types support more partitions.

For m5.4xl+ or m7g.4xl+, optimize throughput by tuning `num.io.threads` and `num.network.threads`. **Important**: Do not increase `num.network.threads` without first increasing `num.io.threads` — this can cause queue saturation.

| Instance Size | Recommended `num.io.threads` | Recommended `num.network.threads` |
|---|---|---|
| m5.4xlarge / m7g.4xlarge | 16 | 8 |
| m5.8xlarge / m7g.8xlarge | 32 | 16 |
| m5.12xlarge / m7g.12xlarge | 48 | 24 |
| m5.16xlarge / m7g.16xlarge | 64 | 32 |
| m5.24xlarge | 96 | 48 |

### Step 3: Calculate number of brokers

Divide total write IO by per-broker throughput capacity. Per-broker throughput is limited by the lowest of: EBS volume throughput, EC2-to-EBS network bandwidth, and EC2 egress bandwidth. **The broker count must be a multiple of the number of AZs (2 or 3).** Round up to the next valid multiple to ensure even partition distribution across availability zones.

**EBS throughput is often the bottleneck.** Default GP2/GP3 volumes cap at 250 MiB/s. For higher throughput, enable provisioned storage throughput (GP3) on broker sizes `kafka.m5.4xlarge`+ or `kafka.m7g.2xlarge`+. Check the [MSK storage throughput documentation](https://docs.aws.amazon.com/msk/latest/developerguide/msk-provision-throughput-management.html) for current max provisioned throughput per broker size.

Without provisioned throughput, a broker with RF=3 and 83 MiB/s client ingress already hits the 250 MiB/s ceiling (83 × 3 = 249 MiB/s write IO). Factor this into your broker count calculation. See [manage-storage.md](manage-storage.md) for provisioning details.

**Pair PST with `num.replica.fetchers`.** Provisioned storage throughput does not deliver its full benefit until you also raise `num.replica.fetchers` from the default of 2. Both changes must be in effect for the cluster to reach the new throughput target. AWS recommends ([source](https://docs.aws.amazon.com/msk/latest/developerguide/msk-provision-throughput-management.html#provisioned-throughput-config)):

| Broker size | `num.replica.fetchers` |
|---|---|
| kafka.m5.4xlarge | 4 |
| kafka.m5.8xlarge | 8 |
| kafka.m5.12xlarge | 14 |
| kafka.m5.16xlarge | 16 |
| kafka.m5.24xlarge | 16 |

For M7g sizes, use the value for the equivalent M5 size as a starting point. After flipping PST on, expect a transitional period (up to 24 hours; ~6 hours per fully utilized 1 TiB volume) where the new throughput ramps in.

**Important**: Maintain CPU utilization (CpuUser + CpuSystem) under 60% to retain headroom for operational events. For a precise per-instance broker count and cost breakdown, run [`scripts/msk_sizing.py`](../scripts/msk_sizing.py) with `--explain` (see top of this document). The [MSK Sizing and Pricing spreadsheet](https://view.officeapps.live.com/op/view.aspx?src=https%3A%2F%2Fdy7oqpxkwhskb.cloudfront.net%2FMSK_Sizing_Pricing.xlsx) is an alternative for offline what-if analysis.

### Step 4: Size EBS storage

Calculate: `client_ingress_per_broker × retention_seconds × RF` for each broker, where `client_ingress_per_broker` is the client write rate divided by broker count (excluding replication). Add 20% headroom. Maximum 16,384 GiB per broker. Enable auto-scaling with utilization target of 50-60%.

## Sizing Express Clusters

### Step 1: Determine throughput requirement

Calculate ingress AND egress separately. **Egress includes all consumer groups**: if 5 consumer groups each read the full stream, egress = 5 × ingress. This read amplification is the primary sizing driver for Express.

### Step 2: Choose broker size and count

Check the [MSK Express broker quotas](https://docs.aws.amazon.com/msk/latest/developerguide/limits.html#msk-express-quota) for current per-broker throughput limits (sustained and maximum) per Express broker size. Each size has a sustained ingress/egress threshold (no degradation) and a maximum quota (hard throttle).

Size using the **sustained performance** values. If throughput exceeds sustained limits, you may experience degraded performance. If it reaches the maximum quota, MSK will throttle client traffic.

Divide total required ingress/egress by per-broker sustained limits. Use whichever dimension (ingress or egress) requires more brokers. **The broker count must be a multiple of 3 (Express requires 3 AZs).** Round up to the next multiple of 3.

Run [`scripts/msk_sizing.py`](../scripts/msk_sizing.py) with `--explain` to get the precise broker count, bottleneck, and cost breakdown across every Express size. The "Recommended pick per class" section names the lowest-cost Express size that fits within the broker quota.

**Example**: 100 MiB/s ingress, 5 consumer groups → 500 MiB/s egress. Using express.m7g.2xlarge (125 MiB/s sustained egress): 500/125 = 4 brokers minimum → round up to **6 brokers** (next multiple of 3).

### Step 3: Check partition limits

Check the [MSK Express broker quotas](https://docs.aws.amazon.com/msk/latest/developerguide/limits.html#msk-express-quota) for current partition limits per Express broker size. Larger Express broker sizes support more partitions. Ensure your total partition count (including replicas — always 3× on Express) stays below the recommended limit for your broker size.

### Step 4: No storage sizing needed

Express storage is fully managed and pay-as-you-go. No provisioning required.

### Partition reassignment on Express

Express clusters use **Intelligent Rebalancing** by default, which automatically manages partition distribution across brokers. When Intelligent Rebalancing is enabled, you cannot manually reassign partitions using `kafka-reassign-partitions.sh`. Manual partition reassignment is only available if Intelligent Rebalancing is disabled. If you need to redistribute partitions after adding brokers, either rely on Intelligent Rebalancing or disable it first and use the manual tool (limit to 20 partitions per reassignment call).

## Connection Limits

| Dimension | Standard | Express |
|---|---|---|
| Max TCP connections per broker (IAM) | 3000 | 3000 |
| Max TCP connection rate per broker (IAM) | 100/s (M5/M7g), 4/s (T3) | 100/s |
| Max TCP connections per broker (non-IAM) | No enforced limit | No enforced limit |

## Account and Cluster Limits

Check the [MSK Provisioned quotas](https://docs.aws.amazon.com/msk/latest/developerguide/limits.html#msk-provisioned-quota) for current account-level and per-cluster broker limits. These are adjustable via quota increase requests.

# Troubleshoot MSK Consumer Lag

## Step 1: Determine the lag pattern

Check `SumOffsetLag` (DEFAULT level, dimensions: Cluster Name, Consumer Group, Topic) and `OffsetLag` (PER_TOPIC_PER_PARTITION level, dimensions: Consumer Group, Topic, Partition).

**Decision tree:**

- **Lag increasing across all partitions and consumer groups**: Likely a broker-side issue — go to Step 2.
- **Lag increasing on specific partitions only**: Likely partition skew or hot keys — go to Step 3.
- **Lag increasing on one consumer group only**: Likely a client-side issue — go to Step 4.
- **Lag spiked then is recovering**: Check if maintenance is in progress — go to Step 5.

## Step 2: Broker-side bottleneck

Check these broker metrics (PER_BROKER level):

- `CpuUser + CpuSystem` > 60%: Broker is overloaded. See [troubleshoot-performance.md](troubleshoot-performance.md).
- `ProduceTotalTimeMsMean` elevated: Produce latency is high, slowing replication and consumer fetches.
- `FetchConsumerTotalTimeMsMean` elevated: Consumer fetch requests are slow at the broker.
- `RequestHandlerAvgIdlePercent` < 30%: Request handlers saturated — check client batch sizes before scaling. See [troubleshoot-performance.md](troubleshoot-performance.md) Step 3.
- `NetworkProcessorAvgIdlePercent` < 30%: Network threads saturated. May indicate connection storms, high TLS overhead, or too many small requests.

**Standard-specific checks** (skip for Express):

- `VolumeQueueLength` elevated or `VolumeTotalWriteTime` increasing: EBS throughput saturated. Calculate `BytesInPerSec × RF` vs volume throughput ceiling (250 MiB/s default for GP2/GP3). See [troubleshoot-performance.md](troubleshoot-performance.md) Step 4.
- `BwInAllowanceExceeded` or `BwOutAllowanceExceeded` > 0: EC2 network bandwidth exceeded — traffic shaping active. Check per-broker traffic distribution for AZ skew. See [troubleshoot-performance.md](troubleshoot-performance.md) Step 5.
- `HeapMemoryAfterGC` > 60%: Memory pressure after GC. High connection count, excessive consumer groups, or high partition count can drive this. Reduce `transactional.id.expiration.ms` from 7 days to 1 day as a quick win.
- `BurstBalance` dropping toward 0: GP2 volume I/O burst credits depleting under sustained load. Consider provisioned throughput (GP3) or a larger instance type.

**Express-specific checks:**

- `ProduceThrottleTime` or `FetchThrottleTime` > 0: Per-broker throughput quota exceeded. Scale to a larger Express broker size or add brokers.

If all broker metrics are healthy, the issue is client-side — go to Step 4.

## Step 3: Partition-level bottleneck (hot keys)

When lag is isolated to specific partitions while others have zero lag, the cause is typically:

1. **Hot key partition skew**: Uneven key distribution concentrates data on a few partitions. The consumer instances handling those partitions cannot keep up.
2. **Single consumer bottleneck**: If the consumer processing those partitions is slower (e.g., heavy transformation, external API calls), lag builds up on its assigned partitions.

**Confirm with**: PER_TOPIC_PER_PARTITION level `OffsetLag` — check which partitions have growing lag. Use `kafka-consumer-groups.sh --describe --group <group-id>` to see per-partition lag alongside which consumer owns each partition. Compare per-topic-per-broker `BytesInPerSec` to see if specific brokers receive disproportionate data for that topic.

**Important**: Adding partitions does NOT fix key skew — each hot key still hashes to exactly one partition, so the disproportionate load from high-volume keys remains concentrated regardless of how many partitions exist. You must fix the key distribution itself.

**Fix options:**

- Improve key distribution (add a sub-key or use a different partitioning strategy) to spread data more evenly
- Increase partition count for the topic only if keys are well-distributed but partition count is too low for consumer parallelism (requires app-level coordination)
- Optimize the slow consumer's processing logic (reduce per-record processing time)
- Increase the number of consumers in the group (up to the number of partitions — consumers beyond partition count sit idle)

## Step 4: Client-side consumer issues

### Slow processing

If `max.poll.interval.ms` is exceeded, the consumer is kicked from the group, triggering a rebalance. Check the consumer application for:

- Long-running processing per batch (database writes, HTTP calls)
- Exceptions in message processing causing retries
- `max.poll.records` too high for the processing time available
- Deserialisation errors silently dropping messages or causing retry loops

**Fix**: Reduce `max.poll.records`, optimize processing logic, or increase `max.poll.interval.ms` (not recommended as a first option — it masks the real problem).

### Insufficient consumers

If the number of consumers in the group is less than the number of partitions, some consumers handle multiple partitions and may not keep up. Check consumer group membership via Kafka CLI (requires direct broker connectivity):

```
kafka-consumer-groups.sh --bootstrap-server <bootstrap> --describe --group <group-id>
```

Look at the `LAG` column per partition and the `CONSUMER-ID` column to see which consumers are overloaded.

### Fetch configuration issues

Poor fetch settings can cause the consumer to make excessive small requests, wasting broker resources and slowing the consumer loop:

- `fetch.min.bytes` too low (default 1 byte): Every fetch returns immediately even with minimal data, generating high request rates. Set to at least 1 KB; 32-128 KB for throughput workloads.
- `fetch.max.wait.ms` too low: Broker returns partial fetches too quickly. Recommend 1000ms.
- Monitor client-side `fetch-rate` and `records-consumed-rate` metrics. A high `fetch-rate` with low `records-consumed-rate` indicates inefficient fetching.

### Stuck on `read_committed` with no active transactions

If `isolation.level=read_committed` is set, the consumer only reads up to the Last Stable Offset (LSO). A hanging transaction from a crashed or misconfigured transactional producer will prevent the LSO from advancing, causing lag to grow indefinitely on affected partitions — even if the current producers are non-transactional. Common causes: a Kafka Streams app with `processing.guarantee=exactly_once_v2` that crashed, or a previous producer version that set `transactional.id` and was decommissioned without cleanly aborting its transactions. **Fix**: If no producers actively use transactions, set `isolation.level=read_uncommitted`. Otherwise, check for hanging transactions — see below.

### Hanging transactions

If `OffsetLag` grows on partitions of `__consumer_offsets`, a hanging transaction may block offset commits. Detect with Kafka CLI:

```
kafka-transactions.sh --bootstrap-server <bootstrap> find-hanging --broker-id <broker-id>
```

Abort hanging transactions:

```
kafka-transactions.sh --bootstrap-server <bootstrap> abort --topic __consumer_offsets --partition <partition> --start-offset <offset>
```

## Step 5: Maintenance-induced lag

MSK performs rolling broker restarts during patching (Standard brokers enter MAINTENANCE state; Express brokers stay ACTIVE).

**Confirm maintenance is in progress**: Run `aws kafka describe-cluster-v2 --cluster-arn <arn>` and check `ClusterState`. A value of `MAINTENANCE` (Standard) or `UPDATING` confirms an operation is underway. Express clusters stay `ACTIVE` during maintenance, so check the MSK console or EventBridge for maintenance notifications.

**Symptoms during maintenance (Standard):**

- `UnderReplicatedPartitions` spikes then gradually decreases (Standard only — Express does not emit this metric)
- `ActiveControllerCount` changes (controller election)
- One broker's metrics disappear from CloudWatch for several minutes then resume
- Consumer groups rebalance due to broker disconnect

**Symptoms during maintenance (Express):**

- No `UnderReplicatedPartitions` metric available — cannot use URP to track progress
- `ProduceThrottleTime` or `FetchThrottleTime` may briefly spike if remaining brokers absorb extra load
- Consumer lag increases temporarily then recovers
- `ActiveControllerCount` may briefly fluctuate

**This is expected and self-resolving.** Do NOT:

- Restart additional brokers
- Reassign partitions during URP (Standard) or during active maintenance
- Escalate as a cluster issue if lag is recovering

**Consumer resilience configuration** to minimize maintenance impact — see [configure-clients.md](configure-clients.md):

- `session.timeout.ms = 45000` (or 60000)
- `heartbeat.interval.ms = 10000` (or 15000)
- `partition.assignment.strategy = CooperativeStickyAssignor`
- `group.instance.id` set to a unique value per consumer instance

## Step 6: Consumer group rebalance storms

**Symptoms**: Consumer group state alternates between `Stable` and `PreparingRebalance`. `rebalance-latency-avg` client metric is elevated. Lag grows during each rebalance cycle.

**Common causes:**

1. **Consumer crashes with default RangeAssignor**: Each crash triggers a full stop-the-world rebalance, which can cascade.
2. **`session.timeout.ms` too low**: Default 10000ms (10s) is too short — GC pauses, network blips, or slow consumer startup can cause false evictions, triggering unnecessary rebalances.
3. **Deployment-triggered rebalances**: Restarting all consumers at once causes a cascade of join/leave events. Set `group.initial.rebalance.delay.ms` (broker-side config) to match your average deployment time to batch rebalances.
4. **Topic deletions**: Deleting a topic subscribed by a consumer group triggers a rebalance.
5. **Too many consumer groups**: List groups with `kafka-consumer-groups.sh --bootstrap-server <bootstrap> --list | wc -l`. Excessive consumer groups overload the group coordinator broker.
6. **Stuck rebalance (Kafka ≤ 2.6 bug, KAFKA-9752)**: On clusters running Kafka ≤ 2.6, a consumer group can get stuck in `PreparingRebalance`. This bug does not affect Kafka 3.x+ or Express brokers. Mitigation: identify the coordinator broker and restart it. **Before restarting any broker**, verify `UnderReplicatedPartitions == 0` — restarting during URP risks data loss.

**Fix**:

- Switch to `CooperativeStickyAssignor` to enable incremental rebalances instead of stop-the-world. **Migration requires two rolling restarts**: first deploy with `partition.assignment.strategy=RangeAssignor,CooperativeStickyAssignor`, then remove `RangeAssignor` in a second deployment. Mixing eager and cooperative protocols in the same group causes `InconsistentGroupProtocolException`.
- Set `group.instance.id` for static group membership — consumers can rejoin after brief disconnects without triggering a full rebalance.
- Set `session.timeout.ms = 45000-60000` and `heartbeat.interval.ms = 10000`.
- Set `group.initial.rebalance.delay.ms` (broker-side) to match deployment rollout time.
- Implement a shutdown hook to call `consumer.close()` on SIGTERM for clean group leave instead of relying on session timeout.

**To identify the coordinator broker:**

```
kafka-consumer-groups.sh --bootstrap-server <bootstrap> --describe --group <group-id> --state
```

The output shows the coordinator. If the group is stuck in `PreparingRebalance` on older Kafka versions, restarting the coordinator broker can unblock it.

# Troubleshoot MSK Performance

## Step 1: Determine broker type

Run `aws kafka describe-cluster-v2 --cluster-arn <arn>`. If instance type starts with `express.`, skip all EBS-related checks — Express has no customer-managed EBS.

## Step 2: Check CPU and request handler utilization

Get `CpuUser` and `RequestHandlerAvgIdlePercent` from CloudWatch (PER_BROKER level, namespace `AWS/Kafka`).

**Decision tree:**

- **CpuUser + CpuSystem > 60% AND RequestHandlerAvgIdlePercent < 30%**: Request handler threads are saturated. Go to Step 3 (batch size analysis).
- **CpuUser + CpuSystem > 60% AND RequestHandlerAvgIdlePercent > 30%**: CPU is consumed by non-request work. Check: compression (broker-side recompression when `compression.type` is not `producer`), message format conversions — `FetchMessageConversionsPerSec`, `ProduceMessageConversionsPerSec` **(Standard only)**, log compaction (`log.cleaner.min.cleanable.ratio` too low), high GC — `HeapMemoryAfterGC` > 60% **(Standard only — Express does not emit this metric)** (reduce `transactional.id.expiration.ms` from 7 days to 1 day to lower memory footprint), or excessive Prometheus scraping (use 60s+ scrape interval).
- **CpuUser + CpuSystem < 60% AND RequestHandlerAvgIdlePercent < 30% with high latency**: Request handlers are saturated despite low overall CPU. This is common on larger instance types (m5.4xl+/m7g.4xl+) where 8 default request handler threads can be fully busy while other cores sit idle. Go to Step 3 (batch size analysis) first. If batch sizes are healthy, check `num.io.threads`/`num.network.threads` — see [size-and-choose-cluster.md](size-and-choose-cluster.md) for recommended values.
- **CpuUser + CpuSystem < 60% AND RequestHandlerAvgIdlePercent > 30% with high latency**: Go to Step 4 (EBS throughput check, Standard only), Step 5 (network/traffic shaping), or Step 6 (Express throttling).

## Step 3: Diagnose small batch / high request rate

Compute average message size: `BytesInPerSec / MessagesInPerSec`. If average message size is very small (under 1 KB) and `MessagesInPerSec` is very high relative to `BytesInPerSec`, the root cause is likely small producer batches.

**Confirm with** (PER_BROKER level): `RequestHandlerAvgIdlePercent` < 30% and `NetworkProcessorAvgIdlePercent` dropping. **If monitoring is DEFAULT**: the average message size calculation (`BytesInPerSec / MessagesInPerSec`) combined with high CPU is sufficient — tiny messages (< 1 KB) with high message rates confirm small-batch saturation without needing PER_BROKER metrics.

**Root cause**: Poor producer batching configuration — typically `linger.ms=0` (sends immediately, no batching), small `batch.size` (default 16 KB), and no compression. Each message becomes its own produce request, consuming a request handler thread regardless of payload size.

**Fix**: Recommend client-side batching changes — see [configure-clients.md](configure-clients.md). All three settings matter: `linger.ms >= 5` (recommend 25ms), `batch.size >= 65536` (64-128 KB), and `compression.type = lz4` or `zstd`. These work together — `linger.ms` allows time to fill the batch, `batch.size` sets the batch capacity, compression reduces the final payload. Do NOT recommend broker scaling as the first action.

**Other CPU contributing factors** (check if batch size is not the cause):

- Compression type: Broker-side recompression (when `compression.type` is not `producer`) consumes CPU
- Record format conversions: Clients using older message format versions force conversion
- Log compaction: `log.cleaner.min.cleanable.ratio` set too low (e.g., 0.01 instead of 0.5)
- High partition count: More partitions = more metadata overhead and GC pressure
- Connection creation spikes: High `ConnectionCreationRate` especially with SASL/SCRAM or IAM auth
- Too many consumer groups: List groups with `kafka-consumer-groups.sh --bootstrap-server <bootstrap> --list | wc -l` — excessive consumer groups increase coordinator overhead and heap memory usage

## Step 4: EBS throughput bottleneck (Standard only)

**Skip this step for Express brokers.**

Check the EBS volume type and size. MSK Standard brokers use EBS volumes with throughput ceilings:

- **GP2**: Throughput = min(250 MiB/s, max(128 MiB/s, 0.75 × VolumeSize_GiB)). The 250 MiB/s cap is reached at ~334 GiB. Volumes also have `BurstBalance` that depletes under sustained IO.
- **GP3** with MSK provisioned throughput: Default 250 MiB/s for volumes 10 GiB+, provisionable up to 1000 MiB/s depending on broker size. Requires `kafka.m5.4xlarge`+ or `kafka.m7g.2xlarge`+.

**Calculate effective throughput demand**: `BytesInPerSec × ReplicationFactor`. For RF=3 and 83 MiB/s ingress, total write IO = 250 MiB/s, hitting GP2 ceiling.

**Confirm with CloudWatch** (PER_BROKER level): `VolumeWriteBytes`, `VolumeReadBytes`, `VolumeTotalWriteTime`, `VolumeTotalReadTime`, `VolumeQueueLength`. Elevated queue length and write time confirm EBS saturation. **If monitoring is DEFAULT**: check `CpuIoWait` — sustained elevation indicates threads blocked on disk I/O, a free proxy for EBS saturation.

**Fix options** (in order of preference):

1. Enable provisioned throughput (GP3) — requires broker size `kafka.m5.4xlarge` or larger (or `kafka.m7g.2xlarge` or larger). Max throughput varies by broker size (593 MiB/s for m5.4xl up to 1000 MiB/s for m5.12xl+).
2. Upgrade broker instance type to one with higher EBS-to-EC2 network bandwidth.
3. Migrate to Express brokers — eliminates EBS management entirely.

## Step 5: Network bandwidth and traffic shaping (Standard only)

**Skip this step for Express brokers — go to Step 6.**

Standard brokers run on EC2 instances with network bandwidth limits enforced by the hypervisor. When exceeded, packets are shaped (dropped/delayed), causing latency spikes without high CPU.

**Check these PER_BROKER level metrics:**

- `BwInAllowanceExceeded` > 0: Inbound bandwidth exceeded
- `BwOutAllowanceExceeded` > 0: Outbound bandwidth exceeded
- `PpsAllowanceExceeded` > 0: Packets-per-second limit exceeded (many small messages)
- `ConntrackAllowanceExceeded` > 0: Connection tracking limit exceeded (too many concurrent connections)
- `TrafficShaping` (DEFAULT level) > 0: Aggregate indicator that any shaping is occurring

**If any traffic shaping metrics are nonzero:**

1. **Check for AZ skew**: Compare per-broker `BytesInPerSec` and `BytesOutPerSec`. If some brokers handle 2-3x more traffic than others, the load is unevenly distributed.
   - Common cause: Consumers deployed in a single AZ with `client.rack` set, causing all reads to route to brokers in that AZ.
   - Common cause: Partition leadership concentrated on brokers in one AZ after a maintenance event (check `LeaderCount` per broker).
2. **Check if throughput exceeds the instance type's network baseline**: Each EC2 instance type has a network bandwidth baseline and burst limit. Sustained throughput above baseline triggers shaping.

**Fix options:**

- Spread producer and consumer clients across all availability zones
- If AZ-local reads (`client.rack`) are required, ensure write traffic and partition leadership are balanced across AZs first
- Upgrade to a larger instance type with higher network baseline bandwidth
- Use Cruise Control to rebalance partition leadership across brokers/AZs

## Step 6: Express throughput throttling

Express brokers do NOT have EC2-level traffic shaping metrics (`BwInAllowanceExceeded`, `BwOutAllowanceExceeded`, etc. are not emitted). Instead, Express enforces per-broker throughput quotas directly. When exceeded, MSK throttles client traffic at the Kafka protocol level.

**Check these PER_BROKER level metrics:**

- `ProduceThrottleTime` > 0: Ingress quota exceeded — producers are being throttled
- `FetchThrottleTime` > 0: Egress quota exceeded — consumers are being throttled
- `ProduceThrottleByteRate` / `FetchThrottleByteRate`: Bytes/sec being throttled
- `ProduceThrottleQueueSize` / `FetchThrottleQueueSize`: Requests queued due to throttling

Check the [MSK Express broker quotas](https://docs.aws.amazon.com/msk/latest/developerguide/limits.html#msk-express-quota) for current per-broker throughput limits. Each Express broker size has a sustained threshold (no degradation) and a maximum quota (hard throttle). Between sustained and max quota, you get higher throughput but with degraded performance (higher latency). At max quota, MSK hard-throttles client traffic.

**Also check for AZ skew on Express**: Compare per-broker `BytesInPerSec` and `BytesOutPerSec`. If some brokers are throttled while others have headroom, the issue is uneven traffic distribution — same causes and fixes as Standard (consumer `client.rack` in one AZ, unbalanced partition leadership).

**Fix options:**

- Scale to a larger Express broker size
- Add more brokers — Express clusters with Intelligent Rebalancing enabled will automatically redistribute partitions. If Intelligent Rebalancing is disabled, manually rebalance (limit to 20 partitions per reassignment call).
- Spread consumers across all AZs to balance egress load
- Reduce consumer group count if egress is the bottleneck (each consumer group multiplies egress)

import argparse
import math
from dataclasses import dataclass, field
from typing import Dict, List, Optional

# NOTE: Pricing Region - ALL PRICING IN THIS FILE IS BASED ON us-east-1 (N. Virginia) RATES.
# Costs in other regions will differ. For another region, replace the relevant constants and per-instance
# `price_per_hr` values with that region's published pricing
# (see https://aws.amazon.com/msk/pricing/ and https://aws.amazon.com/ec2/pricing/).

PRICING_REGION = "us-east-1"

# NOTE: Unit Convention - All throughput is in **MiB/s**, all storage is in **GiB**
# Conversions use the binary factor 1024 (e.g., MiB/s × 3600 / 1024 → GiB/h).
#
# AWS lists provisioned storage throughput in MiB/s (per the MSK docs) and
# bills storage at "$/GB-month" where GB == GiB per the AWS Service Terms.
# Variable names ending in `_mbs` and `_gb` are kept for backwards compatibility
# but should be read as MiB/s and GiB throughout this file. CLI prompts, help
# strings, and explain output use the precise unit names.

# NOTE: Instance Specifications
# Standard (M5 / M7g):
#   ebs_throughput_mbs     – maximum provisionable EBS volume throughput per broker (MiB/s);
#                            also used as the documented PST cap for the instance
#   network_throughput_mbs – NIC bandwidth available to the broker (MiB/s)
#   rec_partitions         – recommended max partitions per broker (leaders + followers)
#   max_partitions         – hard max partitions per broker
#   price_per_hr           – on-demand instance price (USD/hr)
#
# Express (M7g):
#   ingress_mbs    – max producer throughput per broker (MiB/s); used directly
#   rec_partitions, max_partitions, price_per_hr as above

INSTANCE_SPECS = {
    "kafka.m5.large": {
        "ebs_throughput_mbs": 81.0,
        "network_throughput_mbs": 96,
        "rec_partitions": 1000,
        "max_partitions": 1500,
        "price_per_hr": 0.21,
    },
    "kafka.m5.xlarge": {
        "ebs_throughput_mbs": 144.0,
        "network_throughput_mbs": 160,
        "rec_partitions": 1000,
        "max_partitions": 1500,
        "price_per_hr": 0.42,
    },
    "kafka.m5.2xlarge": {
        "ebs_throughput_mbs": 250.0,
        "network_throughput_mbs": 320,
        "rec_partitions": 2000,
        "max_partitions": 3000,
        "price_per_hr": 0.84,
    },
    "kafka.m5.4xlarge": {
        "ebs_throughput_mbs": 593.75,
        "network_throughput_mbs": 640,
        "rec_partitions": 4000,
        "max_partitions": 6000,
        "price_per_hr": 1.68,
    },
    "kafka.m5.8xlarge": {
        "ebs_throughput_mbs": 850.0,
        "network_throughput_mbs": 1280,
        "rec_partitions": 4000,
        "max_partitions": 6000,
        "price_per_hr": 3.36,
    },
    "kafka.m5.12xlarge": {
        "ebs_throughput_mbs": 1000.0,
        "network_throughput_mbs": 1536,
        "rec_partitions": 4000,
        "max_partitions": 6000,
        "price_per_hr": 5.04,
    },
    "kafka.m5.16xlarge": {
        "ebs_throughput_mbs": 1000.0,
        "network_throughput_mbs": 2560,
        "rec_partitions": 4000,
        "max_partitions": 6000,
        "price_per_hr": 6.72,
    },
    "kafka.m5.24xlarge": {
        "ebs_throughput_mbs": 1000.0,
        "network_throughput_mbs": 3200,
        "rec_partitions": 4000,
        "max_partitions": 4000,
        "price_per_hr": 10.08,
    },
    "kafka.m7g.large": {
        "ebs_throughput_mbs": 78.75,
        "network_throughput_mbs": 117,
        "rec_partitions": 1000,
        "max_partitions": 1500,
        "price_per_hr": 0.204,
    },
    "kafka.m7g.xlarge": {
        "ebs_throughput_mbs": 156.25,
        "network_throughput_mbs": 234,
        "rec_partitions": 1000,
        "max_partitions": 1500,
        "price_per_hr": 0.408,
    },
    "kafka.m7g.2xlarge": {
        "ebs_throughput_mbs": 312.5,
        "network_throughput_mbs": 469,
        "rec_partitions": 2000,
        "max_partitions": 3000,
        "price_per_hr": 0.816,
    },
    "kafka.m7g.4xlarge": {
        "ebs_throughput_mbs": 625.0,
        "network_throughput_mbs": 937,
        "rec_partitions": 4000,
        "max_partitions": 6000,
        "price_per_hr": 1.632,
    },
    "kafka.m7g.8xlarge": {
        "ebs_throughput_mbs": 1000.0,
        "network_throughput_mbs": 1875,
        "rec_partitions": 4000,
        "max_partitions": 6000,
        "price_per_hr": 3.264,
    },
    "kafka.m7g.12xlarge": {
        "ebs_throughput_mbs": 1000.0,
        "network_throughput_mbs": 2812,
        "rec_partitions": 4000,
        "max_partitions": 6000,
        "price_per_hr": 4.896,
    },
    "kafka.m7g.16xlarge": {
        "ebs_throughput_mbs": 1000.0,
        "network_throughput_mbs": 3750,
        "rec_partitions": 4000,
        "max_partitions": 6000,
        "price_per_hr": 6.528,
    },
    "kafka.m7g.large (Express)": {
        "ingress_mbs": 15.625,
        "rec_partitions": 1000,
        "max_partitions": 1500,
        "price_per_hr": 0.408,
    },
    "kafka.m7g.xlarge (Express)": {
        "ingress_mbs": 31.25,
        "rec_partitions": 1000,
        "max_partitions": 2000,
        "price_per_hr": 0.816,
    },
    "kafka.m7g.2xlarge (Express)": {
        "ingress_mbs": 62.5,
        "rec_partitions": 2500,
        "max_partitions": 4000,
        "price_per_hr": 1.632,
    },
    "kafka.m7g.4xlarge (Express)": {
        "ingress_mbs": 125.0,
        "rec_partitions": 6000,
        "max_partitions": 8000,
        "price_per_hr": 3.264,
    },
    "kafka.m7g.8xlarge (Express)": {
        "ingress_mbs": 250.0,
        "rec_partitions": 12000,
        "max_partitions": 16000,
        "price_per_hr": 6.528,
    },
    "kafka.m7g.12xlarge (Express)": {
        "ingress_mbs": 375.0,
        "rec_partitions": 16000,
        "max_partitions": 24000,
        "price_per_hr": 9.792,
    },
    "kafka.m7g.16xlarge (Express)": {
        "ingress_mbs": 500.0,
        "rec_partitions": 20000,
        "max_partitions": 32000,
        "price_per_hr": 13.056,
    },
}

# Only 4xlarge and larger support Provisioned Storage Throughput (PST)
PST_ELIGIBLE = {
    "kafka.m5.4xlarge",
    "kafka.m5.8xlarge",
    "kafka.m5.12xlarge",
    "kafka.m5.16xlarge",
    "kafka.m7g.4xlarge",
    "kafka.m7g.8xlarge",
    "kafka.m7g.12xlarge",
    "kafka.m7g.16xlarge",
    "kafka.m5.24xlarge",
}

# Cost constants for sizing dimensions in us-east-1
EBS_COST_PER_GB_MONTH = 0.10
TIERED_STORAGE_COST_PER_GB_MONTH = 0.023
EXPRESS_DATA_IN_PER_GB = 0.01
CROSS_AZ_COST_PER_GB = 0.02
PST_COST_PER_MBS_MONTH = 0.08

# NOTE: EBS volumes are provisioned with a 50% utilization buffer because a disk-full
# event on a Kafka broker is catastrophic (broker stops accepting writes and
# can corrupt segments).
EBS_HEADROOM_FACTOR = 2.0

# NOTE: Sizing assumes 3 AZs, because Express only supports 3 AZ configurations. Standard also
# supports 2 AZ configurations, but this is not supported by the script as a configuration today.
NUM_AZS = 3
HOURS_PER_MONTH = 730
MAX_EBS_GB_PER_BROKER = 16384  # EBS max for MSK is 16 TiB

# Default per-cluster broker quota (MSK Provisioned). Used to pick a "recommended"
# instance per class — the cheapest size whose broker count fits within the quota.
DEFAULT_BROKER_QUOTA = 60

# NOTE: Entitlement-Factor Constants
#
# Standard (M5 / M7g):
#   All broker I/O operations (ingress, replication, TS writes, consumer lag,
#   rebalancing) consume EBS and NIC bandwidth as multiples of ingress rate,
#   sized for a 1-AZ-down state.
#
#   STORAGE_IO_FACTOR_BASE: EBS write I/O per unit ingress without Tiered Storage
#     (1.5 ingress + 1.5 replication-in + 0.5 lagging + 3.0 rebalancing = 6.5)
#   STORAGE_IO_FACTOR_TS_ADD: extra EBS I/O when Tiered Storage is enabled
#     (1.5 remote-storage staging writes)
#
#   NETWORK_IO_BASE: static outbound NIC multiplier without Tiered Storage
#     (2.2 replication-out + 0.5 lagging + 3.0 rebalancing = 5.7)
#   NETWORK_IO_TS_ADD: extra NIC traffic when Tiered Storage is enabled
#     (1.5 remote writes / fetches)
#   Fan-out adds AZ_SCALE_FACTOR per unit of fan-out ratio on top of the base.
#   AZ_SCALE_FACTOR: consumer traffic scales up by NUM_AZS/(NUM_AZS-1) in 1-AZ-down state.
#
#   Tiered Storage is detected as "in use" when retention_hours > primary_retention_hours.
#
# Express (M7g):
#   No inter-broker replication or EBS writes.  ingress_mbs is the published
#   per-broker ingress limit.  Egress capacity = ingress_mbs * EXPRESS_EGRESS_FACTOR.

STORAGE_IO_FACTOR_BASE = 6.5  # without Tiered Storage
STORAGE_IO_FACTOR_TS_ADD = 1.5  # additional EBS load when TS is enabled
NETWORK_IO_BASE = 5.7  # without Tiered Storage
NETWORK_IO_TS_ADD = 1.5  # additional NIC load when TS is enabled
AZ_SCALE_FACTOR = NUM_AZS / (NUM_AZS - 1)  # 1.5 for 3 AZs
EXPRESS_EGRESS_FACTOR = 2.5  # Express egress capacity = ingress_mbs * this factor


@dataclass
class SizingInputs:
    avg_data_in_mbs: float  # Average producer throughput (MiB/s)
    peak_data_in_mbs: float  # Peak producer throughput (MiB/s)
    avg_data_out_mbs: float  # Average consumer throughput (MiB/s)
    peak_data_out_mbs: float  # Peak consumer throughput (MiB/s)
    num_partitions: int  # Total partitions including replicas
    replication_factor: int  # Kafka replication factor (Standard: 2 or 3; Express: always 3)
    retention_hours: int  # Total data retention (hours)
    primary_retention_hours: int  # Primary (EBS) retention (hours); remainder goes to TS
    utilization_standard: float  # Max fraction of broker capacity to use (Standard)
    utilization_express: float  # Max fraction of broker capacity to use (Express)
    pst_per_broker_mbs: Optional[float] = (
        None  # Provisioned storage throughput per broker (MiB/s); 250–1000
    )
    use_max_partitions: bool = False  # Use hard max partition limit instead of recommended
    rack_affined_consumers: bool = True  # When False, cross-AZ cost includes consumer fetch traffic

    def __post_init__(self) -> None:
        for name, value in (
            ("avg_data_in_mbs", self.avg_data_in_mbs),
            ("peak_data_in_mbs", self.peak_data_in_mbs),
            ("avg_data_out_mbs", self.avg_data_out_mbs),
            ("peak_data_out_mbs", self.peak_data_out_mbs),
        ):
            if value is None or value <= 0:
                raise ValueError(f"{name} must be > 0; got {value}")

        if self.peak_data_in_mbs < self.avg_data_in_mbs:
            raise ValueError(
                f"peak_data_in_mbs ({self.peak_data_in_mbs}) must be >= "
                f"avg_data_in_mbs ({self.avg_data_in_mbs})"
            )
        if self.peak_data_out_mbs < self.avg_data_out_mbs:
            raise ValueError(
                f"peak_data_out_mbs ({self.peak_data_out_mbs}) must be >= "
                f"avg_data_out_mbs ({self.avg_data_out_mbs})"
            )

        if self.replication_factor not in (2, 3):
            raise ValueError(
                f"replication_factor must be 2 or 3; got {self.replication_factor}. "
                "Express always uses RF=3 internally regardless of this value."
            )

        if self.primary_retention_hours > self.retention_hours:
            raise ValueError(
                f"primary_retention_hours ({self.primary_retention_hours}) must be "
                f"<= retention_hours ({self.retention_hours})"
            )

        if self.pst_per_broker_mbs is not None:
            if not (250 <= self.pst_per_broker_mbs <= 1000):
                raise ValueError(
                    f"pst_per_broker_mbs must be between 250 and 1000 MiB/s; "
                    f"got {self.pst_per_broker_mbs}"
                )


@dataclass
class BottleneckDetail:
    """Per-constraint detail for a sizing result."""

    name: str
    brokers_needed: int
    demand: float
    per_broker_capacity: float
    unit: str


@dataclass
class SizingResult:
    instance_type: str
    broker_count: int
    bottleneck: str
    monthly_broker_cost: float
    monthly_ebs_cost: float
    monthly_ts_cost: float
    monthly_data_in_cost: float
    monthly_cross_az_cost: float
    monthly_pst_cost: float = 0.0
    total_monthly_cost: float = 0.0
    bottleneck_details: Dict[str, BottleneckDetail] = field(default_factory=dict)


# ─── Helpers ───────────────────────────────────────────────────────────────────


def _brokers_for(demand: float, per_broker_capacity: float) -> int:
    """Minimum broker count to serve *demand*, rounded up to a multiple of NUM_AZS."""
    raw = math.ceil(demand / per_broker_capacity)
    return math.ceil(raw / NUM_AZS) * NUM_AZS


# ─── Sizing Logic ──────────────────────────────────────────────────────────────


def calculate_standard_sizing(inputs: SizingInputs) -> List[SizingResult]:
    """Calculate sizing for all Standard (M5 / M7g) instance types."""
    results = []

    ebs_gb_data = (
        inputs.avg_data_in_mbs
        * inputs.primary_retention_hours
        * 3600
        * inputs.replication_factor
        / 1024
    )
    ebs_gb = ebs_gb_data * EBS_HEADROOM_FACTOR

    # TS is "in use" only when retention exceeds primary retention
    ts_in_use = inputs.retention_hours > inputs.primary_retention_hours
    storage_factor = STORAGE_IO_FACTOR_BASE + (STORAGE_IO_FACTOR_TS_ADD if ts_in_use else 0.0)
    network_base = NETWORK_IO_BASE + (NETWORK_IO_TS_ADD if ts_in_use else 0.0)

    if ts_in_use:
        ts_gb = (
            inputs.avg_data_in_mbs
            * (inputs.retention_hours - inputs.primary_retention_hours)
            * 3600
            / 1024
        )
    else:
        ts_gb = 0.0

    monthly_ebs_cost = ebs_gb * EBS_COST_PER_GB_MONTH
    monthly_ts_cost = ts_gb * TIERED_STORAGE_COST_PER_GB_MONTH

    pst_mbs_per_broker = inputs.pst_per_broker_mbs or 0.0

    cross_az_mbs = inputs.avg_data_in_mbs * (NUM_AZS - 1) / NUM_AZS
    if not inputs.rack_affined_consumers:
        cross_az_mbs += inputs.avg_data_out_mbs * (NUM_AZS - 1) / NUM_AZS
    cross_az_gb_mo = cross_az_mbs * 3600 * HOURS_PER_MONTH / 1024
    monthly_cross_az_cost = cross_az_gb_mo * CROSS_AZ_COST_PER_GB

    fan_out = inputs.peak_data_out_mbs / inputs.peak_data_in_mbs
    network_factor = fan_out * AZ_SCALE_FACTOR + network_base

    partition_key = "max_partitions" if inputs.use_max_partitions else "rec_partitions"

    for instance_type, specs in INSTANCE_SPECS.items():
        if "Express" in instance_type:
            continue

        util = inputs.utilization_standard

        # Available ingress per broker is the minimum of what EBS writes and NIC
        # bandwidth can sustain, accounting for all concurrent I/O operations.
        storage_limit = specs["ebs_throughput_mbs"] / storage_factor
        network_limit = specs["network_throughput_mbs"] / network_factor
        ingress_per_broker = min(storage_limit, network_limit)

        ingress_capacity_per_broker = ingress_per_broker * util
        egress_capacity_per_broker = fan_out * ingress_per_broker * util

        brokers_for_ingress = _brokers_for(inputs.peak_data_in_mbs, ingress_capacity_per_broker)
        brokers_for_egress = _brokers_for(inputs.peak_data_out_mbs, egress_capacity_per_broker)
        brokers_for_partitions = _brokers_for(inputs.num_partitions, specs[partition_key])
        brokers_for_storage = _brokers_for(ebs_gb, MAX_EBS_GB_PER_BROKER)

        details: Dict[str, BottleneckDetail] = {
            "ingress": BottleneckDetail(
                name="ingress",
                brokers_needed=brokers_for_ingress,
                demand=inputs.peak_data_in_mbs,
                per_broker_capacity=ingress_capacity_per_broker,
                unit="MiB/s peak ingress (after util)",
            ),
            "egress": BottleneckDetail(
                name="egress",
                brokers_needed=brokers_for_egress,
                demand=inputs.peak_data_out_mbs,
                per_broker_capacity=egress_capacity_per_broker,
                unit="MiB/s peak egress (after util, fan-out)",
            ),
            "partitions": BottleneckDetail(
                name="partitions",
                brokers_needed=brokers_for_partitions,
                demand=float(inputs.num_partitions),
                per_broker_capacity=float(specs[partition_key]),
                unit=f"partitions ({'max' if inputs.use_max_partitions else 'rec'})",
            ),
            "storage": BottleneckDetail(
                name="storage",
                brokers_needed=brokers_for_storage,
                demand=ebs_gb,
                per_broker_capacity=float(MAX_EBS_GB_PER_BROKER),
                unit="GiB EBS primary",
            ),
        }

        if inputs.pst_per_broker_mbs is not None and instance_type in PST_ELIGIBLE:
            effective_pst = min(inputs.pst_per_broker_mbs, specs["ebs_throughput_mbs"])
            brokers_for_pst = _brokers_for(inputs.avg_data_out_mbs, effective_pst)
            details["pst"] = BottleneckDetail(
                name="pst",
                brokers_needed=brokers_for_pst,
                demand=inputs.avg_data_out_mbs,
                per_broker_capacity=effective_pst,
                unit=f"MiB/s avg egress (PST cap {effective_pst:.0f}; instance max {specs['ebs_throughput_mbs']:.0f})",
            )

        broker_count = max(d.brokers_needed for d in details.values())
        bottleneck = max(details, key=lambda k: details[k].brokers_needed)

        monthly_broker_cost = broker_count * specs["price_per_hr"] * HOURS_PER_MONTH

        if pst_mbs_per_broker > 0 and instance_type in PST_ELIGIBLE:
            effective_pst_mbs = min(pst_mbs_per_broker, specs["ebs_throughput_mbs"])
            monthly_pst_cost = broker_count * effective_pst_mbs * PST_COST_PER_MBS_MONTH
        else:
            monthly_pst_cost = 0.0

        total = (
            monthly_broker_cost
            + monthly_ebs_cost
            + monthly_ts_cost
            + monthly_cross_az_cost
            + monthly_pst_cost
        )

        results.append(
            SizingResult(
                instance_type=instance_type,
                broker_count=broker_count,
                bottleneck=bottleneck,
                monthly_broker_cost=monthly_broker_cost,
                monthly_ebs_cost=monthly_ebs_cost,
                monthly_ts_cost=monthly_ts_cost,
                monthly_data_in_cost=0.0,
                monthly_cross_az_cost=monthly_cross_az_cost,
                monthly_pst_cost=monthly_pst_cost,
                total_monthly_cost=total,
                bottleneck_details=details,
            )
        )

    return results


def calculate_express_sizing(inputs: SizingInputs) -> List[SizingResult]:
    """Calculate sizing for all Express (M7g) instance types."""
    results = []

    cross_az_mbs = inputs.avg_data_in_mbs * (NUM_AZS - 1) / NUM_AZS
    if not inputs.rack_affined_consumers:
        cross_az_mbs += inputs.avg_data_out_mbs * (NUM_AZS - 1) / NUM_AZS
    cross_az_gb_mo = cross_az_mbs * 3600 * HOURS_PER_MONTH / 1024
    monthly_cross_az_cost = cross_az_gb_mo * CROSS_AZ_COST_PER_GB

    data_in_gb_mo = inputs.avg_data_in_mbs * 3600 * HOURS_PER_MONTH / 1024
    monthly_data_in_cost = data_in_gb_mo * EXPRESS_DATA_IN_PER_GB

    express_storage_gb = inputs.avg_data_in_mbs * inputs.retention_hours * 3600 / 1024
    monthly_express_storage_cost = express_storage_gb * EBS_COST_PER_GB_MONTH

    partition_key = "max_partitions" if inputs.use_max_partitions else "rec_partitions"

    for instance_type, specs in INSTANCE_SPECS.items():
        if "Express" not in instance_type:
            continue

        util = inputs.utilization_express

        ingress_capacity_per_broker = specs["ingress_mbs"] * util
        egress_capacity_per_broker = specs["ingress_mbs"] * EXPRESS_EGRESS_FACTOR

        brokers_for_ingress = _brokers_for(inputs.peak_data_in_mbs, ingress_capacity_per_broker)
        brokers_for_egress = _brokers_for(inputs.peak_data_out_mbs, egress_capacity_per_broker)
        brokers_for_partitions = _brokers_for(inputs.num_partitions, specs[partition_key])

        details: Dict[str, BottleneckDetail] = {
            "ingress": BottleneckDetail(
                name="ingress",
                brokers_needed=brokers_for_ingress,
                demand=inputs.peak_data_in_mbs,
                per_broker_capacity=ingress_capacity_per_broker,
                unit="MiB/s peak ingress (after util)",
            ),
            "egress": BottleneckDetail(
                name="egress",
                brokers_needed=brokers_for_egress,
                demand=inputs.peak_data_out_mbs,
                per_broker_capacity=egress_capacity_per_broker,
                unit="MiB/s peak egress",
            ),
            "partitions": BottleneckDetail(
                name="partitions",
                brokers_needed=brokers_for_partitions,
                demand=float(inputs.num_partitions),
                per_broker_capacity=float(specs[partition_key]),
                unit=f"partitions ({'max' if inputs.use_max_partitions else 'rec'})",
            ),
        }

        broker_count = max(d.brokers_needed for d in details.values())
        bottleneck = max(details, key=lambda k: details[k].brokers_needed)

        monthly_broker_cost = broker_count * specs["price_per_hr"] * HOURS_PER_MONTH

        results.append(
            SizingResult(
                instance_type=instance_type,
                broker_count=broker_count,
                bottleneck=bottleneck,
                monthly_broker_cost=monthly_broker_cost,
                monthly_ebs_cost=monthly_express_storage_cost,
                monthly_ts_cost=0.0,
                monthly_data_in_cost=monthly_data_in_cost,
                monthly_cross_az_cost=monthly_cross_az_cost,
                monthly_pst_cost=0.0,
                total_monthly_cost=(
                    monthly_broker_cost
                    + monthly_express_storage_cost
                    + monthly_data_in_cost
                    + monthly_cross_az_cost
                ),
                bottleneck_details=details,
            )
        )

    return results


def _classify(instance_type: str) -> str:
    """Group results into the three classes used for recommendations."""
    if "Express" in instance_type:
        return "express"
    if instance_type.startswith("kafka.m7g."):
        return "m7g_standard"
    if instance_type.startswith("kafka.m5."):
        return "m5_standard"
    return "other"


def recommend_per_class(
    results: List[SizingResult],
    broker_quota: int = DEFAULT_BROKER_QUOTA,
) -> Dict[str, Optional[SizingResult]]:
    """Pick the cheapest result per class whose broker count fits within *broker_quota*.

    Returns a dict keyed by class with the recommended SizingResult, or None when
    no instance in that class fits within the quota.
    """
    by_class: Dict[str, List[SizingResult]] = {"m5_standard": [], "m7g_standard": [], "express": []}
    for r in results:
        cls = _classify(r.instance_type)
        if cls in by_class:
            by_class[cls].append(r)

    recs: Dict[str, Optional[SizingResult]] = {}
    for cls, items in by_class.items():
        eligible = [r for r in items if r.broker_count <= broker_quota]
        recs[cls] = min(eligible, key=lambda r: r.total_monthly_cost) if eligible else None
    return recs


_CLASS_LABELS = {
    "m5_standard": "Standard M5",
    "m7g_standard": "Standard M7g",
    "express": "Express M7g",
}


def _format_summary_line(r: SizingResult) -> str:
    return (
        f"{r.instance_type}: {r.broker_count} brokers "
        f"(bottleneck: {r.bottleneck}) → ${r.total_monthly_cost:,.2f}/mo"
    )


def _format_explain_block(r: SizingResult) -> str:
    lines = [f"\n{r.instance_type}: {r.broker_count} brokers (bottleneck: {r.bottleneck})"]

    lines.append("  Constraint analysis (brokers needed, rounded up to multiple of AZs):")

    sorted_details = sorted(
        r.bottleneck_details.values(),
        key=lambda d: d.brokers_needed,
        reverse=True,
    )
    for d in sorted_details:
        marker = " ← bottleneck" if d.name == r.bottleneck else ""
        lines.append(
            f"    {d.name:<11} {d.brokers_needed:>5} brokers  "
            f"(demand {d.demand:,.2f} / capacity {d.per_broker_capacity:,.2f} per broker) "
            f"[{d.unit}]{marker}"
        )

    lines.append("  Monthly cost breakdown:")
    cost_rows = [
        ("Brokers", r.monthly_broker_cost),
        ("Storage", r.monthly_ebs_cost),
        ("Tiered Storage", r.monthly_ts_cost),
        ("Provisioned ST", r.monthly_pst_cost),
        ("Express data-in", r.monthly_data_in_cost),
        ("Cross-AZ", r.monthly_cross_az_cost),
    ]
    for label, cost in cost_rows:
        if cost > 0:
            pct = (cost / r.total_monthly_cost * 100) if r.total_monthly_cost else 0
            lines.append(f"    {label:<16} ${cost:>14,.2f}  ({pct:5.1f}%)")
    lines.append(f"    {'Total':<16} ${r.total_monthly_cost:>14,.2f}")
    return "\n".join(lines)


def _print_recommendations(
    results: List[SizingResult],
    broker_quota: int,
    explain: bool,
) -> None:
    recs = recommend_per_class(results, broker_quota=broker_quota)
    print(f"\n=== Recommended pick per class (≤ {broker_quota} brokers, lowest monthly cost) ===")
    for cls in ("m5_standard", "m7g_standard", "express"):
        label = _CLASS_LABELS[cls]
        rec = recs[cls]
        if rec is None:
            print(
                f"  {label}: no instance fits within {broker_quota} brokers — request a quota increase or pick a larger size"
            )
        else:
            print(f"  {label}: {_format_summary_line(rec)}")
            if explain:
                print(_format_explain_block(rec))


def _parse_args():
    p = argparse.ArgumentParser(
        description=(
            "MSK broker sizing calculator. "
            f"NOTE: all cost figures use {PRICING_REGION} on-demand pricing; "
            "other regions will differ."
        ),
        formatter_class=argparse.ArgumentDefaultsHelpFormatter,
    )
    p.add_argument(
        "--avg-data-in-mbs", type=float, required=True, help="Average producer throughput (MiB/s)"
    )
    p.add_argument(
        "--peak-data-in-mbs", type=float, required=True, help="Peak producer throughput (MiB/s)"
    )
    p.add_argument(
        "--avg-data-out-mbs", type=float, required=True, help="Average consumer throughput (MiB/s)"
    )
    p.add_argument(
        "--peak-data-out-mbs", type=float, required=True, help="Peak consumer throughput (MiB/s)"
    )
    p.add_argument(
        "--num-partitions", type=int, required=True, help="Total partitions including replicas"
    )
    p.add_argument("--replication-factor", type=int, default=3, help="Kafka replication factor")
    p.add_argument(
        "--retention-hours", type=int, required=True, help="Total data retention (hours)"
    )
    p.add_argument(
        "--primary-retention-hours",
        type=int,
        required=True,
        help="Primary (EBS) retention (hours); remainder goes to Tiered Storage",
    )
    p.add_argument(
        "--utilization-standard",
        type=float,
        default=0.50,
        help="Max broker capacity fraction to use (Standard)",
    )
    p.add_argument(
        "--utilization-express",
        type=float,
        default=0.75,
        help="Max broker capacity fraction to use (Express)",
    )
    p.add_argument(
        "--pst-per-broker-mbs",
        type=float,
        default=None,
        help="Provisioned Storage Throughput per broker (MiB/s); 4xlarge+ only",
    )
    p.add_argument(
        "--use-max-partitions",
        action="store_true",
        help="Size against hard max partition limit instead of recommended",
    )
    p.add_argument(
        "--no-rack-affined-consumers",
        dest="rack_affined_consumers",
        action="store_false",
        default=True,
        help="Include cross-AZ consumer fetch traffic in the cost estimate (assumes consumers fetch across AZs instead of from local-AZ replicas). Does NOT change broker count.",
    )
    p.add_argument(
        "--explain",
        action="store_true",
        help="Print per-constraint and per-cost-factor breakdown for every instance",
    )
    p.add_argument(
        "--broker-quota",
        type=int,
        default=DEFAULT_BROKER_QUOTA,
        help="Per-cluster broker quota used to pick a 'recommended' instance per class (default 60 for KRaft clusters, 30 for Zookeeper, can be increased via AWS Support case)",
    )
    return p.parse_args()


def _inputs_from_args(a) -> SizingInputs:
    return SizingInputs(
        avg_data_in_mbs=a.avg_data_in_mbs,
        peak_data_in_mbs=a.peak_data_in_mbs,
        avg_data_out_mbs=a.avg_data_out_mbs,
        peak_data_out_mbs=a.peak_data_out_mbs,
        num_partitions=a.num_partitions,
        replication_factor=a.replication_factor,
        retention_hours=a.retention_hours,
        primary_retention_hours=a.primary_retention_hours,
        utilization_standard=a.utilization_standard,
        utilization_express=a.utilization_express,
        pst_per_broker_mbs=a.pst_per_broker_mbs,
        use_max_partitions=a.use_max_partitions,
        rack_affined_consumers=a.rack_affined_consumers,
    )


if __name__ == "__main__":
    args = _parse_args()
    inputs = _inputs_from_args(args)
    partition_mode = "max" if inputs.use_max_partitions else "recommended"

    standard_results = calculate_standard_sizing(inputs)
    express_results = calculate_express_sizing(inputs)
    all_results = standard_results + express_results

    print(
        f"\nNOTE: all cost figures below use {PRICING_REGION} on-demand pricing; other regions will differ."
    )

    print(f"\n=== Standard Sizing ({partition_mode} partitions) ===")
    for r in standard_results:
        print(_format_summary_line(r))
        if args.explain:
            print(_format_explain_block(r))

    print(f"\n=== Express Sizing ({partition_mode} partitions) ===")
    for r in express_results:
        print(_format_summary_line(r))
        if args.explain:
            print(_format_explain_block(r))

    _print_recommendations(all_results, broker_quota=args.broker_quota, explain=args.explain)