Event-Driven Architecture Tools: A Comprehensive Guide to Kafka, SQS, EventBridge and Cloud Alternatives

// SQS with proper error handling and DLQconst params = {  QueueUrl: 'https://sqs.us-east-1.amazonaws.com/123/my-queue',  ReceiveMessageWaitTimeSeconds: 20,  // Long polling  MaxNumberOfMessages: 10,  VisibilityTimeout: 30,  // Processing window  MessageAttributeNames: ['All']};
// DLQ configurationconst dlqParams = {  QueueName: 'my-queue-dlq',  Attributes: {    MessageRetentionPeriod: '1209600',  // 14 days    RedrivePolicy: JSON.stringify({      deadLetterTargetArn: dlqArn,      maxReceiveCount: 3  // Retry 3 times before DLQ    })  }};

import { CloudTasksClient } from '@google-cloud/tasks';
const client = new CloudTasksClient();const parent = client.queuePath(project, location, queue);
const task = {    httpRequest: {        httpMethod: 'POST',        url: 'https://example.com/process',        headers: { 'Content-Type': 'application/json' },        body: Buffer.from(JSON.stringify(payload))    },    scheduleTime: { seconds: Math.floor(timestamp / 1000) } // Delayed execution};
const response = await client.createTask({ parent, task });

// SNS with filter policies for smart routingconst publishParams = {  TopicArn: 'arn:aws:sns:us-east-1:123:my-topic',  Message: JSON.stringify(event),  MessageAttributes: {    eventType: { DataType: 'String', StringValue: 'ORDER_CREATED' },    priority: { DataType: 'Number', StringValue: '1' }  }};
// Subscription with filterconst subscriptionPolicy = {  eventType: ['ORDER_CREATED', 'ORDER_UPDATED'],  priority: [{ numeric: ['>', 0] }]};

import { PubSub } from '@google-cloud/pubsub';
const pubsub = new PubSub();const topic = pubsub.topic(topicId);
// Publishing with ordering keyconst messageId = await topic.publishMessage({    data: Buffer.from(data),    orderingKey: 'user-123', // Ensures order per key    attributes: {        event_type: 'user_updated',        version: '2'    }});

// EventBridge with content-based routingconst rule = {  Name: 'OrderProcessingRule',  EventPattern: JSON.stringify({    source: ['order.service'],    'detail-type': ['Order Created'],    detail: {      amount: [{ numeric: ['>', 100] }],      country: ['US', 'UK', 'DE']    }  }),  Targets: [    {      Arn: lambdaArn,      RetryPolicy: {        MaximumRetryAttempts: 2,        MaximumEventAge: 3600      },      DeadLetterConfig: {        Arn: dlqArn      }    }  ]};

// EventBridge event bus sharingconst eventBusPolicy = {  StatementId: 'AllowAccountAccess',  Action: 'events:PutEvents',  Principal: '987654321098',  // Target account  Condition: {    StringEquals: {      'events:detail-type': 'Order Created'    }  }};

// Kinesis with KCL for processingconst kinesisClient = new AWS.Kinesis();
// Enhanced fan-out for low latencyconst consumer = await kinesisClient.registerStreamConsumer({  StreamARN: streamArn,  ConsumerName: 'low-latency-consumer'}).promise();
// Kinesis Data Analytics for SQL processingconst sqlQuery = `CREATE STREAM orders_above_100 ASSELECT * FROM SOURCE_SQL_STREAM_001WHERE orderAmount > 100;`;

// Dataflow for stream processingimport { Pipeline } from '@google-cloud/dataflow';
const pipeline = new Pipeline(options);
pipeline    .readFromPubSub({ topic })    .window({ type: 'fixed', duration: 60 })    .map((data: string) => JSON.parse(data))    .filter((item: any) => item.amount > 100)    .writeToBigQuery({ tableSpec });

// Simple DLQ implementationconst dlqParams = {  QueueName: 'my-queue-dlq',  Attributes: {    MessageRetentionPeriod: '1209600',  // 14 days    RedrivePolicy: JSON.stringify({      deadLetterTargetArn: dlqArn,      maxReceiveCount: 3  // Retry 3 times before DLQ    })  }};

// Edge-optimized event processingclass EdgeEventProcessor {  private localQueue: Queue[] = [];  private cloudBuffer: Message[] = [];
  async processEvent(event: Event) {    // Process locally first    const processed = await this.localProcess(event);
    // Batch for cloud sync    if (this.shouldSyncToCloud(processed)) {      this.cloudBuffer.push(processed);
      if (this.cloudBuffer.length >= 100 ||          Date.now() - this.lastSync > 60000) {        await this.syncToCloud();      }    }  }
  private async syncToCloud() {    try {      // Compress and batch send      const compressed = this.compress(this.cloudBuffer);      await this.cloudClient.sendBatch(compressed);      this.cloudBuffer = [];      this.lastSync = Date.now();    } catch (error) {      // Store locally if cloud unreachable      await this.localStorage.store(this.cloudBuffer);    }  }}

// Cloudflare Workers Queue Handlerexport default {  async queue(batch: MessageBatch, env: Env): Promise<void> {    for (const message of batch.messages) {      try {        // Process at edge        const result = await processMessage(message.body);
        // Store in Durable Objects or KV        await env.KV.put(          `processed:${message.id}`,          JSON.stringify(result),          { expirationTtl: 3600 }        );
        message.ack();      } catch (error) {        // Retry with backoff        message.retry({ delaySeconds: 30 });      }    }  }};

// AWS IoT Core with Greengrassimport { GreengrassCoreIPCClient } from 'aws-iot-device-sdk-v2';import { PublishToTopicRequest, PublishMessage, BinaryMessage } from 'aws-iot-device-sdk-v2';
class EdgeIoTProcessor {    private ipcClient: GreengrassCoreIPCClient;
    constructor() {        this.ipcClient = new GreengrassCoreIPCClient();    }
    async publishEdgeEvent(event: any): Promise<void> {        // Local processing        const processed = this.processLocally(event);
        // Publish to IoT Core        const request: PublishToTopicRequest = {            topic: `edge/${this.deviceId}/events`,            publishMessage: {                binaryMessage: {                    message: Buffer.from(JSON.stringify(processed))                }            }        };
        await this.ipcClient.publishToTopic(request);    }}

// Abstract multi-cloud event interfaceinterface CloudEventAdapter {  publish(event: CloudEvent): Promise<void>;  subscribe(handler: EventHandler): Promise<void>;}
class MultiCloudEventBridge {  private adapters: Map<string, CloudEventAdapter> = new Map();
  constructor() {    this.adapters.set('aws', new AWSEventBridgeAdapter());    this.adapters.set('azure', new AzureEventGridAdapter());    this.adapters.set('gcp', new GCPEventarcAdapter());  }
  async publishToAll(event: CloudEvent) {    const promises = Array.from(this.adapters.values())      .map(adapter => adapter.publish(event));
    const results = await Promise.allSettled(promises);
    // Handle partial failures    const failures = results.filter(r => r.status === 'rejected');    if (failures.length > 0) {      await this.handleFailures(failures, event);    }  }}

// Solution: Claim check patternclass LargeMessageHandler {  async send(largePayload: any) {    if (JSON.stringify(largePayload).length > 256000) {      // Store in S3      const s3Key = await this.uploadToS3(largePayload);
      // Send reference      return this.queue.send({        type: 'large_message',        s3Key,        size: largePayload.length      });    }
    return this.queue.send(largePayload);  }}

// Solution: Poison message detectionclass PoisonMessageDetector {  private messageAttempts = new Map<string, number>();
  async process(message: Message) {    const messageId = message.id;    const attempts = this.messageAttempts.get(messageId) || 0;
    if (attempts >= 3) {      // Identified as poison message      await this.quarantine(message);      return;    }
    try {      await this.processMessage(message);      this.messageAttempts.delete(messageId);    } catch (error) {      this.messageAttempts.set(messageId, attempts + 1);
      // Check if specific error pattern      if (this.isPoisonPattern(error)) {        await this.quarantine(message);      } else {        throw error; // Retry      }    }  }}

// Solution: Partition key strategyclass OrderedEventProcessor {  async publishOrdered(events: Event[]) {    // Group by entity ID for ordering    const grouped = this.groupBy(events, e => e.entityId);
    for (const [entityId, entityEvents] of grouped) {      // Sort by timestamp      entityEvents.sort((a, b) => a.timestamp - b.timestamp);
      // Send with same partition key      for (const event of entityEvents) {        await this.kafka.send({          topic: 'events',          key: entityId,  // Ensures ordering          value: event        });      }    }  }}

// Comprehensive metrics collectionclass EventMetrics {  private metrics = {    messagesPublished: new Counter('messages_published_total'),    messagesConsumed: new Counter('messages_consumed_total'),    messagesFailed: new Counter('messages_failed_total'),    processingDuration: new Histogram('message_processing_duration_seconds'),    queueDepth: new Gauge('queue_depth'),    consumerLag: new Gauge('consumer_lag'),    dlqDepth: new Gauge('dlq_depth')  };
  async recordProcessing(message: Message, processor: Function) {    const timer = this.metrics.processingDuration.startTimer();
    try {      const result = await processor(message);      this.metrics.messagesConsumed.inc();      return result;    } catch (error) {      this.metrics.messagesFailed.inc({        error_type: error.constructor.name,        queue: message.source      });      throw error;    } finally {      timer();    }  }}