Managed Flink and Amazon MSK

By the end of this lesson you will be able to

• Explain what a stream processor does that a stream alone does not: filters, windows, joins and state, results within seconds.
• Describe Amazon Managed Service for Apache Flink: its old names, its sources (Kinesis Data Streams and MSK, never Data Firehose), its sinks, languages, KPUs, automatic scaling, checkpoints and snapshots.
• Explain Apache Kafka at the level the exam needs: brokers, topics, partitions, offsets, producers, consumer groups, replication and retention.
• Describe what Amazon MSK manages for you (brokers, ZooKeeper or KRaft controllers, Multi-AZ, recovery, EBS storage) and choose between Standard brokers, Express brokers and MSK Serverless.
• Name the consumers of MSK (Flink, Glue Streaming ETL, Lambda, Firehose, MSK Connect, apps on EC2, ECS and EKS) and pick the right one for a job.
• Compare Kinesis Data Streams and MSK (message size, shards vs partitions, scaling, encryption, retention, operations) and answer “Kinesis or Kafka?” exam questions.

#Collecting a stream is not processing it

#What a stream processor adds

In chapter 13 you met the two Kinesis services that move streaming data. Kinesis Data Streams (lesson 13.5) stores the records in shards so many consumers can read and replay them. Amazon Data Firehose (lesson 13.6) batches records and delivers them to S3, Redshift or OpenSearch. Neither of them thinks about the data. Neither can answer “how many ride requests came from Mirpur in the last minute, compared with free drivers there?”.

A stream processor is a program that runs all the time on the stream and does that thinking:

• Filter and transform: drop test events, mask phone numbers, turn raw pings into clean rows.
• Aggregate over windows: counts, sums and averages per minute, per five minutes, per session.
• Join: match a payment with the login that came just before it, or pings from one stream with trips from another.
• Detect patterns: five failed PIN attempts from three districts within two minutes.
• Keep state: running totals and recent history that survive from one event to the next.

The answer arrives within seconds, while the event still matters. A Lambda function on the stream can do simple per-batch work, but it forgets everything between invocations; anything that needs windows and state over many events wants a stream processor. On AWS, that is Amazon Managed Service for Apache Flink.

#Names you will meet

#Amazon Managed Service for Apache Flink

#What it is

Apache Flink is an open-source framework for processing data streams: you write a job that reads one or more sources, transforms the events and writes to one or more sinks, and Flink runs it in parallel across many workers, keeping state safely. Running Flink yourself means a cluster of job managers and task managers, state backends, upgrades and failure handling.

Managed Service for Apache Flink runs any Apache Flink application on a managed cluster for you. You upload your code (a JAR for Java, a zip for Python) to S3, create an application that points at it, give it an IAM role, and start it. AWS provisions the compute, runs the application in parallel, scales it, takes backups and restarts it after failures. There are no servers for you to see. You can use any Apache Flink programming feature: the low-level stream API, the Table API and Flink SQL, windows, timers, state, joins.

#Sources and sinks

On AWS a Flink application reads from:

• Kinesis Data Streams, with the Flink Kinesis connector.
• Amazon MSK (or any Apache Kafka cluster), with the Flink Kafka connector.

It writes to a long list of sinks: Kinesis Data Streams, Data Firehose, S3, OpenSearch, DynamoDB, MSK, and more through Flink connectors.

#Windows, time and state

Most real-time questions are about windows: a finite slice of an infinite stream.

Flink can measure time by the clock of the machine (processing time) or by the timestamp inside each event (event time). Event time gives the right answer even when a rider’s phone loses signal in a tunnel and sends its pings two minutes late; Flink uses watermarks to decide how long to wait for late events before closing a window.

All of this needs state: the counts inside an open window, the last location of each driver, the previous login of each wallet. Flink keeps state locally next to the computation (fast) and backs it up in checkpoints (safe).

#Capacity, scaling and cost

Capacity is measured in Kinesis Processing Units (KPUs). One KPU is 1 vCPU and 4 GB of memory, with 50 GB of running application storage. You choose:

• Parallelism: how many parallel copies of each operator run (default 1).
• ParallelismPerKPU: how many of those fit in one KPU (default 1, up to 8; raise it for jobs that mostly wait on I/O).

KPUs allocated = Parallelism ÷ ParallelismPerKPU. An application can use up to 64 KPUs by default (a quota you can raise).

Automatic scaling is on by default and works on CPU:

