Hadoop ecosystems with NestJS

Overview

The Hadoop ecosystem offers a wide range of tools for addressing big data challenges. This article will guide you through setting up a robust development environment for big data processing and application development. We'll begin by installing VirtualBox and Ubuntu Linux as our base system. Then, we'll use Docker to create an efficient and reproducible development setup. Docker will allow us to easily connect and manage several key components:

1. A Kafka broker for high-throughput message streaming
2. A Cassandra NoSQL database for scalable data storage

By utilizing Docker's port forwarding feature, we can seamlessly interact with these services from our host machine. NestJS will serve as our primary framework for implementing business logic and managing connections between Kafka, Cassandra, and any other services we might need. This setup provides a foundation for developing and testing big data applications, with NestJS offering a modular and extensible architecture that integrates well with the Hadoop ecosystem.

VirtualBox and Ubuntu Linux

VirtualBox is a powerful tool to run multiple operating systems on the computer without changing the main system. VirtualBox creates a safe, isolated environment where capable of experimenting with different operating systems and software. Please download and install it.

On the other hand, Ubuntu Linux is a popular, user-friendly version of the Linux operating system. It's known for its ease of use, stability, and large community support. Ubuntu is a great choice for those who want to explore the world of Linux or need a robust, free operating system for their work or personal projects. I’m using Ubuntu desktop for better developer experiences.

Docker

What is Docker

Docker packages software and all its dependencies into a neat, portable container. Imagine a new house and instead of packing everything into separate boxes, somehow it could shrink the entire room into a single suitcase - furniture, decorations, and all. Docker containers are lightweight, standalone, and executable packages that include everything needed to run a piece of software, including the code, runtime, system tools, libraries, and settings.

Docker compose

Docker Compose is like a conductor for the Docker orchestra. It's a tool that helps manage multiple Docker containers as a single application. For instance, building a complex app with several parts - maybe a web server, a database, and a caching service. Instead of starting each of these containers individually, Docker Compose defines and runs all of them with a single command. After installing the Linux system, let’s install Docker in Ubuntu and please be aware of the supported Ubuntu version. We’ll use Docker to host Kafka and Cassandra instances.

Kafka

What is Kafka

Apache Kafka is an efficient, high-speed messaging system for big data, capable of handling millions of messages per second! It was originally developed by LinkedIn and later became an open-source project under the Apache Software Foundation. It's designed to solve a common problem in the world of big data: how to reliably and quickly move large amounts of data from one place to another. Here, we’ll use Kafka to build a publish-subscribe message queue. Please download and install the docker image.

Cassandra

What is Cassandra

Apache Cassandra is a highly scalable, high-performance distributed NoSQL database designed to handle large amounts of structured data across many commodity servers. It provides a robust solution for companies requiring scalability and high availability without compromising performance.

Cassandra was originally developed by Facebook and later became an open-source project under the Apache Software Foundation. It's designed to handle big data challenges with ease, providing high availability and fault tolerance without compromising performance. Please download and install the docker image on the Ubuntu OS.

The reason for using Cassandra and the introduction to CQL (Cassandra Query Language)

CQL is Cassandra's friendly way of talking to your data. It's designed to be familiar to SQL users, making it easier for people with traditional database experience to transition to Cassandra. There are more details in NestJS implementations.

Set up port forwarding in Virtualbox and docker compose

Once VirtualBox, Ubuntu, docker, Kafka, and Cassandra are installed, the last step would be constructing port forwarding and drafting a docker-compose.yaml image.

Port forwarding

In the VirtualBox → Settings → Network → Advanced → Port Forwarding

• Kafka host port and guest port: 9092
• Cassandra host port and guest port: 9042
• Cassandra host port and guest port: 7000

Docker compose

Create a folder called url-shortener and create a file called docker-compose.yaml within the Ubuntu OS.

# ./url-shortener/docker-compose.yaml
services:
  kafka:
    image: apache/kafka:3.8.0 # locate your image accordingly
    ports:
      - 9092:9092
  cassandra:
    image: cassandra:latest
    container_name: my-cassandra
    ports:
      - 9042:9042
      - 7000:7000
    restart: always
    environment:
      - CASSANDRA_CLUSTER_NAME=MyCluster
      - CASSANDRA_ENDPOINT_SNITCH=SimpleSnitch
      - CASSANDRA_DC=datacenter1        

