Event Processing Pattern

এই লেসন শেষে তুমি পারবে

• SQS, SNS আর Lambda-কে শুরু থেকে শেষ পর্যন্ত জুড়তে, আর প্রশ্নের শব্দ দেখে ঠিকটা বাছতে: buffer করা আর decouple করা (SQS), অনেকের কাছে push করা (SNS), কোনো event হলে code চালানো (Lambda)।
• একটা Lambda function-এর পেছনের তিন রকম retry বুঝিয়ে বলতে: SQS + Lambda (poller, visibility timeout, maxReceiveCount, queue-এর DLQ, partial batch response), SQS FIFO + Lambda (message group ধরে ক্রম, failed message নিজের group আটকে রাখে), আর SNS + Lambda (asynchronous invoke, Lambda-র দুটো retry, on-failure destination, subscription DLQ)।
• প্রতিটা shape-এর জন্য dead-letter queue ঠিকভাবে বসাতে, আর সমস্যা মেটার পর message redrive করতে।
• Code থেকে কয়েকটা queue-তে লেখার বদলে SNS → SQS fan-out বেছে নিতে, আর সেটা চালু করতে (queue policy, filter policy, SNS হয়ে S3 event)।
• S3 event notification design করতে (event type, prefix আর suffix, গন্তব্য, overlap-এর নিয়ম, loop), আর কখন S3 event-এর বদলে EventBridge-এ পাঠাতে হয় বুঝতে।
• CloudTrail + EventBridge দিয়ে যেকোনো AWS API call-এ সাড়া দিতে, আর Region ও read-only-এর ফাঁদ এড়াতে।
• AWS service integration দিয়ে API Gateway থেকে সোজা Kinesis Data Streams-এ stream ঢোকাতে, আর throttling সামলাতে।

#Event processing: architect-এর চোখে

এই লেসনের প্রতিটা ইটের নিজের আলাদা লেসন আগেই হয়ে গেছে। এখানে প্রশ্নটা অন্য: কয়েকটা জিনিস একসাথে কীভাবে খাপ খায়, প্রতিটা ধাপ কী পারে আর কী পারে না, আর পুরো ব্যবস্থাটা কোথায় ভাঙে। নোটও শুরুতে একই সতর্কতা দেয়: এই অংশ থেকে অনেক scenario প্রশ্ন আসে, আর সেগুলো জেতা যায় shape-টা চিনে ফেলতে পারলে।

#মূল তিনজন: Lambda, SNS আর SQS

নোট এই তিনটাকে AWS-এ event-driven architecture-এর ভিত্তি বলে, আর লেসন ১৩.৭-এ এদের বিস্তারিত তুলনা আছে। ছোট করে বললে, সাথে পরীক্ষার প্রিয় একটা সংশোধন:

• SQS producer আর consumer-কে আলাদা করে। এটা pull-based: EC2 worker, container বা Lambda তৈরি হলে message টেনে নেয়।
• SNS হলো pub/sub, push-based: একবার publish করলেই সব subscriber একসাথে পায় (email, SMS, HTTPS, SQS, Lambda, Firehose)।
• Lambda কিছু ঘটলে code চালায়। SNS-এর বেলায় SNS নিজে function invoke করে। SQS-এর বেলায় কেউ push করে না: Lambda service একটা event source mapping চালায়, যার poller-রা তোমার হয়ে queue পড়ে।

পরের service-কে সরাসরি call করলেই তো হয়, তাই না? সরাসরি call মানে যে ডাকে আর যাকে ডাকে, দুজন বাঁধা পড়ে যায়: যে ডাকে সে অপেক্ষা করে, ওপাশ ব্যর্থ হলে সে-ও ব্যর্থ হয়, আর নতুন কোনো consumer এলেই তাকে বদলাতে হয়। নিচের প্রতিটা pattern এই বাঁধনগুলোর একটা করে খুলে দেয়।

#একটা Lambda function-এর পেছনে তিন রকম retry

এই অংশের প্রথম slide একই function-কে তিনটা আলাদা source-এর পেছনে আঁকে, আর প্রত্যেকটার পাশে লেখে সে কীভাবে retry করে। এটাই এই লেসনের প্রাণ, কারণ কে retry করবে, কতবার, কোন ক্রমে, আর failed message শেষে কোথায় যাবে, সবই ঠিক করে দেয় source।

#SQS + Lambda: poller-রা চেষ্টা করে, আবার চেষ্টা করে

