Connectors, Transforms & SMTs

Name: StreamFlow
Author: StreamFlow

Build custom source connectors, sink connectors, and single message transforms for StreamFlow.

1.Create a Source Connector
2.Create a Sink Connector
3.Create a Transformation (SMT)
4.Chain Transforms
5.Configure Connectors
6.Deploy a Custom Connector
7.Built-in Connectors

1. Create a Source Connector

A source connector captures data from an external system and sends it into StreamFlow.

Complete Example: Weather API Connector

This connector calls a REST API every 30 seconds and publishes weather data into StreamFlow.

Java

1package com.example.connector;
2
3import com.streamflow.connect.source.sdk.EventCollector;
4import com.streamflow.connect.source.sdk.SourceConnectorPlugin;
5import java.net.URI;
6import java.net.http.*;
7import java.nio.charset.StandardCharsets;
8import java.util.Map;
9import java.util.concurrent.*;
10
11/**
12 * Source connector that polls a weather API and publishes
13 * readings to a StreamFlow topic.
14 *
15 * Configuration (in connect-source.properties):
16 *   source.meteo.type=weather-api
17 *   source.meteo.topic=weather-readings
18 *   source.meteo.api.url=https://api.meteo.example.com/current
19 *   source.meteo.api.key=${WEATHER_API_KEY}
20 *   source.meteo.cities=Paris,Lyon,Marseille
21 *   source.meteo.poll.interval.seconds=30
22 */
23public class WeatherSourceConnector implements SourceConnectorPlugin {
24
25    private String apiUrl;
26    private String apiKey;
27    private String[] cities;
28    private int pollIntervalSeconds;
29    private String topic;
30
31    private HttpClient httpClient;
32    private ScheduledExecutorService scheduler;
33    private EventCollector collector;
34    private volatile boolean healthy = true;
35
36    // ── Connector identity ──
37
38    @Override
39    public String id() {
40        return "weather-api";  // used in source.<name>.type=weather-api
41    }
42
43    @Override
44    public String name() {
45        return "Weather API Source Connector";
46    }
47
48    // ── Configuration ──
49    // The "source.<name>." prefix is automatically stripped.
50    // You receive: {"api.url": "...", "api.key": "...", "cities": "..."}
51
52    @Override
53    public void configure(Map<String, String> config) {
54        this.apiUrl = config.getOrDefault("api.url",
55            "https://api.meteo.example.com/current");
56        this.apiKey = config.getOrDefault("api.key", "");
57        this.topic = config.getOrDefault("topic", "weather-readings");
58        this.cities = config.getOrDefault("cities", "Paris").split(",");
59        this.pollIntervalSeconds = Integer.parseInt(
60            config.getOrDefault("poll.interval.seconds", "30"));
61    }
62
63    // ── Startup ──
64    // The `collector` is provided by StreamFlow. Call collector.collect()
65    // to send events to the topic.
66
67    @Override
68    public void start(EventCollector collector) {
69        this.collector = collector;
70        this.httpClient = HttpClient.newHttpClient();
71        this.scheduler = Executors.newSingleThreadScheduledExecutor();
72
73        // Poll the API at regular intervals
74        scheduler.scheduleAtFixedRate(
75            this::pollWeather,
76            0,                          // start immediately
77            pollIntervalSeconds,        // then every N seconds
78            TimeUnit.SECONDS
79        );
80
81        System.out.printf("[%s] Started: %d cities, interval %ds%n",
82            name(), cities.length, pollIntervalSeconds);
83    }
84
85    private void pollWeather() {
86        for (String city : cities) {
87            try {
88                // 1. Call the API
89                HttpRequest request = HttpRequest.newBuilder()
90                    .uri(URI.create(apiUrl + "?city=" + city.trim()
91                        + "&key=" + apiKey))
92                    .GET()
93                    .build();
94
95                HttpResponse<String> response = httpClient.send(
96                    request, HttpResponse.BodyHandlers.ofString());
97
98                if (response.statusCode() == 200) {
99                    // 2. Send to StreamFlow
100                    collector.collect(
101                        city.trim(),                                      // key
102                        response.body().getBytes(StandardCharsets.UTF_8),  // payload
103                        Map.of(                                           // headers
104                            "source", "weather-api",
105                            "city", city.trim(),
106                            "timestamp", String.valueOf(
107                                System.currentTimeMillis())
108                        )
109                    );
110                }
111
112            } catch (Exception e) {
113                System.err.printf("[%s] Error for %s: %s%n",
114                    name(), city, e.getMessage());
115                healthy = false;
116            }
117        }
118
119        // 3. Force flush (otherwise events accumulate in the batch)
120        collector.flush();
121    }
122
123    // ── Health check (called by StreamFlow for monitoring) ──
124
125    @Override
126    public boolean isHealthy() {
127        return healthy;
128    }
129
130    // ── Graceful shutdown ──
131
132    @Override
133    public void stop() {
134        if (scheduler != null) {
135            scheduler.shutdown();
136            try {
137                scheduler.awaitTermination(5, TimeUnit.SECONDS);
138            } catch (InterruptedException e) {
139                Thread.currentThread().interrupt();
140            }
141        }
142    }
143
144    @Override
145    public void close() {
146        stop();
147    }
148}