• If CPU stays at or above 75% for 15 minutes, the service doubles the parallelism, and so the KPUs.
• If CPU stays below 10% for six hours, it halves them, but never below the parallelism you configured.
• While it scales, the application restarts from a snapshot, so output pauses briefly. If that pause is unacceptable, turn automatic scaling off and set parallelism yourself (by hand, on a schedule before a known rush, or with your own metric-based automation).

Pricing (us-east-1): $0.11 per KPU-hour, billed per second, plus one extra KPU per application for orchestration, plus running storage at $0.10 per GB-month and durable backups per GB-month. A small application with parallelism 2 and ParallelismPerKPU 1 uses 2 + 1 = 3 KPUs: about 3 × $0.11 × 730 hours ≈ $241 a month, running 24 hours a day whether events arrive or not. There is no pay-per-event model; a stream processor is always on.

#Checkpoints and snapshots

The slides call these “application backups”. There are two kinds:

A checkpoint stores the state and the position in each source (the sequence number in each Kinesis shard, the offset in each Kafka partition). After a crash, Flink reloads the last checkpoint and re-reads the source from those positions. Every event then affects the state exactly once, even though some were read twice. That only works because the source can replay: one more reason Flink reads Kinesis Data Streams or Kafka, not Firehose.

#Apache Kafka at a high level

#The parts of Kafka

Apache Kafka is the most widely used open-source platform for event streaming, created at LinkedIn. Many companies in Bangladesh and abroad already run it on their own servers. Its parts:

• Broker: a Kafka server that stores data and serves clients. A cluster is several brokers.
• Topic: a named stream of records, such as orders or payments.
• Partition: a topic is split into partitions, each an ordered, append-only log stored on a broker. A partition is to Kafka what a shard is to Kinesis: the unit of ordering and of parallelism.
• Offset: the position of a record inside its partition (0, 1, 2…).
• Producer: your code that writes records to a topic. A record has an optional key; records with the same key go to the same partition, so they stay in order.
• Consumer and consumer group: your code that reads (polls) records. Consumers that share a group ID split the partitions between them: each partition is read by exactly one consumer of the group. Different groups each read everything, with their own offsets.
• Replication: each partition has a leader on one broker and followers on others (replication factor 3 is the usual choice). Producers write to the leader; followers copy it. If the leader’s broker dies, an in-sync follower takes over.
• Retention: reading does not delete anything. Records stay for the topic’s retention time (7 days by default in Kafka, configurable to much longer or forever), so consumers can replay.

The flow the slides show: data comes from many places (Kinesis, IoT devices, RDS…), producers (your code) write to a topic on the cluster, the brokers replicate it, consumers (your code) poll from the topic, and send results on to EMR, S3, SageMaker, Kinesis, RDS and so on.

#Partitions decide your parallelism

Because a partition goes to exactly one consumer of a group, the number of partitions is the maximum number of consumers that can work in parallel in one group. Six consumers on a four-partition topic leave two consumers idle. When a consumer crashes or a new one joins, the group rebalances: partitions are reassigned, and the new owner continues from the last committed offset. Records that were processed but not committed are processed again, so Kafka consumers are at-least-once by default, like Kinesis consumers, and must be idempotent.

You can add partitions to a topic later, but you can never remove them. Adding partitions changes which partition a key maps to, so for a while the events of one key can be split across an old and a new partition and lose their order. Choose the partition count for the busiest day you expect, up front.

#Amazon MSK: managed Kafka

#What MSK does for you

Amazon Managed Streaming for Apache Kafka (MSK) is fully managed Apache Kafka on AWS, and the slides present it as the alternative to Amazon Kinesis. It runs real, open-source Kafka, so your existing producers, consumers, tools and skills work unchanged. MSK:

• Lets you create, update and delete clusters with an API call or a few clicks.
• Creates and manages the Kafka broker nodes and the metadata nodes for you: ZooKeeper nodes on older clusters, KRaft controllers on newer ones.
• Deploys the cluster in your VPC, spread over two or three Availability Zones for high availability (three is recommended; Express brokers require three).
• Recovers automatically from common Kafka failures: it detects a failed broker, replaces it and reattaches its storage.
• Stores data on EBS volumes, up to 16 TiB per broker, for as long as you want: retention is your choice, not a service limit. Storage can grow automatically, and tiered storage moves older data to a cheaper tier.
• Encrypts data at rest with KMS always, and in transit with TLS by default.
• Patches and upgrades Kafka versions when you ask, and sends metrics to CloudWatch (or Prometheus).

What you still decide on a provisioned cluster: broker instance type and count, storage size, Kafka version, topic settings (partitions, replication factor, retention), and client authentication.