1. Event source mapping queue-টাকে long poll করে একটা batch receive করে (default-এ সর্বোচ্চ ১০টা message; batching window দিলে আরও বেশি)।
2. Batch দিয়ে function-কে synchronous ভাবে invoke করে। Message-গুলো এখন in flight, queue-এর visibility timeout পর্যন্ত লুকানো।
3. Function সফল হলে Lambda message-গুলো delete করে।
4. ব্যর্থ হলে কিছুই delete হয় না। Visibility timeout শেষে message-গুলো আবার দেখা যায়, আবার receive হয়। প্রতিটা receive-এ message-এর receive count বাড়ে।
5. Receive count redrive policy-র maxReceiveCount পার হলে SQS message-টাকে queue-এর DLQ-তে সরিয়ে দেয়।

দুটো setting এই shape-কে ভালোভাবে চালায়:

• Partial batch response। Default-এ একটা exception মানেই পুরো batch failed, তাই খারাপটার সাথে ৯টা ভালো message-ও আবার process হয়। Event source mapping-এ ReportBatchItemFailures চালু করো, আর শুধু failed message-গুলোর ID ফেরত দাও: বাকিগুলো Lambda delete করে দেয়।
• Visibility timeout অন্তত function timeout-এর ৬ গুণ। নইলে প্রথম চেষ্টা চলতে চলতেই message আবার দেখা দেয়, আর একই message-এর দুটো কপি একসাথে চলে।

Standard queue একটা খারাপ message-এর পেছনে কখনো আটকায় না: বাকিরা চলতেই থাকে। আর DLQ বসে queue-এর ওপর, function-এর ওপর না, কারণ message-এর মালিক function কখনোই না; queue।

#SQS FIFO + Lambda: চেষ্টা, আবার চেষ্টা, আর আটকে থাকা

একই polling, সাথে ক্রম (১৩.৩):

• Message আসে message group ID ধরে ক্রম মেনে, আর এক group-এর একসাথে একটাই batch in flight থাকে। Lambda বাড়ে active group-এর সংখ্যা ধরে, message-এর সংখ্যা ধরে না।
• একটা message fail করলে সে ফিরে আসে নিজের group-এর মাথায়। ওই group-এর পরের সব message তার পেছনে অপেক্ষা করে: head-of-line blocking। Slide-এর “blocking” শব্দটা এটাই। অন্য group-এর কিছু হয় না।
• FIFO queue-তে partial batch response দিলে function-কে প্রথম failure-এই থামতে হবে, আর ওই message ও batch-এর পরের সবগুলোকে failed বলতে হবে, নইলে ক্রম ভেঙে যায়।
• maxReceiveCount পার হলে message যায় এমন একটা DLQ-তে, যেটাকে নিজেও FIFO হতে হয়। Group ছাড়া পায়, কিন্তু পরের message-গুলো এখন failed message-এর আগেই চলে। এগিয়ে চলার চেয়ে ক্রম বেশি জরুরি হলে (এমন ledger, যেখানে payment ৩ নির্ভর করে payment ২-এর ওপর), DLQ দিয়ো না, আটকে থাকা group-এর ওপর alarm দাও, আর কারণটা ঠিক করো।

#SNS + Lambda: asynchronous retry

SNS function-কে asynchronous ভাবে invoke করে (১৫.১):

1. Lambda event-টা নিজের internal queue-তে রেখে সঙ্গে সঙ্গে 202 উত্তর দেয়। SNS-এর কাছে delivery সফল।
2. Function exception দিলে Lambda নিজে default-এ দুবার retry করে (প্রায় ১ মিনিট পরে, তারপর আরও প্রায় ২ মিনিট পরে); ০, ১ বা ২ সেট করা যায়। Throttle হওয়া event retry হয় maximum event age পর্যন্ত (default ৬ ঘণ্টা)।
3. Retry শেষ হলে Lambda একটা record পাঠায় on-failure destination-এ (SQS, SNS, EventBridge, আরেকটা Lambda function, বা S3): event, error আর কারণ সহ। পুরোনো বিকল্প হলো function DLQ (একটা SQS queue বা SNS topic), যেটা শুধু event-টা পায়। কোনোটাই না থাকলে event ফেলে দেওয়া হয়।

আলাদা এক রকম failure-এর জন্য SNS-এর নিজের একটা জাল আছে: subscription DLQ (subscription-এর ওপর বসানো একটা SQS queue) সেই message ধরে, যা SNS পৌঁছাতেই পারেনি, যেমন কেউ function-এর resource policy মুছে দিলে (client-side error, retry হয় না)। যে message পৌঁছে গেছে কিন্তু code-এ fail করেছে, সেটা কখনো SNS DLQ-তে যায় না।

এই shape-এ কোনো ক্রম নেই (SNS FIFO topic শুধু SQS queue-তে পাঠায়, Lambda-তে না), আর asynchronous delivery অন্তত একবার, তাই function-কে idempotent হতে হবে।

#তিনটা shape পাশাপাশি

তিনটা shape নিজে চালিয়ে দেখো: source বাছো, message-টাকে poison pill বানাও, জাল চালু-বন্ধ করো, আর timeline মিলিয়ে দেখো।

