কাইনেসিস স্ট্রীমে রেকর্ড যোগ করার সময় AWS Lambda টাইমআউট সমস্যা সমাধান করা

Kinesis ডেটা স্ট্রীমের জন্য AWS Lambda টাইমআউটের সমস্যা সমাধান করা

কল্পনা করুন যে আপনি AWS-এ একটি রিয়েল-টাইম ডেটা পাইপলাইন তৈরি করছেন, এমন একটি সেটআপ সহ যা SQS থেকে একটি Lambda ফাংশনে বার্তা পাঠায় এবং শেষ পর্যন্ত একটি Kinesis ডেটা স্ট্রীমে। 📨 এই প্রবাহ তাত্ত্বিকভাবে নির্বিঘ্নে কাজ করে, কিন্তু কখনও কখনও বাস্তবতার অন্য পরিকল্পনা থাকে। আপনি যখন আরাম করতে চলেছেন, তখন আপনার Lambda ফাংশন লগগুলিতে একটি ETIMEDOUT ত্রুটি ক্রপ আপ হয়।

এই ত্রুটিটি দেখা হতাশাজনক হতে পারে, বিশেষ করে যখন আপনি অনুমতিগুলি যাচাই করেছেন এবং ফাংশনটি একাধিকবার পরীক্ষা করেছেন৷ প্রকৃতপক্ষে, কাইনেসিস স্ট্রীমে এই বিরতিহীন ETIMEDOUT সমস্যাটি সাধারণত অপ্রত্যাশিতভাবে ঘটে, আপনার অগ্রগতিকে বাধা দেয়। Lambda একটি পুনঃস্থাপনের পরে পুরোপুরি কাজ করতে পারে কিন্তু তারপর আবার ব্যর্থ হয়, আপাতদৃষ্টিতে কারণ ছাড়াই।

এই ধরনের পরিস্থিতিতে, অনেক ডেভেলপার "Runtime.UnhandledPromiseRejection" এবং "ERR_HTTP2_STREAM_CANCEL" এর মতো রহস্যময় বার্তা দ্বারা স্তব্ধ হয়ে পড়েছে। রাস্তা অবরোধ

এখানে, আমরা এই টাইমআউটের কারণগুলি, সেগুলি পরিচালনা করার ব্যবহারিক উপায়গুলি এবং আপনার AWS কনফিগারেশনের সামঞ্জস্যগুলি নিয়ে যাব যা আপনার স্ট্রীমকে স্থিতিশীল করার চাবিকাঠি হতে পারে৷ 🛠️ শেষ পর্যন্ত, আপনি কীভাবে ETIMEDOUT ত্রুটির সমস্যা সমাধান এবং সমাধান করবেন এবং আপনার Lambda এবং Kinesis প্রবাহকে সুচারুভাবে চালিয়ে যেতে পারবেন তা জানতে পারবেন।

কাইনেসিস স্ট্রীমের সাথে AWS Lambda নির্ভরযোগ্যতা বৃদ্ধি করা

প্রদত্ত জাভাস্ক্রিপ্ট স্ক্রিপ্টগুলি একটি দক্ষ AWS Lambda ফাংশন তৈরি করার জন্য ডিজাইন করা হয়েছে যা একটি SQS সারি থেকে বার্তাগুলি পুনরুদ্ধার করে এবং তারপর সেগুলিকে একটি Amazon Kinesis ডেটা স্ট্রীমে প্রকাশ করে৷ এই সমাধানের মূলে রয়েছে ল্যাম্বডা ফাংশনের মাধ্যমে বার্তাগুলিকে অ্যাসিঙ্ক্রোনাসভাবে পরিচালনা করার ক্ষমতা এবং সংযোগের সমস্যাগুলিকে মোকাবেলা করার সময় যা প্রায়শই ঘটে থাকে ETIMEDOUT ত্রুটি স্ক্রিপ্টের একটি মূল অংশ হল এর আরম্ভ করা কাইনেসিস ক্লায়েন্ট, যা অঞ্চল, পুনরায় চেষ্টা গণনা এবং সংযোগের সময়সীমার মতো প্রয়োজনীয় বৈশিষ্ট্যগুলি কনফিগার করে। এই কনফিগারেশনগুলি একটি ক্লাউড সেটআপে গুরুত্বপূর্ণ, কারণ তারা অ্যাপ্লিকেশনটির প্রতিক্রিয়াশীলতা নিয়ন্ত্রণ করে এবং সময় শেষ হওয়ার আগে এটি কতক্ষণ সংযোগ করার চেষ্টা করবে। উচ্চতর সেট করে কানেক্ট টাইমআউট অথবা পুনঃপ্রয়াসের প্রয়াস সামঞ্জস্য করে, আমরা ফাংশনটিকে আরও কার্যকরভাবে নেটওয়ার্ক বিলম্ব পরিচালনা করতে সাহায্য করতে পারি।

