Forem: Memphis.dev team

Ingesting Webhooks From Stripe – The Better Way

Memphis.dev team — Thu, 18 Jan 2024 12:22:37 +0000

Introduction

Learn what webhooks are, and how you use them with Stripe to react to events quickly and in real-time with greater reliability using a streaming platform. Below, we’re answering these questions and more. This post will show you everything you need to know about webhooks, including what they are, how they work, examples, and how they can be improved using Memphis.

What are Webhooks?

Imagine a world where information flows seamlessly between systems. In this world, there’s no need for constant browser refreshing or sending numerous requests for updates. Welcome to the domain of webhooks, where real-time communication glides with the rhythm of efficiency and automation.

Webhooks stand out for their effectiveness for both the provider and the user. The main challenge with webhooks, however, is the complexity involved in their initial setup.

Often likened to reverse APIs, webhooks offer something akin to an API specification, requiring you to craft an API for the webhook to interact with. When the webhook initiates an HTTP request to your application, usually through a POST method, your task is to interpret and handle this incoming data effectively.

The downsides of webhooks

Push-based: Webhooks deliver or push events to your clients’ services, requiring them to handle the resulting back pressure. While understandable, this approach can impede your customers’ progress.
Implementing a server: For your client’s services to receive webhooks, they need a server that listens to incoming events. This involves managing CORS, middleware, opening ports, and securing network access, which adds extra load to their service by increasing overall memory consumption.
Retry: Services experience frequent crashes or unavailability due to various reasons. While some triggered webhooks might lead to insignificant events, others can result in critical issues, such as incomplete datasets where orders fail to be documented in CRM or new shipping instructions are not being processed. Hence, having a robust retry mechanism becomes crucial. 4 Persistent: Standard webhook systems generally lack event persistence for future audits and replays.
Replay: Similarly, it boils down to the user or developer experience you aim to provide. While setting up an endpoint for users to retrieve past events is feasible, it demands meticulous handling, intricate business logic, an extra database, and increased complexity for the client.
Throttling: Throttling is a technique used in computing and networking to control data flow, requests, or operations to prevent overwhelming a system or service. It limits the rate or quantity of incoming or outgoing data, recommendations, or actions. The primary challenge lies not in implementing throttling but in managing distinct access levels for various customers. Consider having an enterprise client with notably higher throughput needs compared to others. To accommodate this, you’d require a multi-tenant webhook system tailored to support diverse demands.

Why to use webhooks with Stripe

When you’re piecing together Stripe integrations, it’s crucial to have your applications tuned in to live events from your Stripe accounts. This way, your backend systems are always ready to spring into action based on these events.

To get this real-time event flow, you’ll need to set up webhook endpoints in your application. Once you’ve registered these endpoints, Stripe becomes your real-time informant, pushing event data directly to your application’s webhook endpoint as things happen in your Stripe account. Stripe uses HTTPS to deliver these events, packaging them as JSON payloads that feature an Event object.

Webhook events are your go-to for monitoring asynchronous activities. They’re perfect for keeping tabs on events like a customer’s bank giving the green light on a payment, charge disputes from customers, successful recurring payments, or managing subscription billing. With webhooks, you’re not just informed; you’re always a step ahead.

Why use Memphis as your Stripe’s webhook destination

Convert the push to pull: Memphis.dev operates as a pull-based message broker where clients actively pull and consume data from the broker.
Retry: Memphis provides a built-in retry system that maintains client states and offsets, even during disconnections. This configurable mechanism resends unacknowledged events until they’re acknowledged or until the maximum number of retries is reached.
Persistent: Memphis ensures message persistence by assigning a retention policy to each topic and message.
Replay: The client has the flexibility to rotate the active offset, enabling easy access to read and replay any past event that complies with the retention policy and is still stored.
Back pressure: Let Memphis handle the back pressure and scale from your team and clients.
Backup: You can easily enable automatic backup that will back up each and every message to an external S3-compatible storage.
Dead-letter: They enable you to preserve unconsumed messages, rather than discarding them, to diagnose why their processing was unsuccessful.

How to get started

Head to Stripe’s webhook dashboard

Create a Memphis account
Create a new Memphis station

Create a new client-type user and generate a URL for producing data

Copy the produce URL to the Stripe dashboard and click “Add endpoint”

Once a selected event will occur, it will trigger an event that will be sent to your Memphis Station

Join 4500+ others and sign up for our data engineering newsletter.

Originally published at Memphis.dev By Shoham Roditi Elimelech, software engineer at @Memphis.dev

Event Sourcing with Memphis.dev: A Beginner’s Guide

Memphis.dev team — Thu, 04 Jan 2024 12:14:54 +0000

Introduction

In the realm of modern software development, managing and maintaining data integrity is paramount. Traditional approaches often involve updating the state of an application directly within a database. However, as systems grow in complexity, ensuring data consistency and traceability becomes more challenging. This is where Event Sourcing, coupled with a powerful distributed streaming platform like Memphis.dev, emerges as a robust solution and a great data structure to work with.

What is Event Sourcing?

At its core, Event Sourcing is a design pattern that captures every change or event that occurs in a system as an immutable and sequentially ordered log of events. Instead of persisting the current state of an entity, Event Sourcing stores a sequence of state-changing events. These events serve as a single source of truth for the system’s state at any given point in time.

Understanding Event Sourcing in Action

Imagine a banking application that tracks an account’s balance. Instead of only storing the current balance in a database, Event Sourcing captures all the events that alter the balance. Deposits, withdrawals, or any adjustments are recorded as individual events in chronological order.

Let’s break down how this might work:

Event Creation: When a deposit of $100 occurs in the account, an event, such as FundsDeposited, with relevant metadata (timestamp, amount, account number) is created.
Event Storage: These events are then appended to an immutable log, forming a sequential history of transactions specific to that account.
State Reconstruction: To obtain the current state of an account, the application replays these events sequentially to compute the current balance. Each event is applied in order to derive the current balance, enabling the system to rebuild state at any given point in time.

Leveraging Memphis in Event Sourcing

Memphis.dev, an open-source distributed event streaming platform, is perfect for implementing Event Sourcing due to its features:

Scalability and Fault Tolerance: Memphis’ distributed nature allows for horizontal scalability and ensures fault tolerance by replicating data across multiple brokers (nodes).
Ordered and Immutable Event Logs: Memphis’ log-based architecture aligns seamlessly with the principles of Event Sourcing. It maintains ordered, immutable logs, preserving the sequence of events.
Real-time Event Processing: Memphis Functions offers a serverless framework, built within the Memphis platform to handle high-throughput, real-time event streams. Applications can process events as they occur, enabling near real-time reactions to changes in state.
Managing Schemas: One of the major challenges in event sourcing is maintaining schemas across the different events to avoid upstream breaks and client crashes.

Benefits of Event Sourcing with Kafka

Temporal Queries and Auditing: By retaining a complete history of events, it becomes possible to perform temporal queries and reconstruct past states, aiding in auditing and compliance.
Flexibility and Scalability: As the system grows, Event Sourcing with Memphis allows for easy scalability, as new consumers can independently process the event log.
Fault Tolerance and Recovery: In the event of failures, the ability to rebuild state from events ensures resiliency and quick recovery.

Let’s see what it looks like via code

Events occur and are pushed by their order of creation into some Memphis Station (=topic)

Event Log:

class EventLog:
    def __init__(self):
        self.events = []

    def append_event(self, event):
        self.events.append(event)

    def get_events(self):
        return self.events

Memphis Producer:

from __future__ import annotations
import asyncio
from memphis import Memphis, Headers, MemphisError, MemphisConnectError, MemphisHeaderError, MemphisSchemaError
import json