#প্রতিটা shape-এ dead-letter queue

নোট ধারণাটা এভাবে দেয়: SQS, SNS বা Lambda কোনো message-এ বারবার fail করলে আর সর্বোচ্চ retry শেষ হয়ে গেলে, message-টাকে একটা আলাদা queue-তে সরিয়ে দেওয়া হয়, যাতে সেটা হারিয়ে না যায়, আর আটকে বা ঘুরতে না থাকে। পরে developer-রা সেটা পড়ে debug করে। কলকবজা ১৩.২-এ আছে; এখানে আছে একজন architect-কে যা ঠিক করতেই হবে।

• SQS-এ DLQ মানে source queue-এর ওপর একটা redrive policy: deadLetterTargetArn আর maxReceiveCount (১ থেকে ১,০০০; সাধারণত ৩ থেকে ১০)। DLQ একটা সাধারণ queue, যেটা আগে বানিয়ে নিতে হয়।
• Poison pill এমন message, যেটা কখনো সফল হবে না (ভুল type, মুছে যাওয়া customer, এমন bug যেটা শুধু এই input-এই ধরা পড়ে)। DLQ না থাকলে retention শেষ হওয়া পর্যন্ত এটা ঘোরে, তারপর উধাও। DLQ থাকলে maxReceiveCount বার চেষ্টার পর সরে যায়।
• একই type, একই জায়গা: FIFO queue-এর লাগবে FIFO DLQ, standard queue-এর standard DLQ, একই account আর Region-এ।
• Source-এর চেয়ে বেশি retention (সাধারণত ১৪ দিন): standard queue-তে message তার আসল enqueue সময় নিয়েই যায়, তাই সমান retention-এর DLQ সেটা প্রায় সঙ্গে সঙ্গে মুছে ফেলতে পারে।
• নজর রাখো: DLQ-এর ApproximateNumberOfMessagesVisible ০-এর বেশি হলে একটা CloudWatch alarm। যে DLQ-র দিকে কেউ তাকায় না, সেটা message হারানোর একটা ধীর উপায় মাত্র।
• সমস্যা মিটলে redrive: console-এর “Start DLQ redrive” বা StartMessageMoveTask message-গুলো source queue-তে (বা একই type-এর অন্য queue-তে) ফেরত পাঠায়, চাইলে একটা নির্দিষ্ট গতিতে।

#Fan-out: এক event অনেক queue-তে পৌঁছে দেওয়া

দ্বিতীয় slide একটা event-কে তিনটা SQS queue-তে পৌঁছানোর দুটো উপায় দেখায়।

Option 1: producer নিজে প্রতিটা queue-তে লেখে। Code তিনবার SendMessage call করে, একটার পর একটা queue-তে। ভালো দিনে চলে, কিন্তু:

• তিনটা queue জুড়ে কোনো transaction নেই। দ্বিতীয় লেখার পর process মারা গেলে দুজন consumer event পায়, তৃতীয়জন কখনো পায় না। পুরো কাজটা আবার চালালে প্রথম দুজন duplicate পায়।
• প্রতিটা লেখা producer-এর latency বাড়ায়, আর প্রতিটা নতুন consumer মানে producer বদলে আবার deploy।

Option 2: fan-out। Producer একটা SNS topic-এ একবার publish করে; queue-গুলো তাতে subscribe করা; SNS প্রত্যেক queue-তে একটা করে কপি push করে। Producer একটাই call করে, কাউকে চেনে না। প্রত্যেক consumer-এর নিজের queue, তাই নিজের গতি, নিজের retry আর DLQ; কোনো consumer বন্ধ থাকলে তার queue-তে শুধু backlog জমে: পরীক্ষার বাক্যের “without data loss” এটাই। SNS থেকে SQS-এ delivery SNS ২৩ দিন পর্যন্ত retry করে, তাই কপি পৌঁছানো কার্যত নিশ্চিত।

বাস্তবে fan-out চালু রাখতে যা লাগে (১৩.৪):

• Queue-এর access policy-কে topic-কে পাঠানোর অনুমতি দিতে হবে: principal sns.amazonaws.com, action sqs:SendMessage, শর্ত aws:SourceArn = topic। Subscribe করার সময় console এটা যোগ করে দেয়; CLI, CloudFormation আর Terraform দেয় না, তখন delivery access denied দিয়ে fail করে। SSE-KMS দিয়ে encrypt করা queue-এর জন্য এমন customer managed key লাগে, যেটা SNS-কে অনুমতি দেয়।
• Subscription-এর ওপর filter policy প্রত্যেক queue-কে শুধু তার দরকারি message পাঠায় (message attribute বা body দেখে); নইলে প্রত্যেক worker-কে পাওয়া বেশিরভাগ message ফেলে দিতে হয়।
• Raw message delivery queue-কে ঠিক publish করা body-টাই দেয়, SNS-এর JSON খাম ছাড়া।
• অন্য Region আর অন্য account-এর subscription-ও চলে (queue policy আর topic policy দুটোই লাগে)।
• শুরু থেকে শেষ FIFO: SNS FIFO topic আর SQS FIFO queue মিলে fan-out-এর মধ্যেও ক্রম আর duplicate বাদ দেওয়া ধরে রাখে।
• S3 event → SNS → অনেক queue হলো S3-এর “এক event, এক গন্তব্য” নিয়ম এড়ানোর পুরোনো উপায় (পরের অংশ)।

