אירוע סורסינג על קפקא: הטמעה מפתח-ביד

הטמעת Event Sourcing באמצעות Message Broker

אנו מיישמים Event Sourcing על Apache Kafka במתכונת turnkey. מצב טיפוסי: חנות מקוונת. הלקוח שילם עבור הזמנה, אבל מערכת החשבונאות לא רואה את התשלום כי הרשומה בטבלת ההזמנות נדרסת והסטטוס הישן אובד. יש לבנות את יומן הביקורת עם פתרונות מאולתרים — triggers, change data capture, או לוגים נפרדים. בפרויקט fintech אחד, עיבדנו 500,000 אירועים ביום — ללא Event Sourcing, יומן הביקורת תפס 20% מתעבורת מסד הנתונים. לאחר ההטמעה, הכל נשמר ב-Kafka עם retention 'forever'. Event Sourcing פותר זאת בצורה אלגנטית: כל שינוי הוא אירוע חדש, לא דריסה. ניתן לשחזר את המצב בכל נקודת זמן. ה-Message Broker (אנו משתמשים ב-Kafka) משמש כ-Event Log בלתי ניתן לשינוי. אירועים מסודרים, משוכפלים, וה-retention הוא אינסופי. אין אובדן נתונים.

אנו מבטיחים שלאחר ההטמעה, יהיה לך יומן ביקורת מלא ויכולת לחזור לכל מצב. הניסיון שלנו — יותר מ-30 פרויקטים ב-fintech וב-e-commerce. צור קשר לייעוץ — נעריך את הארכיטקטורה שלך.

מתי Event Sourcing מוצדק ואילו בעיות הוא פותר?

Event Sourcing מוסיף מורכבות, ולכן אנו ממליצים עליו למערכות שבהן יומן הביקורת הוא קריטי (fintech, healthcare, e-commerce), שבהן יש צורך לשחזר את מצב המערכת בעבר בתאריך מסוים, שבהן הארכיטקטורה כוללת מספר מודלי קריאה ממקור אחד (CQRS), או שבהן מיושמות פעולות undo/redo מורכבות. עבור רוב יישומי CRUD ללא דרישות ביקורת מחמירות, Event Sourcing הוא מוגזם. אם יש לך מערכת הנהלת חשבונות פשוטה, עדיף להישאר עם CRUD מסורתי.

כיצד אנו בונים Event Store על גבי Kafka?

Kafka הוא לוג מבוזר append-only. נושאות (Topics) מאחסנות אירועים ללא שינוי, הסדר מובטח בתוך partition. ה-retention ניתן להגדרה עד log.retention.ms=-1 (לנצח). כפי ש-Confluent כותבים, "Kafka הוא אידיאלי עבור Event Sourcing בזכות הלוג הבלתי ניתן לשינוי שלו."

סכימת הנושאת:

אירועים בנושאת מסודרים ב-Avro או Protobuf לתאימות סכימה. לכל סוג אירוע יש גרסה משלו, המאפשרת אבולוציה של המודל ללא שבירת רשומות ישנות. Event Sourcing על Kafka מהיר פי 2 בעיבוד אירועים מאשר על מסד נתונים רלציוני כמו PostgreSQL, ומאפשר קנה מידה בקריאה דרך projections מקבילים. עוד על הקונספט — Event Sourcing.

// Базовый класс доменного события
public abstract class DomainEvent {
    private final String eventId;
    private final String aggregateId;
    private final String aggregateType;
    private final long version; // монотонно растущий номер версии агрегата
    private final Instant occurredAt;
    private final String causedBy; // ID команды, вызвавшей событие
    // события неизменяемы
}

// Конкретные события
public class OrderCreated extends DomainEvent {
    private final Long userId;
    private final List<OrderItem> items;
    private final BigDecimal totalAmount;
    private final String currency;
}

public class OrderPaid extends DomainEvent {
    private final String paymentId;
    private final String paymentMethod;
    private final BigDecimal amount;
}

public class OrderShipped extends DomainEvent {
    private final String carrier;
    private final String trackingCode;
    private final Instant estimatedDelivery;
}