Lambda হ্যান্ডলারের মধ্যে, স্ক্রিপ্ট লিভারেজ Promise.all Settled(), একাধিক অ্যাসিঙ্ক্রোনাস অনুরোধ প্রক্রিয়া করার সময় একটি অমূল্য টুল। যখন একাধিক রেকর্ড একবারে প্রসেস করা হয়, তখন প্রত্যেকটি সম্পূর্ণ হয়েছে তা নিশ্চিত করা অত্যাবশ্যক, তা সফলভাবে হোক বা একটি ত্রুটি সহ। Promise.all Settled() একটি অনুরোধ ব্যর্থ হলে ফাংশন প্রক্রিয়াকরণ বন্ধ করে না তা নিশ্চিত করে; পরিবর্তে, এটি প্রতিটি ফলাফল পৃথকভাবে লগ করে। এই পদ্ধতিটি এমন পরিস্থিতিতে বিশেষভাবে কার্যকর যেখানে নেটওয়ার্ক সংযোগ অপ্রত্যাশিত হতে পারে। উদাহরণস্বরূপ, যদি একটি নেটওয়ার্ক সমস্যার কারণে একটি রেকর্ড ব্যর্থ হয় কিন্তু অন্যরা সফল হয়, ফাংশনটি ব্যর্থ রেকর্ডগুলিকে আলাদাভাবে লগ করতে পারে, যা বিকাশকারীদের বার্তাগুলির সম্পূর্ণ ব্যাচকে ব্যর্থ করার পরিবর্তে সমস্যার উদাহরণগুলিকে আলাদা করতে দেয়৷ 🛠️

দ প্রক্রিয়া ইভেন্ট স্ক্রিপ্টের মধ্যে ফাংশনটি মডুলার এবং প্রধান ডেটা রূপান্তর এবং প্রেরণ প্রক্রিয়া পরিচালনা করে। এই ফাংশনটি SQS বার্তা গ্রহণ করে, এটিকে পার্স করে এবং এটিকে বাইট বিন্যাসে এনকোড করে যা কাইনেসিস প্রয়োজন। এখানে, দ TextEncoder().encode() পদ্ধতিটি গুরুত্বপূর্ণ কারণ কাইনেসিস শুধুমাত্র বাইনারি ডেটা গ্রহণ করে; JSON একটি সামঞ্জস্যপূর্ণ বিন্যাসে রূপান্তর করা আবশ্যক। ফাংশনের এই অংশটি নিশ্চিত করে যে ল্যাম্বডা সঠিকভাবে ডেটা পাঠায়, অমিল ডেটা ফরম্যাট থেকে উদ্ভূত ত্রুটির সম্ভাবনা হ্রাস করে। ফাংশনটি একটি কাস্টম পার্টিশন কী জেনারেটর ফাংশনও ব্যবহার করে, যা কাইনেসিস স্ট্রিমের শার্ড জুড়ে রেকর্ড বিতরণ করে। গতিশীল পার্টিশন কী ব্যবহার করে (যেমন এলোমেলো কী), স্ক্রিপ্ট একই শার্ড বারবার আঘাত করার সম্ভাবনা কমিয়ে দেয়, যা "হট শার্ড" প্রতিরোধ করতে পারে যা বাধা সৃষ্টি করে।

