Skip to main content

​
Overview

The Lightstreamer Sink Connector consumes records from Kafka topics, converts them into Lightstreamer field updates, and sends them through the Lightstreamer Proxy Adapter to web, mobile, or other clients that subscribe to real-time Items. It sits between Kafka Connect and Lightstreamer Server: Kafka Connect reads records and manages offsets, the Connector routes topics to Items and maps records to fields, and Lightstreamer Server distributes updates to clients. A topic can map to one or more static Items, or Item templates can route updates to parameterized Items based on record keys, values, headers, or metadata. Field mappings can extract the key, value, headers, topic, partition, offset, and timestamp, converting the result to a string or null. The Connector does not provide an initial snapshot. In the normal active connection mode, records produce Item updates only when a matching client subscription exists. This Connector is suitable for pushing market data, device status, business events, and real-time dashboard data from Kafka to Lightstreamer clients. Lightstreamer Item updates and Kafka offset commits are not transactional across the two systems, so recovery may produce duplicate updates. Keep stable identifiers in events and handle duplicates in the client or business layer as needed.

​
Prerequisites

  • Deploy Lightstreamer Server 7.4.2 or later, and configure a Proxy Data Adapter in the target Adapter Set that can communicate with the Connector. Its request/reply port, connection direction, and optional authentication settings must match the Connector configuration.
  • Lightstreamer clients must connect to the correct Adapter Set and Data Adapter, and subscribe to the Items defined by topic.mappings or item.templates and the fields defined by record.mappings. No client updates are produced without a matching subscription.
  • When extracting with KEY or VALUE paths, the selected Kafka Connect Converter must produce a non-null Connect Schema consistent with the record structure. When using a DLQ, prepare a Kafka topic that the Connector can write to and that is outside the input subscription scope.

​
License

Licensed under the Apache License 2.0.

​
Quick Start

Prepare a Connect Cluster, Kafka topic, Lightstreamer Server, Proxy Adapter, and subscribing client, and confirm network connectivity and access permissions. For cluster preparation and Connector management, see Manage Connectors. The following example consumes string messages and sends each record’s value as the message field to the static Item events.
Replace both occurrences of <kafka-topic> with the same input topic, and replace the Lightstreamer Proxy Adapter address. The client must subscribe to the events Item and the message field in the corresponding Adapter Set. If authentication is enabled on the Proxy Adapter, also configure a username and password. Inject credentials through the deployment environment’s secure configuration mechanism; do not store them in version control or shared logs.

​
Configuration

​
Connector Identity, Input, and Tasks

​
connector.class

Specifies the Lightstreamer Sink Connector implementation class.
  • Type: string
  • Default: None
  • Importance: High
  • Valid values / Notes: Use com.lightstreamer.kafka.connect.LightstreamerSinkConnector. Every Worker that may run a Task must be able to load the plugin.
  • Required: Yes

​
topics

Specifies the Kafka topics to consume.
  • Type: list
  • Default: Empty list
  • Importance: High
  • Valid values / Notes: Separate multiple names with commas. Configure either this property or topics.regex, but not both. Input topics must also match a literal or regular expression rule in topic.mappings. When a DLQ is enabled, do not subscribe to the DLQ topic.

​
topics.regex

Selects Kafka topics to consume by regular expression.
  • Type: string
  • Default: Empty string
  • Importance: High
  • Valid values / Notes: Use Java regular expression syntax. This property is mutually exclusive with topics. When a DLQ is configured, the expression must not match the DLQ topic. This property controls only the Kafka Connect input subscription; it does not automatically enable regular expression interpretation for topic.mappings.

​
tasks.max

Sets the maximum number of Tasks Kafka Connect may create.
  • Type: int
  • Default: 1
  • Importance: High
  • Valid values / Notes: Must be at least 1. The Connector always returns only one Task configuration, so increasing this value does not increase Task parallelism for a single Connector instance.

​
Proxy Adapter Connection and Authentication

​
lightstreamer.server.proxy_adapter.address

