Create a Custom Connector Pipeline

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:

  Copy

  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 version 2.25.5.

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

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. For Gradle projects, use the Gradle Shadow Plugin.

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:

   Copy

   <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

   Copy

   SET GLOBAL experimental_features_config = 'connector_pipelines_enabled=true';

Create the Pipeline

Create a pipeline using the following command/syntax:

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:

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

  Copy

  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

  Copy

  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:

SET GLOBAL pipelines_extractor_debug_logging = ON;

SHOW PIPELINES;

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

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

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

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

Related Topics

• Debugging Pipeline Errors

• Connector Development Guide