Connector

In short

The connector role collects data that workspaces set up under Settings → Data sources:

• it reads Kafka topics institutions publish to (see Kafka topics);
• it fetches files from SFTP servers and S3-compatible buckets (see File drops);
• it follows database changes: Debezium change events on Kafka topics, and Postgres databases through logical replication (see Change data capture).

It loads everything in process, through the same ingest service as the ingest role: the same checks, duplicate protection, currency conversion, encryption, publishing and per-row results.

It is a role of its own, not part of ingest, for two reasons. ingest is stateless HTTP behind a load balancer, while the connector is a Kafka consumer-group member and a scheduler. And the connector needs outbound access to hosts that workspace administrators choose, which the nodes that face the internet shouldn’t have.

--role=all includes connector. Without a running connector, data sources can be set up but nothing is collected.

What it needs

Needs	Why
Postgres (system role)	Reads every workspace’s data sources and records each file and its results, and the change data capture checkpoints and ledger
Kafka (SpeculaPro’s cluster)	Publishes transactions and account events, like ingest, and checks how far behind the persister and stream evaluator are
Redis	The same idempotency and feature updates as ingest
The encryption keys	Decrypts SFTP keys and S3 secrets, and encrypts personal data in what it loads
Outbound network access	SFTP (port 22, or the port a data source sets), S3 endpoints (443), the institution Kafka cluster if it isn’t SpeculaPro’s own (9092, 9093), schema registries for Debezium Avro (the registry’s port, often 443 or 8081), and institution Postgres databases for replication sources (5432, or their port)
Local disk	Each file is staged on disk while it loads, plus its results file: SP_CONNECTOR_SPOOL_DIR needs room for the largest file times SP_CONNECTOR_CONCURRENCY

Settings

All settings are in the configuration reference. The ones that matter most:

Setting	Default	Notes
SP_CONNECTOR_ALLOW_PRIVATE	false	true lets data sources use private-network SFTP servers and buckets. Set it on a self-hosted installation whose servers sit on your network. Never on an installation shared by several institutions: with SP_DEPLOYMENT=saas, SpeculaPro refuses to start with it (config: SP_CONNECTOR_ALLOW_PRIVATE can't be set when SP_DEPLOYMENT=saas)
SP_CONNECTOR_ALLOW_HTTP	false	true allows plain-http S3 endpoints (an on-premises MinIO without TLS). The secret then travels in the clear: a private network only
SP_CONNECTOR_SPOOL_DIR	The OS temporary directory	Where files are staged. Put it on a volume with room for the largest file times the concurrency
SP_CONNECTOR_MAX_FILE_BYTES	2 GiB	Larger files are recorded as failed without being downloaded. Also caps a .gz file’s decompressed content
SP_CONNECTOR_CONCURRENCY	2	File drops run at once by this process
SP_CONNECTOR_FILE_WORKERS	4	Batches of one file loaded in parallel
SP_CONNECTOR_KAFKA_BROKERS	Empty	The cluster institutions publish to. Empty: SpeculaPro’s own (SP_KAFKA_*)
SP_CONNECTOR_KAFKA_TOPIC_NAMESPACE	Empty	Set it on any installation shared by several institutions. See Who may read which topic
SP_CONNECTOR_KAFKA_SHARED_TOPICS	Empty	Topics shared by several workspaces, read by the sp-workspace header. For producers you control only
SP_CONNECTOR_CDC_ALLOW_PLAINTEXT	false	By default a Postgres source must verify the server: ssl_mode verify-full, or verify-ca with the institution’s CA certificate. true also accepts require (TLS, but any certificate is accepted) and disable (no TLS): the password and every row could then reach an impostor, or travel in the clear. Development or a private network only; refused when SP_DEPLOYMENT=saas
SP_CONNECTOR_CDC_SLOT_LAG_ALERT_BYTES	1 GiB	A Postgres replication slot further behind than this raises the alert. See Postgres replication for DBAs

Set SP_CONNECTOR_ALLOW_PRIVATE, SP_CONNECTOR_ALLOW_HTTP, SP_CONNECTOR_CDC_ALLOW_PLAINTEXT, the topic namespace and shared topics the same on the api role: the API checks data sources against them when they are saved.

Running more than one

Two or more connector processes are safe, and give failover:

• File drops are leased. A run claims its data source for two minutes and keeps extending the claim while it works. Each file is claimed by name, size and SHA-256, so it is loaded once even if two connectors see it. If a connector stops mid-file, another reloads that file once the claim runs out; rows already loaded are recognised as duplicates.
• Kafka reading is a consumer group (SP_CONNECTOR_KAFKA_GROUP, speculapro-connector by default), so partitions are shared out between connectors.

Network access

Connectors connect to hosts that workspace administrators choose, so SpeculaPro guards those connections itself:

• Private and internal addresses are refused unless SP_CONNECTOR_ALLOW_PRIVATE is set: localhost, *.local, *.internal, loopback, private, carrier-grade NAT, link-local (including the cloud metadata address 169.254.169.254), multicast, reserved and documentation ranges, and IPv6 forms that carry an IPv4 address (mapped, NAT64 with a private address inside, 6to4, Teredo). The check runs when a data source is saved, and again on every connection, after DNS resolution, so a name that later points at an internal address is still refused. Even with it, link-local addresses (169.254.0.0/16, fe80::/10), cloud metadata addresses and names (169.254.169.254, fd00:ec2::254, 100.100.100.200, metadata.google.internal and similar), 0.0.0.0/8 and carrier-grade NAT (100.64.0.0/10) stay refused, also when written in IPv4-mapped or NAT64 form: allow-private opens only RFC 1918 networks, IPv6 unique-local and loopback. See Which addresses are refused.
• No proxy is used for S3, so the real destination is always the one checked.
• SFTP host keys are pinned. An unknown or changed key is refused; the error shows the key the server presented, so you can check it with the server’s administrator.
• Timeouts: connecting (including the SSH handshake) gives up after SP_CONNECTOR_CONNECT_TIMEOUT (15 seconds); a transfer that makes no progress for SP_CONNECTOR_READ_TIMEOUT (60 seconds) fails.

Your own firewall should still limit the connector’s outbound traffic to the ports above.

Test connection runs on a connector

Test connection in Settings → Data sources is made by a connector, never by the web tier, so only connector nodes need outbound access to SFTP servers, buckets and institution brokers. When the api and connector roles run in one process (all), the test runs there. Otherwise the API queues it and waits up to 25 seconds for any running connector to pick it up (each test gives up after 20 seconds). If no connector answers in time, the screen says No connector answered within 25 seconds. Tests run on the connector role (the only one allowed to reach outside systems): check that it is running and healthy, then try again.

S3 with your own certificate authority

S3 endpoints are checked against the system’s trusted certificate authorities. For an on-premises MinIO or Ceph whose certificate comes from your own CA, give SpeculaPro the CA bundle with the standard SSL_CERT_FILE (a PEM file) or SSL_CERT_DIR environment variable on the connector processes. Connection tests run on connectors too, so the API doesn’t need it.

The institution Kafka cluster

Left empty, SP_CONNECTOR_KAFKA_BROKERS means SpeculaPro’s own cluster and its settings. For a separate cluster, set the brokers and, as needed, SP_CONNECTOR_KAFKA_SASL_MECHANISM (PLAIN, SCRAM-SHA-256 or SCRAM-SHA-512), _USERNAME, _PASSWORD, and SP_CONNECTOR_KAFKA_TLS_ENABLED with _TLS_CA_FILE, _TLS_CERT_FILE, _TLS_KEY_FILE and _TLS_SERVER_NAME. Give the connector credentials that can read the institutions’ topics, write their dead-letter topics, and manage the consumer group.

Packaging

Kubernetes (Helm)

The chart runs the connector as roles.connector, with 2 replicas. See Kubernetes (Helm).

• With the network policies on, connector pods may reach any address on networkPolicy.egress.connectorPorts ([22, 443, 9092, 9093]). Set it to [] to allow none, or add the ports your SFTP servers use.

• api pods get no egress to SFTP servers, buckets or institution brokers: connection tests run on connector pods.

• Change data capture has its own rule, networkPolicy.egress.cdc: connector pods may reach postgresPorts ([5432, 6432]) for the databases they replicate from and schemaRegistryPorts ([8081]) for Debezium Avro schema registries. Its to is empty (any address) by default; narrow it to the institutions’ addresses when they are fixed. It is separate from networkPolicy.egress.postgres, so narrowing that to SpeculaPro’s own database doesn’t cut off replication sources.

• Two alerts cover change data capture: SpeculaProCDCSlotBehind and SpeculaProCDCSourceFailed. See Monitoring.

• Files are staged in /tmp, an emptyDir that the chart sizes at 3 GiB for connector pods (roles.connector.tmpSizeLimit), with SP_CONNECTOR_MAX_FILE_BYTES at 900 MiB. Each pod loads up to SP_CONNECTOR_CONCURRENCY (2) files at once, each with its results file beside it (allow half the file again), so the rule is:

  tmpSizeLimit ≥ SP_CONNECTOR_CONCURRENCY × 1.5 × SP_CONNECTOR_MAX_FILE_BYTES (here 2 × 1.5 × 900 MiB = 2.7 GiB)

  To accept bigger files or more at once, raise them together, or mount a volume (extraVolumes, extraVolumeMounts) and point SP_CONNECTOR_SPOOL_DIR at it. In mode: all, the single deployment’s /tmp is sized the same way.

Docker Compose

In the split profile, the background container runs stream, persist, worker and connector. Files are staged in the spool volume, mounted at /var/lib/speculapro/spool (SP_CONNECTOR_SPOOL_DIR); the all-in-one container uses the same volume. See Docker Compose (production).

Sizing

• Files: the connector parses as a stream and never holds a file in memory. Memory grows with SP_CONNECTOR_CONCURRENCY × SP_CONNECTOR_FILE_WORKERS batches in flight, not with file size. A single record is capped too: 1 MiB per CSV row and JSON Lines line, 8 MiB per ISO 20022 entry or instruction, so one malformed or hostile record can’t exhaust memory. Disk needs room for the largest file times the concurrency.
• Throughput depends on Postgres (each row’s duplicate check), Redis and Kafka, as for the ingest API. Add file workers before adding connectors.
• Kafka: one connector reads every mapped topic; add connectors to share partitions.
• Change data capture: Debezium topics are read like other topics. A Postgres replication source is read by one connector at a time (others wait, so two never share a replication slot); spread many sources over several connectors. As an indicative laptop measurement only, decoding Postgres changes ran at about 14,500 a second on one core, and about 10,600 a second end to end into SpeculaPro. See Postgres replication for DBAs.

An indicative measurement

On a developer laptop, a 128 MiB CSV of 1,000,000 transactions went from an SFTP server to SpeculaPro’s Kafka topic (download, checksum, parsing, checks, duplicate claims, conversion, encryption, publishing, the results file and the move) in 37.4 seconds, about 26,700 rows a second, with the default four file workers. This is not a guarantee: measure on your own hardware with your own files. The design target is one million rows in at most ten minutes.

The measurement is a test in the source tree, TestOneMillionRows in backend/internal/filedrop. It takes minutes, so the regular test suite skips it; run it with SP_PERF=1 (and SP_PERF_ROWS to change the number of rows), with the same test databases the rest of the suite uses.

Metrics

On the metrics listener:

Metric	Type	Labels	What it counts
speculapro_connector_records_total	Counter	topic, result (accepted, duplicate, stale, rejected)	Records read from institution topics. rejected records went to the dead-letter topic
speculapro_connector_lag_records	Gauge	topic	Records waiting on each institution topic for the connector’s group
speculapro_connector_paused	Gauge		1 while reading is paused because the pipeline is behind
speculapro_connector_pipeline_lag_records	Gauge		How far the persister and stream evaluator are behind on SpeculaPro’s transaction topic, as the connector sees it

| speculapro_cdc_changes_total | Counter | source_kind (cdc, postgres_cdc), result (accepted, duplicate, stale, rejected, skipped, replayed, superseded) | Row changes read from change data capture sources | | speculapro_cdc_slot_lag_bytes | Gauge | source | WAL the institution’s database keeps for SpeculaPro’s replication slot | | speculapro_cdc_slot_behind | Gauge | source | 1 while a slot is further behind than SP_CONNECTOR_CDC_SLOT_LAG_ALERT_BYTES | | speculapro_cdc_source_failed | Gauge | source | 1 while a change data capture source is stopped | | speculapro_cdc_last_event_timestamp_seconds | Gauge | source | When the last applied change was committed at the source |

File drops are tracked per file in each data source’s File history, and per row through the ingest metrics. The source label is the data source’s ID.

Logs

Message	Meaning
kafka connector started, file drops started	The role is running
connector: consuming	The set of mapped topics changed; lists them
connector: pipeline lagging; reading paused / connector: pipeline caught up; reading resumed	Back-pressure from the persister and stream evaluator
connector: pipeline unavailable, retrying	Ingest couldn’t take records (Postgres, Redis or Kafka down). Nothing is committed until it can
connector: dead-letter publish failed, retrying	Rejected records couldn’t be written to the dead-letter topic. Reading stalls until they can
connector: mapping ignored: topic mode doesn't match SP_CONNECTOR_KAFKA_SHARED_TOPICS	A data source’s topic was made shared, or unshared, after it was saved. Edit the data source
filedrop: file processed	A file finished, with its status and counts
filedrop: run finished with a problem	A run, or a file in it, failed. The reason is also on the data source
filedrop: lease lost; stopping this run	Another connector took over the data source, usually after a long pause. The run stops safely
filedrop: file failed after repeated interrupted loads	A file’s load stopped three times (out of memory, killed, restarted), so it was failed and is skipped. See Troubleshooting
filedrop: skipping a file	A file name was refused as unsafe (path separators, .., control characters, too long)
cdc: snapshot started, cdc: snapshot done; streaming	A Postgres source’s first copy began and finished
cdc: source stopped	A change data capture source stopped, with the reason (a mapped column gone, the slot lost). Data-source managers are notified
cdc: REPLICATION SLOT FALLING BEHIND: …	The institution’s database is keeping more WAL for SpeculaPro than the alert threshold. Fix the connector, or drop the slot
cdc: replication slot dropped; source disabled	Drop slot was carried out
cdc: pipeline unavailable, retrying	Ingest couldn’t take changes. The checkpoint doesn’t move until it can
connector: schema registry unavailable, retrying	A Debezium Avro source can’t reach its schema registry. Its records wait; nothing is dead-lettered
cdc: truncate at the source ignored: SpeculaPro never deletes	A mapped table was truncated at the source

Troubleshooting

Symptom	Likely cause	What to do
Data sources never run; Run now does nothing	No connector role running	Start one. Check SP_ROLES or --role
destination not allowed: … when saving or testing	The host is private or internal	On a self-hosted installation whose servers sit on your network, set SP_CONNECTOR_ALLOW_PRIVATE=true on the connector and API
host key mismatch: the server presented SHA256:…	The SFTP server’s key changed, or something is in the middle	Confirm the new fingerprint with the server’s administrator before updating the data source
staging: … no space left on device	The spool directory is too small	Give SP_CONNECTOR_SPOOL_DIR more room, or lower SP_CONNECTOR_MAX_FILE_BYTES or SP_CONNECTOR_CONCURRENCY
Test connection says No connector answered within 25 seconds	No connector role is running, or it is overloaded	Start or check the connector. See Test connection runs on a connector
A file shows Failed with loading this file stopped without finishing 3 times	The file made a connector run out of memory, or connectors were restarted while loading it	Check the file (for example a missing closing quote, or one enormous entry), then upload it again under a new name. See File drops
x509: certificate signed by unknown authority (S3)	Your CA isn’t trusted	See S3 with your own certificate authority
Records pile up on a topic; speculapro_connector_paused is 1	Monitoring is behind	Normal after a burst. If it lasts, scale the persister or stream evaluator
Records pile up; not paused	The pipeline is unavailable, or the connector can’t write dead letters	Check the logs above
A file loads again and again	Its rows keep getting status retry	The pipeline was unavailable each time. Check Postgres, Redis and Kafka
A change data capture source shows Stopped	A mapped column was dropped or renamed, or the replication slot was dropped or invalidated	Read the reason on the source. Fix the mapping (saving a new version resumes it), or select Read everything again. See Change data capture
another connector is reading this source	Two connectors; only one reads a Postgres source at a time	Normal: the other takes over if the first stops
ssl_mode: TLS is required	A Postgres source without TLS	Use TLS. SP_CONNECTOR_CDC_ALLOW_PLAINTEXT=true on a private development network only
ssl_mode: require doesn’t verify the server’s certificate	A Postgres source with ssl_mode require	Use verify-full, or verify-ca with your CA certificate. require is accepted only with SP_CONNECTOR_CDC_ALLOW_PLAINTEXT=true, never on SaaS
another connector is reading this source (debug log)	Two connector replicas; one holds the source’s lease	Nothing: the other replica takes over within a minute if the holder stops

Related

• Data sources
• File drops
• Change data capture
• Postgres replication for DBAs
• Kafka topics
• Configuration: connectors
• Running roles