Pub/Sub and Event Streaming

import java.nio.charset.StandardCharsets;
import java.util.*;
import java.util.zip.CRC32;

class PartitionedLog {
    private final List<List<String>> partitions = new ArrayList<>();
    private final Map<String, Integer> offsets = new HashMap<>(); // "group/partition" -> next index

    PartitionedLog(int count) {
        for (int i = 0; i < count; i++) {
            partitions.add(new ArrayList<>());
        }
    }

    synchronized int append(String key, String event) {
        CRC32 crc = new CRC32();
        crc.update(key.getBytes(StandardCharsets.UTF_8));
        int p = (int) (crc.getValue() % partitions.size());
        partitions.get(p).add(event);
        return p;
    }

    // Returns events after the group's offset and moves the offset (commits) past them.
    synchronized List<String> poll(String group, int partition, int max) {
        String k = group + "/" + partition;
        List<String> log = partitions.get(partition);
        int from = offsets.getOrDefault(k, 0);
        int to = Math.min(log.size(), from + max);
        offsets.put(k, to);
        return new ArrayList<>(log.subList(from, to));
    }

    synchronized void seek(String group, int partition, int offset) {
        offsets.put(group + "/" + partition, offset);
    }
}

import threading
import zlib

class PartitionedLog:
    def __init__(self, count):
        self.partitions = [[] for _ in range(count)]
        self.offsets = {}  # (group, partition) -> next index
        self.lock = threading.Lock()

    def append(self, key, event):
        with self.lock:
            p = zlib.crc32(key.encode()) % len(self.partitions)
            self.partitions[p].append(event)
            return p

    def poll(self, group, partition, max_events):
        """Returns events after the group's offset and moves the offset past them."""
        with self.lock:
            log = self.partitions[partition]
            start = self.offsets.get((group, partition), 0)
            end = min(len(log), start + max_events)
            self.offsets[(group, partition)] = end
            return log[start:end]

    def seek(self, group, partition, offset):
        with self.lock:
            self.offsets[(group, partition)] = offset