অবশেষে, বিভিন্ন পরিস্থিতিতে এই সেটআপটি সঠিকভাবে কাজ করে তা নিশ্চিত করতে, স্ক্রিপ্টগুলি অন্তর্ভুক্ত করে ইউনিট পরীক্ষা জেস্ট ব্যবহার করে। ইউনিট পরীক্ষাগুলি লাইভ AWS সংস্থানগুলির প্রয়োজন ছাড়াই কাইনেসিস ক্লায়েন্টের আচরণকে অনুকরণ করা সম্ভব করে, একটি নিয়ন্ত্রিত পরিবেশে সময়সীমা বা ডেটা রূপান্তর সমস্যাগুলি পরিচালনা করার ল্যাম্বডার ক্ষমতা পরীক্ষা করার একটি নির্ভরযোগ্য উপায় অফার করে। উদাহরণস্বরূপ, যদি Kinesis ক্লায়েন্ট সংযোগ করতে অক্ষম হয়, Jest mocks একটি টাইমআউট ত্রুটি অনুকরণ করতে পারে, যাচাই করে যে ত্রুটিটি পরিচালনা করা হচ্ছে প্রক্রিয়া ইভেন্ট উদ্দেশ্য হিসাবে কাজ করে। এই কৌশলটি মজবুত বৈধকরণের অনুমতি দেয়, নিশ্চিত করে যে ল্যাম্বডা একাধিক নেটওয়ার্ক অবস্থার মধ্যে নির্ভরযোগ্য। 🧪 এই উপাদানগুলিকে একত্রিত করে, Lambda ফাংশন SQS থেকে Kinesis পর্যন্ত দক্ষতার সাথে টাইমআউট এবং অন্যান্য সাধারণ স্ট্রিমিং ত্রুটিগুলি হ্রাস করার সময় ডেটা পরিচালনা করতে পারে।

import { KinesisClient, PutRecordCommand } from "@aws-sdk/client-kinesis";
const KINESIS_CLIENT = new KinesisClient({
  region: 'us-west-2',
  maxRetries: 3,
  requestHandler: {
    connectionTimeout: 5000, // Lower timeout for quicker retries
  },
});
export const handler = async (event) => {
  const promises = event.Records.map(record => processEvent(record.body));
  const results = await Promise.allSettled(promises);
  results.forEach(result => {
    if (result.status !== "fulfilled") {
      console.error("Failed:", result.reason);
    }
  });
};
async function processEvent(body) {
  const data = JSON.parse(body);
  const partitionKey = generatePartitionKey(data);
  const encodedData = new TextEncoder().encode(JSON.stringify(data));
  try {
    await KINESIS_CLIENT.send(new PutRecordCommand({
      Data: encodedData,
      StreamName: 'InputEventStream',
      PartitionKey: partitionKey
    }));
  } catch (error) {
    console.error("Error putting record:", error);
    throw error;
  }
}
function generatePartitionKey(data) {
  return data.userId ? data.userId.toString() : Date.now().toString();
}

import { KinesisClient, PutRecordCommand } from "@aws-sdk/client-kinesis";
const KINESIS_CLIENT = new KinesisClient({
  region: 'us-west-2',
  maxRetries: 5,
  httpOptions: {
    connectTimeout: 15000, // Extended timeout
    timeout: 20000 // Total request timeout
  }
});
export const handler = async (event) => {
  const results = await Promise.allSettled(event.Records.map(async (record) => {
    await processEvent(record.body);
  }));
  results.forEach((result) => {
    if (result.status !== "fulfilled") {
      console.log("Unsuccessful attempt:", result.reason);
    }
  });
};
async function processEvent(body) {
  const parsedData = JSON.parse(body);
  const partitionKey = `pk-${Math.random()}`;
  try {
    await KINESIS_CLIENT.send(new PutRecordCommand({
      StreamName: "InputEventStream",
      Data: new TextEncoder().encode(JSON.stringify(parsedData)),
      PartitionKey: partitionKey
    }));
  } catch (err) {
    if (err.name === "TimeoutError") {
      console.warn("Retry on timeout:", err);
    }
    throw err;
  }
}

import { handler, processEvent } from './your-lambda-file.js';
import { KinesisClient } from "@aws-sdk/client-kinesis";
jest.mock("@aws-sdk/client-kinesis");
describe('Lambda Handler and Kinesis Integration', () => {
  it('should call processEvent for each record in the event', async () => {
    const mockEvent = {
      Records: [{ body: '{"userId": 1, "data": "test"}' }]
    };
    await handler(mockEvent);
    expect(KinesisClient.prototype.send).toHaveBeenCalledTimes(1);
  });
  it('should handle timeout errors gracefully', async () => {
    KinesisClient.prototype.send.mockRejectedValueOnce(new Error('TimeoutError'));
    await expect(processEvent('{"userId": 2}')).rejects.toThrow('TimeoutError');
  });
});

AWS Lambda-Kinesis ইন্টিগ্রেশনে টাইমআউট ত্রুটি বোঝা