SPI Registration

Create the file META-INF/services/com.streamflow.connect.source.sdk.SourceConnectorPlugin:

1com.example.connector.WeatherSourceConnector

Maven Structure

XML

1<!-- pom.xml -->
2<project>
3    <groupId>com.example</groupId>
4    <artifactId>weather-source-connector</artifactId>
5    <version>1.0.0</version>
6    <packaging>jar</packaging>
7
8    <dependencies>
9        <dependency>
10            <groupId>com.streamflow</groupId>
11            <artifactId>streamflow-connect</artifactId>
12            <version>0.1.0-SNAPSHOT</version>
13            <scope>provided</scope>  <!-- provided by StreamFlow at runtime -->
14        </dependency>
15    </dependencies>
16</project>

Project Structure

weather-source-connector/
├── pom.xml
└── src/main/
    ├── java/com/example/connector/
    │   └── WeatherSourceConnector.java
    └── resources/META-INF/services/
        └── com.streamflow.connect.source.sdk.SourceConnectorPlugin

2. Create a Sink Connector

A sink connector receives events from StreamFlow and sends them to an external system.

Complete Example: Slack Connector (notifications)

Java

1package com.example.connector;
2
3import com.streamflow.connect.connector.*;
4import com.streamflow.connect.config.ConnectorConfig;
5import com.streamflow.connect.sink.*;
6import java.net.URI;
7import java.net.http.*;
8import java.nio.charset.StandardCharsets;
9import java.util.List;
10import java.util.Map;
11
12/**
13 * Sink connector that sends events to a Slack channel.
14 *
15 * Configuration:
16 *   sink.slack.type=slack
17 *   sink.slack.webhook.url=${SLACK_WEBHOOK_URL}
18 *   sink.slack.channel=#alerts
19 *   sink.slack.username=StreamFlow Bot
20 */
21public class SlackSinkConnector implements SinkConnector {
22
23    private String webhookUrl;
24    private String channel;
25    private String username;
26    private HttpClient httpClient;
27    private ConnectorStatus status =
28        new ConnectorStatus(ConnectorStatus.State.STOPPED, null);
29
30    @Override
31    public String connectorId() {
32        return "slack-sink";
33    }
34
35    @Override
36    public ConnectorType type() {
37        return ConnectorType.SINK;
38    }
39
40    @Override
41    public void initialize(ConnectorConfig config) {
42        this.webhookUrl = config.getString("webhook.url");
43        this.channel = config.getString("channel", "#general");
44        this.username = config.getString("username", "StreamFlow");
45    }
46
47    @Override
48    public void start() {
49        this.httpClient = HttpClient.newHttpClient();
50        this.status = new ConnectorStatus(
51            ConnectorStatus.State.RUNNING, null);
52        System.out.printf("[Slack Sink] Started → channel %s%n", channel);
53    }
54
55    // ── Receiving events ──
56    // StreamFlow calls put() with a batch of records
57
58    @Override
59    public void put(List<SinkRecord> records) {
60        for (SinkRecord record : records) {
61            try {
62                String message = new String(
63                    record.value(), StandardCharsets.UTF_8);
64
65                // Build Slack payload
66                String slackPayload = """
67                    {
68                      "channel": "%s",
69                      "username": "%s",
70                      "text": "*[%s]* key=`%s`\n```%s```"
71                    }
72                    """.formatted(channel, username,
73                        record.topic(), record.key(), message);
74
75                // Send via webhook
76                HttpRequest request = HttpRequest.newBuilder()
77                    .uri(URI.create(webhookUrl))
78                    .header("Content-Type", "application/json")
79                    .POST(HttpRequest.BodyPublishers.ofString(slackPayload))
80                    .build();
81
82                httpClient.send(request,
83                    HttpResponse.BodyHandlers.ofString());
84
85            } catch (Exception e) {
86                System.err.printf("[Slack Sink] Error: %s%n",
87                    e.getMessage());
88            }
89        }
90    }
91
92    @Override
93    public void flush(Map<String, SinkOffset> offsets) {
94        // Slack is fire-and-forget, no checkpoint needed
95    }
96
97    @Override
98    public SinkOffset currentOffset(String topic, int partition) {
99        return null;  // no offset resume for Slack
100    }
101
102    @Override
103    public ConnectorStatus status() {
104        return status;
105    }
106
107    @Override
108    public ConnectorMetadata metadata() {
109        return new ConnectorMetadata(
110            "slack-sink", "Slack Sink Connector",
111            "Sends events to a Slack channel via webhook",
112            "1.0.0", "My Company", ConnectorType.SINK,
113            List.of("json"), List.of()
114        );
115    }
116
117    @Override
118    public void stop() {
119        status = new ConnectorStatus(
120            ConnectorStatus.State.STOPPED, null);
121    }
122
123    @Override
124    public void close() {
125        stop();
126    }
127}