Sets the Lightstreamer Proxy Adapter host and request/reply port.
  • Type: string
  • Default: None
  • Importance: High
  • Valid values / Notes: Use the host:port format. The host must not be empty, and the port must be a positive integer without leading zeros. This property is always required. It must still be provided when connection inversion is enabled, but its host and port are not used for an active outbound connection. Bracketed IPv6 addresses do not satisfy this property’s validation format.
  • Required: Yes

​
lightstreamer.server.proxy_adapter.socket.connection.setup.timeout.ms

Sets how long to wait when establishing the Proxy Adapter socket in active connection mode, in milliseconds.
  • Type: int
  • Default: 5000
  • Importance: Low
  • Valid values / Notes: Must be greater than or equal to 0; 0 means no connection timeout is set. Applies only when connection.inversion.enable=false.

​
lightstreamer.server.proxy_adapter.socket.connection.setup.max.retries

Sets the maximum number of retries after an initial connection failure in active connection mode.
  • Type: int
  • Default: 1
  • Importance: Medium
  • Valid values / Notes: Must be greater than or equal to 0; 0 disables retries. This property controls only connection establishment during Task startup. It is not a record-level retry and does not continuously reconnect after a runtime disconnection.

​
lightstreamer.server.proxy_adapter.socket.connection.setup.retry.delay.ms

Sets the wait time between initial connection attempts in active connection mode, in milliseconds.
  • Type: long
  • Default: 5000
  • Importance: Low
  • Valid values / Notes: Must be greater than or equal to 0. Applies only when connection.inversion.enable=false and the maximum retry count is greater than 0.

​
lightstreamer.server.proxy_adapter.username

Sets the username for Proxy Adapter remote connection authentication.
  • Type: string
  • Default: null
  • Importance: Medium
  • Valid values / Notes: Configure this property only when remote Adapter authentication is enabled on the Proxy Adapter. An empty string is different from leaving it unset. It can be used in both active connection and connection inversion modes.

​
lightstreamer.server.proxy_adapter.password

Sets the password for Proxy Adapter remote connection authentication.
  • Type: password
  • Default: null
  • Importance: Medium
  • Valid values / Notes: Typically configured together with lightstreamer.server.proxy_adapter.username. Store it through a secure configuration mechanism, and do not expose the actual value in logs, documentation, or version control.

​
connection.inversion.enable

Sets whether to reverse the direction in which the connection between the Proxy Adapter and the Connector is established.
  • Type: boolean
  • Default: false
  • Importance: Low
  • Valid values / Notes: When false, the Connector actively connects to lightstreamer.server.proxy_adapter.address. When true, the Connector listens on request_reply.port, and the Proxy Adapter connects to it. In inversion mode, also configure the Connector’s remote host on the Proxy Adapter.

​
request_reply.port

Sets the request/reply port on which the Connector listens in connection inversion mode.
  • Type: int
  • Default: 6661
  • Importance: Low
  • Valid values / Notes: Applies only when connection.inversion.enable=true. Configuration validation requires only a value greater than or equal to 0; in practice, select a port the operating system can bind that is within the valid TCP port range.

​
max.proxy.adapter.connections

Sets the number of Proxy Adapter connections that may be active simultaneously in connection inversion mode.
  • Type: int
  • Default: 1
  • Importance: Low
  • Valid values / Notes: Must be greater than or equal to 1, and applies only when connection.inversion.enable=true. The same batch of records is sent to all connections active at that time, but no transactional or ordering guarantee is provided across connections.

​
Topic and Item Routing

​
topic.mappings

Defines mappings from Kafka topics to static Lightstreamer Items or Item templates.
  • Type: string
  • Default: None
  • Importance: High
  • Valid values / Notes: Use the format topic:item1,item2;other-topic:item3. Each topic or regular expression key must be unique, and a mapping must not be empty. Duplicate Items within the same mapping are deduplicated. By default, topics are matched literally. When topic.mappings.regex.enable is enabled, Java regular expressions are used for full matches. An item-template.<name> reference must be defined in item.templates.
  • Required: Yes

