common_wal/config/kafka/
datanode.rs

1// Copyright 2023 Greptime Team
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use std::time::Duration;
16
17use common_base::readable_size::ReadableSize;
18use serde::{Deserialize, Serialize};
19
20use crate::config::kafka::common::{
21    DEFAULT_TOPIC_LATEST_OFFSET_FETCH_INTERVAL, KafkaConnectionConfig, KafkaTopicConfig,
22};
23
24/// Kafka wal configurations for datanode.
25#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
26#[serde(default)]
27pub struct DatanodeKafkaConfig {
28    /// The kafka connection config.
29    #[serde(flatten)]
30    pub connection: KafkaConnectionConfig,
31    /// TODO(weny): Remove the alias once we release v0.9.
32    /// The max size of a single producer batch.
33    #[serde(alias = "max_batch_size")]
34    pub max_batch_bytes: ReadableSize,
35    /// The consumer wait timeout.
36    #[serde(with = "humantime_serde")]
37    pub consumer_wait_timeout: Duration,
38    /// The kafka topic config.
39    #[serde(flatten)]
40    pub kafka_topic: KafkaTopicConfig,
41    // Automatically create topics for WAL.
42    pub auto_create_topics: bool,
43    // Create index for WAL.
44    pub create_index: bool,
45    #[serde(with = "humantime_serde")]
46    pub dump_index_interval: Duration,
47    /// Internal interval for fetching latest Kafka topic offsets.
48    #[serde(with = "humantime_serde")]
49    pub topic_latest_offset_fetch_interval: Duration,
50    /// Ignore missing entries during read WAL.
51    pub overwrite_entry_start_id: bool,
52}
53
54impl Default for DatanodeKafkaConfig {
55    fn default() -> Self {
56        Self {
57            connection: KafkaConnectionConfig::default(),
58            // Warning: Kafka has a default limit of 1MB per message in a topic.
59            max_batch_bytes: ReadableSize::mb(1),
60            consumer_wait_timeout: Duration::from_millis(100),
61            kafka_topic: KafkaTopicConfig::default(),
62            auto_create_topics: true,
63            create_index: false,
64            dump_index_interval: Duration::from_secs(60),
65            topic_latest_offset_fetch_interval: DEFAULT_TOPIC_LATEST_OFFSET_FETCH_INTERVAL,
66            overwrite_entry_start_id: false,
67        }
68    }
69}
common_wal/config/kafka/datanode.rs

common_wal/config/kafka/
datanode.rs
