camel-master 0.55.0

Master/leader-election component — wraps a delegate Consumer with distributed lock-based leadership coordination via LeadershipService
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152

//! leadership lifecycle tests (delegate started/stopped/recreated on epochs). Split from tests.rs by concern (bd rc-ubk1v).
//! Shared mocks/helpers live in the tests module root (tests.rs).

use super::*;

fn build_master_consumer(
    platform_service: Arc<dyn PlatformService>,
    create_consumer_calls: Arc<AtomicUsize>,
    start_calls: Arc<AtomicUsize>,
    delegate_retry_max_attempts: Option<u32>,
) -> MasterConsumer {
    let reconnect = match delegate_retry_max_attempts {
        Some(max) => NetworkRetryPolicy {
            max_attempts: max,
            ..NetworkRetryPolicy::default()
        },
        None => NetworkRetryPolicy {
            max_attempts: 0,
            ..NetworkRetryPolicy::default()
        },
    };
    MasterConsumer::new(
        "lock-a".to_string(),
        "fake:delegate".to_string(),
        Arc::new(FakeDelegateComponent {
            create_consumer_calls,
            start_calls,
        }),
        Arc::new(NoOpMetrics),
        platform_service,
        Duration::from_millis(500),
        reconnect,
        Arc::new(PanicRuntimeObservability) as Arc<dyn camel_component_api::RuntimeObservability>,
    )
}

#[tokio::test]
async fn starts_delegate_only_after_started_leading() {
    let leadership = Arc::new(FakeLeadershipService::new(None));
    let platform_service = Arc::new(FakePlatformService::new(leadership.clone()));
    let create_consumer_calls = Arc::new(AtomicUsize::new(0));
    let start_calls = Arc::new(AtomicUsize::new(0));
    let mut master = build_master_consumer(
        platform_service,
        Arc::clone(&create_consumer_calls),
        Arc::clone(&start_calls),
        Some(30),
    );

    let (tx, mut rx) = tokio::sync::mpsc::channel(16);
    let cancel = CancellationToken::new();
    let ctx = ConsumerContext::new(tx, cancel.clone(), "master-test-route".to_string());

    master.start(ctx).await.unwrap();

    sleep(Duration::from_millis(80)).await;
    assert!(rx.try_recv().is_err());
    assert_eq!(create_consumer_calls.load(Ordering::SeqCst), 0);

    leadership.emit(LeadershipEvent::StartedLeading).await;

    let first = timeout(Duration::from_millis(500), rx.recv())
        .await
        .unwrap()
        .unwrap();
    assert_eq!(first.exchange.input.body.as_text(), Some("epoch-1"));
    assert_eq!(create_consumer_calls.load(Ordering::SeqCst), 1);
    assert_eq!(start_calls.load(Ordering::SeqCst), 1);

    cancel.cancel();
    master.stop().await.unwrap();
}

#[tokio::test]
async fn stops_delegate_on_stopped_leading() {
    let leadership = Arc::new(FakeLeadershipService::new(None));
    let platform_service = Arc::new(FakePlatformService::new(leadership.clone()));
    let create_consumer_calls = Arc::new(AtomicUsize::new(0));
    let start_calls = Arc::new(AtomicUsize::new(0));
    let mut master = build_master_consumer(
        platform_service,
        Arc::clone(&create_consumer_calls),
        Arc::clone(&start_calls),
        Some(30),
    );

    let (tx, mut rx) = tokio::sync::mpsc::channel(16);
    let cancel = CancellationToken::new();
    let ctx = ConsumerContext::new(tx, cancel.clone(), "master-test-route".to_string());

    master.start(ctx).await.unwrap();
    leadership.emit(LeadershipEvent::StartedLeading).await;
    let _ = timeout(Duration::from_millis(500), rx.recv())
        .await
        .unwrap()
        .unwrap();

    leadership.emit(LeadershipEvent::StoppedLeading).await;
    sleep(Duration::from_millis(100)).await;
    while rx.try_recv().is_ok() {}
    assert!(
        timeout(Duration::from_millis(120), rx.recv())
            .await
            .is_err()
    );

    cancel.cancel();
    master.stop().await.unwrap();
}

#[tokio::test]
async fn recreates_delegate_on_new_leadership_epoch() {
    let leadership = Arc::new(FakeLeadershipService::new(None));
    let platform_service = Arc::new(FakePlatformService::new(leadership.clone()));
    let create_consumer_calls = Arc::new(AtomicUsize::new(0));
    let start_calls = Arc::new(AtomicUsize::new(0));
    let mut master = build_master_consumer(
        platform_service,
        Arc::clone(&create_consumer_calls),
        Arc::clone(&start_calls),
        Some(30),
    );

    let (tx, mut rx) = tokio::sync::mpsc::channel(16);
    let cancel = CancellationToken::new();
    let ctx = ConsumerContext::new(tx, cancel.clone(), "master-test-route".to_string());

    master.start(ctx).await.unwrap();

    leadership.emit(LeadershipEvent::StartedLeading).await;
    let first = timeout(Duration::from_millis(500), rx.recv())
        .await
        .unwrap()
        .unwrap();
    assert_eq!(first.exchange.input.body.as_text(), Some("epoch-1"));

    leadership.emit(LeadershipEvent::StoppedLeading).await;
    sleep(Duration::from_millis(120)).await;

    leadership.emit(LeadershipEvent::StartedLeading).await;
    let second = timeout(Duration::from_millis(500), rx.recv())
        .await
        .unwrap()
        .unwrap();
    assert_eq!(second.exchange.input.body.as_text(), Some("epoch-2"));

    assert_eq!(create_consumer_calls.load(Ordering::SeqCst), 2);
    assert_eq!(start_calls.load(Ordering::SeqCst), 2);

    cancel.cancel();
    master.stop().await.unwrap();
}