risingwave_connector/source/kafka/
split.rs1use risingwave_common::types::JsonbVal;
16use serde::{Deserialize, Serialize};
17
18use crate::error::ConnectorResult;
19use crate::source::{SplitId, SplitMetaData};
20
21#[derive(Clone, Serialize, Deserialize, Debug, PartialEq, Hash)]
22pub struct KafkaSplit {
23 pub(crate) topic: String,
24 pub(crate) partition: i32,
25 pub(crate) start_offset: Option<i64>,
34 pub(crate) stop_offset: Option<i64>,
35}
36
37impl SplitMetaData for KafkaSplit {
38 fn id(&self) -> SplitId {
39 format!("{}", self.partition).into()
41 }
42
43 fn restore_from_json(value: JsonbVal) -> ConnectorResult<Self> {
44 serde_json::from_value(value.take()).map_err(Into::into)
45 }
46
47 fn encode_to_json(&self) -> JsonbVal {
48 serde_json::to_value(self.clone()).unwrap().into()
49 }
50
51 fn update_offset(&mut self, last_seen_offset: String) -> ConnectorResult<()> {
52 self.start_offset = Some(last_seen_offset.as_str().parse::<i64>().unwrap());
53 Ok(())
54 }
55}
56
57impl KafkaSplit {
58 pub fn new(
59 partition: i32,
60 start_offset: Option<i64>,
61 stop_offset: Option<i64>,
62 topic: String,
63 ) -> KafkaSplit {
64 KafkaSplit {
65 topic,
66 partition,
67 start_offset,
68 stop_offset,
69 }
70 }
71
72 pub fn get_topic_and_partition(&self) -> (String, i32) {
73 (self.topic.clone(), self.partition)
74 }
75}