Then, please run the command in the Ubuntu:

sudo docker compose up        

Please note the setup doesn’t specify the mounted volume, which saves the data. It’s for simplicity purposes.

NestJS implementations

Overview

We’re using NestJS as the backend to interact with Kafka and Cassandra. However, it’s better to understand the design under the hood before proceeding with programming. I’ll provide a helicopter view and introduce each concept gradually to make sense of it.

What is and why NestJS

Compared to plain NodeJS, NestJS provides TypeScript as the default language, making it type-safe. Further, it supports a clear MVC (Model-View-Controller) structure to foster development efficiency and smooth maintenance. There are three typical types of files within a module:

• .module.ts for managing services and controllers
• .controller.ts for endpoint construction
• .service.ts serves as a model for business logic

We’ll follow the NestJS convention for the following implementations.

Monorepo and PNPM

The concept of monorepo is managing multiple projects under one Git repository. Benefits might include but are not limited to:

1. Issue separation for each project
2. Flexibility and scalability when each project grows
3. One commit may have implementations across multiple projects, which makes it easy to maintain

To manage multiple projects within one workspace, PNPM is one of the best options for JavaScript-based projects. It’s easy to understand and saves disk space. Here is the package.json for reference at the moment writing the article. Here is the root package.json

{
  "name": "url-shortener",
  "version": "1.0.0",
  "description": "",
  "scripts": {
    "dev:kafka-manager": "pnpm --filter kafka-manager start --watch",
    "dev:producer": "pnpm --filter producer start --watch",
    "dev:consumer": "pnpm --filter consumer start --watch"
  },
  "keywords": [],
  "author": "",
  "license": "ISC",
  "dependencies": {
    "@nestjs/axios": "^3.0.2",
    "@nestjs/common": "^9.0.0",
    "@nestjs/config": "^3.2.3",
    "@nestjs/core": "^9.0.0",
    "@nestjs/microservices": "^10.3.10",
    "@nestjs/platform-express": "^9.0.0",
    "axios": "^1.7.2",
    "cassandra-driver": "^4.7.2",
    "kafkajs": "^2.2.4",
    "reflect-metadata": "^0.1.13",
    "rxjs": "^7.2.0",
  },
  "devDependencies": {
    "@nestjs/cli": "^9.0.0",
    "@nestjs/schematics": "^9.0.0",
    "@nestjs/testing": "^9.0.0",
    "@types/express": "^4.17.13",
    "@types/jest": "28.1.4",
    "@types/node": "^16.18.103",
    "@types/supertest": "^2.0.11",
    "@typescript-eslint/eslint-plugin": "^5.0.0",
    "@typescript-eslint/parser": "^5.0.0",
    "eslint": "^8.0.1",
    "eslint-config-prettier": "^8.3.0",
    "eslint-plugin-prettier": "^4.0.0",
    "jest": "28.1.2",
    "prettier": "^2.3.2",
    "source-map-support": "^0.5.20",
    "supertest": "^6.1.3",
    "ts-jest": "28.0.5",
    "ts-loader": "^9.2.3",
    "ts-node": "^10.0.0",
    "tsconfig-paths": "4.0.0",
    "typescript": "^4.3.5"
  }
}        

In each sub-project’s package.json, we use the PNPM workspace, which is stated in the dependencies and devDependencies. Here is one of the sub-project. Please follow the NestJS documentation for how to initiate a new project.

{
  "name": "consumer",
  "version": "0.0.1",
  "description": "",
  "author": "",
  "private": true,
  "license": "UNLICENSED",
  "scripts": {
    "prebuild": "rimraf dist",
    "build": "nest build",
    "format": "prettier --write \"src/**/*.ts\" \"test/**/*.ts\"",
    "start": "nest start",
    "start:dev": "nest start --watch",
    "start:debug": "nest start --debug --watch",
    "start:prod": "node dist/main",
    "lint": "eslint \"{src,apps,libs,test}/**/*.ts\" --fix",
    "test": "jest",
    "test:watch": "jest --watch",
    "test:cov": "jest --coverage",
    "test:debug": "node --inspect-brk -r tsconfig-paths/register -r ts-node/register node_modules/.bin/jest --runInBand",
    "test:e2e": "jest --config ./test/jest-e2e.json"
  },
  "dependencies": {
    "url-shortener": "workspace:*"
  },
  "devDependencies": {
    "url-shortener": "workspace:*"
  },
  "jest": {
    "moduleFileExtensions": [
      "js",
      "json",
      "ts"
    ],
    "rootDir": "src",
    "testRegex": ".*\\.spec\\.ts$",
    "transform": {
      "^.+\\.(t|j)s$": "ts-jest"
    },
    "collectCoverageFrom": [
      "**/*.(t|j)s"
    ],
    "coverageDirectory": "../coverage",
    "testEnvironment": "node"
  }
}        

