Skip to main content

​
Overview

The Server Sent Events Source Connector connects to an HTTP endpoint that supports Server-Sent Events (SSE), continuously receives events, and writes them to a specified Kafka topic. It is suitable for ingesting service status updates, notification streams, or other real-time data published over a long-lived SSE connection into Kafka for downstream stream processing, storage, or distribution systems. Each SSE event maps to one Kafka record. The record value uses a fixed structure with three string fields: event, id, and data. The data field remains a string; JSON or XML within it is not parsed automatically. The record key is null, and the SSE event ID is not stored as a Kafka Connect offset.

​
License

Uses Apache License 2.0.

​
Quick Start

Prepare a Connect Cluster, Kafka, and an SSE endpoint that does not require authentication, and confirm network connectivity and access permissions. For preparation and management, see Manage Connector.
Replace <sse-endpoint-url> with an SSE endpoint accessible to the Connect Worker, and replace <kafka-topic> with the Kafka topic that receives events. After applying the configuration, check both the Connector and Task status and verify actual events in the target topic. A running Task alone does not mean that the SSE connection has produced data.

​
Configuration

​
Kafka Connect Framework

​
connector.class

The Server Sent Events Source Connector implementation class to load.
  • Type: string
  • Default: None
  • Importance: High
  • Valid Values / Notes: Use com.github.cjmatta.kafka.connect.sse.ServerSentEventsSourceConnector.
  • Required: Yes

​
tasks.max

The maximum number of Tasks that Kafka Connect requests the Connector to create.
  • Type: int
  • Default: 1
  • Importance: High
  • Valid Values / Notes: Must be at least 1. This Connector always creates only one Task; increasing this value does not increase the number of SSE connections or consumption concurrency.

​
SSE Connection and Output

​
sse.uri

The SSE stream address to which the Task connects.
  • Type: string
  • Default: None
  • Importance: High
  • Valid Values / Notes: Use an SSE endpoint accessible to every candidate Worker. The Connector does not validate the URI format, protocol, or reachability during configuration parsing.
  • Required: Yes

​
topic

The Kafka topic that receives all SSE events.
  • Type: string
  • Default: None
  • Importance: High
  • Valid Values / Notes: One Connector instance can write to only one fixed topic. It does not support configuring a topic list or routing rules based on event content.
  • Required: Yes

​
HTTP Authentication and Request Headers

​
http.basic.auth

Controls whether the Connector reads the username and password for HTTP Basic Authentication.
  • Type: boolean
  • Default: false
  • Importance: Medium
  • Valid Values / Notes: Can be set to true or false. When enabled, the Task reads the password-typed configuration as a string and may fail with a type-conversion error during startup. Even if an Authorization header is constructed, the current implementation does not pass that request builder to the actual SSE handshake. Do not treat this as an available authentication method.

​
http.basic.auth.username

The username for HTTP Basic Authentication.
  • Type: string
  • Default: null
  • Importance: Medium
  • Valid Values / Notes: Read only when http.basic.auth=true. The Connector does not validate that the value is nonempty or that it is configured together with the password. Because of the Basic Auth implementation limitations, do not rely on this configuration for endpoints that require authentication.

​
http.basic.auth.password

The password for HTTP Basic Authentication.
  • Type: password
  • Default: null
  • Importance: Medium
  • Valid Values / Notes: Read only when http.basic.auth=true. Enabling Basic Auth and providing a password may cause a type-conversion error when the Task starts. Do not expose real passwords in plaintext configurations, logs, or documentation.

​
http.header.*

Reads custom HTTP request-header configurations. The name after http.header. is used as the request header name.
  • Type: dynamic-prefix string values
  • Default: {}
  • Importance: Medium
  • Valid Values / Notes: For example, http.header.User-Agent represents the User-Agent request header. The Connector does not validate header names or values, and values may contain sensitive credentials. The current implementation adds these headers to a request builder that is not passed to SseEventSource, so do not rely on this configuration for authentication, access control, or User-Agent overrides.

​
HTTP Client

​
compression.enabled

Controls the gzip encoding property of the Jersey client.
  • Type: boolean
  • Default: true
  • Importance: Low
  • Valid Values / Notes: Can be set to true or false. When enabled, the Connector sets Jersey’s gzip encoding property, but the explicitly constructed Accept-Encoding header is not connected to the actual SSE handshake path. Validate compression negotiation and response decompression behavior against the target SSE service.

​
Connection Rate Limiting

​
rate.limit.requests.per.second

Controls the minimum interval before the Connector opens or reopens an SSE connection.
  • Type: double
  • Default: null
  • Importance: Low
  • Valid Values / Notes: A positive value is used to calculate the wait time between connection-opening requests. null, 0, or a negative value does not enable this wait. It does not limit SSE event throughput and is not a Worker-wide request rate limit.

​
rate.limit.max.concurrent

Declares the maximum number of concurrent connections.
  • Type: int
  • Default: null
  • Importance: Low
  • Valid Values / Notes: Although the current implementation accepts and stores this value, it does not use it to limit connection, Task, or request concurrency. Do not use it as a concurrency control mechanism.

​
Reconnection Parameters

​
retry.backoff.initial.ms