#S3 event notification

একটা bucket নিজের object-এর সাথে কী ঘটছে, তা জানিয়ে দিতে পারে (পুরো বিবরণ ১০.৭-এ)।

• Event type: s3:ObjectCreated:* (Put, Post, Copy, CompleteMultipartUpload), s3:ObjectRemoved:*, s3:ObjectRestore:*, s3:Replication:*, আর lifecycle, tagging, ACL ও Intelligent-Tiering-এর event। শুধু :Put না শুনে s3:ObjectCreated:* শোনো, নইলে বড় multipart upload বাদ পড়বে।
• Filter: শুধু key-এর prefix আর/অথবা suffix, যেমন prefix uploads/ আর suffix .jpg। কোনো wildcard নেই (slide-এর *.jpg লিখতে হয় suffix .jpg হিসেবে), size বা metadata দেখে filter নেই।
• গন্তব্য: একটা SNS topic, একটা SQS queue বা একটা Lambda function, bucket-এর একই Region-এ। শুধু standard queue আর topic: native গন্তব্য হিসেবে FIFO চলে না। চতুর্থ উপায় EventBridge, যেটা পুরো bucket-এর জন্য চালু করতে হয় (পরের অংশ)।
• Permission: S3 পাঠায় s3.amazonaws.com service হিসেবে, তাই গন্তব্যের resource policy-কে অনুমতি দিতে হয় (sns:Publish, sqs:SendMessage, lambda:InvokeFunction), ভালো হয় aws:SourceArn আর aws:SourceAccount দিয়ে শুধু ওই bucket-এ সীমিত রাখলে। Permission না থাকলে configuration save-ই হয় না।
• সময় আর delivery: সাধারণত কয়েক second-এ, কখনো এক মিনিট বা তারও বেশি; অন্তত একবার, ক্রম মেনে না (একটা key-এর জন্য sequencer মিলিয়ে দেখো)। Key আসে URL-encoded হয়ে।
• কয়টা rule: যত দরকার বানাও, শুধু কোনো দুটো যেন overlap না করে।

Overlap-এর নিয়ম, নিখুঁত করে বললে: দুটো rule একসাথে থাকতে পারে না, যদি তাদের event type overlap করে, prefix overlap করে, আর suffix overlap করে। তাই নোটের “একই prefix আর suffix-এর জন্য overlapping rule বানানো যায় না” কথাটা সত্যি event type ধরে: একই prefix-এ ObjectCreated একটা function-এ আর ObjectRemoved একটা queue-তে, কোনো সমস্যা নেই; uploads/-এ ObjectCreated একবার function-এ আর একবার queue-তে, বাতিল। Native notification-এ এক event-এর এক গন্তব্য। কয়েক জায়গায় পাঠাতে চাইলে একটা SNS topic-এ পাঠিয়ে সবাইকে subscribe করাও, অথবা EventBridge ব্যবহার করো।

নিজেকে নিজে ডাকা loop: uploads/ দিয়ে trigger হওয়া function যদি নিজের output আবার uploads/-এ লেখে, তাহলে সে নিজেকেই বারবার trigger করে। অন্য prefix-এ (thumbs/) বা অন্য bucket-এ লেখো। Lambda এখন এমন S3 loop ধরে প্রায় ১৬টা invocation-এর পর থামিয়ে দেয়, কিন্তু সেটা নিরাপত্তার জাল, design না।

#Amazon EventBridge হয়ে S3 event

Bucket-এর একটা setting, Send notifications to Amazon EventBridge, চালু করলে S3 তার সব event একই Region-এর default event bus-এ পাঠায়। S3-এর দিকে কিছুই বাছতে হয় না, S3-এর কোনো permission-ও লাগে না। এরপর filter আর routing-এর কাজ EventBridge-এর rule-এর (১৯.৫)।