SPI Registration

META-INF/services/com.streamflow.connect.connector.Connector:

1com.example.connector.SlackSinkConnector

3. Create a Transformation (SMT)

An SMT (Single Message Transform) modifies or filters each event individually.

Example 1: Add an ISO Timestamp

Java

1package com.example.transform;
2
3import com.streamflow.connect.source.sdk.MessageTransform;
4import java.time.Instant;
5import java.time.ZoneId;
6import java.time.format.DateTimeFormatter;
7import java.util.*;
8
9/**
10 * Adds a "timestamp-iso" header with the ISO-8601 formatted date.
11 *
12 * Configuration:
13 *   source.my-source.transforms=add-timestamp
14 *   source.my-source.transforms.add-timestamp.type=timestamp-iso
15 *   source.my-source.transforms.add-timestamp.timezone=Europe/Paris
16 */
17public class TimestampIsoTransform implements MessageTransform {
18
19    private DateTimeFormatter formatter;
20
21    @Override
22    public String type() {
23        return "timestamp-iso";
24    }
25
26    @Override
27    public void configure(Map<String, String> config) {
28        String timezone = config.getOrDefault("timezone", "UTC");
29        this.formatter = DateTimeFormatter.ISO_OFFSET_DATE_TIME
30            .withZone(ZoneId.of(timezone));
31    }
32
33    @Override
34    public TransformResult apply(
35            String key, byte[] payload, Map<String, String> headers) {
36        // Add header without modifying the payload
37        Map<String, String> newHeaders = new HashMap<>(headers);
38        newHeaders.put("timestamp-iso", formatter.format(Instant.now()));
39        return new TransformResult(key, payload, newHeaders);
40    }
41
42    @Override
43    public void close() {}
44}

Example 2: Filter by Amount (drop if < threshold)

Java

1package com.example.transform;
2
3import com.streamflow.connect.source.sdk.MessageTransform;
4import java.nio.charset.StandardCharsets;
5import java.util.Map;
6
7/**
8 * Filters events whose amount is below the threshold.
9 * Returning null = the event is dropped.
10 *
11 * Configuration:
12 *   source.orders.transforms=min-amount
13 *   source.orders.transforms.min-amount.type=min-amount-filter
14 *   source.orders.transforms.min-amount.field=amount
15 *   source.orders.transforms.min-amount.threshold=100.0
16 */
17public class MinAmountFilter implements MessageTransform {
18
19    private String field;
20    private double threshold;
21
22    @Override
23    public String type() {
24        return "min-amount-filter";
25    }
26
27    @Override
28    public void configure(Map<String, String> config) {
29        this.field = config.getOrDefault("field", "amount");
30        this.threshold = Double.parseDouble(
31            config.getOrDefault("threshold", "0"));
32    }
33
34    @Override
35    public TransformResult apply(
36            String key, byte[] payload, Map<String, String> headers) {
37        String json = new String(payload, StandardCharsets.UTF_8);
38
39        // Simple amount extraction (use Jackson in production)
40        String search = "\"" + field + "\":";
41        int idx = json.indexOf(search);
42        if (idx >= 0) {
43            String numStr = json.substring(
44                idx + search.length()).trim();
45            StringBuilder sb = new StringBuilder();
46            for (char c : numStr.toCharArray()) {
47                if (Character.isDigit(c) || c == '.') sb.append(c);
48                else break;
49            }
50            double amount = Double.parseDouble(sb.toString());
51
52            if (amount < threshold) {
53                return null;  // ← DROP: event is removed
54            }
55        }
56
57        return new TransformResult(key, payload, headers);  // ← KEEP
58    }
59
60    @Override
61    public void close() {}
62}