class MemphisEventProducer:
    def __init__(self,host="my.memphis.dev"):
        try:
        self.memphis = Memphis()
        await self.memphis.connect(host=host, username="<application type username>", password="<password>")

    def send_event(self, topic, event):
        await self.memphis.produce(station_name=topic, producer_name='prod_py',
  message=event,nonblocking=False)
        except (MemphisError, MemphisConnectError, MemphisHeaderError, MemphisSchemaError) as e:
          print(e)
        finally:
          await self.memphis.close()

Usage:

# Initialize Event Log
event_log = EventLog()

# Initialize Memphis Producer
producer = MemphisEventProducer()

# Append events to the event log and produce them to Memphis
events_to_publish = [
    {"type": "Deposit", "amount": 100},
    {"type": "Withdrawal", "amount": 50},
    # Add more events as needed
]

for event in events_to_publish:
    event_log.append_event(event)
    producer.send_event('account-events', event)

Criteria to choose the right event streaming platform for the job

When implementing Event Sourcing with a message broker, several key features are crucial for a streamlined and efficient system:

Persistent Message Storage:
Durability: Messages should be reliably stored even in the event of failures. This ensures that no events are lost and the event log remains intact.
Ordered and Immutable Event Logs:
Sequential Order: Preserving the order of events is critical for accurate state reconstruction. Events must be processed in the same sequence they were produced.
Immutability: Once an event is stored, it should not be altered. This guarantees the integrity and consistency of the event log.
Scalability and Performance:
Horizontal Scalability: The message broker should support horizontal scaling to accommodate increased event volume without sacrificing performance.
Low Latency: Minimizing message delivery time ensures near real-time processing of events, enabling quick reactions to state changes.
Fault Tolerance and High Availability:
Redundancy: Ensuring data redundancy across multiple nodes or partitions prevents data loss in the event of node failures.
High Availability: Continuous availability of the message broker is essential to maintain system functionality.
Consumer Flexibility and State Rebuilding:
Consumer Groups: Support for consumer groups allows multiple consumers to independently process the same set of events, aiding in parallel processing and scalability.
State Rebuilding: The broker should facilitate easy rebuilding of the application state by replaying events, enabling historical data retrieval.
Retention Policies and Archiving:
Retention Policies: Configurable retention policies allow managing the duration or size of stored messages. This ensures efficient storage management.
Archiving: Ability to archive or offload older events to long-term storage for compliance or historical analysis purposes.
Monitoring and Management:
Metrics and Monitoring: Providing insights into message throughput, latency, and system health helps in monitoring and optimizing system performance.
Admin Tools: Easy-to-use administrative tools for managing topics, partitions, and configurations streamline system management.
Security and Compliance:
Encryption and Authentication: Support for encryption and authentication mechanisms ensures the confidentiality and integrity of transmitted events.
Compliance Standards: Adherence to compliance standards (such as GDPR, SOC2) ensures that sensitive data is handled appropriately.
Seamless Integration and Ecosystem Support:
Compatibility and Integrations: Seamless integration with various programming languages and frameworks, along with support for diverse ecosystems, enhances usability.
Ecosystem Tools: Availability of connectors, libraries, and frameworks that facilitate Event Sourcing simplifies implementation and reduces development efforts.

Choosing a message broker that aligns with these critical features is essential for implementing robust Event Sourcing, ensuring data integrity, scalability, and resilience within your application architecture.

Event Sourcing using a Database vs a Message Broker (Streaming Platform)

Use Case Complexity: For simpler applications or where scalability isn’t a primary concern, databases might suffice. For higher reliability, distributed systems needing high scalability, and real-time processing, a message broker can be more suitable.

Replay: In event streaming platforms or message brokers, events are stored in a FIFO manner, one after the other as they first appear. That nature also makes it easier for the consumer on the other side to understand the natural flow of events and replay the entire “scene,” whereas in databases, it is not the case, and additional fields must be added, like timestamps, to organize the data based on time. It also requires additional logic to understand the latest state of an entity.

Continue your learning: read how and why event sourcing outgrows the database.

Join 4500+ others and sign up for our data engineering newsletter.

Originally published at Memphis.dev By Idan Asulin, Co-Founder & CTO at @Memphis.dev.

Comparing Webhooks and Event Consumption: A Comparative Analysis

Memphis.dev team — Thu, 28 Dec 2023 14:59:11 +0000

Introduction

In event-driven architecture and API integration, two vital concepts stand out: webhooks and event consumption. Both are mechanisms used to facilitate communication between different applications or services. Yet, they differ significantly in their approaches and functionalities, and by the end of this article, you will learn why consuming events can be a much more robust option than serving them using a webhook.

The foundational premise of the article assumes you function as a platform that wants or already delivers internal events to your clients through webhooks.

Webhooks

Webhooks are user-defined HTTP callbacks triggered by specific events on a service. They enable real-time communication between systems by notifying other applications when a particular event occurs. Essentially, webhooks eliminate the need to do manual polling or checking for updates, allowing for a more efficient, event-driven, and responsive system.

Key Features of Webhooks:

Event-driven: Webhooks are event-driven and are only triggered when a specified event occurs. For example, a webhook can notify an application when a new user signs up or when an order is placed.
Outbound Requests: They use HTTP POST requests to send data payloads to a predefined URL the receiving application provides.
Asynchronous Nature: Webhooks operate asynchronously, allowing the sending and receiving systems to continue their processes independently.

Event Consumption

Event consumption involves receiving, processing, and acting upon events emitted by various systems or services. This mechanism facilitates the seamless integration and synchronization of data across different applications.

Key Features of Event Consumption*:

Message Queues or Brokers: Event consumption often involves utilizing message brokers like Memphis.dev, Kafka, RabbitMQ, or AWS SQS to manage and distribute events.
Subscriber-Driven: Unlike webhooks, event consumption relies on subscribers who listen to event streams and process incoming events.
Scalability: Event consumption systems are highly scalable, efficiently handling large volumes of events.

Architectural questions for a better decision

Push-based VS. Pull based:
Webhooks deliver or push events to your clients’ services, requiring them to handle the resulting back pressure. While understandable, this approach can impede your customers’ progress. Using a scalable message broker to support consumption can alleviate this burden for your clients. How? By allowing clients to pull events based on their availability.
I*mplementing a server vs. Implementing a broker SDK:*
For your client’s services to receive webhooks, they need a server that listens to incoming events. This involves managing CORS, middleware, opening ports, and securing network access, which adds extra load to their service by increasing overall memory consumption.
Opting for pull-based consumption eliminates most of these requirements. With pull-based consumption, as the traffic is egress (outgoing) rather than ingress (incoming), there’s no need to set up a server, open ports, or handle CORS. Instead, the client’s service initiates the communication, significantly reducing complexity.
Retry:
Services experience frequent crashes or unavailability due to various reasons. While some triggered webhooks might lead to insignificant events, others can result in critical issues, such as incomplete datasets where orders fail to be documented in CRM or new shipping instructions are not being processed. Hence, having a robust retry mechanism becomes crucial. This can be achieved by incorporating a retry mechanism within the webhook system or introducing an endpoint within the service.
In contrast, when utilizing a message broker, events are acknowledged only after processing. Although implementing a retry mechanism is necessary in most cases, it’s typically more straightforward and native than handling retries with webhooks.
Persistent:
Standard webhook systems generally lack event persistence for future audits and replays, a capability inherently provided by persisted message brokers.
Replay:
Similarly, it boils down to the user or developer experience you aim to provide. While setting up an endpoint for users to retrieve past events is feasible, it demands meticulous handling, intricate business logic, an extra database, and increased complexity for the client. In contrast, using a message broker supporting this feature condenses the process to just a line or two of code, significantly reducing complexity.
Throttling:
Throttling is a technique used in computing and networking to control data flow, requests, or operations to prevent overwhelming a system or service. It limits the rate or quantity of incoming or outgoing data, recommendations, or actions. The primary challenge lies not in implementing throttling but in managing distinct access levels for various customers. Consider having an enterprise client with notably higher throughput needs compared to others. To accommodate this, you’d require a multi-tenant webhook system tailored to support diverse demands or opt for a message broker or streaming platform designed to handle such differential requirements.