উৎসগুলোর দুটো সংশোধন। Slide বলে “over 18 AWS services as destinations”, আর নোট বলে “১৮+”: আজ EventBridge-এর target তালিকায় ২০-এর বেশি ধরনের target আছে, আর সংখ্যাটা বাড়ছেই, তাই মনে রাখো “অনেক, Step Functions আর Kinesis সহ”। আর দুটোই বলে EventBridge metadata দেখে filter করতে পারে: আসলে সে filter করে event-এর ভেতরের field দেখে (key, size, ETag, version, requester, source IP, reason…), object-এর নিজের দেওয়া x-amz-meta- metadata দেখে না, কারণ সেটা event-এ থাকেই না।

#EventBridge আর CloudTrail দিয়ে API call ধরে ফেলা

অনেক পরিবর্তনের নিজস্ব কোনো service event হয় না: DynamoDB table মুছে গেছে বা tag ছাড়া একটা instance চালু হয়েছে, কেউ ঘোষণা দেয় না। কিন্তু এমন প্রতিটা পরিবর্তনই একটা API call, আর CloudTrail প্রতিটা API call লিখে রাখে (১৯.৬)। CloudTrail সেই record-গুলো default event bus-এ পাঠায় AWS API Call via CloudTrail detail-type-এর event হিসেবে, তাই একটা rule যেকোনো call-এ সাড়া দিতে পারে।

Slide-এর উদাহরণ: একজন user DynamoDB-তে DeleteTable call করলেন → CloudTrail সেটা log করল → একটা EventBridge rule মিলে গেল → SNS team-কে সতর্ক করল। নোটের উদাহরণ: কেউ RunInstances call করলেন → rule একটা Lambda function invoke করে, যেটা tag দেখে, আর দরকারি tag না থাকলে instance বন্ধ করে দেয় (বা tag বসিয়ে দেয়, বা একটা ticket খোলে)।

{
  "source": ["aws.ec2"],
  "detail-type": ["AWS API Call via CloudTrail"],
  "detail": {
    "eventSource": ["ec2.amazonaws.com"],
    "eventName": ["RunInstances"]
  }
}

এটা কাজ করতে যা সত্যি হতে হবে:

• Account জুড়ে logging চালু একটা trail থাকতে হবে (organization trail হলেও চলে)। Trail-এ management event default-এ log হয়; data event (S3 object call, Lambda invoke) শুধু যোগ করলে।
• Rule regional। Call যে Region-এ লেখা হয়েছে, event আসে সেই Region-এর bus-এ। ap-southeast-1-এর rule eu-west-1-এ করা call কখনো দেখে না।
• IAM-এর মতো global service নিজের call লেখে us-east-1-এ, তাই IAM-এর rule-কে সেখানেই থাকতে হবে। Root-এর console sign-in লেখা হয় us-east-1, us-east-2 বা us-west-2-এ।
• Read-only call (Get, List, Describe) সাধারণ enabled অবস্থার rule-এ পৌঁছায় না। Rule-এর state করো ENABLED_WITH_ALL_CLOUDTRAIL_MANAGEMENT_EVENTS (CLI বা CloudFormation দিয়ে), আর অনেক বেশি event-এর জন্য তৈরি থাকো।
• প্রায় real time: কয়েক second থেকে কয়েক মিনিট। এটা ধরে আর সাড়া দেয়; call আটকাতে চাইলে IAM, SCP বা resource-এর নিজের সুরক্ষা (DynamoDB deletion protection) লাগবে।

#API Gateway থেকে সোজা Kinesis Data Streams-এ

API Gateway-এর পেছনে আমরা সাধারণত একটা Lambda function রাখি। কিন্তু API Gateway বেশিরভাগ AWS API সরাসরি call করতে পারে, মাঝখানে কোনো function ছাড়াই: একে বলে AWS service integration (১৫.৬)। Slide-এর উদাহরণ একটা streaming pipeline:

client → API Gateway → Kinesis Data Streams → Data Firehose → Amazon S3 (.json file)

কীভাবে বানানো হয়:

• একটা REST API method, যার integration type AWS, service Kinesis, action PutRecord (batch-এর জন্য PutRecords)।
• একটা mapping template (VTL), যেটা request-কে action-এর input-এ বদলে দেয়: stream-এর নাম, partition key (যেমন rider ID, যাতে এক rider-এর point এক shard-এ ক্রম মেনে থাকে), আর body-কে Base64 করে Data।
• একটা IAM role, যেটা API Gateway assume করে, আর যেটা ওই stream-এ ঠিক kinesis:PutRecord-এরই অনুমতি দেয়।
• Integration response, যেটা Kinesis-এর error-কে ঠিক HTTP status code-এ বদলায়। Non-proxy integration-এ কোনো error-এর জন্য মেলানো integration response না থাকলে default response দিয়ে সেটা 200 হয়ে ফিরতে পারে, error থাকে body-তে।
• Stream-এর পেছনে: Firehose record জমিয়ে S3-এ file বানায় (পরে Athena দিয়ে query), আর real-time consumer-রা stream সরাসরি পড়ে।

