Low Level Design repository

Author: kousiknath

.gitignore

@@ -0,0 +1,29 @@
+### IntelliJ IDEA ###
+out/
+!**/src/main/**/out/
+!**/src/test/**/out/
+
+### Eclipse ###
+.apt_generated
+.classpath
+.factorypath
+.project
+.settings
+.springBeans
+.sts4-cache
+bin/
+!**/src/main/**/bin/
+!**/src/test/**/bin/
+
+### NetBeans ###
+/nbproject/private/
+/nbbuild/
+/dist/
+/nbdist/
+/.nb-gradle/
+
+### VS Code ###
+.vscode/
+
+### Mac OS ###
+.DS_Store

.idea/.gitignore

@@ -0,0 +1,11 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<module type="JAVA_MODULE" version="4">
+  <component name="NewModuleRootManager" inherit-compiler-output="true">
+    <exclude-output />
+    <content url="file://$MODULE_DIR$">
+      <sourceFolder url="file://$MODULE_DIR$/src" isTestSource="false" />
+    </content>
+    <orderEntry type="inheritedJdk" />
+    <orderEntry type="sourceFolder" forTests="false" />
+  </component>
+</module>

.idea/misc.xml

@@ -0,0 +1,53 @@
+package lld.kafka;
+
+import lld.kafka.model.Consumer;
+import lld.kafka.model.ConsumerGroup;
+import lld.kafka.model.Producer;
+import lld.kafka.model.Topic;
+import lld.kafka.service.ConsumerService;
+import lld.kafka.service.TopicService;
+import lld.kafka.service.impl.ConsumerServiceImpl;
+import lld.kafka.service.impl.TopicServiceImpl;
+import lld.kafka.worker.ConsumerWorker;
+import lld.kafka.worker.ProducerWorker;
+
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+
+public class MainApplication {
+
+    private static void test1() {
+        TopicService topicService = new TopicServiceImpl();
+        Topic topic = topicService.createTopic("my-topic", 3);
+        ExecutorService executorService = Executors.newFixedThreadPool(10);
+
+        Producer producer = new Producer();
+        executorService.execute(new ProducerWorker(topic, producer, topicService));
+
+        ConsumerGroup consumerGroup = new ConsumerGroup();
+        ConsumerService consumerService = new ConsumerServiceImpl();
+
+        Consumer consumer1 = new Consumer(consumerGroup);
+        consumerService.subscribe(topic, consumer1);
+        executorService.execute(new ConsumerWorker(consumer1, consumerService));
+
+        Consumer consumer2 = new Consumer(consumerGroup);
+        consumerService.subscribe(topic, consumer2);
+        executorService.execute(new ConsumerWorker(consumer2, consumerService));
+
+        Consumer consumer3 = new Consumer(consumerGroup);
+        consumerService.subscribe(topic, consumer3);
+        executorService.execute(new ConsumerWorker(consumer3, consumerService));
+
+        // Consumer 4 should remain idle as we have only 3 partitions in the topic
+        Consumer consumer4 = new Consumer(consumerGroup);
+        consumerService.subscribe(topic, consumer4);
+        executorService.execute(new ConsumerWorker(consumer4, consumerService));
+
+        executorService.shutdown();
+    }
+
+    public static void main(String[] args) {
+        test1();
+    }
+}

.idea/modules.xml

@@ -0,0 +1,9 @@
+<b> Design a message broker which has the following properties: </b>
+
+1. Producers should be able to produce messages.
+2. Producers can be divided into multiple partitions.
+3. There can be multiple consumers in a consumer group.
+4. A single consumer can consume from multiple topics.
+5. In a consumer group, at a time, a consumer can consume from a single partition. In a consumer group, a 
+6. single consumer always maps to a single partition.
+7. If possible, messages should be persisted.

.idea/vcs.xml

@@ -0,0 +1,42 @@
+package lld.kafka.model;
+
+import java.util.UUID;
+
+public class Consumer {
+    String id;
+    ConsumerGroup consumerGroup;
+
+    public Consumer(ConsumerGroup consumerGroup) {
+        this.id = UUID.randomUUID().toString();
+        this.consumerGroup = consumerGroup;
+        this.consumerGroup.addConsumer(this);
+    }
+
+    public ConsumerGroup getConsumerGroup() {
+        return this.consumerGroup;
+    }
+
+    @Override
+    public int hashCode() {
+        return id.hashCode();
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (o == null || this.getClass() != o.getClass()) {
+            return false;
+        }
+
+        if (o == this) {
+            return true;
+        }
+
+        Consumer other = (Consumer) o;
+        return this.id.equals(other.id);
+    }
+
+    @Override
+    public String toString() {
+        return this.id;
+    }
+}

LowLevelDesign.iml

@@ -0,0 +1,101 @@
+package lld.kafka.model;
+
+import java.util.*;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+
+public class ConsumerGroup {
+    private String groupId;
+    private Set<Consumer> consumers;
+    private Map<Partition, Integer> currentOffset;
+    private Map<Partition, Integer> committedOffset; // Can be used to contain the committed offsets just for crash recovery purpose
+    private Map<Consumer, Partition> consumerPartitionMap;
+
+    private static ConsumerGroup defaultInstance;
+
+    private Lock lock;
+
+    public static ConsumerGroup getDefaultInstance() {
+        if (defaultInstance == null) {
+            defaultInstance = new ConsumerGroup();
+            defaultInstance.groupId = "default-consumer-group";
+        }
+
+        return defaultInstance;
+    }
+
+    public ConsumerGroup() {
+        this.groupId = UUID.randomUUID().toString();
+        this.consumers = new HashSet<>();
+        this.currentOffset = new HashMap<>();
+        this.committedOffset = new HashMap<>();
+        this.consumerPartitionMap = new HashMap<>();
+        this.lock = new ReentrantLock();
+    }
+
+    public void addConsumer(Consumer consumer) {
+        Objects.requireNonNull(consumer);
+        this.consumers.add(consumer); // Irrespective of partition assignment, add the consumer to the group
+    }
+
+    public int getConsumerCount() {
+        return this.consumers.size();
+    }
+
+    public boolean isPartitionAllocated(Partition partition) {
+        return new HashSet<>(this.consumerPartitionMap.values()).contains(partition);
+    }
+
+    public Partition allocatePartition(Partition partition, Consumer consumer) {
+        this.currentOffset.put(partition, -1);
+        this.consumerPartitionMap.put(consumer, partition);
+        return partition;
+    }
+
+    public Message getMessageForConsumer(Consumer consumer) {
+        Message message = null;
+        Partition partition = this.consumerPartitionMap.get(consumer);
+        if (partition == null) {
+            return message;
+        }
+
+        try {
+            lock.lock(); // Check if we need this lock really as in our design, a single consumer is always
+            // mapped to a single partition. So, there is no chance of overriding partition data by multiple threads
+
+            int lastOffset = this.currentOffset.get(partition);
+            int nextOffset = lastOffset + 1;
+
+            message = partition.getMessageAtOffset(nextOffset);
+            if (message == null) {
+                // no more message
+                return null;
+            }
+
+            this.currentOffset.put(partition, nextOffset);
+        } finally {
+            lock.unlock();
+        }
+
+        return message;
+    }
+
+    @Override
+    public int hashCode() {
+        return groupId.hashCode();
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (o == null || this.getClass() != o.getClass()) {
+            return false;
+        }
+
+        if (o == this) {
+            return true;
+        }
+
+        ConsumerGroup other = (ConsumerGroup) o;
+        return this.groupId.equals(other.groupId);
+    }
+}