#Three ways to run MSK

MSK Serverless is what the slides describe: run Kafka on MSK without managing capacity; MSK provisions and scales compute and storage for you. It is the Kafka counterpart of Kinesis on-demand mode.

#Security

• Network: brokers have network interfaces in your private subnets, protected by security groups. Public access can be switched on for provisioned clusters, but most clusters stay private.
• Authentication: IAM access control (IAM policies decide who may write or read which topic), SASL/SCRAM (usernames and passwords kept in Secrets Manager), mutual TLS (client certificates from AWS Private CA), or unauthenticated access if you allow it. MSK Serverless supports IAM only.
• Authorization: IAM policies with IAM auth; Kafka ACLs with SASL/SCRAM or mTLS.
• Encryption in flight: TLS between clients and brokers by default; PLAINTEXT, or both, only if you choose it. Traffic between brokers is TLS by default.
• Encryption at rest: always on, with an AWS managed KMS key or your own customer managed key.

#MSK Connect and MSK Replicator

• MSK Connect runs Kafka Connect workers for you. Kafka Connect is Kafka’s standard way to move data in and out without writing consumer code: a source connector brings data in (for example Debezium reading a database’s change log, which is change data capture), a sink connector takes it out (to S3, OpenSearch, a database). MSK Connect scales the workers automatically.
• MSK Replicator copies topics continuously from one MSK cluster to another, in the same Region or another Region, for disaster recovery or to bring data closer to users.

#Who reads from MSK

The slides list four consumer types; Firehose and MSK Connect complete the picture:

A Lambda consumer keeps order per partition: if one batch keeps failing, that partition waits behind it. Set a retry limit and an on-failure destination (SQS, SNS or S3) on the event source mapping, so a bad record is set aside and the partition moves on. Lesson 15.1 explains event source mappings.

#Kinesis Data Streams or Amazon MSK

#Side by side

#The exam trick

The notes give the rule, and it holds:

• An AWS-native, serverless solution with the least to manage → Kinesis Data Streams + Managed Service for Apache Flink.
• Open-source or an existing Kafka-based ecosystem (“the company already uses Apache Kafka on premises and wants to migrate with minimal code changes”, “uses Kafka Connect”, “needs Kafka APIs”) → Amazon MSK.

Other clues: messages larger than 1 MB (a classic reason for MSK, although Kinesis can now take up to 10 MiB per stream), retention longer than a year (MSK), “must run inside our VPC” (MSK, or Kinesis with a VPC endpoint), “no capacity planning, spiky” (Kinesis on-demand or MSK Serverless; the native choice wins unless Kafka is mentioned).

#What goes wrong

• Flink pointed at Firehose: impossible; Firehose has no consumer API. Use Kinesis Data Streams or MSK as the source.
• A Flink application that falls behind: millisecond lag in the source keeps growing; parallelism is too low, or one operator is a bottleneck. Watch the source lag metrics (millisBehindLatest for Kinesis, consumer lag for Kafka) and CPU, and raise parallelism or let automatic scaling act.
• Output pauses during an Eid rush: automatic scaling restarted the application. Scale before the known peak, or disable automatic scaling and size it yourself.
• A Flink job that cannot recover: checkpoints were disabled, or the source retention is shorter than the outage, so the positions in the checkpoint no longer exist.
• Idle Kafka consumers: more consumers in a group than partitions. Add partitions (planning for key order) or accept the extra consumers as standbys.
• Duplicates after a rebalance: a consumer crashed after processing but before committing. Make processing idempotent; commit after processing, not before.
• Order broken after adding partitions: keys were remapped. Choose the partition count up front.
• Broker disks full: retention too long for the EBS size. Alarm on disk use, grow storage, use tiered storage or Express brokers, or shorten retention.
• A cluster in one AZ, or replication factor 1: an AZ failure makes partitions unavailable or loses data. Use three AZs, replication factor 3, min.insync.replicas 2 and acks=all for important data.
• Clients cannot connect: wrong port or authentication method (IAM, SCRAM, TLS each use their own bootstrap string), security group closed, or an IAM policy without kafka-cluster permissions on the topic.
• A Lambda consumer stuck on one bad record: configure retries and an on-failure destination.

#Flink and MSK in system design

Typical patterns:

• Real-time analytics: producers → Kinesis Data Streams or MSK → Flink (windows, joins) → OpenSearch dashboard, DynamoDB, alerts through SNS; plus Firehose or Glue Streaming → S3 for the data lake. The full serverless pipeline, from IoT devices to dashboards, is the subject of lesson 17.8.
• Fraud detection: payments → stream → Flink keyed by wallet, comparing each payment with the wallet’s recent history kept in state → block or flag within a second.
• Change data capture: database → MSK Connect (Debezium) → MSK → consumers that update search indexes, caches and the data warehouse.
• Migrating an on-premises Kafka: replicate topics into MSK, move consumers, then producers; MSK Replicator for later cross-Region copies.
• Decoupling microservices with events: services publish domain events to MSK topics; each service reads what it needs in its own consumer group.

#Try it yourself

These commands use the AWS CLI v2 and the open-source Kafka command-line tools. MSK and Flink applications are billed per hour while they exist, so delete them at the end. Subnet, security group and account values are examples.

# 1. Create an MSK Serverless cluster (IAM authentication only)
aws kafka create-cluster-v2 --cluster-name rides-example \
  --serverless '{"VpcConfigs":[{"SubnetIds":["subnet-0a1b2c3d","subnet-0e5f6a7b","subnet-0c8d9e0f"],
                 "SecurityGroupIds":["sg-0123456789abcdef0"]}],
                 "ClientAuthentication":{"Sasl":{"Iam":{"Enabled":true}}}}'
# { "ClusterArn": "arn:aws:kafka:ap-southeast-1:111122223333:cluster/rides-example/1a2b3c4d-...-s1",
#   "ClusterName": "rides-example", "State": "CREATING", "ClusterType": "SERVERLESS" }

# 2. Get the bootstrap servers once the cluster is ACTIVE
aws kafka get-bootstrap-brokers \
  --cluster-arn arn:aws:kafka:ap-southeast-1:111122223333:cluster/rides-example/1a2b3c4d-...-s1
# { "BootstrapBrokerStringSaslIam":
#     "boot-abc12345.c1.kafka-serverless.ap-southeast-1.amazonaws.com:9098" }

# 3. On an EC2 instance in the same VPC: client settings for IAM auth (needs the aws-msk-iam-auth jar)
cat > client.properties <<'EOF'
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
EOF
BS=boot-abc12345.c1.kafka-serverless.ap-southeast-1.amazonaws.com:9098

# 4. Create a topic with 6 partitions (Serverless sets the replication itself)
kafka-topics.sh --bootstrap-server $BS --command-config client.properties \
  --create --topic ride-pings --partitions 6
# Created topic ride-pings.

# 5. Produce keyed records, then read them in a consumer group
kafka-console-producer.sh --bootstrap-server $BS --producer.config client.properties \
  --topic ride-pings --property parse.key=true --property key.separator=:
# > driver-17:{"area":"Mirpur","lat":23.80,"lon":90.36}
kafka-console-consumer.sh --bootstrap-server $BS --consumer.config client.properties \
  --topic ride-pings --group surge --from-beginning

# 6. See partitions, offsets and lag per consumer of the group
kafka-consumer-groups.sh --bootstrap-server $BS --command-config client.properties \
  --describe --group surge
# GROUP  TOPIC       PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG  CONSUMER-ID
# surge  ride-pings  0          1204            1204            0    console-consumer-5f1...
# surge  ride-pings  1          977             981             4    console-consumer-5f1...

# 7. Add partitions (allowed) - removing them is not possible
kafka-topics.sh --bootstrap-server $BS --command-config client.properties \
  --alter --topic ride-pings --partitions 12

# 8. A Lambda function as an MSK consumer
aws lambda create-event-source-mapping --function-name update-ride-status \
  --event-source-arn arn:aws:kafka:ap-southeast-1:111122223333:cluster/rides-example/1a2b3c4d-...-s1 \
  --topics ride-pings --starting-position LATEST --batch-size 100

A provisioned cluster takes a --provisioned block instead, with the broker type, the client subnets in two or three AZs and the EBS size:

aws kafka create-cluster-v2 --cluster-name payments-example --provisioned '{
  "KafkaVersion": "3.7.x.kraft", "NumberOfBrokerNodes": 3,
  "BrokerNodeGroupInfo": { "InstanceType": "kafka.m7g.large",
    "ClientSubnets": ["subnet-0a1b2c3d","subnet-0e5f6a7b","subnet-0c8d9e0f"],
    "SecurityGroups": ["sg-0123456789abcdef0"],
    "StorageInfo": { "EbsStorageInfo": { "VolumeSize": 1000 } } },
  "EncryptionInfo": { "EncryptionInTransit": { "ClientBroker": "TLS", "InCluster": true } },
  "ClientAuthentication": { "Sasl": { "Iam": { "Enabled": true } } } }'