Memphis as a tailor-made solution for the task

We are still iterating on the subject, so if you have any thoughts or ideas, I would love to learn from them: idan@memphis.dev

How to handle API rate limitations with a queue

Memphis.dev team — Wed, 20 Dec 2023 14:57:10 +0000

Introduction

Rate limitation refers to restricting the number of times a specific action can be performed within a certain time frame. For example, an API might have rate limitations restricting user or app requests within a given period. This helps prevent server overload, ensures fair usage, and maintains system stability and security.

Rate limitation is also a challenge for the apps that encounter it, as it requires to “slow down” or pause. Here’s a typical scenario:

Initial Request: When the app initiates communication with the API, it requests specific data or functionality.
API Response: The API processes the request and responds with the requested information or performs the desired action.
Rate-Limitation: If the app has reached the limit, it will usually need to wait until the next designated time frame (like a minute to an hour) before making additional requests. If it is a “soft” rate limitation and timeframes are known and linear, it’s easier to handle. Often, the waiting time climbs and increases in every block, requiring a whole different and custom handling per each API.
Handling Rate Limit Exceedances: If the app exceeds the rate limit, it might receive an error response from the API (such as a “429 Too Many Requests” status code). The app needs to handle this gracefully, possibly by queuing requests, implementing backoff strategies (waiting for progressively more extended periods before retrying), or informing the user about the rate limit being reached.

To effectively operate within rate limitations, apps often incorporate strategies like:

Throttling: Regulating the rate of outgoing requests to align with the API’s rate limit.
Caching: Storing frequently requested data locally to reduce the need for repeated API calls.
Exponential Backoff: Implementing a strategy where the app waits increasingly longer between subsequent retries after hitting a rate limit to reduce server load and prevent immediate retries.
Queue? More in the next section

Using a queue

A queue serves as an excellent “sidekick” or tool for helping services manage rate limitations due to its ability to handle tasks systematically. However, while it offers significant benefits, it’s not a standalone solution for this purpose.

In constructing a robust architecture, the service or app used to interact with an external API subject to rate limitations often handles tasks asynchronously. This service is typically initiated by tasks derived from a queue. When the service encounters a rate limit, it can easily return the job to the main queue, or assign it to a separate queue designated for delayed tasks, and revisit it after a specific waiting period, say X seconds.

This reliance on a queue system is highly advantageous, primarily because of its temporary nature and ordering. However, the queue alone cannot fully address rate limitations; it requires additional features or help from the service itself to effectively handle these constraints.

Challenges may arise when utilizing a queue:

Tasks re-entering the queue might return earlier than necessary, as their timing isn’t directly controlled by your service.
Exceeding rate limitations due to frequent calls within restricted timeframes. This may necessitate implementing sleep or wait mechanisms, commonly considered poor practice due to their potential impact on performance and responsiveness.

Here is what it will look like with RabbitMQ:

const amqp = require('amqplib');
const axios = require('axios');

// Function to make API requests, simulating rate limitations
async function makeAPICall(url) {
  try {
    const response = await axios.get(url);
    console.log('API Response:', response.data);
  } catch (error) {
    console.error('API Error:', error.message);
  }
}

// Connect to RabbitMQ server
async function connect() {
  try {
    const connection = await amqp.connect('amqp://localhost');
    const channel = await connection.createChannel();

    const queue = 'rateLimitedQueue';
    channel.assertQueue(queue, { durable: true });

    // Consume messages from the queue
    channel.consume(queue, async msg => {
      const { url, delayInSeconds } = JSON.parse(msg.content.toString());

      // Simulating rate limitation
      await new Promise(resolve => setTimeout(resolve, delayInSeconds * 1000));

      await makeAPICall(url); // Make the API call

      channel.ack(msg); // Acknowledge message processing completion
    });
  } catch (error) {
    console.error('RabbitMQ Connection Error:', error.message);
  }
}

// Function to send a message to the queue
async function addToQueue(url, delayInSeconds) {
  try {
    const connection = await amqp.connect('amqp://localhost');
    const channel = await connection.createChannel();

    const queue = 'rateLimitedQueue';
    channel.assertQueue(queue, { durable: true });

    const message = JSON.stringify({ url, delayInSeconds });
    channel.sendToQueue(queue, Buffer.from(message), { persistent: true });

    console.log('Task added to the queue');
  } catch (error) {
    console.error('RabbitMQ Error:', error.message);
  }
}

// Usage example
addToQueue('https://api.example.com/data', 5); // Add an API call with a delay of 5 seconds

// Start the consumer
connect();

Or with Kafka

const { Kafka } = require('kafkajs');
const axios = require('axios');

// Function to make API requests, simulating rate limitations
async function makeAPICall(url) {
  try {
    const response = await axios.get(url);
    console.log('API Response:', response.data);
  } catch (error) {
    console.error('API Error:', error.message);
  }
}

// Kafka configuration
const kafka = new Kafka({
  clientId: 'my-app',
  brokers: ['localhost:9092'], // Replace with your Kafka broker address
});

// Create a Kafka producer
const producer = kafka.producer();

// Connect to Kafka and send messages
async function produceToKafka(topic, message) {
  await producer.connect();
  await producer.send({
    topic,
    messages: [{ value: message }],
  });
  await producer.disconnect();
}

// Create a Kafka consumer
const consumer = kafka.consumer({ groupId: 'my-group' });

// Consume messages from Kafka topic
async function consumeFromKafka(topic) {
  await consumer.connect();
  await consumer.subscribe({ topic });
  await consumer.run({
    eachMessage: async ({ message }) => {
      const { url, delayInSeconds } = JSON.parse(message.value.toString());

      // Simulating rate limitation
      await new Promise(resolve => setTimeout(resolve, delayInSeconds * 1000));

      await makeAPICall(url); // Make the API call
    },
  });
}

// Usage example - Sending messages to Kafka topic
async function addToKafka(topic, url, delayInSeconds) {
  const message = JSON.stringify({ url, delayInSeconds });
  await produceToKafka(topic, message);
  console.log('Message added to Kafka topic');
}

// Start consuming messages from Kafka topic
const kafkaTopic = 'rateLimitedTopic';
consumeFromKafka(kafkaTopic);

// Usage example - Adding messages to Kafka topic
addToKafka('rateLimitedTopic', 'https://api.example.com/data', 5); // Add an API call with a delay of 5 seconds

Both approaches are legitimate, yet they necessitate your service to incorporate a ‘sleep’ mechanism.

With Memphis, you can offload the delay from the client to the queue using a simple feature made
just for that purpose and called “Delayed Messages”. Delayed messages allow you to send a received message back to the broker when your consumer application requires extra processing time.

What sets apart Memphis’ implementation is the consumer’s capability to control this delay independently and atomically.
Within the station, the count of unconsumed messages doesn’t impact the consumption of delayed messages. For instance, if a 60-second delay is necessary, it precisely configures the invisibility time for that specific message.

Memphis.dev Delayed Messages

message is received by the consumer group.
An event occurs, prompting the consumer group to pause processing the message.
Assuming the maxMsgDeliveries hasn’t hit its limit, the consumer will activate message.delay(delayInMilliseconds), bypassing the message. Instead of immediately reprocessing the same message, the broker will retain it for the specified duration.
The subsequent message will be consumed.
Once the requested delayInMilliseconds has passed, the broker will halt the primary message flow and reintroduce the delayed message into circulation.

