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
153
154
155
156
157
158
159
160
161
162
163
164
165
166
//! Public `TopologyDeclarer<B>` + `Topics` tuple trait.
use crate::backend::{Backend, TopologyImpl};
use crate::error::Result;
#[cfg(feature = "kafka")]
use crate::markers::Kafka;
use crate::topic::Topic;
#[cfg(feature = "kafka")]
use crate::topology::KafkaCleanupPolicy;
#[cfg(feature = "kafka")]
use std::time::Duration;
#[must_use]
pub struct TopologyDeclarer<B: Backend> {
pub(crate) inner: B::TopologyImpl,
}
impl<B: Backend> TopologyDeclarer<B> {
pub(crate) fn new(inner: B::TopologyImpl) -> Self {
Self { inner }
}
pub async fn declare<T: Topic>(&self) -> Result<()> {
self.inner.declare::<T>().await
}
pub async fn declare_all<Ts: Topics>(&self) -> Result<()> {
Ts::declare_all(self).await
}
}
#[cfg(feature = "kafka")]
#[cfg_attr(docsrs, doc(cfg(feature = "kafka")))]
impl TopologyDeclarer<Kafka> {
/// Replication factor applied to every topic this declarer creates (main +
/// DLQ). Defaults to `1` (single-broker dev) when unset; production
/// clusters should set `>= 3` (or whatever the cluster's quorum demands).
/// `declare` is idempotent and will not alter the replication of an
/// already-existing topic.
///
/// # Panics
///
/// Panics if `n < 1`.
pub fn with_replication_factor(mut self, n: i32) -> Self {
self.inner = self.inner.with_replication_factor(n);
self
}
/// Partition floor for the topics this declarer creates: each topic gets
/// `max(topology_default, n)` partitions so Kafka can spread load across at
/// least `n` consumers. `declare` only ever expands partition counts,
/// never shrinks them.
pub fn with_min_partitions(mut self, n: i32) -> Self {
self.inner = self.inner.with_min_partitions(n);
self
}
/// Kafka topic-level config entry (e.g. `retention.ms`) applied to every
/// **main** topic this declarer creates or reconciles. Repeatable; later
/// calls for the same key win. Per-topic entries set via
/// [`TopologyBuilder::with_topic_config`](crate::topology::TopologyBuilder::with_topic_config)
/// override these key-by-key. DLQ topics keep cluster defaults.
///
/// On an already-existing topic, `declare` compares the declared keys
/// against the live values and issues an alter when they drift, preserving
/// the topic's other dynamic config entries.
///
/// ```ignore
/// broker.topology()
/// .with_topic_config("retention.ms", "604800000")
/// .declare::<IngestionTopic>()
/// .await?;
/// ```
pub fn with_topic_config(mut self, key: impl Into<String>, value: impl Into<String>) -> Self {
self.inner = self.inner.with_topic_config(key, value);
self
}
/// Sets `retention.ms` from a [`Duration`]. Sugar for
/// [`with_topic_config`](Self::with_topic_config); the same scope and
/// reconcile semantics apply. Sub-millisecond precision is truncated.
///
/// # Panics
///
/// Panics if combined with
/// [`with_retention_forever`](Self::with_retention_forever).
pub fn with_retention(mut self, retention: Duration) -> Self {
self.inner = self.inner.with_retention(retention);
self
}
/// Sets `retention.ms = -1`, retaining messages forever (typically paired
/// with [`with_cleanup_policy`](Self::with_cleanup_policy) and
/// [`KafkaCleanupPolicy::Compact`]). Sugar for
/// [`with_topic_config`](Self::with_topic_config).
///
/// # Panics
///
/// Panics if combined with [`with_retention`](Self::with_retention).
pub fn with_retention_forever(mut self) -> Self {
self.inner = self.inner.with_retention_forever();
self
}
/// Sets `retention.bytes`, the maximum partition size before old segments
/// are discarded. Sugar for
/// [`with_topic_config`](Self::with_topic_config).
pub fn with_retention_bytes(self, bytes: u64) -> Self {
self.with_topic_config("retention.bytes", bytes.to_string())
}
/// Sets `cleanup.policy`. Sugar for
/// [`with_topic_config`](Self::with_topic_config).
pub fn with_cleanup_policy(self, policy: KafkaCleanupPolicy) -> Self {
self.with_topic_config("cleanup.policy", policy.as_str())
}
/// Sets `max.message.bytes`, the largest record batch the topic accepts.
/// Sugar for [`with_topic_config`](Self::with_topic_config).
pub fn with_max_message_bytes(self, bytes: u32) -> Self {
self.with_topic_config("max.message.bytes", bytes.to_string())
}
}
/// Multi-topic declaration via tuples. Arities 1 through 16.
pub trait Topics: Sized {
fn declare_all<B: Backend>(
declarer: &TopologyDeclarer<B>,
) -> impl Future<Output = Result<()>> + Send;
}
macro_rules! impl_topics_for_tuple {
($( ($($T:ident),+) ),+ $(,)?) => {
$(
impl<$($T: Topic),+> Topics for ($($T,)+) {
async fn declare_all<B: Backend>(d: &TopologyDeclarer<B>) -> Result<()> {
$( d.declare::<$T>().await?; )+
Ok(())
}
}
)+
};
}
impl_topics_for_tuple!(
(T1),
(T1, T2),
(T1, T2, T3),
(T1, T2, T3, T4),
(T1, T2, T3, T4, T5),
(T1, T2, T3, T4, T5, T6),
(T1, T2, T3, T4, T5, T6, T7),
(T1, T2, T3, T4, T5, T6, T7, T8),
(T1, T2, T3, T4, T5, T6, T7, T8, T9),
(T1, T2, T3, T4, T5, T6, T7, T8, T9, T10),
(T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11),
(T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12),
(T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13),
(T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14),
(
T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15
),
(
T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16
),
);