Lambda function সরানো কেন: সে শুধু request কপি করত, তবু যোগ করত latency (চলার সময়, আর মাঝে মাঝে cold start), প্রতি request-এ খরচ, আর account-এর ভাগাভাগি করা pool থেকে concurrency। এটা না থাকলে লেখার, patch করার বা scale করার মতো কোনো code নেই। Function রাখো শুধু তখন, যখন সে সত্যিকারের কাজ করে (এমন validation যা model দিয়ে বলা যায় না, তথ্য যোগ করা, routing)।

একই কৌশল অনেক service-এ চলে: SQS SendMessage, SNS Publish, Step Functions StartExecution, DynamoDB PutItem, S3 object, EventBridge PutEvents। HTTP API এর কয়েকটাকে বানানো-তৈরি first-class integration হিসেবে দেয় (SQS, Kinesis Data Streams, Step Functions, EventBridge আর আরও কিছু)।

এই পথে যে সীমাগুলো গুরুত্বপূর্ণ:

Stream ছোট হলে Kinesis বাড়তি লেখা throttle করে। সমাধান: বেশি shard বা on-demand capacity mode, এমন partition key যা load ছড়িয়ে দেয় (শহরকে key বানালে একটা hot shard হয়), client-এর backoff দিয়ে retry আর PutRecords দিয়ে batch, আর API Gateway throttling, যাতে হঠাৎ ভিড় দরজাতেই ফিরিয়ে দেওয়া যায়।

#প্রশ্ন পড়ে ঠিক pattern চেনা

#কোথায় গোলমাল হয়

• একটা poison message দিনের পর দিন ঘোরে: redrive policy নেই। যুক্তিসঙ্গত maxReceiveCount সহ DLQ আর তার ওপর একটা alarm দাও।
• Batch-এ একটা failure-এর পর ভালো message-ও দুবার process হয়: partial batch response নেই। ReportBatchItemFailures চালু করো, তবু consumer-কে idempotent রাখো।
• একই message দুটো invocation একসাথে process করে: visibility timeout function timeout-এর চেয়ে ছোট। অন্তত function timeout-এর ছয় গুণ দাও।
• একটা FIFO group আটকে আছে: তার মাথার message বারবার fail করছে। কারণটা ঠিক করো, অথবা ক্রম ভাঙা মেনে নিয়ে FIFO DLQ দাও।
• DLQ বসানো হয়েছে function-এ, তবু SQS message ঘুরছে: SQS source-এর জন্য DLQ বসে queue-এর ওপর।
• SNS দিয়ে চালু হওয়া failed event উধাও, অথচ SNS DLQ খালি: delivery সফল হয়েছিল; failure ছিল code-এ। Function-এ on-failure destination দাও।
• Fan-out-এর একটা queue কিছুই পায় না: তার access policy topic-কে অনুমতি দেয় না, বা তার KMS key SNS-কে দেয় না।
• S3 notification configuration save হয় না: একই event type-এর overlap করা rule, অথবা গন্তব্যের policy S3-কে অনুমতি দেয় না, অথবা গন্তব্য একটা FIFO queue।
• Deploy-এর পর bill আকাশছোঁয়া: function নিজের trigger হওয়া prefix-এই লিখছে।
• API call-এর rule কখনো চলে না: trail নেই, ভুল Region, global service-এর event us-east-1-এ, অথবা call-টা read-only।
• ভিড়ের সময় ping হারিয়ে যায়: Kinesis throttle করছে, আর integration error-টাকে 200 বানিয়ে দিচ্ছে। Integration response, বেশি shard বা on-demand mode, আর client-এর retry দাও।

#System design-এ event processing

একটা wallet-এর জন্য সব টুকরো জুড়ে দেখো:

• Mobile API বিপুল telemetry (screen event, location) পাঠায় API Gateway → Kinesis → Firehose → S3 দিয়ে, কোনো code ছাড়া।
• একটা সম্পন্ন payment একবার একটা SNS topic-এ publish হয়; ledger, SMS আর fraud, প্রত্যেকের নিজের DLQ সহ একটা SQS queue; ledger-এর queue FIFO, account ধরে group করা, কারণ সেখানে ক্রম জরুরি।
• Customer-দের upload করা KYC কাগজ যায় S3-এ; EventBridge rule বড় file পাঠায় একটা Step Functions review workflow-এ, আর প্রতিটা upload একটা audit FIFO queue-তে।
• CloudTrail + EventBridge rule DeleteTable, security group পরিবর্তন আর root sign-in-এ security-কে সতর্ক করে, ব্যবহার করা প্রতিটা Region-এ আর us-east-1-এ।
• প্রতিটা asynchronous Lambda function-এর একটা on-failure destination আছে; প্রতিটা queue-এর DLQ আর সবচেয়ে পুরোনো message-এর বয়সের ওপর alarm আছে।