Then, please run the command pnpm install at the root directory level.

Kafka Manager

The Kafka manager manages Kafka topics and partitions.

// kafka-admin.module.ts
import { Module } from '@nestjs/common';
import { KafkaAdminController } from './kafka-admin.controller';
import { KafkaAdminService } from './kafka-admin.service';
import { ClientsModule, Transport } from '@nestjs/microservices';

@Module({
  imports: [
    ClientsModule.register([
      {
        name: 'ADMIN_SERVICE',
        transport: Transport.KAFKA,
        options: {
          producerOnlyMode: true, // not joining the consumer group
          client: {
            clientId: 'admin',
            brokers: ['localhost:9092'], // default localhost port
          },
        },
      },
    ]),
  ],
  controllers: [KafkaAdminController],
  providers: [KafkaAdminService],
})
export class KafkaAdminModule {}        

// kafka-admin-controller.ts
import { Controller, Post, Body, Get, Param } from '@nestjs/common';
import { KafkaAdminService } from './kafka-admin.service';

@Controller('kafka-admin')
export class KafkaAdminController {
  constructor(private readonly kafkaAdminService: KafkaAdminService) {}

  @Post('create-topic')
  async createTopic(
    @Body('topic') topic: string,
    @Body('numPartitions') numPartitions: number,
    @Body('replicationFactor') replicationFactor: number,
  ) {
    await this.kafkaAdminService.createTopic(
      topic,
      numPartitions,
      replicationFactor,
    );
    return { message: `Topic ${topic} created successfully` };
  }

  @Get('topics')
  async getTopics() {
    return this.kafkaAdminService.getTopics();
  }

  @Get('topic-metadata/:topicName')
  async getTopicMetadata(@Param('topicName') topicName: string) {
    return this.kafkaAdminService.getTopicMetadata(topicName);
  }

  @Get('topic-partition-counts/:topicName')
  async getTopicPartitionCounts(@Param('topicName') topicName: string) {
    return await this.kafkaAdminService.getTopicPartitionCounts(topicName);
  }
}        

// kafka-admin.service.ts
// use kafkajs
import {
  Injectable,
  OnModuleDestroy,
  Logger,
  Inject,
  OnApplicationBootstrap,
} from '@nestjs/common';
import { ConfigService } from '@nestjs/config';
import { ClientKafka } from '@nestjs/microservices';
import { Admin } from 'kafkajs';