const { memphis } = require('memphis-dev');

// Function to make API requests, simulating rate limitations 
async function makeAPICall(message) 
{ 
  try { 
    const response = await axios.get(message.getDataAsJson()['url']); 
    console.log('API Response:', response.data); 
    message.ack();
  } catch (error) { 
    console.error('API Error:', error.message); 
    console.log("Delaying message for 1 minute"); 
    message.delay(60000);
  } 
}

(async function () {
    let memphisConnection;

    try {
        memphisConnection = await memphis.connect({
            host: '<broker-hostname>',
            username: '<application-type username>',
            password: '<password>'
        });

        const consumer = await memphisConnection.consumer({
            stationName: '<station-name>',
            consumerName: '<consumer-name>',
            consumerGroup: ''
        });

        consumer.setContext({ key: "value" });
        consumer.on('message', (message, context) => {
            await makeAPICall(url, message);
        });

        consumer.on('error', (error) => { });
    } catch (ex) {
        console.log(ex);
        if (memphisConnection) memphisConnection.close();
    }
})();

Wrapping up

Understanding and adhering to rate limitations is crucial for app developers working with APIs. It involves managing request frequency, handling errors when limits are reached, implementing backoff strategies to prevent overloading the API servers, and utilizing rate limit information provided by the API to optimize app performance, and now you know how to do it with a queue as well!