Sets the initial wait time for a reconnection triggered by the idle health check and contributes to calculating the SSE client’s own reconnection interval.
  • Type: long
  • Default: 2000
  • Importance: Low
  • Valid Values / Notes: The unit is milliseconds. The Connector does not validate that the value is nonnegative. The interval used by the SSE client itself does not exceed 2000 milliseconds, while an idle health-check reconnection uses the configured value as the first exponential-backoff wait.

​
retry.backoff.max.ms

Sets the maximum exponential-backoff wait time for a reconnection triggered by the idle health check.
  • Type: long
  • Default: 30000
  • Importance: Low
  • Valid Values / Notes: The unit is milliseconds. The Connector does not validate that the value is nonnegative or require it to be greater than or equal to retry.backoff.initial.ms. It does not control every reconnection path of the SSE client itself.

​
retry.max.attempts

Limits the number of idle health-check reconnection attempts managed by the Connector.
  • Type: int
  • Default: -1
  • Importance: Low
  • Valid Values / Notes: -1 means unlimited attempts, and 0 rejects the first reconnection of this type. Values less than -1 are not treated as unlimited. This configuration does not guarantee coverage of asynchronous SSE errors, every HTTP status, or recovery after a Task failure.

​
Monitoring

​
What to Monitor

Monitor Kafka Connect health, Connector and Task status, throughput, latency, offset commits, errors, retries, and Worker JVM signals. Also monitor logs related to SSE connection errors and queue growth to prevent heap memory pressure when the source continuously outpaces Kafka. Monitor DLQ activity only when the corresponding error handling is enabled, but do not rely on the DLQ to capture connection and event-processing errors internal to the Connector.

​
Import the Grafana Dashboard

Ensure that Connect metrics are available in a Grafana data source and that the collected labels match the dashboard filters; download the Kafka Connect Dashboard, import the JSON into Grafana, and select the corresponding data source.

​
Limitations

  • One Connector instance connects to one SSE URI, writes to one fixed topic, and always creates only one Task. To connect to multiple SSE endpoints, create a separate Connector instance for each endpoint.
  • The Connector does not provide event-type filtering or parse data content. SSE data is always written as a string field in the fixed structure.
  • The Connector does not store SSE event IDs as Kafka Connect offsets and does not provide persistent deduplication. Task or Worker restarts and reconnections may cause duplicates or omissions; do not claim exactly-once, duplicate-free, omission-free, or unconditional at-least-once delivery.
  • SSE events first enter an unbounded in-memory queue, and a single poll has no record-count or byte-size limit. If the source continuously outpaces Kafka, heap memory pressure may occur.
  • Basic Auth has a password type-conversion failure path. The Authorization header, custom headers, and explicit Accept-Encoding header are added to a request builder that is not passed to SseEventSource. Do not rely on these configurations to access protected endpoints or override the User-Agent.
  • rate.limit.max.concurrent does not enforce a concurrency limit. rate.limit.requests.per.second affects only the wait before the Connector actively opens a connection; it does not limit event throughput or every internal reconnection performed by the SSE client.
  • The reconnection parameters cover only specific connection lifecycle paths. They cannot guarantee that every initial connection failure, asynchronous error, or HTTP response is retried and recovered automatically.
  • An SSE message without an event name may be skipped or emitted with event=unknown, depending on its position within a batch.

​
FAQ

​
The Connector and Task Are Running, but the Target Topic Has No Messages

A successfully started Task does not necessarily mean that the SSE stream is connected and continuously producing events. Confirm that sse.uri is accessible from the Worker running the Task, that the endpoint returns a valid SSE response, and that it is publishing events with data. Check the Worker logs for connection, protocol, and asynchronous errors, and verify output in the target topic. For endpoints that require Basic Auth or custom request headers, do not assume that the current configuration can complete the handshake. Prefer a controlled endpoint without authentication when isolating the problem.

​
Duplicate Messages Appear or a Segment of Events Is Missing After a Restart

The Connector does not store a recoverable SSE consumption position or write event IDs to Kafka Connect offsets. Even if the SSE service supports Last-Event-ID, after a Task or Worker restart the Connector has no persisted event ID from which to resume. Use stable business identifiers for idempotent processing or deduplication downstream, and monitor both duplicates and data gaps during restart windows. Do not rely on offset commits to eliminate this risk.

​
Increasing tasks.max Still Produces Only One Task

This is the task-allocation boundary of the Connector. It always returns one Task configuration, so increasing tasks.max does not increase the number of SSE connections or throughput. To ingest multiple endpoints, create an independent Connector instance for each endpoint. Validate capacity planning for a single endpoint against the source event rate, Kafka write capacity, and Worker memory.

​
Worker Memory Usage Keeps Growing

The SSE callback writes events to an unbounded in-memory queue, and the Connector provides neither backpressure nor a batch limit. Check the source event rate, Kafka produce latency, Task errors, and queue-related logs. Reduce the source send rate, restore Kafka write capacity, or separate different endpoints into independent Connector instances, and configure memory and alert thresholds for the Worker. rate.limit.requests.per.second limits only how often connections are opened; it cannot reduce the event rate of an established SSE stream.