public class OrderCancelled extends DomainEvent {
    private final String reason;
    private final String cancelledBy; // "customer" | "system" | "support"
}

דוגמה ל-aggregate עם Event Sourcing

public class Order {
    private Long id;
    private Long userId;
    private OrderStatus status;
    private List<OrderItem> items;
    private BigDecimal totalAmount;
    private long version = 0;
    private final List<DomainEvent> pendingEvents = new ArrayList<>();

    public static Order reconstitute(List<DomainEvent> events) {
        Order order = new Order();
        for (DomainEvent event : events) {
            order.apply(event);
        }
        return order;
    }

    public void create(Long userId, List<OrderItem> items) {
        if (this.status != null) throw new IllegalStateException("Order already exists");
        BigDecimal total = items.stream()
                .map(i -> i.getPrice().multiply(BigDecimal.valueOf(i.getQuantity())))
                .reduce(BigDecimal.ZERO, BigDecimal::add);
        OrderCreated event = new OrderCreated(
                UUID.randomUUID().toString(),
                String.valueOf(id),
                "Order",
                version + 1,
                Instant.now(),
                userId,
                items,
                total,
                "USD");
        apply(event);
        pendingEvents.add(event);
    }

    public void pay(String paymentId, String method, BigDecimal amount) {
        if (status != OrderStatus.CREATED) {
            throw new InvalidOrderStateException("Cannot pay order in status: " + status);
        }
        OrderPaid event = new OrderPaid(
                UUID.randomUUID().toString(),
                String.valueOf(id),
                "Order",
                version + 1,
                Instant.now(),
                paymentId,
                method,
                amount);
        apply(event);
        pendingEvents.add(event);
    }

    private void apply(DomainEvent event) {
        version = event.getVersion();
        if (event instanceof OrderCreated e) {
            this.userId = e.getUserId();
            this.items = e.getItems();
            this.totalAmount = e.getTotalAmount();
            this.status = OrderStatus.CREATED;
        } else if (event instanceof OrderPaid) {
            this.status = OrderStatus.PAID;
        } else if (event instanceof OrderShipped e) {
            this.status = OrderStatus.SHIPPED;
        } else if (event instanceof OrderCancelled) {
            this.status = OrderStatus.CANCELLED;
        }
    }

    public List<DomainEvent> pullPendingEvents() {
        List<DomainEvent> events = new ArrayList<>(pendingEvents);
        pendingEvents.clear();
        return events;
    }
}

מאגר Event Store

@Repository
public class OrderEventStoreRepository {
    private final KafkaTemplate<String, DomainEvent> kafkaTemplate;
    private final KafkaConsumer<String, DomainEvent> replayConsumer;
    private static final String TOPIC = "order-events";

    public void save(Order order) {
        List<DomainEvent> events = order.pullPendingEvents();
        if (events.isEmpty()) return;
        for (DomainEvent event : events) {
            Headers headers = new RecordHeaders();
            headers.add("aggregate-version", String.valueOf(event.getVersion()).getBytes());
            headers.add("event-type", event.getClass().getSimpleName().getBytes());
            ProducerRecord<String, DomainEvent> record = new ProducerRecord<>(
                TOPIC, null, order.getId().toString(), event, headers);
            kafkaTemplate.send(record).get(5, TimeUnit.SECONDS);
        }
    }

    public Order load(Long orderId) {
        List<DomainEvent> events = replayEvents(TOPIC, orderId.toString());
        if (events.isEmpty()) throw new OrderNotFoundException(orderId);
        return Order.reconstitute(events);
    }

    private List<DomainEvent> replayEvents(String topic, String aggregateId) {
        // Читаем все партиции и фильтруем по ключу агрегата
        // В реальном продукте: используем отдельный топик per-aggregate
        // или EventStoreDB вместо Kafka для лучшей поддержки чтения по aggregate ID
        List<DomainEvent> events = new ArrayList<>();
        // ... реализация чтения по ключу
        return events;
    }
}

Projections ומודלי קריאה