@Injectable()
export class KafkaAdminService
  implements OnModuleDestroy, OnApplicationBootstrap
{
  private admin: Admin;

  constructor(
    private configService: ConfigService,
    @Inject('ADMIN_SERVICE') kafkaClient: ClientKafka,
  ) {
    this.admin = kafkaClient.createClient().admin();
  }
  
  // NestJS life cycle
  async onApplicationBootstrap() {
    await this.admin.connect();
    await this.createTopics();
  }

  async onModuleDestroy() {
    await this.admin.disconnect();
  }

  async createTopic(
    topic: string,
    numPartitions = 1,
    replicationFactor = 1,
  ): Promise<void> {
    const topics = await this.admin.listTopics();
    if (!topics.includes(topic)) {
      await this.admin.createTopics({
        topics: [{ topic, numPartitions, replicationFactor }],
      });
      Logger.log(
        `Topic ${topic} created successfully`,
        `${KafkaAdminService.name}.${KafkaAdminService.prototype.createTopic.name}`,
      );
    } else {
      Logger.log(
        `Topic ${topic} already exists`,
        `${KafkaAdminService.name}.${KafkaAdminService.prototype.createTopic.name}`,
      );
    }
  }

  async createTopics() {
    try {
      Logger.log(
        'Creating default topics...',
        `${KafkaAdminService.name}.${KafkaAdminService.prototype.createTopic.name}`,
      );
      await this.createTopic(
        this.configService.get<string>('KAFKA_TOPIC'),
        this.configService.get<number>('KAFKA_PARTITIONS'),
      );
    } catch (error) {
      Logger.error(
        error,
        `${KafkaAdminService.name}.${KafkaAdminService.prototype.createTopic.name}`,
      );
    }
  }

  async getTopics(): Promise<string[]> {
    return this.admin.listTopics();
  }

  async getTopicMetadata(topicName: string) {
    try {
      const topicMetadata = await this.admin.fetchTopicMetadata({
        topics: [topicName],
      });
      const topic = topicMetadata.topics.find((t) => t.name === topicName);

      if (topic) {
        return topic;
      } else {
        throw new Error(`Topic ${topicName} not found`);
      }
    } catch (error) {
      Logger.error(
        error,
        `${KafkaAdminService.name}.${KafkaAdminService.prototype.getTopicMetadata.name}`,
      );
    }
  }

  async getTopicPartitionCounts(topicName: string): Promise<number> {
    const topic = await this.getTopicMetadata(topicName);
    return topic.partitions.length;
  }

  async deleteTopic(topic: string): Promise<void> {
    const topics = await this.admin.listTopics();
    if (topics.includes(topic)) {
      await this.admin.deleteTopics({ topics: [topic] });
      Logger.log(
        `Topic ${topic} deleted successfully`,
        `${KafkaAdminService.name}.${KafkaAdminService.prototype.deleteTopic.name}`,
      );
    } else {
      Logger.log(
        `Topic ${topic} does not exist`,
        `${KafkaAdminService.name}.${KafkaAdminService.prototype.deleteTopic.name}`,
      );
    }
  }
}        

// kafka-manager.module.ts
// the root module handles all sub-modules
import { Module } from '@nestjs/common';
import { ConfigModule } from '@nestjs/config';
import { join } from 'path';
import { KafkaAdminModule } from './kafka-admin/kafka-admin.module';

@Module({
  imports: [
    KafkaAdminModule,
    // .env configuration for Kafka creation
    ConfigModule.forRoot({
      isGlobal: true,
      envFilePath: join('../', '../.env'),
    }),
  ],
  controllers: [],
  providers: [],
})
export class KafkaManagerModule {}        

// main.ts
// the place where NestJS bootstraps the application and microservices
import { NestFactory } from '@nestjs/core';
import { KafkaManagerModule } from './kafka-manager.module';
import { MicroserviceOptions } from '@nestjs/microservices';

async function bootstrap() {
  const app = await NestFactory.create(KafkaManagerModule);
  app.connectMicroservice<MicroserviceOptions>({});
  await app.listen(3000);
}
bootstrap();        

Producer-consumer pattern

The producer generates messages and the consumer can be notified and process the message afterwards. It’s a common pattern in system design and programming fields. In NestJS, there are two available patterns: message-driven and event-driven. The implementation showcases the message-driven pattern.

Producer

// producer.module.ts
import { Module } from '@nestjs/common';
import { ProducerController } from './producer.controller';
import { ProducerService } from './producer.service';
import { ClientsModule, Transport } from '@nestjs/microservices';

@Module({
  imports: [
    ClientsModule.register([
      {
        name: 'PRODUCER_SERVICE',
        transport: Transport.KAFKA,
        options: {
          producerOnlyMode: true,
          client: {
            clientId: 'hero',
            brokers: ['localhost:9092'],
          },
        },
      },
    ]),
  ],
  controllers: [ProducerController],
  providers: [ProducerService],
})
export class ProducerModule {}        

// producer.controller.ts
import { Body, Controller, Logger, Param, Post } from '@nestjs/common';
import { ProducerService } from './producer.service';

@Controller('producer')
export class ProducerController {
  constructor(private readonly producerService: ProducerService) {}

  @Post(':topic')
  async sendMessage(@Param('topic') topic: string, @Body() message: any) {
    return await this.producerService.publish(topic, message);
  }
}        

// producer.service.ts
import {
  Injectable,
  OnModuleInit,
  OnModuleDestroy,
  Inject,
} from '@nestjs/common';
import { ClientKafka } from '@nestjs/microservices';
import { Producer } from 'kafkajs';

@Injectable()
export class ProducerService implements OnModuleInit, OnModuleDestroy {
  private producer: Producer;