Example 3: Mask Sensitive Fields (PII)

Masks sensitive fields in JSON. "email": "alice@example.com" → "email": "***". This transform is already built-in (MaskFieldsTransform), see the Built-in Transforms section below.

Properties

1source.users.transforms=mask-pii
2source.users.transforms.mask-pii.type=mask-fields
3source.users.transforms.mask-pii.fields=email,phone,ssn
4source.users.transforms.mask-pii.replacement=***

4. Chain Transforms

Transforms are applied in declaration order. If a transform returns null, the event is dropped and subsequent transforms are not called.

Configuration

Properties

1# connect-source.properties
2
3# Source connector
4source.orders.type=http
5source.orders.topic=orders-clean
6source.orders.url=https://api.shop.com/orders
7
8# Chain of 3 transforms (in order)
9source.orders.transforms=filter-cancelled,add-region,mask-card
10
11# Transform 1: Drop cancelled orders
12source.orders.transforms.filter-cancelled.type=filter
13source.orders.transforms.filter-cancelled.not.contains=CANCELLED
14
15# Transform 2: Add a "region" header
16source.orders.transforms.add-region.type=header-enrich
17source.orders.transforms.add-region.header.region=eu-west
18source.orders.transforms.add-region.header.env=production
19
20# Transform 3: Mask credit card number
21source.orders.transforms.mask-card.type=mask-fields
22source.orders.transforms.mask-card.fields=credit_card,cvv
23source.orders.transforms.mask-card.replacement=****

Data Flow

API HTTP → {"status":"ACTIVE","credit_card":"4111-1111-1111-1111","amount":150}
                                    ↓
           Transform 1 (filter)     → KEEP (not "CANCELLED")
                                    ↓
           Transform 2 (enrich)     → headers += {region: eu-west, env: production}
                                    ↓
           Transform 3 (mask)       → {"status":"ACTIVE","credit_card":"****","amount":150}
                                    ↓
           Topic "orders-clean"     → clean, secure, enriched event

5. Configure Connectors

Configuration Format

Properties

1# connect-source.properties
2
3# ── Pattern: source.<name>.<property> ──
4
5# MQTT
6source.iot-sensors.type=mqtt
7source.iot-sensors.topic=sensor-data
8source.iot-sensors.broker=tcp://mqtt.prod.internal:1883
9source.iot-sensors.subscribe=sensors/#
10source.iot-sensors.qos=1
11source.iot-sensors.username=${MQTT_USER}
12source.iot-sensors.password=${MQTT_PASSWORD}
13
14# PostgreSQL CDC (change data capture)
15source.orders-cdc.type=cdc
16source.orders-cdc.topic=cdc-orders
17source.orders-cdc.database.type=postgresql
18source.orders-cdc.database.hostname=db.prod.internal
19source.orders-cdc.database.port=5432
20source.orders-cdc.database.name=shop
21source.orders-cdc.database.user=cdc_user
22source.orders-cdc.database.password=${DB_PASSWORD}
23source.orders-cdc.tables=public.orders,public.order_items
24source.orders-cdc.poll.interval.ms=1000
25
26# HTTP webhook
27source.payments.type=http
28source.payments.topic=payment-events
29source.payments.url=https://payments.example.com/webhook
30source.payments.method=GET
31source.payments.poll.interval.ms=5000
32source.payments.headers.Authorization=Bearer ${PAYMENT_API_KEY}
33
34# AMQP (RabbitMQ)
35source.notifications.type=amqp
36source.notifications.topic=notifications
37source.notifications.host=rabbitmq.internal
38source.notifications.port=5672
39source.notifications.queue=notification-queue
40source.notifications.username=${RABBIT_USER}
41source.notifications.password=${RABBIT_PASSWORD}

Environment Variables

${VAR_NAME} values are replaced by environment variables at startup:

bash

1export MQTT_USER=admin
2export MQTT_PASSWORD=secret123
3export DB_PASSWORD=pg_secure_pass
4export PAYMENT_API_KEY=pk_live_abc123