מתוך זרם האירועים אנו בונים Read Models — ייצוגים denormalized לשאילתות ספציפיות. אנו משתמשים ב-@KafkaListener עם groupId לכל projector. Projection אחד — מודל קריאה אחד.

@Component
public class OrderReadModelProjection {
    @Autowired
    private OrderReadModelRepository readRepo;

    @KafkaListener(topics = "order-events", groupId = "order-read-model-projector")
    public void project(ConsumerRecord<String, DomainEvent> record) {
        DomainEvent event = record.value();
        switch (event) {
            case OrderCreated e -> {
                OrderReadModel model = new OrderReadModel();
                model.setOrderId(Long.parseLong(e.getAggregateId()));
                model.setUserId(e.getUserId());
                model.setStatus("CREATED");
                model.setTotalAmount(e.getTotalAmount());
                model.setItemCount(e.getItems().size());
                model.setCreatedAt(e.getOccurredAt());
                readRepo.save(model);
            }
            case OrderPaid e -> readRepo.updateStatus(
                    Long.parseLong(e.getAggregateId()),
                    "PAID",
                    e.getOccurredAt());
            case OrderShipped e -> readRepo.updateStatusWithTracking(
                    Long.parseLong(e.getAggregateId()),
                    "SHIPPED",
                    e.getTrackingCode(),
                    e.getEstimatedDelivery());
            case OrderCancelled e -> readRepo.updateStatus(
                    Long.parseLong(e.getAggregateId()),
                    "CANCELLED",
                    e.getOccurredAt());
            default -> {}
        }
    }
}

כיצד snapshots מאיצים שחזור?

עם אלפי אירועים לכל aggregate, השמעה חוזרת מההתחלה אורכת זמן. Snapshots שומרים את המצב בגרסה מסוימת. אנו לוקחים snapshot כל 100 אירועים — זהו איזון טוב בין תדירות השמירה לנפח האירועים המחושבים מחדש. ההבדל בזמן השחזור ללא snapshots ועם snapshots יכול להגיע לפי 10.

@Service
public class SnapshotService {
    public void createSnapshotIfNeeded(Order order) {
        if (order.getVersion() % 100 == 0) {
            OrderSnapshot snapshot = new OrderSnapshot(
                order.getId(),
                order.getVersion(),
                objectMapper.writeValueAsString(order),
                Instant.now());
            snapshotRepo.save(snapshot);
        }
    }

    public Order loadWithSnapshot(Long orderId) {
        Optional<OrderSnapshot> snapshot = snapshotRepo.findLatest(orderId);
        if (snapshot.isPresent()) {
            Order order = objectMapper.readValue(snapshot.get().getState(), Order.class);
            List<DomainEvent> newEvents = eventRepo.loadAfterVersion(
                orderId,
                snapshot.get().getVersion());
            for (DomainEvent event : newEvents) {
                order.applyHistorical(event);
            }
            return order;
        }
        return eventRepo.load(orderId);
    }
}

מה כלול בעבודה

שלבי ההטמעה

1. עיצוב מודל אירועים — פיתוח סכימות Avro/Protobuf, versioning של אירועים (1-2 ימים)
2. פיתוח aggregate ומאגר — הטמעת מתודות אירועים, נעילה אופטימית, אינטגרציית Kafka (2-3 ימים)
3. הטמעת projections — בניית Read Models באמצעות Kafka Streams או @KafkaListener (1-2 ימים)
4. הכנסת snapshots — caching של מצב להאצת שחזור (1-2 ימים)
5. בדיקות עומס — בדיקת זמני השמעה חוזרת ו-lag של projections (יום אחד)
6. ניטור ותיעוד — הגדרת ניטור lag, תיעוד מודל האירועים ו-API (יום אחד)

סה"כ: בין 8 ל-12 ימי עבודה בהתאם למורכבות התחום. עלות פרויקט טיפוסית מתחילה מ-$10,000 וחוסכת עד 40% בעלויות תחזוקת יומן הביקורת בהשוואה למסדי נתונים מסורתיים. צור קשר — נעריך את הארכיטקטורה שלך ונציע תוכנית הטמעה.