fluvio-spu 0.2.0

Fluvio Stream Processing Unit
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

use tracing::trace;
use tracing::error;
use tracing::debug;
use tracing::warn;
use fluvio_socket::FlvSocketError;
use fluvio_socket::FlvStream;
use fluvio_socket::FlvSocket;
use fluvio_service::api_loop;
use fluvio_types::SpuId;

use crate::core::DefaultSharedGlobalContext;

use super::LeaderReplicaControllerCommand;
use super::FollowerOffsetUpdate;
use super::LeaderPeerApiEnum;
use super::LeaderPeerRequest;
use super::UpdateOffsetRequest;
use super::ReplicaOffsetRequest;

/// Handle connection from follower to leader
pub struct LeaderConnection {
    ctx: DefaultSharedGlobalContext,
    follower_id: SpuId,
}

impl LeaderConnection {
    /// manage connection from follower
    pub async fn handle(
        ctx: DefaultSharedGlobalContext,
        follower_id: SpuId,
        socket: FlvSocket,
    ) -> Result<(), FlvSocketError> {
        let (sink, stream) = socket.split();
        ctx.follower_sinks().insert_sink(follower_id, sink);

        let connection = Self {
            ctx: ctx.clone(),
            follower_id,
        };

        connection.main_loop(stream).await?;

        ctx.follower_sinks().clear_sink(&follower_id);
        Ok(())
    }

    async fn main_loop(&self, mut stream: FlvStream) -> Result<(), FlvSocketError> {
        trace!(
            "starting connection handling from follower: {} for leader: {}",
            self.follower_id,
            self.ctx.local_spu_id()
        );

        let mut api_stream = stream.api_stream::<LeaderPeerRequest, LeaderPeerApiEnum>();

        api_loop!(
            api_stream,
            LeaderPeerRequest::UpdateOffsets(request) => {
                self.route_offset_request(request.request).await
            }
        );

        Ok(())
    }

    /// route offset update request from follower to replica leader controller
    async fn route_offset_request(&self, request: UpdateOffsetRequest) {
        debug!("receive offset request from follower: {}", self.follower_id);
        for replica in request.replicas {
            route_replica_offset(self.ctx.clone(), self.follower_id, replica).await
        }
    }
}

/// send route replica offsets to leader replica controller
/// it spawn request
async fn route_replica_offset(
    ctx: DefaultSharedGlobalContext,
    follower_id: SpuId,
    replica: ReplicaOffsetRequest,
) {
    let replica_key = replica.replica;
    let follower_update = FollowerOffsetUpdate {
        follower_id,
        leo: replica.leo,
        hw: replica.hw,
    };

    match ctx
        .leaders_state()
        .send_message(
            &replica_key,
            LeaderReplicaControllerCommand::FollowerOffsetUpdate(follower_update),
        )
        .await
    {
        Ok(success_stat) => {
            if success_stat {
                trace!("send offset data to: replica leader: {}", replica_key);
            } else {
                warn!("replica leader: {} was not found", replica_key); // this could happen when leader controller is not happen
            }
        }
        Err(err) => error!(
            "Error writing message to replica {}, err: {}",
            replica_key, err
        ),
    }
}