Head to our blog or [docs(https://docs.memphis.dev/memphis/getting-started/readme) for more examples like that!

Join 4500+ others and sign up for our data engineering newsletter.

Originally published at Memphis.dev By Idan Asulin, Co-Founder & CTO at @Memphis.dev.

Real-Time Data Scrubbing Before Storing In A Data Warehouse

Memphis.dev team — Wed, 20 Dec 2023 08:22:03 +0000

Between January 2023 and May 2023, companies violating general data processing principles incurred fines totaling 1.86 billion USD (!!!).

In today’s data-driven landscape, the importance of data accuracy and compliance cannot be overstated. As businesses amass vast amounts of information, the need to ensure data integrity, especially PII storing, becomes paramount. Data scrubbing emerges as a crucial process, particularly in real-time scenarios, before storing information in a data warehouse.

Data Scrubbing in the context of compliance
Data scrubbing, often referred to as data cleansing or data cleaning, involves the identification and rectification of errors or inconsistencies in a dataset. In the context of compliance, it means removing certain values that qualify as PII that cannot be stored or should be handled differently.

Real-time data scrubbing takes the cleansing process a step further by ensuring that incoming data is cleaned and validated instantly, before being stored in a data warehouse.

Compliance standards, such as GDPR, HIPAA, or industry-specific regulations, mandate stringent requirements for data accuracy, privacy, and security. Failure to adhere to these standards can result in severe repercussions, including financial penalties and reputational damage. Real-time data scrubbing acts as a robust preemptive measure, ensuring that only compliant data is integrated into the warehouse.

Event-driven Scrubbing
Event-driven applications stand as stateful systems that intake events from one or multiple streams and respond to these incoming events by initiating computations, updating their state, or triggering external actions.

They represent a progressive shift from the conventional application structure that segregates computation and data storage into distinct tiers. In this novel architecture, these applications retrieve data from and save data to a remote transactional database.

In stark contrast, event-driven applications revolve around stateful stream processing frameworks. This approach intertwines data and computation, facilitating localized data access either in-memory or through disk storage. To ensure resilience, these applications implement fault-tolerance measures by periodically storing checkpoints in remote persistent storage.

In the context of Scrubbing, it means that the actual action of scrubbing will take place for each ingested event, in real-time, powering up only when new events arrive, and immediately after, not based on constant times, usually performed on top of the database, after being stored, meaning the potential violation already took place.

How does Memphis Functions support such a use case?
At times, a more comprehensive policy-driven cleansing may be necessary. However, if a quick, large-scale ‘eraser’ is what you require, Memphis Functions can offer an excellent solution. The diagram illustrates two options: data sourced from either a Kafka topic or a Memphis station, potentially both concurrently. This data passes through a Memphis Function named ‘remove-fields‘ before progressing to the data warehouse for further storage.

Behind the curtain, events or streaming data are grouped into batches, a configuration determined by the user’s specifications. These batches then undergo processing via a serverless function, specifically the ‘remove-fields’ function, meticulously designed to cleanse the ingested data according to pre-established rules. Following this scrubbing process, the refined data is either consumed internally or routed to a different Kafka topic, alternatively being swiftly directed straight to the Data Warehouse (DWH) for immediate utilization.

Usage example
Before

{
  "id": 123456789,
  "full_name": "Peter Parker",
  "gender": "male"
}
After (Removing ‘gender’)

{
  "id": 123456789,
  "full_name": "Peter Parker",
}

Next steps
An ideal follow-up action would involve implementing schema enforcement. Data warehouses are renowned for their rigorous schema enforcement practices. By integrating both a transformation layer and schema validation, it’s possible to significantly elevate data quality while reducing the risk of potential disruptions or breaks in the system. This can simply take place by attaching a Schemaverse schema to the station.

Start by signing up to Memphis Cloud. We have a great free plan that can get you up and running in no time, and try to build a pipeline yourself.

Join 4500+ others and sign up for our data engineering newsletter.

Originally published at Memphis.dev By Idan Asulin, Co-Founder & CTO at @Memphis.dev.

gRPC vs Message Broker

Memphis.dev team — Wed, 03 Aug 2022 13:12:00 +0000

What is gRPC? How does it differ from using a message broker? And what are the perfect use cases for each?

Microservices architecture allows large software systems to be broken down into smaller independent units called services. These services usually come with their own servers and databases. The services in a microservices system need a way to effective way to communicate such that the operations on one service does not affect other services.

There are several ways to connect services together. If you haven’t designed a microservices system before, the first thing that may come to mind is to create REST endpoints for one service that other services can call. While this can work in systems with a few services, it’s not scalable in larger systems. This is because REST works based on a blocking request-response mechanism.

A better way to connect microservices is to use a protocol that offers faster request times or use a non-blocking mechanism to get tasks done. Two technologies that enable this are gRPC and message brokers. gRPC sends binary data over the network and so has a faster request time. But gRPC cannot always be used because it requires memory to be allocated on the receiving end to handle response. One way to mitigate such memory allocation is to use a message broker like Memphis. A message broker allows you to queue messages that will be processed by different services in a microservices system.

This article goes over the similarities and differences between gRPC and message brokers. You will learn about the pros, cons, and ideal use cases of both technologies.

What is gPRC

gRPC (which is short for gRPC Remote Procedural Call) is a communication protocol that is used in place of REST to call functions between a client and a server. The client and the server can be microservices, mobile applications, or CLI tools.

For a gRPC set up to work, the has to be a client and a server. The client will make a proto request to the server and the server responds with a proto response.

image source: https://grpc.io

gRPC uses HTTP 2.0 which is a faster than HTTP 1.1 that REST depends on. HTTP 2.0 enforces a binary format by default. This means protocols using HTTP 2.0 need to serialize their data to binary before sending requests over the network. gRPC uses Protobuf, a binary serialization format, for defining data schema.

gRPC also supports data streaming which allows a single request to return a lot more data than it would typically be in the case of REST. This data streaming can either be a server to client streaming or bi-direction streaming between client to server. Note that the service called the client is the service that makes the initial request while the server is the service that sends the response.

What is a Message Broker

A message broker is a server that contains a persistent, append-only log that stores messages. Think of the message broker as a server that contains different log files that get updated as new data comes in.

An example use case of is an image processing pipeline that converts a large image into smaller images of various sizes. The conversion task takes an average of 10 seconds per image, but you have a thousand users trying to convert their images into different sizes. You can store each conversion task in a queue within the message broker and the message broker sends the tasks to the conversion server. This process prevents the server from being overwhelmed and keeps your services fast.

There are several message brokers on the market and your choice of a message broker will depend on your use case. If you’d prefer a cloud-native message broker, then Memphis is a perfect choice. You can also consider brokers such as Apache Kafka, Redis, and RabbitMQ.

Similarities between gRPC and Message Brokers

Message Format

gRPC has similar features to message brokers, the most prominent being the message format. Both gRPC and Memphis for example use proto3 data serialization format. The data is serialized to binary and sent over the network to the client. When the data reaches the consuming client, it is deserialized back to a form the client can use, like JSON.

Streaming support

gRPC and message brokers also support streaming. This means data can be sent from server to client as soon as the data is produced. An example of this is sending resized images from the image resizer service to the image watermarking service as soon as the image is resized. The image data can either be queued in a message broker or sent in an RPC call. In either case, the image data is streamed from the image resizer service to the image watermarking service.

Language agnostic

gRPC and message brokers are mostly language agnostic, with support in a variety of languages. You can use gRPC in the most widely used languages like Python, JavaScript, Java, Go, and C#. You can also connect to a message broker like Memphis using SDKs in Python, JavaScript, and Go.

How gPRC differs from a message broker

While gRPC has similar use cases as message brokers, they differ in so many other ways. A message broker typically stores its data on a disk while gRPC operates on the RAM. A message broker is installed as an executable on a server while gRPC depends on HTTP 2.0. This section goes into detail on how gRPC differs from a message broker.

Disk storage and Partitioning

A message broker serves as a persistent log data structure store, and so, it works on main memory or disk storage. This allows messages to be recovered if there is a server outage. A message broker like Memphis stores data in “stations” and these stations are partitioned across multiple brokers. This distributed storage increases the fault tolerance of the system.

gRPC on the other hand works with RAM because it operates at the source code layer. This also means that gRPC calls are not persisted to disk. Ideally, gRPC can be combined with a message broker to increase the overall performance of a distributed system.

Stream buffering

gRPC being a messaging protocol cannot buffer streaming data. This is unlike message brokers that can store millions of streamed messages as they are produced. An example scenario is streaming temperature data from a thermometer in a factory. If real-time processing is required, there has to be some server processing the data as it comes. This streaming process can overwhelm the server and so, there needs to be a way to process streams in real-time without overwhelming the server. A message broker is the ideal tool to handle this situation.

Deployment method

gRPC is a protocol based on HTTP 2.0 and works at the runtime layer of the software stack. Popular language runtimes like Node.js, Python, and Java 8 have already implemented HTTP 2.0 and so, support gRPC. A software library is usually used to enable gRPC connections within a language runtime.

A message broker on the other hand is an executable that is installed on a server and uses the memory and disk space to function. Memphis for example can be run on Docker containers or on Kubernetes. Clients that connect to a message broker can use REST APIs or gRPC. A user will build an SDK in most cases to ease the process of interacting with a message broker. Memphis for example has language support in Go, Python, and JavaScript.

Conclusion

While both gRPC and message brokers are used for inter-service communication. gRPC is more suited for inter-service calls while message brokers are more suited for task queueing.

The Memphis message broker is the perfect go-to broker for your inter-service communication needs. Memphis is easy to deploy and easy to use. You can immediately start using Memphis if you have Docker installed. Head over to the quick start to start using Memphis today.

Special thanks to Idan Asulin for the article

The end of poison messages!

Memphis.dev team — Wed, 20 Jul 2022 18:31:17 +0000

Have you ever got a call in the middle of the night saying “Infrastructure looks ok, but some service is not consuming data / messages get redelivered. Please figure it out”

Redelivered messages are also called “Poison messages”.

Poison messages = Messages that cause a consumer to repeatedly require a delivery (possibly due to a consumer failure) such that the message is never processed completely and acknowledged so that it can be stopped being sent again to the same consumer.

Example: Some message on an arbitrary queue pushed/pulled to or by a consumer. That consumer, for some reason, doesn’t succeed in handling it. It can be due to a bug, an unknown schema, resources issue, etc…

In RabbitMQ, for example, quorum queues keep track of the number of unsuccessful delivery attempts and expose it in the “x-delivery-count” header that is included with any redelivered message. It is possible to set a delivery limit for a queue using a policy argument, delivery-limit. When a message has been returned more times than the limit the message will be dropped or dead-lettered (if a DLX is configured).

It is known to any developer that uses a queue / messaging bus that poison messages should be taken care of, and it’s the developer’s responsibility to do so.

In RabbitMQ, the most common, simple solution would be to enable DLX (dead-lettered queue), but it doesn’t end there.

Recovering a poison message is just the first part, and the developer must also understand what causes this behavior and mitigate the issue.

Solutions
While there is the classic solution of committing/acknowledging a message as soon as possible, it’s not the best option for use cases requiring ensuring messages are acknowledged only when finished being handled.

Other approaches –

How to handle unacknowledged messages in RabbitMQ

Turn on DLX
Configure the DLX
Place a routing key
Build a dedicated consumer pointed to the DLX
Consume the unacknowledged messages
Fix the code/events

How to handle unacknowledged messages in Apache Kafka

There is no out-of-the-box redelivery/recovery of such messages.

Ensure there are logs within the code, tracking exceptions, and export to pagerduty/datadog/new relic/etc…
If the retention of a message is too small, it will be gone before the developer gets the chance to debug it. In most cases, the message will not be unique and can the loss, but in other, like transactions / atomic requests, it is. To mitigate this, a wrapper that provides this functionality should be made. A great example of such that provides that ability, and more is Wix’s GreyHound. Definitely worth taking a look.
There are other use cases that utilize cache DB of some kind to persist the message while being processed before being getting committed.

Unacknowledged messages in Memphis.dev

When we started to refine our approach, understand the needs, validate the experience, and craft the value it will bring to our users – three key objectives led our process –

The entry point of a user to the issue.
Quickly understand the problematic consumer and the root cause of the issue.
Fix code. The developer must debug the fix with a similar message to ensure it works.

1 - Define the trigger

In Memphis broker, at the SDK level, we use a parameter called “maxMsgDeliveries.”
Once this threshold is reached, the broker will not repeatedly send the “failed-to-ack” message to the same consumer group.

2 - Notification

Memphis broker senses the event of crossing the “maxMsgDeliveries = 2” per station, per consumer group.
Persist the time_sent, and payload of the redelivered message to the Memphis file-based store for 3 hours.
Mark the message as poisoned by a specific consumer.
Create an alert.

3 - Identify the consumer group which didn't acknowledge the message

Instead of going around through logs and multiple consumer groups, we wanted to narrow the finding to the minimum, so a simple click on the “real-time tracing” will lead to a graph screen showing the CGs which passed the redelivery threshold.

4 - Fix and Debug

After the developer understand what went wrong and creates a fix, before pushing the code which will lead to probably more adjustments when new messages will arrive, we have created the “Resend” mechanism which will push the unacknowledged message as many times as needed (Until ACK) to the consumer group that was not able to acknowledge the message in the first place.

The unacknowledged message will be retrieved from Memphis internal DB, ingested into a dedicated station per station, per CG, and only upon request. Next, it will be pushed to the unacknowledged CG – WITHOUT ANY CODE CHANGE, using the same already-configured emit.

That’s it. No need to create a persistency process, cache DB, DLQ, or massy logic. You can be sure no message is lost.

Join 4500+ others and sign up for our data engineering newsletter

This article is written by Avraham Neeman

Here is why you need a message broker

Memphis.dev team — Thu, 07 Jul 2022 07:03:23 +0000

Among the open-source projects my college buddies (and my future co-founders of memphis.dev) and I built, you can find “Makhela”, a Hebrew word for choir.
For the sake of simplicity – We will use “Choir”.

“Choir” was an open-source OSINT (Open-source intelligent) project focused on gathering context-based connections between social profiles using AI models like LDA and topic modeling, written in Python to explain what the world discusses over a specific domain and by high-ranking influencers in that domain and focus on what’s going on at the margins. For proof-of-concept or MVP we used a single data source, fairly easy for integrations – Twitter.

The graph below was the “brain” behind “Choir”. The brain autonomously grows and analyzes new vertexes and edges based on incremental changes in the corpus and fresh ingested data.

Each vertex symbolizes a profile, a persona, and each edge emphasizes (a) who connects to who. (b) Similar color = Similar topic.

Purple = Topic 1
Blue = Topic 2
Yellow = Marginal topic

After a reasonable amount of research, dev time, and a lot of troubleshooting & debug, things started to look good.

Among the issues we needed to solve were:

Understand the connection between profiles
Build a ranking algorithm for adding more influencers
Transform the schema of incoming data to a shape the analysis side knows how to handle
Near real-time is crucial - Enrich each tweet with external data
Adaptivity to "Twitter" rate limit
Each upstream or schema change crashed the analysis functions
Sync between collection and analysis, which were two different components
Infrastructure
Scale

As with any startup or early-stage project, we built “Choir” as MVP, Working solely with “Twitter”, and it looked like this -

The “Collector” is a monolith, python-written application that basically collects and refines the data for analysis and visualization in batches and in a static timing every couple of hours.

However, as the collected data and its complexity grew, problems started to arise. Each batch processing cycle analysis took hours for no good reason in terms of the capacity of the collected data (Hundreds of Megabytes at most!!). More on the rest of the challenges in the next sections.

Fast forward a few months later, users started to use “Choir”!!!
Not just using, but engaging, paying, and raising feature requests.
Any creator’s dream!

But then it hit us.

(a) “Twitter” is not the center of the universe, and we need to expand “Choir” to more sources.

(b) Any minor change in the code breaks the entire pipeline.

As with every eager-to-run project that starting to get good traction, fueling that growth and user base is your number 1, 2, and 3 priority,

and the last thing you want to do at this point is to go back and rebuild your framework. You want to continue the momentum.

With that spirit in our mind, we said “Let’s add more data sources and refactor in the future”. A big Mistake indeed.

Challenges in scaling a data-driven application

Each new data source requires a different schema transformation
Each schema change causes a chain of reaction downstream to the rest of the stages in the pipeline
Incremental / climbing collection. While you can wait for an entire batch collection to finalize and then save it to the DB, applications often crash. Imagine you’re doing a very slow collection and in the very last record, the collection process crashes.
In a monolith architecture, it’s hard to scale out the specific functions which require more power
Analysis functions often require modifications, upgrades, and algorithms to get better results, which are made by using or requiring different keys from the collectors.

While there is no quick fix, what we can do is build a framework to support such requirements.

Solutions

Option 1 – Duplicate the entire existing process to another source, for example, “Facebook”.

In addition to duplicating the collector, we needed to –

Maintain two different schemas (Nightmare)
Entirely different analysis functions. The connections between profiles on Facebook and “Twitter” are different and require different objective relationships.
The analyzer should be able to analyze the data in a joined manner, not individually; therefore, any minor change in source X directly affects the analyzer and often crashes it down.
Double maintenance

And the list goes on…
As a result, it cant scale.

Option 2 - Here it comes. Using a message broker!

I want to draw a baseline. A message broker is not the solution but a supporting framework or a tool to enable branched, growing data-driven architectures.

What is a message broker?

“A message broker is an architectural pattern for message validation, transformation, and routing. It mediates communication among applications[vague], minimizing the mutual awareness that applications should have of each other in order to be able to exchange messages, effectively implementing decoupling.[4]”. Wikipedia.

Firstly, let’s translate it to something we can grasp better.

A message broker is a temporary data store. Why temporary? Because each piece of data within it will be removed after a certain time, defined by the user. Therefore, the pieces of data within the message broker are called “messages.” Each message usually weighs a few bytes to a few megabytes.

Around the message broker, we can find producers and consumers.

Producer = The “thing” that pushes the messages into the message broker.
Consumer = The “thing” that consumes the messages from the message broker.

“Thing” means system/service/application/IoT/some objective that connects with the message broker and exchanges data.

Small note the same service/system/app can act as a producer and consumer at the same time.

Messaging queues derive from the same family, but there is a crucial difference between a broker and a queue.

MQ uses the term publish and subscribe. The MQ itself pushes the data to the consumers and not the other way (consumer pulls data from the broker).
Ordering is promised. Messages will be pushed in the order they receive. Some systems require it.
The ratio between a publisher (producers) and subscribers is 1:1. Having said it, modern versions can achieve it by some features like exchange and more.

Famous message brokers/queues are Apache Kafka, RabbitMQ, Apache Pulsar, and our own Memphis.dev. Kafka use cases span from event streaming to real-time data processing. One might consider using Memphis.dev instead of Kafka due to the ease of deployment and developer friendliness it provides.

Still with me? Awesome!

Thus, let’s understand how using a message broker helped “Choir” to scale.

Instead of doing things like this -

By decoupling the app to smaller microservices, and orchestrating the flow using a message broker, it therefore turned into this –

Starting from the top-left corner, each piece of data (tweet/post) inserted into the system automatically triggers the entire process and flows between the different stages.

Collection. The three collectors search each new profile added to the community in parallel. If any more data source/social network is needed – it’s been developed on the side, and once ready, start listening for incoming events. Allows infinite scale of sources, ability to work on the specific source without disrupting the others, micro-scaling for better performance of each source individually, and more.
Transformation. Once the collection is complete, results will be pushed to the next stage, “Schema transformation,” where the schema transformation service will transform the events’ schemas into a shape the analysis function can interpret. It enables a “single source of truth” regarding schema management, so in case of upstream change, all is needed to reach out to this service and debug the issue. In a more robust design, it can also integrate with an external schema registry to make maintenance even more effortless.

Analysis. Each piece of event is sent to the analysis function transformed, and in a shape the analysis function can interpret. In “Choir” we used different AI models. Scaling it was impossible, so moving to analysis per event definitely helped.
Save. Creates an abstraction between “Choir” and the type of database and the ability to batch several insertions to a single batch instead of request per event.

The main reason behind my writing is to emphasize the importance of implementing a message broker pattern and technology as early as possible to avoid painful refactoring in the future. Message brokers, by default, enable you to build scalable architectures because they remove the tight coupling constraints.

Yes, your roadmap and added features are important, Yes it will take a learning curve, yes it might look like an overkill solution for your stage, but when it comes to a data-driven use case, the need for scale will reveal quickly in performance, agility, feature additions, modifications, and more. Bad design decisions or a lack of proper framework will burn out your resources. It is better to build agile foundations, not necessarily enterprise-grade, before reaching the phase you are overwhelmed by users and feature requests.

To conclude, the entry barrier for a message broker is definitely worth your time.

Special thanks to Yaniv Ben-Hemo for the writing

Join 4500+ others and sign up for our data engineering newsletter

How to contribute to an open-source project

Memphis.dev team — Tue, 05 Jul 2022 08:38:20 +0000

One of the During my career as a software developer, I started getting involved in some open-source communities and actively contributor, I never thought to myself that it will leverage my knowledge and experience to that level it did.

Hence, in the spirit of open-source, I co-founded Memphis.dev together with my 3 best friends from college – A message broker for developers made out of devs’ struggles with using message brokers, building complex data/event-driven apps, and troubleshooting them.

What is an open-source software/project?
Open-source software (OSS) is software whose source code in some shape is open to the public, making it available for use, modification, and distribution with its original rights. Therefore, programmers who have access to source code can change the code by adding features to the project or software, changing it, or fixing parts that aren’t working properly. OSS typically includes a license (Apache, BSD, MIT, GNU) that describes what are the constraints around the project and how “flexible” is the project.
Read more about it in Snyk's article.

Where to find interesting open-source projects to contribute?

So usually OSS contributors start to contribute to projects they are making use of. For example, a developer who works with Redis finds it interesting to go deep into Redis internals, understand what’s going under the hood, fix bugs, or add new features.

Specifically for developers without any former experience working with open-source products, my personal suggestion would be to go over the CNCF projects page. CNCF is the foundation of cloud-native, open-source projects. Furthermore, among the backed projects you can find Kubernetes, Prometheus, and much more. Undoubtedly, it is a really good place to find some interesting projects to start contributing to.

The contribution process

Find a nice project for example Redis, NATS, Memphis{dev}
Connect with the project’s community (Slack channels, Discord, website, GitHub repo, etc.)
Search for contribution guidelines. Often a file located within the project’s main repo called CONTRIBUTING.md
Fork the GitHub repository — Create a copy of the entire repo on your GitHub account
Creating a separate branch from the main branch
Code your changes (bug fixes, new features, etc.)
Push
Create a pull request of your branch to the upstream repo
One of the repo maintainers reviews you PR (Usually happens automatically)
Fix issues and comments left by the maintainer
Awaiting your code to be merged
Celebrate your first contribution with some cold beer 🙂

Join 4500+ others and sign up for our data engineering newsletter

Originally posted on Memphis{dev} blog

Special thanks to Idan Asulin for this amazing article!

Fix: invalid apiVersion “client.authentication.k8s.io/v1alpha1”

Memphis.dev team — Wed, 22 Jun 2022 08:31:39 +0000

I recently encountered this error while trying to run ‘kubectl’ commands over our AWS EKS cluster, not sure if it's shared by more k8s vendors.

In short, you are using an incompatible ‘AWS CLI’ version with your current AWS EKS version.

To fix it — Update your ‘AWS CLI’

Installing or updating the latest version of the AWS CLI
This topic describes how to install or update the latest release of the AWS Command Line Interface (AWS CLI) on…
docs.aws.amazon.com

For Linux x86 (64-bit)

curl "https://awscli.amazonaws.com/awscli-exe-linux-x86_64.zip" -o "awscliv2.zip" && unzip awscliv2.zip && sudo ./aws/install

For MacOS

curl "https://awscli.amazonaws.com/AWSCLIV2.pkg" -o "AWSCLIV2.pkg" sudo installer -pkg AWSCLIV2.pkg -target /

For Window

C:\> msiexec.exe /i https://awscli.amazonaws.com/AWSCLIV2.msi

Hope it helped. Cheers.

Thanks to Yaniv Ben-hemo for writing & Idan Asulin For the QA.

Apache Kafka & Memphis.dev

Memphis.dev team — Tue, 31 May 2022 13:30:40 +0000

What is Memphis.dev, what is Apache Kafka, and what are the strength and weaknesses of each framework

Let’s start!

The complete comparison table is at the bottom.

What is Memphis.dev?

A distributed message broker designed and constructed to make developers’ lives who work around event-driven software incredibly easier with event-level observability and troubleshooting tools. memphis.dev started as a fork of nats.io project (since 2011), written in GoLang, and creating its own stream on top.

Memphis.dev uses the same paradigm as Apache Kafka of produce-consume.

Memphis.dev Important objects

Broker
Factories
Stations
Producers
Consumers / Consumer groups
UI
CLI
Kubernetes

Memphis.dev flow

Memphis robust version deployed over Kubernetes using helm chart.

With Memphis deployment, also deployed is the UI, which provides the user with complete control over the entire Memphis cluster.

Compared to other terminologies like topics and queues, Memphis has stations that are grouped under factories.

Apps produce (write) and consume (read) data to and from stations.

To scale the performance and redundancy of consumers in a super easy manner, several consumers of the same kind can be grouped within the same consumer group.

Messages remain in the station until they hit the defined retention (defined at station creation), but will be consumed exactly once by the same consumer group.

Persistency (Where data is being saved)

Memory — Higher performance, higher cost, redundant for up to two failed brokers.
File — Lower performance, lower cost, redundant for an entire cluster breakdown.

Both are replicated across the brokers as defined by the user.

Redundancy / HA

Cluster-mode — Made by multiple brokers (min. 3). Data is replicated across brokers. Redundant across AZs.
Standalone — Made by a single broker.

What is Apache Kafka?

Apache Kafka is a distributed event store and stream-processing platform. It is an open-source system developed by the Apache Software Foundation written in Java and Scala. Well-known and 2nd most popular event store in the world after RabbitMQ.

Apache Kafka Important players

Consumer / Consumer groups
Producer
Kafka connect
Topic and topic partition
Kafka streams
Broker
Zookeeper (Will be moved soon)

Apache Kafka flow

Apache Kafka is one of the most mature message brokers in the market.

Producers send a message record to a topic. A topic is a category or feed name to which records are published. Consumers subscribe to a topic and pull messages from it.

In Kafka, messages remain in the topic, also if they were consumed (limit time is defined by retention policy)

Heavy memory consumer for performance.

Persistency (Where data is being saved)

Memory
File (log files)

Redundancy / HA

Kafka does it by using Zookeeper to manage the state of the cluster.

In version 3.3, KRaft will be available for production use and remove the need for Zookeeper.

Bottom line

Apache Kafka has a strong maturity and large community, as well as battle-tested against large workloads and distributed environments, but it comprises multiple technical challenges in both ops, developer onboarding, troubleshooting, and cost.

Memphis.dev has a much smaller (yet a growing) community, and less documentation, but eliminates almost completely the need for operations and tunings, developers can onboard autonomously, run on any Kubernetes, and comes with observability and troubleshooting features that make it loveable by devs who are starting their way through the forests of event-driven and real-time.

Comparison Table

Thanks for reading!

How to build your own “DoorDash” app

Memphis.dev team — Wed, 25 May 2022 07:33:18 +0000

Apractical, (relatively) easy-to-do tutorial for developing an event-driven, distributed food delivery app, just like “Uber Eats” or “Wolt”.

Many thanks to Dhanush Kamath for the use case and supporting article.

Meet Fastmart — The fastest and most reliable food delivery app ever built.

The technology stack we will use -

Node.js as our primary dev-lang
MongoDB for order persistency
Memphis is a message broker for developers
Kubernetes will host our microservices

High-Level Plan

Install Minikube using brew package manager.
Install Memphis over Minikube.
Clone “Fastmart” GitHub repo.
Review the code, the different services, and how they interact with each other.
Deploy “Fastmart” over Kubernetes.
Order food!

Let’s start!

1.Install Minikube
For the installation commands, please head here: https://minikube.sigs.k8s.io/docs/start/

minikube is local Kubernetes, focusing on making it easy to learn and develop for Kubernetes.

All you need is Docker (or similarly compatible) container or a Virtual Machine environment, and Kubernetes is a single command away: minikube start

What you’ll need

2 CPUs or more
2GB of free memory
20GB of free disk space
Internet connection

Container or virtual machine manager, such as Docker, Hyperkit, Hyper-V, KVM, Parallels, Podman, VirtualBox, or VMware Fusion/Workstation

Output -

Verify minikube is health —

kubectl get ns

Output —

NAME STATUS AGE default Active 31h kube-node-lease Active 31h kube-public Active 31h kube-system Active 31h

2.Install Memphis

helm repo add memphis https://k8s.memphis.dev/charts/ && helm install memphis memphis/memphis — set connectionToken=”memphis” — create-namespace — namespace memphis

Let’s wait a minute or two, allowing the different components to reach “Running” state.

kubectl get pods -n memphis

Output -

NAME                           READY   STATUS             RESTARTS      AGE
k8s-busybox-68867bb9b7-sqdql   0/1     CrashLoopBackOff   4 (68s ago)   3m13s
memphis-broker-0               3/3     Running            4 (55s ago)   3m13s
memphis-ui-fd54f5bd6-zzqd4     0/1     CrashLoopBackOff   4 (79s ago)   3m13s
mongodb-replica-0              1/1     Running            0             3m13s

NAME                           READY   STATUS             RESTARTS      AGE
k8s-busybox-68867bb9b7-sqdql   0/1     CrashLoopBackOff   4 (76s ago)   3m21s
memphis-broker-0               3/3     Running            4 (63s ago)   3m21s
memphis-ui-fd54f5bd6-zzqd4     1/1     Running            5 (87s ago)   3m21s
mongodb-replica-0              1/1     Running            0             3m21s

k8s-busybox can be ignored. It will be fixed in the coming versions

3.Clone the Fastmart repo

git clone https://github.com/yanivbh1/FastMart.git

4.System Architecture, Code, and Flow

Follow the numbers to understand the flow.

FastMart has three main components:

order-service - Exposes REST endpoints that allow clients to fetch the food menu, place an order and track the order in real-time.

A new order will be saved in mongo with the status “Pending” and will be produced (Pushed) into the “orders” station

GET: /<orderId> Example: curl http://order-service:3000/30 POST: /<order_details> Example: curl -X POST http://order-service:3000/api/orders -d ‘{“items”:[{“name”:”burger”,”quantity”:1}], “email”:”yaniv@memphis.dev”}’ -H ‘Content-Type: application/json’

The code responsible for communicating with Memphis will be found on -

./order-service/src/services/mqService.js

const memphis = require(“memphis-dev”); const { logger } = require(‘./loggerService’) const MEMPHIS_HOST = process.env.MEMPHIS_HOST || ‘localhost’; // create MQ connection string using environment variable const MEMPHIS_USERNAME = process.env.MEMPHIS_USERNAME; const MEMPHIS_TOKEN = process.env.MEMPHIS_TOKEN; let ordersStation_producer = null; const memphisConnect = async () => { try { logger.info(Memphis — trying to connect) await memphis.connect({ host: MEMPHIS_HOST, username: MEMPHIS_USERNAME, connectionToken: MEMPHIS_TOKEN }); logger.info(Memphis — connection established) ordersStation_producer = await memphis.producer({ stationName: “orders”, producerName: “order_service”, }); logger.info(ordersStation_producer created) } catch(ex) { logger.log(‘fatal’,Memphis — ${ex}`);
memphis.close();
process.exit();
}
}
/**

Publish order to station
@param {Object} order — order object containing order details / const publishOrderToStation = (order) => { ordersStation_producer.produce({message: Buffer.from(JSON.stringify(order))}); logger.info(Memphis — order ${order._id} placed); } /*
An express middleware for injecting queue services into the request object.
@param {Object} req — express request object.
@param {Object} res — express response object.
@param {Function} next — express next() function. */ const injectPublishService = (req, res, next) => { // add all exchange operations here const stationServices = { publishOrderToStation: publishOrderToStation } // inject exchangeServices in request object req.stationServices = stationServices; next(); } module.exports = { injectPublishService: injectPublishService, memphisConnect: memphisConnect, }`

email-service - Responsible for notifying the client of the different stages.

email-service consumer messages from two stations: orders and notifications.

As soon as an order is inserted into the station, the email service notifies the client with an order confirmation.

At the same time listens for new notification requests of other services

resturant-service - Responsible for fulfilling an order.

Consume an order
Process the order
Change order status at the MongoDB level to “Accepted”
Using constant sleep time to mimic the preparation of the food by the restaurant
Change order status at the MongoDB level to “Delivered”
Sending notification to the client

5.Deploy “Fastmart” over Kubernetes

Fastmart repo tree -

To deploy Fastmart namespace and different services,

please run kubectl apply -f k8s-deployment.yaml

kubectl get pods -n fastmart

Output -

READY   STATUS             RESTARTS   AGE
email-service-5ddb9b58d6-bq2xd       0/1     CrashLoopBackOff   3          103s
fastmart-ui-5c9bc497bd-kn4lk         1/1     Running            0          11m
orders-service-5b689b66-4h8t9        0/1     CrashLoopBackOff   7          11m
resturant-service-6d97cf6fdc-c9mvs   0/1     Completed          4          103s

Let’s understand why Fastmart services cant start

kubectl logs email-service-5ddb9b58d6-bq2xd -n fastmart

Output -

> email-service@1.0.0 start
> node ./index.js

17-05-2022 07:10:09 PM - info: Sleeping for 300ms before connecting to Memphis.
17-05-2022 07:10:09 PM - info: Memphis - trying to connect
17-05-2022 07:10:09 PM - info: email-service started
17-05-2022 07:10:09 PM - fatal: Memphis - User is not exist

It appears that the services try to connect to “Memphis” with the user “fastmart” which does not exist and we require to create it.

The simplest way to add a new user would be through the UI, but let’s do it via CLI.

Please install Memphis CLI via here.

$ mem Usage: index <command> [options] Options: -V, — version output the version number -h, — help display help for command Commands: connect Connection to Memphis factory Factories usage commands station Stations usage commands user Users usage commands producer Producers usage commands consumer Consumer usage commands init Creates an example project for working with Memphis help display help for command Factory is the place to bind stations that have some close business logic Factory Commands: ls List of factories create Create new factory edit Edit factory name and/or description del Delete a factory Station is Memphis’ queue/topic/channel/subject Station Commands: ls List of stations create Create new station info Specific station’s info del Delete a station Manage users and permissions User Commands: ls List of users add Add new user del Delete user Producer is the entity who can send messages into stations Producer Commands: ls List of Producers Consumer is the entity who can consume messages from stations Consumer Commands: ls List of Consumers

To connect the CLI with Memphis control plane we need —

root password

kubectl get secret memphis-creds -n memphis -o jsonpath=”{.data.ROOT_PASSWORD}” | base64 — decode OqEO9AbncKFF93r9Qd5V

memphis control-plane url

kubectl port-forward service/memphis-cluster 7766:7766 6666:6666 5555:5555 — namespace memphis > /dev/null &

Now, connect the CLI

mem connect — user root — password bpdASQlhwWNzFt4JwLQo — server localhost:5555

Add the user “fastmart”

mem user add -u fastmart — type application

Or via the UI

**Soon after we will create the user,

the pods will restart automatically and reconnect with Memphis.**

6.Order food!
To expose the orders-service through localhost, run -

kubectl port-forward service/orders 9001:80 — namespace fastmart > /dev/null &

Get the menu

curl localhost:9001/api/menu

Output -

{“items”:[{“name”:”burger”,”price”:50},{“name”:”fries”,”price”:20},{“name”:”coke”,”price”:10}]}

Make an order

curl -X POST localhost:9001/api/orders -d ‘{“items”:[{“name”:”burger”,”quantity”:1}], “email”:”test@gmail.com”}’ -H ‘Content-Type: application/json’

An email should arrive shortly at the email address specified above.

Thanks!