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
use std::sync::Arc;
use async_trait::async_trait;
use dataflow_rs::engine::error::DataflowError;
use dataflow_rs::engine::functions::AsyncFunctionHandler;
use dataflow_rs::engine::functions::PublishKafkaConfig;
use dataflow_rs::engine::task_context::TaskContext;
use dataflow_rs::engine::task_outcome::TaskOutcome;
use serde_json::Value;
use super::schema::{FieldKind, FieldSchema};
use crate::connector::ConnectorRegistry;
/// This handler's name in metrics, profiles and error messages (F48).
const NAME: &str = "publish_kafka";
/// 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 AsyncFunctionHandler for PublishKafkaHandler {
type Input = PublishKafkaConfig;
async fn execute(
&self,
ctx: &mut TaskContext<'_>,
input: &PublishKafkaConfig,
) -> dataflow_rs::Result<TaskOutcome> {
// F40: read the channel before the body borrows `ctx` mutably.
let channel = super::extract_channel(ctx.message()).to_string();
super::connector_helpers::guarded_handler(
NAME,
&self.registry,
&input.connector,
&channel,
async move {
let connector =
super::connector_helpers::resolve_connector(&self.registry, &input.connector)
.await?;
let kafka_config = super::connector_helpers::require_kafka_connector(
connector.as_ref(),
&input.connector,
)?;
// 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.
super::connector_helpers::require_op(
kafka_config.operations.publish,
"publish",
&input.connector,
)?;
let producers = match &self.producers {
Some(p) => p,
None => {
return Err(DataflowError::FunctionExecution {
context: format!(
"Kafka publishing to topic '{}' is not available. \
Enable Kafka in configuration to use publish_kafka.",
input.topic
),
source: None,
});
}
};
// 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(
&input.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}",
input.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(&input.topic, key.as_deref(), value.as_bytes())
.await
.map_err(|e| {
DataflowError::function_execution(
format!("Kafka publish to '{}' failed: {e}", input.topic),
None,
)
})?;
tracing::debug!(
topic = %input.topic,
"Published message to Kafka"
);
Ok(TaskOutcome::Success)
},
)
.await
}
}
// -- 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.",
kind: FieldKind::String,
required: true,
resolvable: false,
alias: None,
},
FieldSchema {
name: "topic",
description: "Target topic name.",
kind: FieldKind::String,
required: true,
resolvable: false,
alias: None,
},
FieldSchema {
name: "key_logic",
description: "JSONLogic expression to derive the message key.",
kind: FieldKind::Any,
required: false,
resolvable: false,
alias: None,
},
FieldSchema {
name: "value_logic",
description: "JSONLogic expression to derive the message value.",
kind: FieldKind::Any,
required: false,
resolvable: false,
alias: None,
},
];