  constructor(
    @Inject('PRODUCER_SERVICE') private readonly kafkaClient: ClientKafka,
  ) {
    this.producer = this.kafkaClient.createClient().producer();
  }

  async onModuleInit() {
    // Connect when the module initializes
    await this.producer.connect();
  }

  async onModuleDestroy() {
    // Disconnect when the module is destroyed
    await this.producer.disconnect();
  }

  async publish(topic: string, message: any) {
    // No need to connect here, as we're already connected
    return await this.producer.send({
      topic: topic,
      messages: [{ value: JSON.stringify(message) }],
    });
  }
}        

// main.ts
import { NestFactory } from '@nestjs/core';
import { ProducerModule } from './producer.module';
import { MicroserviceOptions } from '@nestjs/microservices';

async function bootstrap() {
  const app = await NestFactory.create(ProducerModule);

  app.connectMicroservice<MicroserviceOptions>({});

  await app.startAllMicroservices();
  await app.listen(9002);
}
bootstrap();        

Consumer

Cassandra module

// cassandra.module.ts
import { Module } from '@nestjs/common';
import { CassandraService } from './cassandra.service';
import { CassandraController } from './cassandra.controller';

@Module({
  controllers: [CassandraController],
  providers: [CassandraService],
  exports: [CassandraService],
})
export class CassandraModule {}        

// cassandra.controller.ts
import { Controller, Get, Param } from '@nestjs/common';
import { CassandraService } from './cassandra.service';

@Controller('cassandra')
export class CassandraController {
  constructor(private readonly cassandraService: CassandraService) {}

  @Get()
  async getKeyspaces() {
    return await this.cassandraService.getKeyspaces();
  }

  @Get(':keyspace')
  async getTables(@Param('keyspace') keyspace: string) {
    return await this.cassandraService.getTables(keyspace);
  }

  @Get(':keyspace/:table')
  async getTable(
    @Param('keyspace') keyspace: string,
    @Param('table') table: string,
  ) {
    return await this.cassandraService.getTable(keyspace, table);
  }
}        

// cassandra.service.ts
// use Cassandra's formal Node library: Cassandra driver
import {
  Injectable,
  Logger,
  OnModuleInit,
  OnModuleDestroy,
} from '@nestjs/common';
import { Client, types } from 'cassandra-driver';

@Injectable()
export class CassandraService implements OnModuleInit, OnModuleDestroy {
  private client: Client;

  constructor() {
    this.client = new Client({
      contactPoints: ['localhost'],
      localDataCenter: 'datacenter1',
    });
  }

  async onModuleInit() {
    try {
      await this.connect();
      Logger.log('Connected to Cassandra', CassandraService.name);
    } catch (error) {
      Logger.error(
        `Failed to connect to Cassandra: ${error.message}`,
        `${CassandraService.name}.${CassandraService.prototype.onModuleInit.name}`,
      );
      throw error;
    }
  }

  async onModuleDestroy() {
    await this.client.shutdown();
  }

  private async connect() {
    await this.client.connect();
    const query =
      "CREATE KEYSPACE IF NOT EXISTS examples WITH replication = {'class': 'SimpleStrategy', 'replication_factor': '3' }";
    await this.execute(query);
    const createTableQuery =
      'CREATE TABLE IF NOT EXISTS examples.shortened_urls ( \
          url_id text PRIMARY KEY, \
          original_url text, \
          created_at timestamp \
      );';
    await this.execute(createTableQuery);
  }

  async execute(query: string, params?: any[]): Promise<types.ResultSet> {
    try {
      return await this.client.execute(query, params, { prepare: true });
    } catch (error) {
      Logger.error(
        `Query execution error: ${error.message}`,
        `${CassandraService.name}.${CassandraService.prototype.execute.name}`,
      );
      throw error;
    }
  }

  async getKeyspaces(): Promise<types.ResultSet> {
    const query = 'SELECT keyspace_name FROM system_schema.keyspaces';
    return this.execute(query);
  }

  async getTables(keyspace: string): Promise<types.ResultSet> {
    const query =
      'SELECT table_name FROM system_schema.tables WHERE keyspace_name = ?';
    return this.execute(query, [keyspace]);
  }

  async getTable(keyspace: string, table: string): Promise<types.ResultSet> {
    const query = `SELECT * FROM ${keyspace}.${table}`;
    return this.execute(query);
  }
}        