​
topic.mappings.regex.enable

Sets whether each topic key in topic.mappings is interpreted as a Java regular expression.
  • Type: boolean
  • Default: false
  • Importance: Medium
  • Valid values / Notes: When enabled, matching uses the full topic name. An invalid regular expression causes a failure while the Task builds its mapper. This property and topics.regex are independent; both sets of rules must cover the intended input topics.

​
item.templates

Defines parameterized Lightstreamer Item templates that filter subscriptions based on record content.
  • Type: string
  • Default: null
  • Importance: Medium
  • Valid values / Notes: Use the format template-name:item-prefix-#{param=EXPRESSION};other-template:.... Template names must be unique, and values must not be empty. #{...} is the Connector-side template definition; clients must provide literal filter values using the format item-prefix-[param=literalValue]. Template parameters can use only scalar values. topic.mappings references a template as item-template.<template-name>.

​
Record Conversion and Field Mapping

​
key.converter

Sets the Converter for Kafka record keys.
  • Type: class
  • Default: null
  • Importance: Low
  • Valid values / Notes: When unspecified, this property inherits the Worker setting. When using KEY extraction expressions or key-based Item templates, the Converter must match the message encoding and produce a non-null Connect Schema. For simple string keys, use org.apache.kafka.connect.storage.StringConverter.

​
value.converter

Sets the Converter for Kafka record values.
  • Type: class
  • Default: null
  • Importance: Low
  • Valid values / Notes: When unspecified, this property inherits the Worker setting. When using VALUE extraction expressions, the Converter must match the message encoding and produce a non-null Connect Schema. For string messages, use org.apache.kafka.connect.storage.StringConverter. Configure the selected Converter’s own options as required by its implementation.

​
record.mappings

Defines Lightstreamer field names and their corresponding Kafka record extraction expressions.
  • Type: list
  • Default: None
  • Importance: High
  • Valid values / Notes: Use comma-separated field:#{EXPRESSION} entries. Field names must be non-empty and unique. Expressions can extract data from KEY, VALUE, HEADERS, TOPIC, PARTITION, OFFSET, or TIMESTAMP. This configuration fixes the field set; other fields in input records are not published automatically. When an expression needs to contain a comma, also follow properties escaping rules.
  • Required: Yes

​
record.mappings.skip.failed.enable

Sets whether to omit a field when its extraction fails and continue sending the remaining fields.
  • Type: boolean
  • Default: false
  • Importance: Medium
  • Valid values / Notes: When false, field extraction errors are handled by record.extraction.error.strategy. When true, only the failed field is omitted, and other fields can still be updated. This property does not skip Item template parameter extraction failures, and an empty field Map may be sent when every field fails.

​
record.mappings.map.non.scalar.values.enable

Sets whether non-scalar selection results such as Structs, Maps, and arrays may be converted to field text.
  • Type: boolean
  • Default: false
  • Importance: Medium
  • Valid values / Notes: When false, field mappings must select scalar values. When true, Structs are converted to JSON text without a Schema, while other complex Java values are sent using their string representation. This property affects only record.mappings; it does not relax the scalar requirement for Item template parameters.

​
Extraction Errors and DLQ

​
record.extraction.error.strategy

Sets the strategy for handling extraction errors during record mapping.
  • Type: string
  • Default: IGNORE_AND_CONTINUE
  • Importance: Medium
  • Valid values / Notes: Values are case-sensitive: IGNORE_AND_CONTINUE, FORWARD_TO_DLQ, or TERMINATE_TASK. Ignoring an error or successfully writing to the DLQ advances the processing position for that record. To isolate bad records with FORWARD_TO_DLQ and continue processing subsequent records, also configure a non-empty errors.deadletterqueue.topic.name and errors.tolerance=all. A missing DLQ topic or the default tolerance level none causes an extraction error to terminate the Task. This strategy handles only extraction errors during mapping; it does not handle arbitrary connection or communication failures.