টাইমআউট এরর মত ETIMEDOUT AWS Lambda ফাংশন প্রায়ই হতাশাজনক হতে পারে, বিশেষ করে অ্যামাজন কাইনেসিসের সাথে ডেটা স্ট্রিমিং জড়িত একীকরণে। বেশিরভাগ ক্ষেত্রে, ল্যাম্বডা ফাংশন নেটওয়ার্ক সংযোগের সময়সীমা অতিক্রম করার কারণে এই ত্রুটিগুলি ঘটে, সাধারণত একটি সময় KinesisClient অনুরোধ Lambda-তে ডিফল্ট সেটিংস সবসময় এই ধরনের নেটওয়ার্ক অনুরোধগুলিকে মিটমাট নাও করতে পারে, বিশেষ করে যখন উচ্চ-থ্রুপুট স্ট্রিম বা প্রচুর পরিমাণে ডেটা নিয়ে কাজ করা হয়। উদাহরণস্বরূপ, সামঞ্জস্য করা connectTimeout বা maxRetries কনফিগারেশনগুলি এই সমস্যাটি প্রশমিত করতে সাহায্য করতে পারে, ল্যাম্বডাকে কাইনেসিসের সাথে একটি সফল সংযোগের চেষ্টা করার জন্য আরও সময় দেয়। পরিবর্তনশীল নেটওয়ার্ক লেটেন্সি বা উচ্চ চাহিদার অধীনে পরিস্থিতিতে এই ধরনের অপ্টিমাইজেশন প্রায়ই প্রয়োজনীয়। 🛠️

টাইমআউট ত্রুটিগুলি হ্রাস করার আরেকটি মূল দিক হল ডেটা এনকোডিং এবং কার্যকরভাবে পার্টিশন পরিচালনা করা। AWS Kinesis-এর জন্য বাইনারি ফরম্যাটে ডেটা প্রয়োজন, যার মাধ্যমে অর্জন করা যেতে পারে TextEncoder().encode(). এই রূপান্তরটি কাইনেসিসে ডেটা স্থানান্তরের সামঞ্জস্যতা এবং স্ট্রিমলাইন নিশ্চিত করে। উপরন্তু, চিন্তাশীল পার্টিশন কী ব্যবস্থাপনা অত্যন্ত গুরুত্বপূর্ণ। একটি সামঞ্জস্যপূর্ণ বা গতিশীলভাবে জেনারেট করা পার্টিশন কী ব্যবহার করে কাইনেসিস শার্ড জুড়ে সমানভাবে ডেটা বিতরণ করতে সাহায্য করে, "হট শার্ড" এড়িয়ে যায়, যা শার্ডগুলি অসম পরিমাণে রেকর্ড গ্রহণ করে। উচ্চ-ফ্রিকোয়েন্সি স্ট্রিমিং পরিস্থিতিতে, ডায়নামিক কীগুলি বাধা প্রতিরোধ করতে পারে এবং সংযোগের সমস্যাগুলির সম্ভাবনা কমাতে পারে, বিশেষত বড় ডেটাসেটগুলি পরিচালনা করার সময় দরকারী।

এই Lambda-Kinesis মিথস্ক্রিয়াগুলির নির্ভরযোগ্যতা সমস্যা সমাধান এবং উন্নত করতে, ইউনিট পরীক্ষা যোগ করা অপরিহার্য। ইউনিট পরীক্ষাগুলি আপনাকে সম্ভাব্য নেটওয়ার্ক সমস্যাগুলি অনুকরণ করতে, ডেটা এনকোডিং যাচাই করতে এবং ফাংশনটি পুনরায় চেষ্টাগুলি সঠিকভাবে পরিচালনা করতে পারে তা নিশ্চিত করতে দেয়। যেমন উপহাস করে KinesisClient ইউনিট পরীক্ষায়, আপনি কাইনেসিস থেকে বিভিন্ন প্রতিক্রিয়া অনুকরণ করতে পারেন, যেমন সময়সীমা ত্রুটি বা সাফল্যের ক্ষেত্রে, যা ল্যাম্বডা কোডের মধ্যে সূক্ষ্ম-টিউনিং ত্রুটি পরিচালনা এবং সংযোগ ব্যবস্থাপনায় সহায়তা করে। ডেভেলপমেন্টের ক্ষেত্রে এই ধরনের ত্রুটির ক্ষেত্রে পরীক্ষা করার ফলে আরও স্থিতিস্থাপক স্থাপনা হতে পারে, উৎপাদনের সময় শেষ হওয়ার সম্ভাবনা হ্রাস করে এবং আপনার কনফিগারেশনে দুর্বল পয়েন্টগুলি সনাক্ত করা সহজ করে তোলে।