Guidelines for writing a processor

These are processors you'd write yourself in cases the Built-in ones don't meet your needs.

How to write one

Thanks to our Web Assembly (Wasm) processor you can start writing processors in any language that can be compiled to Web Assembly. As a start, Conduit already provides a conduit-processor-sdk that will let you write a processor in Go.

Run conduit processor new <name> to scaffold a building, git-initialized Go processor project in one command, or start from the Conduit processor template directly.

Where to put them

By default, standalone processors are expected to be found in a folder named processors alongside of your pipelines or standalone connectors:

│ # Conduit binary
├── conduit
│ # Folder with pipeline configurations (yaml files)
├── pipelines
│ # Folder with standalone connectors (binary files)
├── connectors
│ # Folder with standalone processors (wasm files)
└── processors

However, in case you need to reference processors in a different location, you could use the -processors.path flag when running Conduit:

./conduit run --processors.path /my-custom-processors-path

Using the conduit-processor-sdk

Assuming you use our conduit-processor-sdk, this is how a processor plugin written in Go could look like. In the following example, we're going to be adding a processed field to each record processed by our pipeline:

//go:build wasm

package main

import (
    "context"

    "github.com/conduitio/conduit-commons/opencdc"
    sdk "github.com/conduitio/conduit-processor-sdk"
)

func main() {
    sdk.Run(sdk.NewProcessorFunc(
        sdk.Specification{Name: "simple-processor", Version: "v1.0.0"},
        func(ctx context.Context, record opencdc.Record) (opencdc.Record, error) {
            record.Payload.After.(opencdc.StructuredData)["processed"] = true
            return record, nil
        },
    ))

Batch processing and stateless processors

The SDK has two opt-in surfaces for hot-path processors. The engine in v0.21 does not use either yet; wiring them in is designed but not scheduled.

• sdk.BatchProcessor — implement ProcessBatch(ctx, []opencdc.Record) ([]ProcessedRecord, error) to process a whole batch at once instead of one record per Process call. The result slice must be the same length as the input (a 1:1 index contract), and ProcessBatch must be idempotent. A processor that doesn't implement it runs unchanged — the engine falls back to Process.
• Specification.Stateless — set Stateless: true on the specification to declare the processor keeps no state between calls, which lets the engine run it in parallel. A processor that declares Stateless but does keep state is a bug; a stateful processor that doesn't declare it is treated as stateful and run serially.

After that, you'd need to compile it, and locate its .wasm file into the desired processors directory as previously mentioned:

GOARCH=wasm GOOS=wasip1 go build -o simple-processor.wasm main.go

Using it in your pipeline

As mentioned in our Getting Started page, in order to use a processor in your pipeline, you need to update its configuration file and reference it accordingly:

version: 2.2
pipelines:
  - id: example-pipeline
    connectors:
    # define source and destination connectors
    # ...
    processors:
      - id: add-processed-field
        plugin: standalone:simple-processor

When running your pipeline again, you should expect seeing a new processed field on every record processed.

info

If you end up writing a standalone processor you'd like to share with the community, please let us know! Post about it on GitHub Discussions.