লেসন ১৫.৮ এই একই টুকরো দিয়ে পুরো serverless application বানায়, আর লেসন ২৪.২ এমন design-এর প্রতিটা স্তরে caching যোগ করে।

#নিজে করে দেখো

এই command-গুলো ap-southeast-1-এ AWS CLI v2 দিয়ে, বানানো নাম ব্যবহার করে; queue, topic আর stream যতক্ষণ থাকে ততক্ষণ টাকা লাগে, তাই শেষে মুছে ফেলো।

# 1. A DLQ and a source queue with a redrive policy (maxReceiveCount 3), DLQ kept 14 days
aws sqs create-queue --queue-name payments-dlq --attributes MessageRetentionPeriod=1209600
aws sqs create-queue --queue-name payments --attributes VisibilityTimeout=180
aws sqs set-queue-attributes \
  --queue-url https://sqs.ap-southeast-1.amazonaws.com/111122223333/payments \
  --attributes '{"RedrivePolicy":"{\"deadLetterTargetArn\":\"arn:aws:sqs:ap-southeast-1:111122223333:payments-dlq\",\"maxReceiveCount\":\"3\"}"}'

# 2. Let Lambda poll it, with partial batch responses
aws lambda create-event-source-mapping --function-name ledger-writer \
  --event-source-arn arn:aws:sqs:ap-southeast-1:111122223333:payments \
  --batch-size 10 --function-response-types ReportBatchItemFailures
# { "UUID": "a1b2c3d4-5678-90ab-cdef-11111EXAMPLE", "BatchSize": 10, "State": "Creating",
#   "FunctionResponseTypes": [ "ReportBatchItemFailures" ], ... }

# 3. After the fix: move the DLQ's messages back to the source queue
aws sqs start-message-move-task \
  --source-arn arn:aws:sqs:ap-southeast-1:111122223333:payments-dlq \
  --destination-arn arn:aws:sqs:ap-southeast-1:111122223333:payments \
  --max-number-of-messages-per-second 50
# { "TaskHandle": "eyJ0YXNrSWQiOiJkYzE2...EXAMPLE" }

# 4. Asynchronous Lambda: 2 retries, 1 hour maximum age, failures to a queue
aws lambda put-function-event-invoke-config --function-name ledger-writer \
  --maximum-retry-attempts 2 --maximum-event-age-in-seconds 3600 \
  --destination-config '{"OnFailure":{"Destination":"arn:aws:sqs:ap-southeast-1:111122223333:ledger-failures"}}'

# 5. Fan-out: subscribe a queue to the topic with a filter policy and raw delivery
#    (the queue policy must already allow sns.amazonaws.com with aws:SourceArn = the topic)
aws sns subscribe \
  --topic-arn arn:aws:sns:ap-southeast-1:111122223333:payment-done \
  --protocol sqs --notification-endpoint arn:aws:sqs:ap-southeast-1:111122223333:sms \
  --attributes '{"RawMessageDelivery":"true","FilterPolicy":"{\"amount\":[{\"numeric\":[\">\",5000]}]}"}'
# { "SubscriptionArn": "arn:aws:sns:ap-southeast-1:111122223333:payment-done:6f1c2b3a-EXAMPLE" }

# 6. S3: one native rule to Lambda, and EventBridge switched on for the bucket
#    (this call replaces the whole notification configuration of the bucket)
aws s3api put-bucket-notification-configuration --bucket news-media-example \
  --notification-configuration '{
    "LambdaFunctionConfigurations": [{
      "LambdaFunctionArn": "arn:aws:lambda:ap-southeast-1:111122223333:function:resize",
      "Events": ["s3:ObjectCreated:*"],
      "Filter": {"Key": {"FilterRules": [{"Name": "prefix", "Value": "uploads/"},
                                         {"Name": "suffix", "Value": ".jpg"}]}}}],
    "EventBridgeConfiguration": {} }'

# 7. A rule that reacts to DeleteTable calls recorded by CloudTrail
aws events put-rule --name alert-delete-table \
  --event-pattern '{"source":["aws.dynamodb"],"detail-type":["AWS API Call via CloudTrail"],"detail":{"eventSource":["dynamodb.amazonaws.com"],"eventName":["DeleteTable"]}}'
# { "RuleArn": "arn:aws:events:ap-southeast-1:111122223333:rule/alert-delete-table" }
aws events put-targets --rule alert-delete-table \
  --targets Id=sns,Arn=arn:aws:sns:ap-southeast-1:111122223333:security-alerts