6. Deploy a Custom Connector

Option A: Copy JAR to connectors/

bash

1# 1. Build the connector
2cd weather-source-connector/
3mvn package
4
5# 2. Copy to StreamFlow plugins directory
6cp target/weather-source-connector-1.0.0.jar /opt/streamflow/connectors/
7
8# 3. Configure
9cat >> /opt/streamflow/conf/connect-source.properties << 'EOF'
10source.meteo-paris.type=weather-api
11source.meteo-paris.topic=weather
12source.meteo-paris.api.url=https://api.meteo.example.com/current
13source.meteo-paris.api.key=${WEATHER_API_KEY}
14source.meteo-paris.cities=Paris,Lyon,Marseille
15source.meteo-paris.poll.interval.seconds=60
16EOF
17
18# 4. Restart StreamFlow (connector is auto-discovered)
19pkill -f StreamFlowApplication
20./scripts/start-streamflow.sh

StreamFlow scans the connectors/ directory at startup and discovers plugins via ServiceLoader.

Option B: Docker (multi-stage)

bash

1# Dockerfile
2FROM streamflow/streamflow:latest
3
4# Add custom connector
5COPY weather-source-connector-1.0.0.jar /opt/streamflow/connectors/
6
7# Add config
8COPY connect-source.properties /opt/streamflow/conf/connect-source.properties

bash

1# docker-compose.yml
2services:
3  streamflow:
4    build: .
5    ports:
6      - "9092:9092"
7      - "8080:8080"
8    environment:
9      WEATHER_API_KEY: "my-api-key"

7. Built-in Connectors

StreamFlow includes 19 connectors ready to use:

Sources (6)

Type	type=	Description
Kafka	kafka	Consumes from an external Kafka cluster
JDBC	jdbc	Polls a database (periodic SELECT)
HTTP	http	Polls a REST API / receives webhooks
CDC	cdc	Change Data Capture (PostgreSQL, MySQL)
MQTT	mqtt	Subscribes to an MQTT broker (IoT)
AMQP	amqp	Consumes from a RabbitMQ queue

Sinks (13)

Type	type=	Description
Kafka	kafka	Produces to an external Kafka cluster
JDBC	jdbc	INSERT/UPSERT into a SQL database
HTTP	http	POST to a REST API
Elasticsearch	elasticsearch	Indexes into Elasticsearch
MongoDB	mongodb	Inserts into MongoDB
Redis	redis	SET into Redis
Datadog	datadog	Sends Datadog metrics
Databricks	databricks	Loads into Databricks Delta
Snowflake	snowflake	Loads into Snowflake
BigQuery	bigquery	Loads into Google BigQuery
RabbitMQ	rabbitmq	Publishes to RabbitMQ
MQTT	mqtt	Publishes to an MQTT broker
AMQP	amqp	Publishes to an AMQP broker

Built-in Transforms (3)

Type	type=	Description
filter	filter	Keep/drop based on content (contains, not.contains)
header-enrich	header-enrich	Adds static headers (header.key=value)
mask-fields	mask-fields	Masks JSON fields (fields=ssn,card, replacement=***)

Summary: Interfaces to Implement

I want to...	Interface	Key Methods
Capture data	SourceConnectorPlugin	configure(), start(collector), stop()
Send data	SinkConnector	initialize(), start(), put(records), flush()
Transform an event	MessageTransform	configure(), apply(key, payload, headers)
Filter an event	MessageTransform	apply() → return null to drop

Each connector/transform is a JAR with an SPI file in META-INF/services/. Drop the JAR in connectors/ and StreamFlow discovers it automatically.

Connectors, Transforms & SMTs

Table of Contents

1. Create a Source Connector

Complete Example: Weather API Connector

SPI Registration

Maven Structure

Project Structure

2. Create a Sink Connector

Complete Example: Slack Connector (notifications)

SPI Registration

3. Create a Transformation (SMT)

Example 1: Add an ISO Timestamp

Example 2: Filter by Amount (drop if < threshold)

Example 3: Mask Sensitive Fields (PII)

4. Chain Transforms

Configuration

Data Flow

5. Configure Connectors

Configuration Format

Environment Variables

6. Deploy a Custom Connector

Option A: Copy JAR to connectors/

Option B: Docker (multi-stage)

7. Built-in Connectors

Sources (6)

Sinks (13)

Built-in Transforms (3)

Summary: Interfaces to Implement