# Create a Custom Connector Pipeline

SingleStore Connector Pipelines use Kafka Connect Source Connector libraries to read data from external sources. While SingleStore does not run Kafka or Kafka Connect, you can build SingleStore pipelines using custom Kafka Connect Source Connectors libraries. Perform the following tasks:

1. Ensure that the [prerequisites](https://docs.singlestore.com/#section-id235862010201613.md) are met.

2. [Create a Custom Kafka Connect Source Connector](https://docs.singlestore.com/#section-id235862033525804.md).

3. [Register the connector](https://docs.singlestore.com/#section-id235862037894952.md).

4. [Configure SingleStore to use the custom connector](https://docs.singlestore.com/#section-id235862039722348.md).

5. [Create the pipeline](https://docs.singlestore.com/#section-id235862040962081.md).

SingleStore recommends having hands-on experience building custom connectors. For more information, refer to the [Connector Development Guide](https://kafka.apache.org/39/kafka-connect/connector-development-guide/).

Existing third-party connector examples for reference:

* [Kafka Connect Wikimedia](https://github.com/conduktor/kafka-connect-wikimedia)
* [File Stream Connector](https://github.com/apache/kafka/tree/trunk/connect/file/src/main/java/org/apache/kafka/connect/file)
* [Kafka Connect Source GitHub](https://github.com/simplesteph/kafka-connect-github-source)

## Prerequisites

* An active SingleStore Self-Managed cluster and access to every host where the connector will run.
* Install Java 25 on each host that runs SingleStore. Set the Java path using the `java_path` engine variable in each node's `memsql.cnf` file:
  ```
  java_path = '<path/to/java-binary>'
  ```

## Create a Custom Kafka Connect Source Connector

SingleStore Pipelines support Kafka Connect API version 3.9.1 and [Log4j/SLF4J](https://mvnrepository.com/artifact/org.apache.logging.log4j/log4j-slf4j-impl/2.25.5) version 2.25.5.

To implement a connector, extend the following two core classes: `SourceConnector` and `SourceTask`.

## SourceConnector Class

The `SourceConnector` task provides metadata, validates configuration properties, and generates the necessary task configurations to instantiate `SourceTask`. SingleStore uses `SourceConnector` as a short-lived class that retrieves the task and then stops. Implement the following `SourceConnector` methods:

| Method                                                | Description                                                                        |
| ----------------------------------------------------- | ---------------------------------------------------------------------------------- |
| `Class<? extends Task> taskClass()`                   | Returns the class of your`SourceTask`implementation.                               |
| `void start(Map<String, String> props)`               | Handles initialization logic, such as storing and validating connector properties. |
| `ConfigDef config()`                                  | Defines supported configuration settings.                                          |
| `List<Map<String, String>> taskConfigs(int maxTasks)` | Builds configurations for each task instance.                                      |
| `void stop()`                                         | Handles cleanup and deinitialization.                                              |

Properties passed to the `start()` method are derived directly from the configuration and credentials specified in the `CREATE PIPELINE` command. By default, SingleStore instantiates four tasks. Use the `tasks.max` configuration parameter to change this limit.

SingleStore Connector Pipelines do not support [dynamic connectors](https://kafka.apache.org/39/kafka-connect/connector-development-guide/#dynamic-connectors) that monitor external systems or perform reconfigurations.

## SourceTask Class

The `SourceTask` class reads data subsets from the source system and translates them into `SourceRecord` objects. Implement the following `SourceTask` methods:

| Method                                  | Description                                                      |
| --------------------------------------- | ---------------------------------------------------------------- |
| `void start(Map<String, String> props)` | Initializes the task, parses configurations, and performs setup. |
| `List<SourceRecord> poll()`             | Fetches new data records from the source system.                 |
| `void stop()`                           | Closes active connections and cleans up resources.               |

Optional methods to implement:

| Method                                                            | Description                                                   |
| ----------------------------------------------------------------- | ------------------------------------------------------------- |
| `void commitRecord(SourceRecord record, RecordMetadata metadata)` | Acknowledges successful ingestion of a specific record.       |
| `void commit()`                                                   | Acknowledges successful ingestion of the latest record batch. |

## Remarks

* Include a `sourcePartition` and a `sourceOffset` in every record returned by `poll()`.
* Each `sourcePartition` must identify a unique data stream. Two stream tasks must not emit records with the same `sourcePartition`.
* The `sourceOffset` identifies the record position in a data stream. SingleStore stores the latest successfully processed `sourceOffset` for each `sourcePartition`.
* Use the `OffsetStorageReader` to retrieve saved offsets and resume processing from the last successfully processed position:
  ```Java
  context.offsetStorageReader().offset(partition)
  ```

SingleStore invokes `poll()` on all `SourceTask` instances during batch processing and waits for every task to complete. If a task fails, the batch fails, and the next `poll()` call starts from the previous offset for each partition. After all tasks (`poll()` calls) succeed, SingleStore writes the batch to the database in a single transaction. Once the transaction succeeds, offsets are updated and tasks are committed.

Because a long-running `poll()` call blocks the entire batch while waiting for data, avoid blocking inside `poll()`. Return an empty list when no records are available.

## Register the Connector

Add a `ServiceLoader` manifest file to the project resources directory at `META-INF/services/org.apache.kafka.connect.source.SourceConnector`.

Add the fully qualified `SourceConnector` class name on a single line. For example:

```
com.example.FileStreamSourceConnector
```

The manifest allows SingleStore to identify the connector by its short class name, `FileStreamSourceConnector`, instead of a fully-qualified class name (for example, `com.example.FileStreamSourceConnector`).

Package the connector as an uber-JAR to avoid dependency conflicts with Java libraries used by SingleStore Connector Pipelines. For Maven projects, use the [Maven Shade Plugin](https://maven.apache.org/plugins/maven-shade-plugin/). For Gradle projects, use the [Gradle Shadow Plugin](https://plugins.gradle.org/plugin/com.gradleup.shadow).

## Test the Custom Connector Locally

Before deploying the custom connector to SingleStore, test it using a local Kafka and Kafka Connect environment. The Kafka ecosystem provides debugging tutorials and observability examples for local environments. Refer to [Kafka Quickstart](https://kafka.apache.org/quickstart/) for setup instructions and [Kafka Connect User Guide](https://kafka.apache.org/43/kafka-connect/user-guide/#plugin-discovery) for connector configuration details.

## Configure SingleStore to Use the Custom Connector

To deploy the custom Kafka Connect Source Connector to the SingleStore cluster:

1. Copy the connector JAR to the following directory on each host:
   ```
   <singlestore-installation-dir>/extractors/lib-extractor-connect/
   ```

2. Verify that the JAR is readable by the SingleStore process.

3. Run the following command to enable custom connector pipelines:
   ```sql
   SET GLOBAL experimental_features_config = 'connector_pipelines_enabled=true';
   ```

## Create the Pipeline

Create a pipeline using the following command/syntax:

```sql
CREATE INFERRED PIPELINE `<pipeline_name>`
AS LOAD DATA CONNECTOR '<connector_class>'
CONFIG '{ <connector_configurations> }'
CREDENTIALS '{ <connector_credentials> }';
```

The `CREATE INFERRED PIPELINE ... AS LOAD DATA CONNECTOR` statement starts the connector and retrieves a sample batch to infer the target schema. SingleStore uses the inferred structure to create the destination table and then drops the sample records without updating offsets. For example:

```sql
CREATE INFERRED PIPELINE `file_events`
AS LOAD DATA CONNECTOR 'FileStreamSourceConnector'
CONFIG '{
  "file": "/tmp/events.json",
  "topic": "events",
  "tasks.max": "1"
}'
CREDENTIALS '{}';
```

SingleStore recommends validating the pipeline configuration with the following two workflows:

* **Check the schema inference**: Run the `CREATE INFERRED PIPELINE` command to create the pipeline and then run the `SHOW CREATE TABLE` command to verify the schema of the inferred table. For example:
  ```sql
  CREATE INFERRED PIPELINE `testEx`
  AS LOAD DATA CONNECTOR 'FileStreamSourceConnector'
  CONFIG '{  "output.source.partition": "true", "output.source.offset": "true" }'
  CREDENTIALS '{}';

  SHOW CREATE TABLE testEx;
  ```
* **Examine record serialization and optional metadata**: Create a table with a structure that can store the records and then create a pipeline that writes the records and optional metadata to the table. For example:
  ```sql
  CREATE TABLE `<table_name>` (
      `data` JSON NOT NULL
  );

  CREATE INFERRED PIPELINE `<pipeline_name>`
  AS LOAD DATA CONNECTOR '<connector_class>'
  CONFIG '{
    "output.key": "true",
    "output.value": "true",
    "output.timestamp": "true",
    "output.headers": "true",
    "output.topic": "true",
    "output.source.partition": "true",
    "output.source.offset": "true",
    <connector_configurations>
  }'
  CREDENTIALS '{<connector_credentials>}'
  INTO TABLE `<table_name>` FORMAT JSON (
      `data` <- %
  );

  START PIPELINE `<pipeline_name>`;
  ```
  Inspect the resulting records to understand data serialization and determine how to define custom column mappings and `SET` clauses.

## Troubleshoot Custom Connector Pipelines

Enable pipeline extractor debug logging:

```sql
SET GLOBAL pipelines_extractor_debug_logging = ON;
```

## Inspect Pipeline Status

Run the [SHOW PIPELINES](https://docs.singlestore.com/db/v10/reference/sql-reference/show-commands/show-pipelines.md) command to view the pipeline status.

```sql
SHOW PIPELINES;
```

## Monitor Batch Processing

Query the [PIPELINES\_BATCHES](https://docs.singlestore.com/db/v10/reference/information-schema-reference/data-ingest/pipelines-batches.md) information schema view to monitor batch processing and track offset movement using the `SOURCE_OFFSETS_POLLED` and `SOURCE_OFFSETS_NEW` columns.

```sql
SELECT * FROM information_schema.PIPELINES_BATCHES
WHERE PIPELINE_NAME = '<pipeline_name>';
```

## Inspect Source Offsets

Query the [PIPELINES\_SOURCE\_OFFSETS](https://docs.singlestore.com/db/v10/reference/information-schema-reference/data-ingest/pipelines-source-offsets.md) information schema view to inspect offset information for each pipeline task and batch.

```sql
SELECT * FROM information_schema.PIPELINES_SOURCE_OFFSETS
WHERE PIPELINE_NAME = '<pipeline_name>';
```

## Reset Pipeline Offsets

Run the [ALTER PIPELINE ... SET OFFSETS](https://docs.singlestore.com/db/v10/reference/sql-reference/pipelines-commands/alter-pipeline.md) command to reset pipeline offsets to resume processing from a previously processed position.

```sql
ALTER PIPELINE '<pipeline_name>'
SET OFFSETS {EARLIEST | LATEST | <json_source_partition_offset>};
```

## Query Pipeline Errors

Query the [PIPELINES\_ERRORS](https://docs.singlestore.com/db/v10/reference/information-schema-reference/data-ingest/pipelines-errors.md) information schema view to view the pipeline errors.

```sql
SELECT * FROM information_schema.PIPELINES_ERRORS
WHERE PIPELINE_NAME = '<pipeline_name>';
```

## Related Topics

* [Debugging Pipeline Errors](https://docs.singlestore.com/cloud/load-data/about-singlestore-pipelines/pipeline-troubleshooting/debugging-pipeline-errors/)
* [Connector Development Guide](https://kafka.apache.org/39/kafka-connect/connector-development-guide/)

***

Modified at: October 27, 2026

Source: [/db/v10/load-data/data-sources/create-a-custom-connector-pipeline/](https://docs.singlestore.com/db/v10/load-data/data-sources/create-a-custom-connector-pipeline/)

(An index of the documentation is available at /llms.txt)