Mantıksal Saatler — Sunucuların Saati Farklıysa Hangisi Önce Oldu?

1
@Component
2
class LamportClock {
3

4
    private final AtomicLong counter = new AtomicLong();
5

6
    /** A local event or a send: one more than before. */
7
    long tick() {
8
        return counter.incrementAndGet();
9
    }
10

11
    /** A receive: at least one more than what the sender had seen. */
12
    long observe(long received) {
13
        return counter.updateAndGet(own -> Math.max(own, received) + 1);
14
    }
15
}

1
@Component
2
class AccountEventPublisher {
3

4
    static final String LAMPORT_HEADER = "lamport";
5
    private final KafkaTemplate<String, AccountEvent> kafka;
6
    private final LamportClock clock;
7

8
    AccountEventPublisher(KafkaTemplate<String, AccountEvent> kafka, LamportClock clock) {
9
        this.kafka = kafka;
10
        this.clock = clock;
11
    }
12

13
    void publish(AccountEvent event) {
14
        // The account number is the key: one account, one partition, one order.
15
        ProducerRecord<String, AccountEvent> record = new ProducerRecord<>("account-events", event.iban(), event);
16
        record.headers().add(LAMPORT_HEADER, Long.toString(clock.tick()).getBytes(StandardCharsets.UTF_8));
17
        kafka.send(record);
18
    }
19
}

1
@Component
2
class AuditListener {
3

4
    private final LamportClock clock;
5
    private final AuditLog audit;
6

7
    AuditListener(LamportClock clock, AuditLog audit) {
8
        this.clock = clock;
9
        this.audit = audit;
10
    }
11

12
    @KafkaListener(topics = "account-events")
13
    void on(AccountEvent event, @Header(AccountEventPublisher.LAMPORT_HEADER) byte[] lamport) {
14
        long seen = clock.observe(Long.parseLong(new String(lamport, StandardCharsets.UTF_8)));
15
        audit.append(event, seen);   // ordered by (lamport, service), never by wall clock
16
    }
17
}

1
class LamportClockTest {
2

3
    @Test
4
    void anEventAfterAReceiveIsAlwaysLaterThanTheSender() {
5
        LamportClock branch = new LamportClock();
6
        LamportClock mobile = new LamportClock();
7

8
        long deposit = branch.tick();                 // 1
9
        long shown = mobile.observe(deposit);         // 2
10
        long withdrawal = mobile.tick();              // 3
11

12
        assertThat(shown).isGreaterThan(deposit);
13
        assertThat(withdrawal).isGreaterThan(shown);
14
    }
15

16
    @Test
17
    void aSlowReceiverJumpsAhead() {
18
        LamportClock receiver = new LamportClock();
19
        assertThat(receiver.observe(41)).isEqualTo(42);
20
    }
21
}