# 8. API Gateway → Kinesis: an AWS service integration calling PutRecord
#    kinesis-template.json holds the mapping template:
#    {"application/json": "{\"StreamName\":\"rider-pings\",\"PartitionKey\":\"$input.path('$.riderId')\",\"Data\":\"$util.base64Encode($input.body)\"}"}
aws apigateway put-integration --rest-api-id a1b2c3d4e5 --resource-id r2s3t4 \
  --http-method POST --type AWS --integration-http-method POST \
  --uri arn:aws:apigateway:ap-southeast-1:kinesis:action/PutRecord \
  --credentials arn:aws:iam::111122223333:role/apigw-kinesis-putrecord \
  --request-templates file://kinesis-template.json
# { "type": "AWS", "httpMethod": "POST",
#   "uri": "arn:aws:apigateway:ap-southeast-1:kinesis:action/PutRecord", ... }
# Map Kinesis 4xx errors (such as throttling) to a 500 the client can see
# (a 500 method response must exist on the method first)
aws apigateway put-integration-response --rest-api-id a1b2c3d4e5 --resource-id r2s3t4 \
  --http-method POST --status-code 500 --selection-pattern '4\d{2}'

# 9. Clean up
aws events remove-targets --rule alert-delete-table --ids sns
aws events delete-rule --name alert-delete-table
aws lambda delete-event-source-mapping --uuid a1b2c3d4-5678-90ab-cdef-11111EXAMPLE
aws sqs delete-queue --queue-url https://sqs.ap-southeast-1.amazonaws.com/111122223333/payments
aws sqs delete-queue --queue-url https://sqs.ap-southeast-1.amazonaws.com/111122223333/payments-dlq

Console-এও চেষ্টা করো: SQS-এ এমন একটা message পাঠাও যেটা তোমার function ফিরিয়ে দেয়, আর Poll for messages-এ Receive count বাড়তে বাড়তে DLQ-তে পৌঁছানো দেখো; তারপর Start DLQ redrive চাপো। EventBridge → Rules-এ EC2-এর জন্য AWS API Call via CloudTrail template থেকে একটা rule বানাও, আর sandbox দিয়ে pattern পরীক্ষা করো।

#সারসংক্ষেপ

মূল কথা

• SQS = pull, buffer, decouple; SNS = অনেকের কাছে push; Lambda = event হলে code। SQS Lambda-কে push করে না: একটা event source mapping poll করে।
• SQS + Lambda: visibility timeout ধরে retry, maxReceiveCount পর্যন্ত, তারপর queue-এর DLQ; partial batch response দাও, আর visibility timeout রাখো function timeout-এর অন্তত ৬ গুণ। অন্যরা আটকায় না।
• SQS FIFO + Lambda: message group ধরে ক্রম; failed message নিজের group আটকে রাখে, সফল হওয়া বা FIFO DLQ-তে যাওয়া পর্যন্ত (তাতে group-এর ক্রম ভাঙে)।
• SNS + Lambda: asynchronous; Lambda default দুবার retry করে, তারপর on-failure destination (পছন্দের, error-এর বিবরণসহ) বা function DLQ, নইলে ফেলে দেয়। SNS subscription DLQ ধরে শুধু delivery-র failure।
• DLQ = source queue-এর ওপর redrive policy (maxReceiveCount); একই type (FIFO→FIFO), একই account আর Region, বেশি retention, একটা alarm, সমস্যা মিটলে redrive। Keyword: poison pill, infinite loop, troubleshoot failed messages।
• Fan-out: SNS-এ একবার publish, অনেক SQS queue subscribe করা; queue policy-কে topic-কে অনুমতি দিতে হবে; filter policy; code থেকে প্রতিটা queue-তে লেখার চেয়ে ভালো (আধা-ব্যর্থতা নেই, producer বদলাতে হয় না)।
• S3 event: ObjectCreated / Removed / Restore / Replication…, শুধু prefix আর suffix, গন্তব্য SNS, SQS, Lambda (শুধু standard), overlap-এর নিয়ম event type ধরে, কয়েক second থেকে এক মিনিট, গন্তব্যের policy-কে S3-কে অনুমতি দিতে হবে, নিজেকে ডাকা loop এড়াও। S3 → EventBridge: সব event, সমৃদ্ধ filter (size, key…), ২০-এর বেশি ধরনের target, এক event-এ অনেক rule, archive আর replay।
• API call ধরা: CloudTrail → EventBridge rule (AWS API Call via CloudTrail) → SNS বা Lambda; trail লাগে; rule regional; IAM-এর event us-east-1-এ; read-only call-এর জন্য rule-কে রাজি হতে হয়।
• API Gateway → Kinesis: IAM role আর mapping template সহ AWS service integration; Lambda নেই মানে কম latency, খরচ আর concurrency; shard-এর throughput আর error mapping-এ চোখ রাখো।

নিজেকে যাচাই করো: Event processing pattern৩০টি প্রশ্ন