You signed in with another tab or window. Reload to refresh your session.You signed out in another tab or window. Reload to refresh your session.You switched accounts on another tab or window. Reload to refresh your session.Dismiss alert
## Summary
Add Kafka named connection support and allow Kafka sink configs to reuse
broker connection settings through `connectionSelector`.
## Changes
- Added a stateless Kafka connection implementation.
- Parses Kafka broker, TLS, and SASL settings.
- Uses `kafka-go` dial as the health check.
- Does not hold producer, consumer, or long-lived Kafka network
resources.
- Registers the `kafka` connection provider when the Kafka
implementation package is loaded.
- Updated Kafka sink metadata.
- Added `connectionSelector`.
- Marked Kafka broker, SASL, and TLS-related fields as
connection-related.
- Added test coverage.
- Unit coverage for Kafka connection provisioning and registration.
- Config merge coverage for Kafka sink `connectionSelector`.
- FVT coverage for `memory source -> Kafka sink` using a Kafka named
connection.
- Added Kafka FVT as a standalone test job in the main test workflow.
- Starts Redpanda locally.
- Creates a Kafka named connection.
- Creates a Kafka sink rule using only `connectionSelector + topic`.
- Verifies the message is written to Kafka.
## Verification
- `go test ./extensions/impl/kafka -run 'TestKafkaConnection'`
- `go test ./internal/topo/node/conf -run
TestOverwriteKafkaConnectionProps`
- `go test -trimpath -tags='full deadlock' -count=1 -v ./fvt -run
TestKafkaConnectionSelectorSinkE2E`
- `go test -tags full ./internal/binder/io`
- `python3 -m json.tool extensions/sinks/kafka/kafka.json`
Note: The final workflow-only adjustment was pushed without rerunning
local tests per maintainer request.
---------
Signed-off-by: Song Gao <disxiaofei@163.com>
Copy file name to clipboardExpand all lines: docs/en_US/api/restapi/connection.md
+16-2Lines changed: 16 additions & 2 deletions
Display the source diff
Display the rich diff
Original file line number
Diff line number
Diff line change
@@ -6,7 +6,7 @@ Source/Sink in rules can be created and managed independently in the form of con
6
6
7
7
### Create connection
8
8
9
-
To create a connection, provide the connection's id, type, and configuration parameters. Currently, `mqtt`/`nng`/`httppush`/`websocket`/`edgex`/`sql` type connections are supported. Here we take creating an mqtt connection as an example.
9
+
To create a connection, provide the connection's id, type, and configuration parameters. Currently, `mqtt`/`nng`/`httppush`/`websocket`/`edgex`/`sql`/`kafka` type connections are supported. Here we take creating an mqtt connection as an example.
10
10
11
11
```shell
12
12
POST http://localhost:9081/connections
@@ -19,9 +19,23 @@ POST http://localhost:9081/connections
19
19
}
20
20
```
21
21
22
+
To create a Kafka connection, use `kafka` as the connection type and provide the Kafka connection properties in `props`.
23
+
24
+
```shell
25
+
POST http://localhost:9081/connections
26
+
{
27
+
"id": "kafka-1",
28
+
"typ": "kafka",
29
+
"props": {
30
+
"brokers": "127.0.0.1:9092",
31
+
"saslAuthType": "none"
32
+
}
33
+
}
34
+
```
35
+
22
36
### Update connection
23
37
24
-
To update a connection, provide the connection's id, type, and configuration parameters. Currently, `mqtt`/`nng`/`httppush`/`websocket`/`edgex`/`sql` types of connections are supported. Here we take updating the mqtt connection as an example. If the connection is referenced by a rule, it cannot be updated.
38
+
To update a connection, provide the connection's id, type, and configuration parameters. Currently, `mqtt`/`nng`/`httppush`/`websocket`/`edgex`/`sql`/`kafka` types of connections are supported. Here we take updating the mqtt connection as an example. If the connection is referenced by a rule, it cannot be updated.
25
39
26
40
```shell
27
41
PUT http://localhost:9081/connections/connection-1
| brokers | false | The broker address list ,split with "," |
59
+
| connectionSelector | true | Reuse the selected Kafka connection. When this property is set, Kafka connection-related properties such as `brokers`, SASL, and TLS settings are copied from the selected connection. |
60
+
| brokers | true | The broker address list, split with ",". Required when `connectionSelector` is not set. |
60
61
| topic | false | The topic of the Kafka |
61
62
| saslAuthType | false | The Kafka sasl authType, support none,plain,scram |
62
63
| saslUserName | true | The sasl user name |
@@ -77,6 +78,46 @@ Restart the eKuiper server to activate the plugin.
77
78
78
79
You can check the connectivity of the corresponding sink endpoint in advance through the API: [Connectivity Check](../../../api/restapi/connection.md#connectivity-check)
79
80
81
+
### Connection Reuse
82
+
83
+
You can create a Kafka connection and reuse its connection-related properties in Kafka sinks through
84
+
`connectionSelector`. The Kafka connection is used to ping the configured brokers and manage connection status. The
85
+
Kafka sink copies the selected connection's configuration and creates its own Kafka producer for publishing messages.
86
+
87
+
Create a Kafka connection:
88
+
89
+
```shell
90
+
POST http://localhost:9081/connections
91
+
{
92
+
"id": "kafka-1",
93
+
"typ": "kafka",
94
+
"props": {
95
+
"brokers": "127.0.0.1:9092",
96
+
"saslAuthType": "none"
97
+
}
98
+
}
99
+
```
100
+
101
+
Use the connection in a Kafka sink:
102
+
103
+
```json
104
+
{
105
+
"id": "kafka",
106
+
"sql": "SELECT * FROM demo_stream",
107
+
"actions": [
108
+
{
109
+
"kafka": {
110
+
"connectionSelector": "kafka-1",
111
+
"topic": "test_topic"
112
+
}
113
+
}
114
+
]
115
+
}
116
+
```
117
+
118
+
When `connectionSelector` is set, the sink ignores connection-related properties configured directly in the sink action,
119
+
including `brokers`, `saslAuthType`, `saslUserName`, `password`, `insecureSkipVerify`, and TLS certificate properties.
120
+
80
121
### Setting Kafka Key and Headers
81
122
82
123
Set the metadata when the Kafka client sends messages through keys and headers:
0 commit comments