​
errors.deadletterqueue.topic.name

Sets the Kafka DLQ topic that receives records with extraction failures.
  • Type: string
  • Default: Empty string
  • Importance: Medium
  • Valid values / Notes: An empty string disables the DLQ. When using record.extraction.error.strategy=FORWARD_TO_DLQ, configure a non-empty value and errors.tolerance=all so processing can continue after the bad record is reported. The DLQ topic must not appear in topics or match topics.regex.

​
errors.tolerance

Sets whether Kafka Connect tolerates record processing failures and keeps the Task running.
  • Type: string
  • Default: none
  • Importance: Medium
  • Valid values / Notes: Valid values are none and all. With none, the Task terminates when the first failed record exceeds the tolerance limit. With all, failed records can be tolerated. For record.extraction.error.strategy=FORWARD_TO_DLQ, set this property to all so processing continues after an extraction failure is reported to the configured DLQ. This setting does not convert Proxy Adapter connection, communication, or other arbitrary errors into tolerable DLQ records.

​
errors.deadletterqueue.topic.replication.factor

Sets the replication factor Kafka Connect uses when it automatically creates the DLQ topic.
  • Type: short
  • Default: 3
  • Importance: Medium
  • Valid values / Notes: Used only when the configured DLQ topic does not exist and Kafka Connect creates it. The value must be suitable for the number of brokers and replication policy in the target Kafka cluster.

​
errors.deadletterqueue.context.headers.enable

Sets whether Kafka Connect error context headers are added to DLQ records.
  • Type: boolean
  • Default: false
  • Importance: Medium
  • Valid values / Notes: Applies only when DLQ records are actually produced. When enabled, context header names use the __connect.errors. prefix. DLQ consumers should not treat these internal diagnostic headers as original business fields.

​
Best Practices

​
Expand the Input Scope by Topic Naming Pattern

Applicable scenario: Initial integration is complete, and additional Kafka topics following a consistent naming pattern will be added over time. You want new topics to enter the same real-time event stream automatically without repeatedly changing a fixed topic list. Configuration example:
Key points: Compared with the Quick Start, this configuration removes topics, uses topics.regex, and also changes the key in topic.mappings to a regular expression. The two sets of rules are independent and should cover the same scope. Otherwise, Kafka Connect may consume a topic whose records do not match any Item route. Before adding a topic, also confirm that its message encoding and field structure are compatible with the existing mapping.

​
Partition Client Subscriptions by Business Entity

Applicable scenario: Clients need updates only for a specific order, device, or account, and not every subscriber should receive all records from the same topic. Use the record key to construct parameterized Items so that clients can subscribe by business entity. Configuration example:
Key points: Compared with the Quick Start, this configuration adds item.templates and a key Converter, and replaces the static Item with a template reference. The Connector template is defined as entity-#{id=KEY}. A client must subscribe with a literal filter value such as entity-[id=42]; only records with the key 42 match that subscription. Do not use entity-42. The template parameter value and Connect Schema extracted from the record must match the subscription parameter. Business entity keys should be stable and non-empty so that key changes do not route updates for the same entity to different Items.

​
Isolate Field Extraction Failures in a DLQ

Applicable scenario: The Connector is already running continuously, and schema evolution or occasional malformed records may make a field path impossible to extract. You want to retain the problematic records for investigation while continuing to process subsequent valid records. Configuration example:
Key points: Compared with the Quick Start, this configuration uses a JSON value with a Schema, a nested field path, and FORWARD_TO_DLQ, and adds errors.tolerance=all and a DLQ topic. all is required to isolate bad records and continue processing subsequent records. If the default none is retained, the Task still terminates after the error is reported because the tolerance limit is exceeded. <dlq-topic> must differ from the input topic and remain outside the subscription scope. After a record is successfully reported to the DLQ, its offset continues to advance. Monitor and consume the DLQ, then replay records according to business rules after correcting the data or mapping. This combination tolerates only extraction failures during mapping; it does not catch arbitrary errors such as Proxy Adapter communication failures.

