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
110
111
112
113
114
115
116
117

use std::sync::Arc;

use futures::future::BoxFuture;
use futures::FutureExt;

use flv_future_core::test_async;
use fluvio_socket::FlvSocketError;
use fluvio_controlplane_metadata::partition::ReplicaKey;
use kf_protocol::api::RequestMessage;
use fluvio_spu_schema::offsets::FlvFetchOffsetsRequest;
use fluvio_spu_schema::offsets::FetchOffsetTopic;
use fluvio_spu_schema::offsets::FetchOffsetPartition;
use kf_protocol::api::FlvErrorCode;

use crate::tests::fixture::TestGenerator;
use crate::tests::fixture::SpuTest;
use crate::tests::fixture::SpuTestRunner;

const TOPIC_ID: &str = "topic1";
const PARTITION_ID: i32 = 0;

fn test_repl_id() -> ReplicaKey {
    ReplicaKey::new("topic1", 0)
}

struct OffsetsFetchTest {}

async fn test_fetch(runner: Arc<SpuTestRunner<OffsetsFetchTest>>) -> Result<(), FlvSocketError> {
    // verify invalid request
    let mut request = FlvFetchOffsetsRequest::default();
    let mut topic = FetchOffsetTopic::default();
    topic.name = "dummy".into();
    let mut partition = FetchOffsetPartition::default();
    partition.partition_index = PARTITION_ID;
    topic.partitions.push(partition);
    request.topics.push(topic);
    let req_msg = RequestMessage::new_request(request).set_client_id("offset fetch tester");
    let produce_resp = runner
        .leader()
        .send_to_public_server(&req_msg)
        .await
        .expect("offset fetch");
    let topic_responses = produce_resp.response.topics;
    assert_eq!(topic_responses.len(), 1);
    let topic_response = &topic_responses[0];
    assert_eq!(topic_response.name, "dummy");
    let partition_responses = &topic_response.partitions;
    assert_eq!(partition_responses.len(), 1);
    let partition = &partition_responses[0];
    assert_eq!(partition.error_code, FlvErrorCode::PartitionNotLeader);

    let produce_req_msg = runner.create_producer_msg("message", TOPIC_ID, PARTITION_ID);
    let _response = runner
        .leader()
        .send_to_public_server(&produce_req_msg)
        .await
        .expect("producer must not fail");

    let mut request = FlvFetchOffsetsRequest::default();
    let mut topic = FetchOffsetTopic::default();
    topic.name = TOPIC_ID.into();
    let mut partition = FetchOffsetPartition::default();
    partition.partition_index = PARTITION_ID;
    topic.partitions.push(partition);
    request.topics.push(topic);
    let req_msg = RequestMessage::new_request(request).set_client_id("offset fetch tester");
    let produce_resp = runner
        .leader()
        .send_to_public_server(&req_msg)
        .await
        .expect("offset fetch");
    let topic_responses = produce_resp.response.topics;
    assert_eq!(topic_responses.len(), 1);
    let topic_response = &topic_responses[0];
    assert_eq!(topic_response.name, TOPIC_ID);
    let partition_responses = &topic_response.partitions;
    assert_eq!(partition_responses.len(), 1);
    let partition = &partition_responses[0];
    assert_eq!(partition.error_code, FlvErrorCode::None);
    assert_eq!(partition.last_stable_offset, 1);
    assert_eq!(partition.start_offset, 0);

    Ok(())
}

impl SpuTest for OffsetsFetchTest {
    type ResponseFuture = BoxFuture<'static, Result<(), FlvSocketError>>;

    fn env_configuration(&self) -> TestGenerator {
        TestGenerator::default()
            .set_base_id(6100)
            .set_base_port(9900)
            .set_base_dir("offset_fetch_test")
            .init()
    }

    fn followers(&self) -> usize {
        0
    }

    fn replicas(&self) -> Vec<ReplicaKey> {
        vec![test_repl_id()]
    }

    fn main_test(&self, runner: Arc<SpuTestRunner<OffsetsFetchTest>>) -> Self::ResponseFuture {
        async move { test_fetch(runner).await }.boxed()
    }
}

#[test_async]
async fn flv_offset_fetch_test() -> Result<(), FlvSocketError> {
    let test = OffsetsFetchTest {};
    SpuTestRunner::run("offset fetch test".to_owned(), test)
        .await
        .expect("test runner should not failer");
    Ok(())
}