Now a Flink application. Build the job into a JAR, upload it to S3, then:

# 9. Create the application: code in S3, an IAM role, parallelism and checkpoints
aws kinesisanalyticsv2 create-application --application-name surge-example \
  --runtime-environment FLINK-1_20 \
  --service-execution-role arn:aws:iam::111122223333:role/flink-surge-role \
  --application-configuration '{
    "ApplicationCodeConfiguration": { "CodeContentType": "ZIPFILE",
      "CodeContent": { "S3ContentLocation": {
        "BucketARN": "arn:aws:s3:::flink-code-example", "FileKey": "surge-1.0.jar" } } },
    "FlinkApplicationConfiguration": {
      "ParallelismConfiguration": { "ConfigurationType": "CUSTOM", "Parallelism": 2,
        "ParallelismPerKPU": 1, "AutoScalingEnabled": true },
      "CheckpointConfiguration": { "ConfigurationType": "DEFAULT" } },
    "ApplicationSnapshotConfiguration": { "SnapshotsEnabled": true } }'
# { "ApplicationDetail": { "ApplicationName": "surge-example", "ApplicationStatus": "READY",
#     "RuntimeEnvironment": "FLINK-1_20", "ApplicationVersionId": 1, ... } }

# 10. Start it, restoring from the latest snapshot if there is one
aws kinesisanalyticsv2 start-application --application-name surge-example \
  --run-configuration '{"ApplicationRestoreConfiguration":{"ApplicationRestoreType":"RESTORE_FROM_LATEST_SNAPSHOT"}}'

# 11. Take a snapshot before a risky change, and list snapshots
aws kinesisanalyticsv2 create-application-snapshot --application-name surge-example \
  --snapshot-name before-eid-release
aws kinesisanalyticsv2 list-application-snapshots --application-name surge-example

# 12. Clean up
aws kinesisanalyticsv2 stop-application --application-name surge-example
aws kafka delete-cluster \
  --cluster-arn arn:aws:kafka:ap-southeast-1:111122223333:cluster/rides-example/1a2b3c4d-...-s1

In the console, the Flink application page opens the Apache Flink dashboard, where you can see each operator, its parallelism and its back pressure. On the MSK side, the Monitoring tab shows bytes in and out per broker and, for provisioned clusters, disk use.

#Summary

Key takeaways

• A stream processor filters, aggregates over windows, joins and keeps state on a live stream, with results in seconds. On AWS that is Amazon Managed Service for Apache Flink, formerly Kinesis Data Analytics for Apache Flink. The SQL-only Kinesis Data Analytics is discontinued (no new apps since 15 October 2025, deleted from 27 January 2026).
• Flink reads from Kinesis Data Streams and Amazon MSK (Kafka). It does not read from Data Firehose; Firehose can be a sink. Code in Java, Python, SQL (Scala deprecated in Flink 2.x); Studio notebooks for interactive SQL.
• Capacity in KPUs (1 vCPU, 4 GB). Automatic scaling doubles parallelism after 15 minutes above 75% CPU and halves it after 6 hours below 10%, with a short restart. Checkpoints (automatic, every 60 s) recover from crashes with exactly-once state; snapshots (manual or on update) restore chosen states.
• Kafka: brokers store topics split into partitions (ordered logs with offsets); producers write by key; a consumer group gives each partition to one consumer; replication factor 3 with leaders and followers; retention, so reading deletes nothing and replay is possible.
• Amazon MSK = managed Kafka, the alternative to Kinesis: it creates and manages brokers and ZooKeeper / KRaft nodes, runs in your VPC across 2 or 3 AZs, recovers failures automatically, stores data on EBS for as long as you want. Options: Standard brokers, Express brokers (more throughput, no storage to manage), MSK Serverless (no capacity management, IAM only).
• MSK consumers: Flink, Glue Streaming ETL, Lambda, apps on EC2 / ECS / EKS, Firehose, MSK Connect. MSK Connect runs Kafka Connect; MSK Replicator copies between clusters.
• Kinesis vs MSK: 1 MB vs 1 MB default (configurable higher); shards (split/merge) vs partitions (only add); TLS always vs TLS or plaintext; KMS at rest for both.
• Exam rule from the notes: AWS-native → Kinesis Data Streams + Managed Flink; open-source or existing Kafka → Amazon MSK.

Self-check: Managed Flink and Amazon MSK32 questions