​
Monitoring

​
What to Monitor

Monitor Kafka Connect health, Connector and Task status, throughput, latency, offset commits, errors, retries, and Worker JVM signals. Monitor DLQ activity only when the corresponding error handling is enabled.

​
Import the Grafana Dashboard

Confirm that Connect metrics are available to a Grafana data source and that the collected labels satisfy the dashboard filters. Download the Kafka Connect Dashboard, import the JSON into Grafana, and select the corresponding data source.

​
Limitations

  • A single Connector instance always creates only one Sink Task. Increasing tasks.max does not increase Task parallelism. To scale by instance, divide input topics across multiple Connector instances.
  • The Connector does not provide an initial snapshot, and field updates are driven by current client subscriptions. In the normal active connection mode, the Connector does not continuously advance its managed safe offset when no subscription exists. A subscription added while the Connector is running does not automatically receive records consumed earlier without a matching subscription, while a restart may replay records from the last committed position.
  • In connection inversion mode, Worker offsets may still be committed when there is no active Proxy Adapter or no subscription. Such records may not produce client updates and are not automatically redelivered based on Kafka offsets.
  • Target Item updates and Kafka offset commits are not atomic. Recovery may produce duplicate updates, and the Connector does not provide exactly-once delivery, target-side deduplication, or confirmation that end clients received an update.
  • KEY and VALUE path extraction requires the Converter to produce a non-null Connect Schema. Field navigation in a schema-less JSON Map must not be treated as a supported use case.
  • record.extraction.error.strategy handles only extraction errors during mapping. Runtime connection loss and other uncaught exceptions may fail the Task. The Connector does not provide record-level retries, asynchronous buffering, or continuous reconnection for active connections.

​
FAQ

​
Why are the Connector and Task running normally, but clients receive no updates?

First confirm that the client is connected to the correct Adapter Set and Data Adapter, and has subscribed to the static Item referenced by topic.mappings or a parameterized Item matching item.templates. Then check that Kafka Connect’s topics or topics.regex covers the input topic, that topic.mappings matches it using the correct literal or regular expression mode, and that the fields requested by the client exist in record.mappings. In the normal active connection mode, no updates are produced without a matching subscription. After establishing the subscription, send a new record to verify the configuration; do not assume that adding a subscription while the Connector is running automatically replays earlier records.

​
Why does the Task fail after reading a record, or why does the record produce no field update?

Check that key.converter and value.converter match the actual message encoding, and confirm that each KEY or VALUE path refers to a real field in a non-null Schema. A missing field, an out-of-bounds array index, navigating into a scalar, or selecting a non-scalar value can cause extraction errors. To keep other fields when one field fails, evaluate record.mappings.skip.failed.enable=true. To isolate the entire problematic record and continue processing subsequent records, configure FORWARD_TO_DLQ, a non-empty DLQ topic, and errors.tolerance=all. To stop processing immediately, use TERMINATE_TASK. This tolerance applies only to record failures that follow the error-handling path; it does not ignore arbitrary connection or communication errors.

​
Why is there still only one Task after increasing tasks.max?

This is the Connector’s Task model. Each Connector instance corresponds to one Remote Adapter Task, and the Connector always generates only one Task configuration. To isolate business flows or scale the deployment, create multiple Connector instances for different topic sets. Keep the input subscriptions and Lightstreamer routing boundaries of each instance clear to avoid unintended duplicate consumption and updates.

​
Why can’t the Connector connect to the Lightstreamer Proxy Adapter?

Check that lightstreamer.server.proxy_adapter.address uses a resolvable host:port, that the port matches the Proxy Adapter’s request/reply port, that the firewall allows the connection, and that the authentication username and password match the Proxy Adapter configuration. If connection.inversion.enable is enabled, also confirm that the Connector can bind request_reply.port and configure the Proxy Adapter’s remote host to point to the Connector. Active connection retry settings apply only during Task startup. After a runtime disconnection, check the Task status and logs, and restart it according to your deployment strategy.