Skip to content

06-msk-serverless-and-connect.md

Latest commit

 

History

History
221 lines (163 loc) · 8.34 KB

06-msk-serverless-and-connect.md

File metadata and controls

221 lines (163 loc) · 8.34 KB
Raw

Amazon MSK Serverless and MSK Connect

This document represents the streaming-ingestion layer of the ecommerce project. Amazon MSK Serverless provides the Kafka cluster, IAM controls producer and connector access, and MSK Connect delivers the ecommerce-orders topic to the S3 Bronze layer.

Implemented streaming flow

EC2 Python Producer
        │ SASL_SSL + IAM · TCP 9098
        ▼
Amazon MSK Serverless
        │ ecommerce-orders
        ▼
Amazon MSK Connect S3 Sink
        │ HTTPS 443
        ▼
s3://adarsh-ecommerce-streaming/bronze/
Setting Project value
AWS Region us-east-1
Cluster type MSK Serverless
Cluster name ecommerce-msk-cluster
Authentication IAM access control
Transport TLS / SASL_SSL
Kafka topic ecommerce-orders
Producer role ecommerce-kafka-client-role
Connector role ecommerce-msk-connect-role
S3 destination s3://adarsh-ecommerce-streaming/bronze/

MSK Serverless requires IAM access control and private VPC connectivity. The project uses two private subnets in separate Availability Zones and security-group references between the producer, connector, and cluster.

Cluster creation in the AWS console

In Amazon MSK, the cluster is represented by these selections:

Creation method: Custom create
Cluster type: Serverless
Cluster name: ecommerce-msk-cluster
Network type: IPv4
VPC: project VPC
Subnets: two private subnets in different Availability Zones
Security group: project MSK security group
Authentication: IAM access control

The MSK security group accepts TCP 9098 from the EC2 producer security group and the MSK Connect security group. It does not expose broker traffic to 0.0.0.0/0.

Cluster creation with the AWS CLI

The reusable request is msk/serverless-cluster.json.example. After replacing the subnet and security-group placeholders, it can be supplied to AWS CLI v2:

aws kafka create-cluster-v2 \
  --region us-east-1 \
  --cli-input-json file://msk/serverless-cluster.json

After the cluster reaches ACTIVE, its ARN and IAM bootstrap servers are available through:

aws kafka list-clusters-v2 \
  --region us-east-1 \
  --cluster-name-filter ecommerce-msk-cluster

aws kafka get-bootstrap-brokers \
  --region us-east-1 \
  --cluster-arn '<MSK_CLUSTER_ARN>'

The private bootstrap string is supplied as a runtime value.

IAM authorization

The producer permissions are defined in iam/producer-msk-policy.json. They allow cluster connection and writes only to ecommerce-orders.

The connector permissions are defined in iam/msk-connect-s3-sink-policy.json. They allow:

  • connection to the project cluster;
  • consumption from ecommerce-orders;
  • connector consumer-group operations;
  • access to the packaged connector plugin;
  • writes only under the bucket's bronze/* prefix.

MSK Serverless uses IAM policies rather than Kafka ACLs.

Kafka client and topic

Kafka administrative tools run on the EC2 client inside the VPC. The IAM client settings are stored in msk/client.properties:

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

The Kafka tools require the AWS MSK IAM authentication JAR on their classpath. With KAFKA_ROOT pointing to the Kafka installation and BOOTSTRAP_SERVERS containing the private IAM endpoint, the topic definition is:

The EC2 role can temporarily use iam/msk-topic-admin-policy.json during topic creation. The narrower producer policy remains attached for normal application traffic.

$KAFKA_ROOT/bin/kafka-topics.sh \
  --create \
  --bootstrap-server "$BOOTSTRAP_SERVERS" \
  --command-config msk/client.properties \
  --partitions 3 \
  --replication-factor 3 \
  --topic ecommerce-orders

Topic verification:

$KAFKA_ROOT/bin/kafka-topics.sh \
  --describe \
  --bootstrap-server "$BOOTSTRAP_SERVERS" \
  --command-config msk/client.properties \
  --topic ecommerce-orders

Producer configuration

The Python producer uses aws-msk-iam-sasl-signer-python and the EC2 instance role. Its bootstrap endpoint is supplied by the host environment:

export MSK_BOOTSTRAP_SERVER='<IAM_BOOTSTRAP_HOST>:9098'
python3 producer/producer.py

The producer sends one JSON order every five seconds and updates one customer approximately every 20 orders. The code is in producer/producer.py.

MSK Connect custom plugin

The S3 sink implementation is packaged as a custom plugin. MSK Connect accepts one JAR or a ZIP containing the connector and all required dependencies. The project stores the archive under:

s3://adarsh-ecommerce-streaming/connectors/confluentinc-kafka-connect-s3.zip

The plugin resource can be created from msk/create-custom-plugin.json.example:

aws kafkaconnect create-custom-plugin \
  --region us-east-1 \
  --cli-input-json file://msk/create-custom-plugin.json

MSK Connect copies the plugin at creation time. Replacing the S3 object does not update an existing plugin revision or connector.

Worker configuration

The implemented connector uses the default MSK Connect worker configuration. String-key and schemaless-JSON converter settings are defined directly in the connector configuration.

S3 sink connector

The human-readable connector settings are in msk/s3-sink-connector.properties. They produce schemaless JSON objects under the Bronze prefix and rotate output regularly so the low-volume portfolio producer does not wait for a very large file.

The complete CLI request is msk/create-connector.json.example. It includes:

  • provisioned connector capacity of one worker and one MCU;
  • IAM cluster authentication and TLS;
  • two private subnets and the connector security group;
  • the custom S3 sink plugin;
  • the default worker configuration;
  • CloudWatch worker logs;
  • the ecommerce-msk-connect-role service execution role.

The kafkaConnectVersion placeholder must be replaced with a version currently offered by MSK Connect and compatible with the packaged plugin.

aws kafkaconnect create-connector \
  --region us-east-1 \
  --cli-input-json file://msk/create-connector.json

Verification evidence

The streaming layer is healthy when all of the following are true:

  1. The MSK Serverless cluster state is ACTIVE.
  2. ecommerce-orders appears in the Kafka topic list.
  3. The Python producer receives successful acknowledgements without authentication or timeout errors.
  4. The connector state is RUNNING.
  5. CloudWatch connector logs contain no authorization or task failures.
  6. New JSON objects appear below s3://adarsh-ecommerce-streaming/bronze/.
  7. A Bronze object contains the original order fields, including customer_updated_at.

Useful connector inspection command:

aws kafkaconnect describe-connector \
  --region us-east-1 \
  --connector-arn '<CONNECTOR_ARN>'

Common failure patterns

Symptom Project-level cause
Kafka connection timeout Producer/connector subnet, DNS, route, or port 9098 security-group path
AccessDenied on Kafka actions Incorrect cluster/topic ARN or missing producer/consumer IAM actions
Connector remains CREATING or becomes FAILED Plugin incompatibility, missing dependency, invalid connector property, or role trust issue
No Bronze files Connector is not consuming, role cannot write bronze/*, or rotation threshold has not been reached
JSON includes a schema envelope value.converter.schemas.enable was not set to false
Plugin changes have no effect Existing custom plugins are copied resources and were not recreated

Deployment variables

The following values are supplied for the deployed AWS environment:

  • account ID;
  • cluster name/UUID ARN components;
  • VPC subnet and security-group IDs;
  • IAM bootstrap server string;
  • custom-plugin ARN;
  • supported Kafka Connect runtime version.

AWS credentials are obtained from service roles and are not stored in the configuration files.