Lider Seçimi ve Split-Brain — Ağ Bölününce Kim Karar Verir?

1
@Configuration
2
class LedgerKafkaConfig {
3

4
    // Three copies; a write is acknowledged only when at least two of them have it.
5
    @Bean
6
    NewTopic ledgerTopic() {
7
        return TopicBuilder.name("ledger-entries")
8
                .partitions(6)
9
                .replicas(3)
10
                .config(TopicConfig.MIN_IN_SYNC_REPLICAS_CONFIG, "2")
11
                .build();
12
    }
13

14
    @Bean
15
    ProducerFactory<String, LedgerEntry> ledgerProducerFactory(KafkaProperties properties) {
16
        Map<String, Object> config = new HashMap<>(properties.buildProducerProperties(null));
17
        config.put(ProducerConfig.ACKS_CONFIG, "all");
18
        config.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
19
        return new DefaultKafkaProducerFactory<>(config);
20
    }
21
}

1
@Configuration
2
class LeadershipConfig {
3

4
    @Bean
5
    DefaultLockRepository lockRepository(DataSource dataSource) {
6
        return new DefaultLockRepository(dataSource);
7
    }
8

9
    @Bean
10
    JdbcLockRegistry lockRegistry(LockRepository repository) {
11
        return new JdbcLockRegistry(repository);
12
    }
13

14
    // Only the instance holding the lock is leader; the others keep trying.
15
    @Bean
16
    LockRegistryLeaderInitiator leaderInitiator(LockRegistry registry) {
17
        return new LockRegistryLeaderInitiator(registry);
18
    }
19
}

1
@Component
2
class InterestPostingJob {
3

4
    private final AtomicBoolean leading = new AtomicBoolean(false);
5
    private final InterestPosting posting;
6

7
    InterestPostingJob(InterestPosting posting) {
8
        this.posting = posting;
9
    }
10

11
    @EventListener
12
    void onGranted(OnGrantedEvent event) {
13
        leading.set(true);
14
    }
15

16
    @EventListener
17
    void onRevoked(OnRevokedEvent event) {
18
        leading.set(false);   // stop at once: someone else may already be leading
19
    }
20

21
    @Scheduled(cron = "0 5 0 * * *", zone = "Europe/Istanbul")
22
    void postDailyInterest() {
23
        if (leading.get()) {
24
            posting.postForYesterday();
25
        }
26
    }
27
}

1
class InterestPostingJobTest {
2

3
    private final InterestPosting posting = mock(InterestPosting.class);
4
    private final InterestPostingJob job = new InterestPostingJob(posting);
5
    private final Context context = mock(Context.class);
6

7
    @Test
8
    void runsOnlyWhileLeading() {
9
        job.onGranted(new OnGrantedEvent(this, context, "ledger"));
10
        job.postDailyInterest();
11
        job.onRevoked(new OnRevokedEvent(this, context, "ledger"));
12
        job.postDailyInterest();
13

14
        verify(posting, times(1)).postForYesterday();
15
    }
16
}