Consumer module

// consumer.module.ts
import { Module } from '@nestjs/common';
import { ConsumerController } from './consumer.controller';
import { ConsumerService } from './consumer.service';
import { ClientsModule, Transport } from '@nestjs/microservices';
import { CassandraModule } from './cassandra/cassandra.module';
import { CassandraService } from './cassandra/cassandra.service';

@Module({
  imports: [
    CassandraModule,
    ClientsModule.register([
      {
        name: 'EXAMPLE_SERVICE',
        transport: Transport.KAFKA,
        options: {
          client: {
            clientId: 'example-consumer',
            brokers: ['localhost:9092'],
          },
          consumer: {
            groupId: 'example-consumer-group',
          },
        },
      },
    ]),
  ],
  controllers: [ConsumerController],
  providers: [ConsumerService, CassandraService],
})
export class ConsumerModule {}        

// consumer.controller.ts
import { Controller, Logger } from '@nestjs/common';
import { ConsumerService } from './consumer.service';
import { MessagePattern, Payload } from '@nestjs/microservices';

@Controller()
export class ConsumerController {
  constructor(private readonly consumerService: ConsumerService) {}

  @MessagePattern('example_topic')
  async handleMessage(@Payload() message: any) {
    return await this.consumerService.processMessage(message);
  }
}        

// consumer.service.ts
import { Injectable, Inject, OnModuleInit, Logger } from '@nestjs/common';
import { ClientKafka } from '@nestjs/microservices';
import { CassandraService } from './cassandra/cassandra.service';

@Injectable()
export class ConsumerService implements OnModuleInit {
  // see the corresponding client avaiable in the module.ts file
  constructor(
    @Inject('EXAMPLE_SERVICE') private readonly kafkaClient: ClientKafka,
    private readonly cassandraService: CassandraService,
  ) {}

  async onModuleInit() {
    // need to subscribe to a topic
    // so that we can get the response from the Kafka microservice
    // the topic will be automatically created if it doesn't exist
    this.kafkaClient.subscribeToResponseOf('example_topic');
    await this.kafkaClient.connect();
  }

  async processMessage(message: any) {
    try {
      const { url_id, original_url } = message;
      const created_at = new Date();

      Logger.log(
        'url_id: ' + url_id,
        `${ConsumerService.name}.${ConsumerService.prototype.processMessage.name}`,
      );
      Logger.log(
        'original_url: ' + original_url,
        `${ConsumerService.name}.${ConsumerService.prototype.processMessage.name}`,
      );
      Logger.log(
        'createdAt: ' + created_at,
        `${ConsumerService.name}.${ConsumerService.prototype.processMessage.name}`,
      );

      await this.cassandraService.execute(
        'INSERT INTO examples.shortened_urls (url_id, original_url, created_at) VALUES (?, ?, ?)',
        [url_id, original_url, created_at],
      );
      return 'Message processed';
    } catch (error) {
      Logger.error(
        error,
        `${ConsumerService.name}.${ConsumerService.prototype.processMessage.name}`,
      );
      throw error;
    }
  }
}        

// main.ts
import { NestFactory } from '@nestjs/core';
import { ConsumerModule } from './consumer.module';
import { MicroserviceOptions, Transport } from '@nestjs/microservices';

async function bootstrap() {
  const app = await NestFactory.create(ConsumerModule);
  app.connectMicroservice<MicroserviceOptions>({
    transport: Transport.KAFKA,
    options: {
      client: {
        brokers: ['localhost:9092'],
      },
      consumer: {
        groupId: 'my-kafka-consumer',
      },
    },
  });
  app.startAllMicroservices();
}
bootstrap();        

Verification

Please run the following commands to run each project at the root level with PNPM workspace.

• pnpm run dev:kafka-manager
• pnpm run dev:producer
• pnpm run dev:consumer

The testing port: `https://localhost:9002/producer/example_topic`
The test body:

```
{
  "url_id": "1",
  "original_url": "original_url1"
}
```        

Conclusion

The article orchestrates several tools to build a big data processing system skeleton with VirtualBox, Ubuntu desktop, Docker, Kafka, Cassandra, and NestJS. The implementations are simplified on purpose to be more understandable. Here is the source code (develop branch for the latest implementation) if you want to learn more.