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
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
use std::sync::Arc;
use async_trait::async_trait;
use dataflow_rs::engine::error::DataflowError;
use dataflow_rs::engine::functions::PublishKafkaConfig;
use dataflow_rs::engine::task_context::TaskContext;
use serde_json::Value;
use super::connector_handler::{ConnectorHandler, Produced};
use super::connector_helpers::{ConnectorCall, require_op};
use super::schema::{FieldKind, FieldSchema};
use crate::connector::ConnectorRegistry;
use crate::engine::HandlerError;
/// Kafka publish handler.
pub struct PublishKafkaHandler {
pub registry: Arc<ConnectorRegistry>,
/// Producer cache keyed by the connector's broker list (F13). `None`
/// when Kafka is disabled.
pub producers: Option<Arc<crate::kafka::producer::KafkaProducerCache>>,
}
#[async_trait]
impl ConnectorHandler for PublishKafkaHandler {
const NAME: &'static str = "publish_kafka";
type Kind = crate::connector::kind::Kafka;
type Input = PublishKafkaConfig;
/// The resolved topic. Since dataflow-rs 3.9 it is a `Template` like the
/// key and the value, so one task can route by message content; it is
/// resolved here rather than in `run` because it names the destination in
/// the "Kafka is not enabled" refusal, which precedes the producer.
type Parsed = String;
fn registry(&self) -> &Arc<ConnectorRegistry> {
&self.registry
}
fn parse(
&self,
_call: &ConnectorCall<'_>,
input: &PublishKafkaConfig,
ctx: &TaskContext<'_>,
) -> Result<Self::Parsed, HandlerError> {
Ok(input.resolve_topic(ctx)?)
}
fn gate(
_parsed: &Self::Parsed,
conn: &crate::connector::KafkaConnectorConfig,
connector: &str,
) -> Result<(), HandlerError> {
// F22e: gate before anything else — whether the deployment has Kafka
// enabled says nothing about whether this connector is allowed to
// publish, and the refusal must be the same either way.
Ok(require_op(conn.operations.publish, "publish", connector)?)
}
async fn run(
&self,
topic: Self::Parsed,
kafka_config: &crate::connector::KafkaConnectorConfig,
call: &ConnectorCall<'_>,
input: &PublishKafkaConfig,
ctx: &mut TaskContext<'_>,
) -> Result<Produced, HandlerError> {
let producers = match &self.producers {
Some(p) => p,
None => {
return Err(DataflowError::FunctionExecution {
context: format!(
"Kafka publishing to topic '{topic}' is not available. Enable Kafka in \
configuration to use publish_kafka."
),
source: None,
}
.into());
}
};
// S6: the broker list is connector data, so judge it against the
// private-address guard before dialling. An empty list means the
// globally configured cluster — operator config, not connector data —
// and is deliberately not re-judged.
if !kafka_config.brokers.is_empty() {
crate::validation::check_broker_endpoints(
call.connector,
&kafka_config.brokers,
kafka_config.allow_private_urls,
)
.await
.map_err(crate::errors::connector_detail_error)?;
}
// F13: publish to the cluster the *connector* names, not the one
// globally configured. Empty brokers keep the previous meaning: the
// global cluster.
let producer = producers
.for_brokers(&kafka_config.brokers)
.await
.map_err(|e| {
DataflowError::function_execution(
format!(
"Failed to create Kafka producer for connector '{}': {e}",
call.connector
),
None,
)
})?;
// `resolve_key` applies the same string coercion this handler used to
// spell out — a JSON string yields its contents, anything else its
// compact form.
let key = input.resolve_key(ctx)?;
// No `value_logic` still means "publish the message data": that is a
// transport decision the config cannot make, so upstream returns `None`
// and the fallback stays here.
let value_json: Value = match input.resolve_value(ctx)? {
Some(value) => value,
None => ctx.data().into(),
};
let value = serde_json::to_string(&value_json).map_err(|e| {
DataflowError::function_execution(
format!("Failed to serialize Kafka message value: {e}"),
None,
)
})?;
producer
.send(&topic, key.as_deref(), value.as_bytes())
.await
.map_err(|e| {
DataflowError::function_execution(
format!("Kafka publish to '{topic}' failed: {e}"),
None,
)
})?;
tracing::debug!(topic = %topic, "Published message to Kafka");
// A publish records nothing at `output`; the send is the whole effect.
Ok(Produced::nothing())
}
}
// -- Input schema (F53) --
//
// The table describing this handler's `function.input` lives next to the
// handler it describes. It used to sit in `schema.rs` with the other nine,
// which is how every schema/handler divergence in the 1.0 audit happened:
// a field was added, renamed or made conditional here and the table saying
// so was in a different file.
pub(super) const PUBLISH_KAFKA_FIELDS: &[FieldSchema] = &[
FieldSchema {
name: "connector",
description: "Name of the Kafka connector to publish through (JSONLogic; a \
computed name is not yet supported).",
kind: FieldKind::String,
required: true,
template_at: &[""],
..FieldSchema::DEFAULT
},
FieldSchema {
name: "topic",
description: "Target topic name (JSONLogic), so one task can route by message \
content.",
kind: FieldKind::String,
required: true,
template_at: &[""],
..FieldSchema::DEFAULT
},
FieldSchema {
name: "key",
description: "Message key (JSONLogic). \
(Was `key_logic`; still accepted, but not alongside `key`.)",
kind: FieldKind::Any,
template_at: &[""],
alias: Some("key_logic"),
..FieldSchema::DEFAULT
},
FieldSchema {
name: "value",
description: "Message value (JSONLogic). Defaults to the message data. \
(Was `value_logic`; still accepted, but not alongside `value`.)",
kind: FieldKind::Any,
template_at: &[""],
alias: Some("value_logic"),
..FieldSchema::DEFAULT
},
];