Help us shape the future of Go! Take the 2026 Go Developer Survey

sdk

package module
v0.17.1 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Oct 17, 2026 License: Apache-2.0 Imports: 57 Imported by: 371

README

Conduit Connector SDK

License Test Go Report Card Go Reference

This repository contains the Go software development kit for implementing a connector for Conduit. If you want to implement a connector in another language please have a look at the connector protocol.

Quickstart

Create a new folder and initialize a fresh go module:

go mod init example.com/conduit-connector-demo

Add the connector SDK dependency:

go get github.com/conduitio/conduit-connector-sdk

With this you can start implementing the connector. To implement a source (a connector that reads from a 3rd party resource and sends data to Conduit) create a struct that implements sdk.Source. To implement a destination (a connector that receives data from Conduit and writes it to a 3rd party resource) create a struct that implements sdk.Destination. You can implement both to make a connector that can be used both as a source or a destination.

Apart from the source and/or destination you should create a global variable of type sdk.Connector that contains references to constructors for sdk.Source, sdk.Destination and sdk.Specification.

The last part is the entrypoint, it needs to call sdk.Serve and pass in the connector mentioned above.

package main

import (
	demo "example.com/conduit-connector-demo"
	sdk "github.com/conduitio/conduit-connector-sdk"
)

func main() {
	sdk.Serve(demo.Connector)
}

Now you can build the standalone connector:

go build path/to/main.go

You will get a compiled binary which Conduit can use as a connector. To run your connector as part of a Conduit pipeline you can create it using the connectors API and specify the path to the compiled connector binary in the field plugin.

Here is an example request to POST /v1/connectors (find more about the Conduit API):

{
  "type": "TYPE_SOURCE",
  "plugin": "/path/to/compiled/connector/binary",
  "pipelineId": "...",
  "config": {
    "name": "my-connector",
    "settings": {
      "my-key": "my-value"
    }
  }
}

Find out more information on building a connector in the Go doc reference.

FAQ

Q: How to identify the source from which a record originated?

A connector can use whatever means available to associate a record with the originating source. However, to promote compatibility between connectors, we highly recommend that a record's metadata is used to indicate from which collection[^1] a record originated.

The metadata key to be used is opencdc.collection, which can be accessed through the sdk.MetadataCollection constant.

For example, if a record was read from a database table called employees, it should have the following in its metadata:

{
  "opencdc.collection": "employees",
  // other metadata
}

Additionally, Conduit automatically adds the following metadata to each record:

  • conduit.source.plugin.name: the source plugin that created a record
  • conduit.source.plugin.version: version of the source plugin that created this record

More information about metadata in OpenCDC records can be found here.

Q: If a destination connector is able to write to multiple tables (topics, collections, indexes, etc.), how should a record be routed to the correct destination?

Similarly to above, we recommend that the metadata key "opencdc.collection" is used.

For example, if a record has the metadata field "opencdc.collection" set to employees, then the PostgreSQL destination connector will write it to the employees table.

Q: Is there a standard format for errors?

Conduit doesn't expect any specific error format. We still encourage developers to follow the conventional error message formatting and include enough contextual information to make debugging as easy as possible (e.g. stack trace, information about the value that caused the error, internal state).

Q: Is there a standard format for logging?

Developers should use sdk.Logger to retrieve a *zerolog.Logger instance. It can be used to emit structured and leveled log messages that will be included in Conduit logs.

Keep in mind that logging in the hot path (e.g. reading or writing a record) can have a negative impact on performance and should be avoided. If you really want to add a log message in the hot path please use the "trace" level.

Q: How do I enable logging in my tests?

By default, logging calls made using the sdk.Logger in your tests will not produce any output. To enable logging while running your connector tests or debugging, you need to pass a custom context with a zerolog logger attached:

func TestFoo(t *testing.T) {
	logger := zerolog.New(zerolog.NewTestWriter(t))
	ctx := logger.WithContext(context.Background())

	// pass ctx to connector functions ...
}

Q: Do I need to worry about ordering?

In case of the destination connector you do not have to worry about ordering. Conduit will supply records one by one in the order they were produced in the source.

On the other hand, the source connector is in charge of producing records and thus dictates the order. That said, you do not have to worry about concurrent reads, the SDK will call Source.Read repeatedly and only in one goroutine, all you have to do is return one record at a time.

Examples

For examples of simple connectors you can look at existing connectors like conduit-connector-generator or conduit-connector-file.

[^1]: Collection is a generic term used in Conduit to describe an entity in a 3rd party system from which records are read from or to which records they are written to. Examples are: topics (in Kafka), tables (in a database), indexes (in a search engine), and collections (in NoSQL databases).

Documentation

Overview

Package sdk implements utilities for implementing a Conduit connector.

Getting started

Conduit connectors can be thought of as the edges of a Conduit pipeline. They are responsible for reading records from and writing records to third party systems. Conduit uses connectors as plugins that hide the intricacies of working with a particular third party system, so that Conduit itself can focus on efficiently processing records and moving them safely from sources to destinations.

To implement a connector, start by defining a global variable of type Connector, preferably in connector.go at the root of your project to make it easy to discover. 

var Connector = sdk.Connector {
    NewSpecification: Specification,  // Specification is my connector's specification
    NewSource:        NewSource,      // NewSource is the constructor for my source
    NewDestination:   NewDestination, // NewDestination is the constructor for my destination
}

Connector will be used as the starting point for accessing three main connector components that you need to provide:

  • Specification contains general information about the plugin like its name and what it does. Writing a specification is relatively simple and straightforward, for more info check the corresponding field docs of Specification.
  • Source is the connector part that knows how to fetch data from the third party system and convert it to a [Record].
  • Destination is the connector part that knows how to write a [Record] to the third party system.

General advice for implementing connectors:

  • The SDK provides a structured logger that can be retrieved with Logger. It allows you to create structured and leveled output that will be included as part of the Conduit logs.
  • If you want to add logging to the hot path (i.e. code that is executed for every record that is read or written) you should use the log level "trace", otherwise it can greatly impact the performance of your connector.

Source

A Source is responsible for continuously reading data from a third party system and returning it in form of a [Record].

Every Source implementation needs to include an UnimplementedSource to satisfy the interface. This allows us to potentially change the interface in the future while remaining backwards compatible with existing Source implementations. 

type Source struct {
  sdk.UnimplementedSource
}

You need to implement the functions required by Source and provide your own implementations. Please look at the documentation of Source for further information about individual functions.

Your source's configuration struct should embed DefaultSourceMiddleware, which contributes the SDK's default source middleware (schema extraction, schema context, encoding and batching) as configuration fields. Return a pointer to that struct from Source.Config so the SDK can populate it from the pipeline configuration: 

type SourceConfig struct {
  sdk.DefaultSourceMiddleware
  // your own configuration fields go here
}

func (s *Source) Config() sdk.SourceConfig {
  return &s.config
}

You should also create a constructor function for your source struct. Note that this is the same function that should be set as the value of Connector.NewSource. The constructor should wrap the source in the middleware declared in its configuration using SourceWithMiddleware, which inspects the struct returned by Config and applies every embedded middleware. Unless you have a very good reason, you should always keep the embedded DefaultSourceMiddleware; add further middleware by embedding additional middleware config structs alongside it. 

func NewSource() sdk.Source {
  return sdk.SourceWithMiddleware(&Source{})
}

Additional tips for implementing a source:

  • The SDK provides utilities for certain operations like creating records in SourceUtil. You can access it through the global variable Util.Source.
  • The function Source.Ack is optional and does not have to be implemented.
  • Source is responsible for creating record positions that should ideally uniquely identify a record. Think carefully about what you will store in the position, it should give the source enough information to resume reading records at that specific position.
  • The SDK provides acceptance tests, if your source doesn't pass it means your implementation has a bug¹.

Destination

A Destination is responsible for writing [Record] to third party systems.

Every Destination implementation needs to include an UnimplementedDestination to satisfy the interface. This allows us to potentially change the interface in the future while remaining backwards compatible with existing Destination implementations. 

type Destination struct {
  sdk.UnimplementedDestination
}

You need to implement the functions required by Destination and provide your own implementations. Please look at the documentation of Destination for further information about individual functions.

Your destination's configuration struct should embed DefaultDestinationMiddleware, which contributes the SDK's default destination middleware (rate limiting, record format, batching and schema extraction) as configuration fields. Return a pointer to that struct from Destination.Config so the SDK can populate it from the pipeline configuration: 

type DestinationConfig struct {
  sdk.DefaultDestinationMiddleware
  // your own configuration fields go here
}

func (d *Destination) Config() sdk.DestinationConfig {
  return &d.config
}

You should also create a constructor function for your destination struct. Note that this is the same function that should be set as the value of Connector.NewDestination. The constructor should wrap the destination in the middleware declared in its configuration using DestinationWithMiddleware, which inspects the struct returned by Config and applies every embedded middleware. Unless you have a very good reason, you should always keep the embedded DefaultDestinationMiddleware; add further middleware by embedding additional middleware config structs alongside it. 

func NewDestination() sdk.Destination {
  return sdk.DestinationWithMiddleware(&Destination{})
}

Additional tips for implementing a destination:

  • The SDK provides utilities for certain operations like routing records based on their operation in DestinationUtil. You can access it through the global variable Util.Destination.
  • If your destination writes records as a whole to the destination you should use [Record].Bytes to get the raw record representation.
  • If possible, make your destination writes idempotent. It is possible that the destination will receive the same record twice after a pipeline restart.
  • Some sources won't be able to distinguish create and update operations. In case your destination is updating data in place, we recommend to upsert the record on a create or update.
  • The SDK provides acceptance tests, if your destination doesn't pass it means your implementation has a bug¹.

Acceptance tests

The SDK provides acceptance tests that can be run in a simple Go test.¹

To run acceptance tests you should create a test file, preferably named acceptance_test.go at the root of your project to make it easy to discover. Inside create a Go test where you trigger the function AcceptanceTest. 

func TestAcceptance(t *testing.T) {
  // set up dependencies here
  sdk.AcceptanceTest(t, sdk.ConfigurableAcceptanceTestDriver{
    Config: sdk.ConfigurableAcceptanceTestDriverConfig{
      Connector: Connector, // Connector is the global variable from your connector
      SourceConfig: map[string]string{ … },
      DestinationConfig: map[string]string{ … },
    },
  }
}

AcceptanceTest uses the AcceptanceTestDriver for certain operations. The SDK already provides a default implementation for the driver with ConfigurableAcceptanceTestDriver, although you can supply your own implementation if you need to adjust the behavior of acceptance tests for your connector.

Some acceptance tests will try to write data using the destination and then read the same data using the source. Because of that you need to make sure that the configurations point both to the same exact data store (e.g. in case of the file connector the source and destination need to read and write to the same file).

If your connector does not implement both sides of the connector (a source and a destination) you will need to write a custom driver that knows how to read or write, depending on which side of the connector is not implemented. Here is an example how to do that: 

type CustomAcceptanceTestDriver struct {
  sdk.ConfigurableAcceptanceTestDriver
}
func (d *CustomAcceptanceTestDriver) ReadFromDestination(t *testing.T, records []opencdc.Record) []opencdc.Record {
  // implement read
}
func (d *CustomAcceptanceTestDriver) WriteToSource(t *testing.T, records []opencdc.Record) []opencdc.Record {
  // implement write
}

For more information about what behavior can be customized please refer to the AcceptanceTestDriver interface.

Acceptance tests v2

AcceptanceTestV2 runs the v1 tests unchanged and then the v2 tests, which drive the connector the way Conduit does (pipelined batches, asynchronous acks, Stop, Teardown) and check what v1 cannot see: that acked records are stored, that a crash and replay loses nothing, that batches of 1, 100 and 10000 records work, that large records survive, that cancellation and a graceful stop do not hang or leak. Passing both groups is what "passed suite v2" means; AcceptanceSuiteV2Revision says which revision of the suite that was. 

func TestAcceptanceV2(t *testing.T) {
  sdk.AcceptanceTestV2(t, sdk.ConfigurableAcceptanceTestDriver{
    Config: sdk.ConfigurableAcceptanceTestDriverConfig{
      Connector: Connector,
      SourceConfig: map[string]string{ … },
      DestinationConfig: map[string]string{ … },
      Capabilities: sdk.AcceptanceCapabilities{
        Destination: sdk.DestinationCapabilities{Idempotence: sdk.IdempotenceYes},
      },
    },
  })
}

Some tests only make sense if the connector has a property, for example idempotent writes, ordering per key, exact resume or a schema evolution policy. They run only if the connector claims the property in AcceptanceCapabilities, and a claim that is made is verified; a claim the suite cannot verify (for example idempotence with a driver that cannot return duplicates) fails the case instead of skipping it. Without claims only the baseline runs, and the report lists what was skipped. Source-only and sink-only connectors declare SourceCapabilities.Synthetic and DestinationCapabilities.WriteOnly.

Set the environment variable AcceptanceReportEnv to a file path to get a JSON AcceptanceReport with the suite revision, the claims, every test's outcome and the recorded throughput numbers. The throughput checks never fail a test.

¹Acceptance tests are currently still experimental.

Index

Constants

View Source 
const (
	// CertificationMinBatchSize is the smallest MaxBatchSize that counts for
	// certification: the default.
	CertificationMinBatchSize = 10000
	// CertificationMinRecordBytes is the smallest MaxRecordBytes that counts
	// for certification: the default.
	CertificationMinRecordBytes = 1 << 20
)

Limits a registry can require for a "certified" result. The suite runs at the declared values (a connector with a hard limit still gets tested), and the report says whether the floors are met.

View Source 
const (
	AcceptanceOutcomePassed  = "passed"
	AcceptanceOutcomeFailed  = "failed"
	AcceptanceOutcomeSkipped = "skipped"
)

Outcomes of an AcceptanceCaseResult.

View Source 
const (
	// SkipNotApplicable: the connector has no such side (no source, no destination).
	SkipNotApplicable = "not_applicable"
	// SkipUnclaimed: the case needs a capability the connector did not claim.
	SkipUnclaimed = "unclaimed"
	// SkipDeclared: the connector declared Synthetic or WriteOnly.
	SkipDeclared = "declared"
	// SkipReplaced: a v1 data path test that suite v2 replaces with a v2 case.
	SkipReplaced = "replaced"
	// SkipInCase: the case skipped itself. Counted as unexplained.
	SkipInCase = "in_case"
	// SkipDriver: skipped before the case ran, by the driver's BeforeTest
	// (for example a Skip regular expression). The reason is not known to the
	// suite. Counted as unexplained.
	SkipDriver = "driver"
)

Kinds of skip, see AcceptanceCaseResult.SkipKind.

View Source 
const AcceptanceReportEnv = "CONDUIT_ACCEPTANCE_REPORT"

AcceptanceReportEnv is the name of the environment variable that makes AcceptanceTestV2 write a machine-readable AcceptanceReport (JSON) to the path it contains. The file is written when the test finishes, regardless of the outcome.

View Source 
const AcceptanceSuiteV2Revision = "2.0.1"

AcceptanceSuiteV2Revision is the revision of the v2 acceptance suite. The major number is the bar ("suite v2"), the minor number is bumped when tests are added that only apply to connectors claiming a capability, and a patch number is bumped for fixes that do not change what a connector has to do. The value is written into the report (see AcceptanceReportEnv) so that whoever consumes the report knows which bar was passed.

View Source 
const DefaultMaxRedelivery = 10

DefaultMaxRedelivery is how many already acked records a source may deliver again after a restart when it does not claim ExactResume.

View Source 
const MaxAcceptedRedelivery = 1000

MaxAcceptedRedelivery is the largest MaxRedelivery the suite accepts. SRC-03 sizes its data set to more than twice the declared bound so that a source that starts over cannot hide inside it; beyond this value the data set would be unreasonably large and the claim says nothing, so SRC-03 fails.

View Source 
const ReportSchemaVersion = 1

ReportSchemaVersion is the version of the JSON layout of AcceptanceReport.

Variables

View Source 
var (
	// ErrBackoffRetry can be returned by Source.Read to signal the SDK there is
	// no record to fetch right now and it should try again later.
	ErrBackoffRetry = errors.New("backoff retry")

	// ErrUnimplemented is returned in functions of plugins that don't implement
	// a certain method.
	ErrUnimplemented = errors.New("the connector plugin does not implement this action, please check the source code of the connector and make sure all required connector methods are implemented")
)
View Source 
var Util = struct {
	// SourceUtil provides utility methods for implementing a source.
	Source SourceUtil
	// SourceUtil provides utility methods for implementing a destination.
	Destination DestinationUtil

	// ParseConfig sanitizes the configuration, applies defaults, validates it and
	// copies the values into the target object. It combines the functionality
	// provided by github.com/conduitio/conduit-commons/config.Config into a single
	// convenient function. It is intended to be used in the Configure method of a
	// connector to parse the configuration map.
	//
	// The configuration parameters should be provided through NewSource().Parameters()
	// or NewDestination().Parameters() so that parameters from SDK middlewares are
	// included too.
	//
	// The function does the following:
	//   - Removes leading and trailing spaces from all keys and values in the
	//     configuration.
	//   - Applies the default values defined in the parameter specifications to the
	//     configuration.
	//   - Validates the configuration by checking for unrecognized parameters, type
	//     validations, and value validations.
	//   - Copies configuration values into the target object. The target object must
	//     be a pointer to a struct.
	ParseConfig func(ctx context.Context, cfg config.Config, target any, parameters config.Parameters) error
}{
	ParseConfig: parseConfig,
}

Util provides utilities for implementing connectors.

Functions

func AcceptanceTest added in v0.3.0 

func AcceptanceTest(t *testing.T, driver AcceptanceTestDriver)

AcceptanceTest is the acceptance test that all connector implementations should pass. It should manually be called from a test case in each implementation: 

func TestAcceptance(t *testing.T) {
    // set up test dependencies ...
    sdk.AcceptanceTest(t, sdk.ConfigurableAcceptanceTestDriver{
        Config: sdk.ConfigurableAcceptanceTestDriverConfig{
            Connector: myConnector,
            SourceConfig: config.Config{...},      // valid source config
            DestinationConfig: config.Config{...}, // valid destination config
        },
    })
}

func AcceptanceTestV2 added in v0.16.0 

func AcceptanceTestV2(t *testing.T, driver AcceptanceTestDriver)

AcceptanceTestV2 runs version 2 of the acceptance suite. The v1 specification and configuration tests run unchanged as the subtest "Core"; the v1 data path tests are replaced by the v2 cases, which cover the same ground through the plugin interface and with acks (see skipCore for why). The v2 cases run as the subtest "V2". A connector has passed suite v2 only if both groups pass. Existing connectors keep calling AcceptanceTest; nothing changes for them.

The v2 tests drive the connector the way Conduit does, through the plugin interface with a pipelined, asynchronously acknowledged stream, so that the SDK's batching and ack handling are exercised together with the connector. Tests that only apply to connectors with a certain capability run only if the driver declares it (see AcceptanceCapabilities). 

func TestAcceptanceV2(t *testing.T) {
    sdk.AcceptanceTestV2(t, sdk.ConfigurableAcceptanceTestDriver{
        Config: sdk.ConfigurableAcceptanceTestDriverConfig{
            Connector:         myConnector,
            SourceConfig:      config.Config{...},
            DestinationConfig: config.Config{...},
            Capabilities: sdk.AcceptanceCapabilities{
                Destination: sdk.DestinationCapabilities{Idempotence: sdk.IdempotenceYes},
            },
        },
    })
}

Set the environment variable AcceptanceReportEnv to a file path to get a JSON AcceptanceReport of the run.

func BenchmarkSource added in v0.9.0 

func BenchmarkSource(
	b *testing.B,
	s Source,
	cfg map[string]string,
)

BenchmarkSource is a benchmark that any source implementation can run to figure out its performance. The benchmark expects that the source resource contains at least b.N number of records. This should be prepared before the benchmark is executed. The function should be manually called from a benchmark function: 

func BenchmarkConnector(b *testing.B) {
    // set up test dependencies and write b.N records to source resource ...
    sdk.BenchmarkSource(
        b,
        mySourceConnector,
        map[string]string{...}, // valid source config
    )
}

The benchmark can be run with a specific number of records by supplying the option -benchtime=Nx, where N is the number of records to be benchmarked (e.g. -benchtime=100x benchmarks reading 100 records).

func ConnectorIDFromContext added in v0.10.1 

func ConnectorIDFromContext(ctx context.Context) string

ConnectorIDFromContext fetches the connector ID from the context. If the context does not contain a connector ID it returns an empty string.

func Logger 

func Logger(ctx context.Context) *zerolog.Logger

Logger returns an instance of a logger that can be used for leveled and structured logging in a plugin. The logger will respect the log level configured in Conduit.

func NewDestinationPlugin 

func NewDestinationPlugin(impl Destination, cfg pconnector.PluginConfig, parameters config.Parameters) pconnector.DestinationPlugin

NewDestinationPlugin takes a Destination and wraps it into an adapter that converts it into a pconnector.DestinationPlugin. If the parameter is nil it will wrap UnimplementedDestination instead.

func NewSourcePlugin 

func NewSourcePlugin(impl Source, cfg pconnector.PluginConfig, parameters config.Parameters) pconnector.SourcePlugin

NewSourcePlugin takes a Source and wraps it into an adapter that converts it into a pconnector.SourcePlugin. If the Source is nil it will wrap UnimplementedSource instead.

func NewSpecifierPlugin 

func NewSpecifierPlugin(specs Specification) pconnector.SpecifierPlugin

NewSpecifierPlugin takes a Specification and wraps it into an adapter that converts it into a pconnector.SpecifierPlugin.

func Serve 

func Serve(c Connector)

Serve starts the plugin and takes care of its whole lifecycle by blocking until the plugin can safely stop running. Any fixable errors will be output to os.Stderr and the process will exit with a status code of 1. Serve will panic for unexpected conditions where a user's fix is unknown.

It is essential that nothing gets written to stdout or stderr before this function is called, as the first output is used to perform the initial handshake.

Plugins should call Serve in their main() functions.

func YAMLSpecification added in v0.13.0 

func YAMLSpecification(rawYaml, version string) func() Specification

YAMLSpecification parses a Specification from the given YAML file. The connector version found in the YAML file can be overridden with the `version` parameter.

Types

type AcceptanceCapabilities added in v0.16.0 

type AcceptanceCapabilities struct {
	Source      SourceCapabilities      `json:"source"`
	Destination DestinationCapabilities `json:"destination"`

	// MaxBatchSize is the largest batch the suite sends in the mixed batch
	// size tests. The default is 10000. Lower it only if the 3rd party system
	// has a hard limit that Conduit's batching configuration cannot work
	// around. The report states whether [CertificationMinBatchSize] is met.
	MaxBatchSize int `json:"max_batch_size,omitempty"`
	// MaxRecordBytes is the size of the record used in the large record tests.
	// The default is 1 MiB. Values above 3 MiB are capped, because the default
	// maximum gRPC message size for connector plugins is 4 MiB. The report
	// states whether [CertificationMinRecordBytes] is met.
	MaxRecordBytes int `json:"max_record_bytes,omitempty"`
	// ThroughputDuration is how long the throughput smoke checks run. The
	// default is 3 seconds. The result is recorded in the report, never
	// asserted.
	ThroughputDuration time.Duration `json:"throughput_duration_ns,omitempty"`

	// Waivers names the limits or exceptions the connector asks the registry
	// maintainers to waive (for example "max_redelivery", "max_batch_size").
	// Declaring a waiver waives nothing: it puts the request in the report, and
	// the registry entry must carry a matching approval from a maintainer
	// (who, when, why) for the Certified check to accept it.
	Waivers []string `json:"waivers,omitempty"`
}

AcceptanceCapabilities declares what a connector claims to support, beyond the baseline every connector must meet. A conditional test runs only when the matching claim is made; a claim that is made is verified. Claims are part of the report, so a registry can show what a "passed" result actually covered.

The zero value claims nothing, which is always safe: only the baseline tests run.

type AcceptanceCapabilitiesProvider added in v0.16.0 

type AcceptanceCapabilitiesProvider interface {
	AcceptanceCapabilities() AcceptanceCapabilities
}

AcceptanceCapabilitiesProvider is optionally implemented by an AcceptanceTestDriver to declare the connector's capabilities. ConfigurableAcceptanceTestDriver implements it using ConfigurableAcceptanceTestDriverConfig.Capabilities. AcceptanceTest (v1) ignores it.

type AcceptanceCaseResult added in v0.16.0 

type AcceptanceCaseResult struct {
	ID      string `json:"id"`
	Name    string `json:"name"`
	Outcome string `json:"outcome"` // one of the AcceptanceOutcome constants
	// SkipKind classifies a skip, see the Skip constants.
	SkipKind string `json:"skip_kind,omitempty"`
	// Reason says why a test was skipped, e.g. which capability was not
	// claimed.
	Reason          string  `json:"reason,omitempty"`
	DurationSeconds float64 `json:"duration_seconds"`
}

AcceptanceCaseResult is the outcome of one case.

type AcceptanceKeyedRecordGenerator added in v0.16.0 

type AcceptanceKeyedRecordGenerator interface {
	GenerateRecordWithKey(t *testing.T, op opencdc.Operation, key string) opencdc.Record
}

AcceptanceKeyedRecordGenerator is optionally implemented by an AcceptanceTestDriver to produce records with a given key, for the ordering tests. Without it the suite takes a generated record and replaces its key with raw data.

type AcceptanceLargeRecordGenerator added in v0.16.0 

type AcceptanceLargeRecordGenerator interface {
	GenerateLargeRecord(t *testing.T, op opencdc.Operation, size int) opencdc.Record
}

AcceptanceLargeRecordGenerator is optionally implemented by an AcceptanceTestDriver to produce a record with a payload of roughly the given size, for connectors where padding a generated record would not fit the target (e.g. a fixed table schema).

type AcceptanceMetric added in v0.16.0 

type AcceptanceMetric struct {
	Name  string  `json:"name"`
	Value float64 `json:"value"`
	Unit  string  `json:"unit"`
	// Detail documents the conditions of the measurement.
	Detail string `json:"detail,omitempty"`
}

AcceptanceMetric is one recorded measurement.

type AcceptanceProcessDeathDriver added in v0.16.0 

type AcceptanceProcessDeathDriver interface {
	// KilledSessionID returns the identifier of the server-side session the
	// connector's source holds, for example "conduitslot".
	KilledSessionID(*testing.T) string
	// ReleaseSession ends the session with that identifier, as process death
	// would.
	ReleaseSession(ctx context.Context, sessionID string) error
}

AcceptanceProcessDeathDriver is optionally implemented by an AcceptanceTestDriver to do, on the system the connector talks to, what the death of a connector process does. The suite kills a source by cancelling its context and leaving it alone (no Stop, no Teardown); a real kill also closes the process's sockets, which ends server-side sessions such as the replication slot walsender of a dead instance. In-process that never happens, and a successor can find the session still attached.

The signature is deliberately narrow. The driver names the session once (AcceptanceProcessDeathDriver.KilledSessionID, for example a replication slot name) and the suite later asks it to release exactly that (AcceptanceProcessDeathDriver.ReleaseSession) with a context and the identifier: the suite hands the hook no source instance, no test handle, no record. That is a property of the signature, not a guarantee: the driver creates the source (Connector().NewSource), so a driver can still reach the instance some other way, and can rescue a connector with or without the hook. With the hook the rescue is flagged and counted in the report; without it, it is not flagged at all and is caught only by the attested kill -9 test. The hook is meant for what the operating system or the server would do on its own (like Postgres's pg_terminate_backend of the walsender holding the slot) and must never persist, commit or flush anything on the connector's behalf.

Using the hook is recorded in the report (exception "process_death_hook"). A connector that needs it has NOT passed a real kill: the in-process kill is an approximation, and the attested kill -9 test stays mandatory for Certified.

type AcceptanceReadAllDriver added in v0.16.0 

type AcceptanceReadAllDriver interface {
	ReadAllFromDestination(*testing.T) []opencdc.Record
}

AcceptanceReadAllDriver is optionally implemented by an AcceptanceTestDriver that can return everything currently stored in the destination, duplicates included. The replay cases need it to tell at-least-once (duplicates are fine) from lost records, and the idempotence claim cannot be verified without it. It is an explicit opt-in: either implement this method on your driver, or set ConfigurableAcceptanceTestDriverConfig.ReadAllFromSource to read the whole source. ConfigurableAcceptanceTestDriver does not satisfy this interface by itself, so a driver that embeds it and overrides ReadFromDestination is never silently read through the source instead. Without either, the suite falls back to AcceptanceTestDriver.ReadFromDestination, which cannot see duplicates.

type AcceptanceReport added in v0.16.0 

type AcceptanceReport struct {
	// ReportSchemaVersion is the version of this layout.
	ReportSchemaVersion int `json:"report_schema_version"`
	// SuiteVersion is the major version of the suite (2).
	SuiteVersion int `json:"suite_version"`
	// SuiteRevision is the full revision, see [AcceptanceSuiteV2Revision]. The
	// revision together with CaseSetHash identifies exactly which cases a
	// connector had to pass.
	SuiteRevision string `json:"suite_revision"`
	// SDKVersion is the version of the SDK module the test binary was built
	// with: from the build information if it lists the SDK, otherwise from the
	// module path of the SDK source (binaries built by go test have no
	// dependency list). "local checkout" means a replace directive or
	// workspace, "unknown" that neither source says. It is informational:
	// compatibility is decided by SuiteRevision and CaseSetHash.
	SDKVersion string `json:"sdk_version"`
	// SDKVersionSource says how SDKVersion was found (build_info,
	// module_cache_path, local_checkout, vendored, unknown). Informational.
	SDKVersionSource string `json:"sdk_version_source"`
	// Connector identifies the connector under test.
	Connector struct {
		Name    string `json:"name"`
		Version string `json:"version"`
	} `json:"connector"`
	// Capabilities are the claims the connector made.
	Capabilities AcceptanceCapabilities `json:"capabilities"`
	// Limits are the batch and record sizes the suite used, and whether they
	// meet the certification floors.
	Limits struct {
		MaxBatchSize          int      `json:"max_batch_size"`
		MaxRecordBytes        int      `json:"max_record_bytes"`
		MaxRedelivery         int      `json:"max_redelivery"`
		CertificationFloorMet bool     `json:"certification_floor_met"`
		BelowFloor            []string `json:"below_floor,omitempty"`
	} `json:"limits"`
	// Goleak records what the driver told the goroutine leak check to ignore.
	// Names are only available for ignores declared through
	// [ConfigurableAcceptanceTestDriverConfig.GoleakIgnoreFunctions].
	// goleak.Option is opaque, so options passed any other way cannot be
	// listed: OpaqueOptions counts them and Unreviewable says that the
	// ignore list is incomplete. A registry should require zero.
	Goleak struct {
		IgnoredFunctions []string `json:"ignored_functions,omitempty"`
		OpaqueOptions    int      `json:"opaque_options"`
		Unreviewable     bool     `json:"unreviewable"`
	} `json:"goleak"`
	// RunID is random, chosen when the run starts. A registry that runs the
	// suite itself can tie a report to its own run through it (for example by
	// passing it in the environment of the job log) and can tell two reports of
	// one connector version apart.
	RunID string `json:"run_id"`
	// TestBinarySHA256 is the hash of the test executable that produced the
	// report; SuiteSourceHash is the hash of the suite's own source files as
	// they were compiled in. A registry-side re-run compares both with the
	// values it computes for the pinned SDK revision and the connector build.
	//
	// The binary hash is stable only in a pinned build environment: the module
	// path of the build, -race, CGO, the toolchain and GOOS/GOARCH all change
	// it (and without -trimpath, so do the directories). It matches a re-run
	// with identical build settings and is informational otherwise; Build
	// records the settings so a registry can tell. CI should build with
	// -trimpath.
	TestBinarySHA256 string `json:"test_binary_sha256"`
	// Build records the settings the test binary was built with.
	Build           BuildSettings `json:"build"`
	SuiteSourceHash string        `json:"suite_source_hash"`
	// Exceptions lists everything that weakens the run and that a maintainer
	// has to approve: "synthetic", "write_only", "process_death_hook",
	// "max_redelivery_above_default", and "waiver:<name>" for every declared
	// waiver. Empty means none. See the design document.
	Exceptions []string `json:"exceptions"`
	// ProcessDeathHookCalls counts how often the driver's
	// [AcceptanceProcessDeathDriver] released a session.
	ProcessDeathHookCalls int `json:"process_death_hook_calls"`
	// ExpectedCaseIDs is every case that must have a result for the run to be
	// complete, in execution order.
	ExpectedCaseIDs []string `json:"expected_case_ids"`
	// ExpectedCaseCount is len(ExpectedCaseIDs).
	ExpectedCaseCount int `json:"expected_case_count"`
	// CaseSetHash is the SHA-256 of the suite revision and the sorted expected
	// IDs.
	CaseSetHash string `json:"case_set_hash"`
	// CaseContentHash is the SHA-256 of the case IDs, names and logic versions
	// (not the revision), see caseContentHash.
	CaseContentHash string `json:"case_content_hash"`
	// Complete is true if every expected case has a result. A run filtered with
	// -run is never complete.
	Complete bool `json:"complete"`
	// UnexplainedSkips counts skips that are not explained by a structural
	// reason, an unclaimed capability, or a declaration. A skip by the driver
	// or inside a case counts.
	UnexplainedSkips int `json:"unexplained_skips"`
	// Finished is false in the report written when the run starts and true in
	// the one written when it ends. A report with finished=false is the trace
	// of a run that did not end (a crash); it never has passed=true.
	Finished bool `json:"finished"`
	// Passed is true if no test failed, the run is complete and there are no
	// unexplained skips.
	Passed bool `json:"passed"`
	// Cases lists every case with its outcome.
	Cases []AcceptanceCaseResult `json:"cases"`
	// Metrics are recorded measurements. They are never pass/fail criteria.
	Metrics []AcceptanceMetric `json:"metrics,omitempty"`
	// StartedAt is when the run started.
	StartedAt time.Time `json:"started_at"`
	// DurationSeconds is the wall-clock duration of the run.
	DurationSeconds float64 `json:"duration_seconds"`
	// contains filtered or unexported fields
}

AcceptanceReport is the machine-readable result of AcceptanceTestV2.

The report is evidence that a run happened, not proof: it is written by the process that ran the tests. A registry that wants an authenticated result re-runs the suite in CI it controls and uses this layout to compare the outcome with the expected case set (AcceptanceReport.ExpectedCaseIDs).

type AcceptanceTestDriver added in v0.3.0 

type AcceptanceTestDriver interface {
	// Context returns the context to use in tests.
	Context() context.Context
	// Connector is the connector to be tested.
	Connector() Connector

	// SourceConfig should be a valid config for a source connector, reading
	// from the same location as the destination will write to.
	SourceConfig(*testing.T) config.Config
	// DestinationConfig should be a valid config for a destination connector,
	// writing to the same location as the source will read from.
	DestinationConfig(*testing.T) config.Config

	// BeforeTest is executed before each acceptance test.
	BeforeTest(*testing.T)
	// AfterTest is executed after each acceptance test.
	AfterTest(*testing.T)

	// GoleakOptions will be applied to goleak.VerifyNone. Can be used to
	// suppress false positive goroutine leaks.
	GoleakOptions(*testing.T) []goleak.Option

	// GenerateRecord will generate a new Record for a certain Operation. It's
	// the responsibility of the AcceptanceTestDriver implementation to provide
	// records with appropriate contents (e.g. appropriate type of payload).
	// The generated record will contain mixed data types in the field Key and
	// Payload (i.e. RawData and StructuredData), unless configured otherwise
	// (see ConfigurableAcceptanceTestDriverConfig.GenerateDataType).
	GenerateRecord(*testing.T, opencdc.Operation) opencdc.Record

	// WriteToSource receives a slice of records that should be prepared in the
	// 3rd party system so that the source will read them. The returned slice
	// will be used to verify the source connector can successfully execute
	// reads.
	// It is encouraged for the driver to return the same slice, unless there is
	// no way to write the records to the 3rd party system, then the returning
	// slice should contain the expected records a source should read.
	WriteToSource(*testing.T, []opencdc.Record) []opencdc.Record
	// ReadFromDestination should return a slice with the records that were
	// written to the destination. The slice will be used to verify the
	// destination has successfully executed writes.
	// The parameter contains records that were actually written to the
	// destination. These will be compared to the returned slice of records. It
	// is encouraged for the driver to only touch the input records to change
	// the order of records and to not change the records themselves.
	ReadFromDestination(*testing.T, []opencdc.Record) []opencdc.Record

	// ReadTimeout controls the time the test should wait for a read operation
	// to return before it considers the operation as failed.
	ReadTimeout() time.Duration
	// WriteTimeout controls the time the test should wait for a write operation
	// to return before it considers the operation as failed.
	WriteTimeout() time.Duration
}

AcceptanceTestDriver is the object that each test uses for fetching the connector and its configurations. The SDK provides a default implementation ConfigurableAcceptanceTestDriver that should fit most use cases. In case more flexibility is needed you can create your own driver, include the default driver in the struct and override methods as needed.

type BuildSettings added in v0.16.0 

type BuildSettings struct {
	GoVersion string `json:"go_version"`
	GOOS      string `json:"goos"`
	GOARCH    string `json:"goarch"`
	Compiler  string `json:"compiler,omitempty"`
	Race      bool   `json:"race"`
	Trimpath  bool   `json:"trimpath"`
	CGO       bool   `json:"cgo"`
	Tags      string `json:"tags,omitempty"`
}

BuildSettings are the build settings of the test binary, from its build information.

type ConfigurableAcceptanceTestDriver added in v0.3.0 

type ConfigurableAcceptanceTestDriver struct {
	Config ConfigurableAcceptanceTestDriverConfig
}

ConfigurableAcceptanceTestDriver is the default implementation of AcceptanceTestDriver. It provides a convenient way of configuring the driver without the need of implementing a custom driver from scratch.

func (ConfigurableAcceptanceTestDriver) AcceptanceCapabilities added in v0.16.0 

func (d ConfigurableAcceptanceTestDriver) AcceptanceCapabilities() AcceptanceCapabilities

AcceptanceCapabilities implements AcceptanceCapabilitiesProvider.

func (ConfigurableAcceptanceTestDriver) AfterTest added in v0.3.0 

func (d ConfigurableAcceptanceTestDriver) AfterTest(t *testing.T)

func (ConfigurableAcceptanceTestDriver) BeforeTest added in v0.3.0 

func (d ConfigurableAcceptanceTestDriver) BeforeTest(t *testing.T)

func (ConfigurableAcceptanceTestDriver) Connector added in v0.3.0 

func (d ConfigurableAcceptanceTestDriver) Connector() Connector

func (ConfigurableAcceptanceTestDriver) Context added in v0.9.0 

func (d ConfigurableAcceptanceTestDriver) Context() context.Context

func (ConfigurableAcceptanceTestDriver) DestinationConfig added in v0.3.0 

func (d ConfigurableAcceptanceTestDriver) DestinationConfig(*testing.T) config.Config

func (ConfigurableAcceptanceTestDriver) GenerateData added in v0.3.0 

func (d ConfigurableAcceptanceTestDriver) GenerateData(t *testing.T) opencdc.Data

GenerateData generates either RawData or StructuredData depending on the configured data type (see ConfigurableAcceptanceTestDriverConfig.GenerateDataType).

func (ConfigurableAcceptanceTestDriver) GenerateRecord added in v0.3.0 

func (d ConfigurableAcceptanceTestDriver) GenerateRecord(t *testing.T, op opencdc.Operation) opencdc.Record

func (ConfigurableAcceptanceTestDriver) GenerateValue added in v0.3.0 

func (d ConfigurableAcceptanceTestDriver) GenerateValue(t *testing.T) interface{}

GenerateValue generates a random value of a random builtin type.

func (ConfigurableAcceptanceTestDriver) GoleakIgnoredFunctions added in v0.16.0 

func (d ConfigurableAcceptanceTestDriver) GoleakIgnoredFunctions() []string

GoleakIgnoredFunctions returns the names in Config.GoleakIgnoreFunctions.

func (ConfigurableAcceptanceTestDriver) GoleakOptions added in v0.3.0 

func (d ConfigurableAcceptanceTestDriver) GoleakOptions(_ *testing.T) []goleak.Option

func (ConfigurableAcceptanceTestDriver) GoleakOtherOptionCount added in v0.16.0 

func (d ConfigurableAcceptanceTestDriver) GoleakOtherOptionCount() int

GoleakOtherOptionCount returns the number of opaque options in Config.GoleakOptions.

func (ConfigurableAcceptanceTestDriver) ReadAllFromSource added in v0.16.0 

func (d ConfigurableAcceptanceTestDriver) ReadAllFromSource(t *testing.T) ([]opencdc.Record, bool)

ReadAllFromSource returns everything the source reads, including duplicates, if the driver opted in with Config.ReadAllFromSource; the second result is false otherwise. It opens the source and reads until the source has been quiet for ReadTimeout.

It deliberately is not called ReadAllFromDestination: embedding this driver must not satisfy AcceptanceReadAllDriver, because a driver that overrides ReadFromDestination would then be read through the source behind its back.

func (ConfigurableAcceptanceTestDriver) ReadFromDestination added in v0.3.0 

func (d ConfigurableAcceptanceTestDriver) ReadFromDestination(t *testing.T, records []opencdc.Record) []opencdc.Record

ReadFromDestination by default opens the source and reads all records from the source. It is expected that the destination is writing to the same location the source is reading from. If the connector does not implement a source the function will fail the test.

func (ConfigurableAcceptanceTestDriver) ReadTimeout added in v0.3.0 

func (d ConfigurableAcceptanceTestDriver) ReadTimeout() time.Duration

func (ConfigurableAcceptanceTestDriver) Skip added in v0.3.0 

func (d ConfigurableAcceptanceTestDriver) Skip(t *testing.T)

func (ConfigurableAcceptanceTestDriver) SourceConfig added in v0.3.0 

func (d ConfigurableAcceptanceTestDriver) SourceConfig(*testing.T) config.Config

func (ConfigurableAcceptanceTestDriver) WriteTimeout added in v0.3.0 

func (d ConfigurableAcceptanceTestDriver) WriteTimeout() time.Duration

func (ConfigurableAcceptanceTestDriver) WriteToSource added in v0.3.0 

func (d ConfigurableAcceptanceTestDriver) WriteToSource(t *testing.T, records []opencdc.Record) []opencdc.Record

WriteToSource by default opens the destination and writes records to the destination. It is expected that the destination is writing to the same location the source is reading from. If the connector does not implement a destination the function will fail the test.

type ConfigurableAcceptanceTestDriverConfig added in v0.3.0 

type ConfigurableAcceptanceTestDriverConfig struct {
	// Context is the context to use in tests. The default is a context with a
	// logger that writes to the test output.
	//nolint:containedctx // We are using this as a configuration option.
	Context context.Context
	// Connector is the connector to be tested.
	Connector Connector

	// SourceConfig should be a valid config for a source connector, reading
	// from the same location as the destination will write to.
	SourceConfig config.Config
	// DestinationConfig should be a valid config for a destination connector,
	// writing to the same location as the source will read from.
	DestinationConfig config.Config

	// BeforeTest is executed before each acceptance test.
	BeforeTest func(t *testing.T)
	// AfterTest is executed after each acceptance test.
	AfterTest func(t *testing.T)

	// GoleakOptions will be applied to goleak.VerifyNone. Can be used to
	// suppress false positive goroutine leaks.
	GoleakOptions []goleak.Option
	// GoleakIgnoreFunctions is like GoleakOptions with goleak.IgnoreAnyFunction
	// for each name, and the names are written to the acceptance report. Prefer
	// it over GoleakOptions: goleak options are opaque, so a registry can only
	// see that they exist, not what they hide.
	GoleakIgnoreFunctions []string

	// ReadAllFromSource makes the driver implement reading everything stored in
	// the destination (duplicates included) by opening the source and reading
	// until it has been quiet for ReadTimeout. It is an explicit opt-in used by
	// AcceptanceTestV2 and is only right if the source reads exactly what the
	// destination wrote. A driver that overrides ReadFromDestination must
	// implement AcceptanceReadAllDriver itself instead.
	ReadAllFromSource bool

	// Skip is a slice of regular expressions used to identify tests that should
	// be skipped. The full test name will be matched against all regular
	// expressions and the test will be skipped if a match is found.
	Skip []string

	// GenerateDataType controls which Data type will be generated in test
	// records. The default is GenerateMixedData which will produce both RawData
	// and StructuredData. To generate only one type of data set this field to
	// GenerateRawData or GenerateStructuredData.
	GenerateDataType GenerateDataType

	// ReadTimeout controls the time the test should wait for a read operation
	// to return a record before it considers the operation as failed. The
	// default timeout is 5 seconds. This value should be changed only if there
	// is a good reason (uncontrollable limitations of the 3rd party system).
	ReadTimeout time.Duration
	// WriteTimeout controls the time the test should wait for a write operation
	// to return a record before it considers the operation as failed. The
	// default timeout is 5 seconds. This value should be changed only if there
	// is a good reason (uncontrollable limitations of the 3rd party system).
	WriteTimeout time.Duration

	// Capabilities declares the optional claims of the connector. It is only
	// used by AcceptanceTestV2, AcceptanceTest ignores it.
	Capabilities AcceptanceCapabilities
}

ConfigurableAcceptanceTestDriverConfig contains the configuration for ConfigurableAcceptanceTestDriver.

type Connector added in v0.3.0 

type Connector struct {
	// NewSpecification should create a new Specification that describes the
	// connector. This field is mandatory, if it is empty the connector won't
	// work.
	NewSpecification func() Specification
	// NewSource should create a new Source plugin. If the plugin doesn't
	// implement a source connector this field can be nil.
	NewSource func() Source
	// NewDestination should create a new Destination plugin. If the plugin
	// doesn't implement a destination connector this field can be nil.
	NewDestination func() Destination
}

Connector combines all constructors for each plugin into one struct.

type Converter added in v0.5.0 

type Converter interface {
	// Name returns the identifier of the converter, used as the first half of a
	// record format name (e.g. "opencdc" or "debezium").
	Name() string
	// Configure returns a new Converter configured with the given options; it
	// does not mutate the receiver.
	Configure(map[string]string) (Converter, error)
	// Convert transforms the record into the target structure. It does not
	// encode it to bytes; that is the Encoder's job.
	Convert(opencdc.Record) (any, error)
}

Converter is a type that can change the structure of a Record. It's used in destination connectors to change the output structure (e.g. opencdc records, debezium records etc.).

type DebeziumConverter added in v0.5.0 

type DebeziumConverter struct {
	SchemaName string
	RawDataKey string
}

DebeziumConverter outputs a Debezium record.

func (DebeziumConverter) Configure added in v0.5.0 

func (c DebeziumConverter) Configure(opt map[string]string) (Converter, error)

func (DebeziumConverter) Convert added in v0.5.0 

func (c DebeziumConverter) Convert(r opencdc.Record) (any, error)

func (DebeziumConverter) Name added in v0.5.0 

func (c DebeziumConverter) Name() string

type DefaultDestinationMiddleware added in v0.3.0 

type DefaultDestinationMiddleware struct {
	UnimplementedDestinationConfig

	DestinationWithRateLimit
	DestinationWithRecordFormat
	DestinationWithBatch
	DestinationWithSchemaExtraction
}

DefaultDestinationMiddleware should be embedded in the DestinationConfig struct to provide a list of default middlewares. Note that if the embedding struct overwrites Validate manually, it should call Validate on this struct as well.

func (*DefaultDestinationMiddleware) Validate added in v0.13.0 

func (c *DefaultDestinationMiddleware) Validate(ctx context.Context) error

Validate validates all the Validatable structs in the middleware.

type DefaultSourceMiddleware added in v0.3.0 

type DefaultSourceMiddleware struct {
	UnimplementedSourceConfig

	SourceWithBatch
	SourceWithEncoding
	SourceWithSchemaContext
	SourceWithSchemaExtraction
}

DefaultSourceMiddleware should be embedded in the SourceConfig struct to provide the SDK's default source middleware as configuration fields. Its fields are picked up by SourceWithMiddleware, which wraps the source in the corresponding middleware. Note that if the embedding struct overwrites Validate manually, it should call Validate on this struct as well.

func (*DefaultSourceMiddleware) Validate added in v0.13.0 

func (c *DefaultSourceMiddleware) Validate(ctx context.Context) error

Validate validates all the Validatable structs in the middleware.

type Destination 

type Destination interface {
	// Config returns the configuration that the destination expects. It should
	// return a pointer to a struct that contains all the configuration keys that
	// the destination expects. The struct should be annotated with the necessary
	// validation tags. The value should be a pointer to allow the SDK to
	// populate it using the values from the configuration.
	//
	// The returned DestinationConfig should contain all the configuration keys
	// that the destination expects, including middleware fields (see
	// [DefaultDestinationMiddleware]).
	Config() DestinationConfig

	// Open is called after Configure to signal the plugin it can prepare to
	// start writing records. If needed, the plugin should open connections in
	// this function.
	Open(context.Context) error

	// Write writes len(r) records from r to the destination right away without
	// caching. It should return the number of records written from r
	// (0 <= n <= len(r)) and any error encountered that caused the write to
	// stop early. Write must return a non-nil error if it returns n < len(r).
	Write(ctx context.Context, r []opencdc.Record) (n int, err error)

	// Teardown signals to the plugin that all records were written and there
	// will be no more calls to any other function. After Teardown returns, the
	// plugin should be ready for a graceful shutdown.
	Teardown(context.Context) error

	// LifecycleOnCreated is called after Configure and before Open when the
	// connector is run for the first time. This call will be skipped if the
	// connector was already started before. This method can be used to do some
	// initialization that needs to happen only once in the lifetime of a
	// connector (e.g. create a bucket). Anything that the connector creates in
	// this method is considered to be owned by this connector and should be
	// cleaned up in LifecycleOnDeleted.
	LifecycleOnCreated(ctx context.Context, config config.Config) error
	// LifecycleOnUpdated is called after Configure and before Open when the
	// connector configuration has changed since the last run. This call will be
	// skipped if the connector configuration did not change. It can be used to
	// update anything that was initialized in LifecycleOnCreated, in case the
	// configuration change affects it.
	LifecycleOnUpdated(ctx context.Context, configBefore, configAfter config.Config) error
	// LifecycleOnDeleted is called when the connector was deleted. It will be
	// the only method that is called in that case. This method can be used to
	// clean up anything that was initialized in LifecycleOnCreated.
	LifecycleOnDeleted(ctx context.Context, config config.Config) error
	// contains filtered or unexported methods
}

Destination receives records from Conduit and writes them to 3rd party resources. All implementations must embed UnimplementedDestination for forward compatibility.

func DestinationWithMiddleware added in v0.3.0 

func DestinationWithMiddleware(d Destination) Destination

DestinationWithMiddleware wraps the destination into the middleware defined in the config.

type DestinationCapabilities added in v0.16.0 

type DestinationCapabilities struct {
	// Idempotence states whether replaying records creates duplicates, see
	// [IdempotencyClaim]. A claim of [IdempotenceYes] needs a driver that can
	// return duplicates (see [AcceptanceReadAllDriver]); without one the case
	// fails instead of silently checking less.
	Idempotence IdempotencyClaim `json:"idempotence,omitempty"`
	// OrdersByKey claims that records with the same key are stored in the order
	// they were written.
	OrdersByKey bool `json:"orders_by_key,omitempty"`
	// SchemaEvolution is the policy for records whose structure changes
	// while the destination is running.
	SchemaEvolution SchemaEvolutionPolicy `json:"schema_evolution,omitempty"`
	// WriteOnly declares that the destination cannot be read back (for
	// example a log or a webhook). Tests that only need acks still run, tests
	// that need to inspect the stored data are skipped and say so in the
	// report. This removes the durability checks, so a registry must treat it
	// as a per-connector exception that maintainers approve and list; a driver
	// can often avoid it by capturing the destination's output itself.
	WriteOnly bool `json:"write_only,omitempty"`
}

DestinationCapabilities are the optional claims of a destination connector.

type DestinationConfig added in v0.13.0 

type DestinationConfig interface {
	Validatable
	// contains filtered or unexported methods
}

DestinationConfig represents the configuration containing all configuration keys that a destination expects. The type needs to implement Validatable, which will be used to automatically validate the config when configuring the connector.

type DestinationMiddleware added in v0.3.0 

type DestinationMiddleware interface {
	Wrap(Destination) Destination
}

DestinationMiddleware wraps a Destination and adds functionality to it.

type DestinationUtil added in v0.3.0 

type DestinationUtil struct{}

DestinationUtil provides utility methods for implementing a destination. Use it by calling Util.Destination.*.

func (DestinationUtil) Route added in v0.3.0 

func (DestinationUtil) Route(
	ctx context.Context,
	rec opencdc.Record,
	handleCreate func(context.Context, opencdc.Record) error,
	handleUpdate func(context.Context, opencdc.Record) error,
	handleDelete func(context.Context, opencdc.Record) error,
	handleSnapshot func(context.Context, opencdc.Record) error,
) error

Route makes it easier to implement a destination that mutates entities in place and thus handles different operations differently. It will inspect the operation on the record and based on that choose which handler to call.

Example usage: 

func (d *Destination) Write(ctx context.Context, r opencdc.Record) error {
  return d.Util.Route(ctx, r,
    d.handleInsert,
    d.handleUpdate,
    d.handleDelete,
    d.handleSnapshot, // we could also reuse d.handleInsert
  )
}
func (d *Destination) handleInsert(ctx context.Context, r opencdc.Record) error {
  ...
}

type DestinationWithBatch added in v0.3.0 

type DestinationWithBatch struct {
	UnimplementedDestinationConfig

	// Maximum size of batch before it gets written to the destination.
	BatchSize int `json:"sdk.batch.size" default:"0" validate:"gt=-1"`
	// Maximum delay before an incomplete batch is written to the destination.
	BatchDelay time.Duration `json:"sdk.batch.delay" default:"0"`
}

DestinationWithBatch adds support for batching on the destination.

func (*DestinationWithBatch) Wrap added in v0.3.0 

func (c *DestinationWithBatch) Wrap(impl Destination) Destination

Wrap a Destination into the middleware.

type DestinationWithRateLimit added in v0.3.0 

type DestinationWithRateLimit struct {
	UnimplementedDestinationConfig

	// Maximum number of records written per second (0 means no rate limit).
	RatePerSecond float64 `json:"sdk.rate.perSecond" default:"0" validate:"gt=-1"`
	// Allow bursts of at most X records (0 or less means that bursts are not
	// limited). Only takes effect if a rate limit per second is set. Note that
	// if `sdk.batch.size` is bigger than `sdk.rate.burst`, the effective batch
	// size will be equal to `sdk.rate.burst`.
	Burst int `json:"sdk.rate.burst" default:"0" validate:"gt=-1"`
}

DestinationWithRateLimit adds support for rate limiting to the destination.

func (*DestinationWithRateLimit) Wrap added in v0.3.0 

func (c *DestinationWithRateLimit) Wrap(impl Destination) Destination

Wrap a Destination into the middleware.

type DestinationWithRecordFormat added in v0.5.0 

type DestinationWithRecordFormat struct {
	UnimplementedDestinationConfig

	// The format of the output record. See the Conduit documentation for a full
	// list of supported formats (https://conduitdata.io/docs/using/connectors/configuration-parameters/output-format).
	RecordFormat *string `json:"sdk.record.format" default:"opencdc/json"`
	// Options to configure the chosen output record format. Options are normally
	// key=value pairs separated with comma (e.g. opt1=val2,opt2=val2), except
	// for the `template` record format, where options are a Go template.
	RecordFormatOptions string `json:"sdk.record.format.options" default:""`

	// RecordSerializers can be set to change the list uf supported formats. It
	// defaults to the output of DestinationWithRecordFormat.DefaultSerializers.
	RecordSerializers []RecordSerializer `json:"-"`
}

DestinationWithRecordFormat adds support for changing the output format of records, specifically of the Record.Bytes method.

func (*DestinationWithRecordFormat) DefaultRecordSerializers added in v0.13.0 

func (c *DestinationWithRecordFormat) DefaultRecordSerializers() []RecordSerializer

DefaultRecordSerializers returns the list of record serializers that are used if DestinationWithRecordFormat.RecordSerializers is nil.

func (*DestinationWithRecordFormat) Validate added in v0.13.0 

func (c *DestinationWithRecordFormat) Validate(context.Context) error

func (*DestinationWithRecordFormat) Wrap added in v0.5.0 

func (c *DestinationWithRecordFormat) Wrap(impl Destination) Destination

Wrap a Destination into the middleware.

type DestinationWithSchemaExtraction added in v0.10.0 

type DestinationWithSchemaExtraction struct {
	UnimplementedDestinationConfig

	// Whether to extract and decode the record payload with a schema.
	PayloadEnabled *bool `json:"sdk.schema.extract.payload.enabled" default:"true"`
	// Whether to extract and decode the record key with a schema.
	KeyEnabled *bool `json:"sdk.schema.extract.key.enabled" default:"true"`
}

DestinationWithSchemaExtraction is a middleware that extracts and decodes the key and/or payload of a record using a schema. It takes the schema subject and version from the record metadata, fetches the schema from the schema service, and decodes the key and/or payload using the schema. If the schema subject and version is not found in the record metadata, it will log a warning and skip decoding the key and/or payload. This middleware is useful when the source connector sends the data with the schema attached. This middleware is the counterpart of SourceWithSchemaExtraction.

func (*DestinationWithSchemaExtraction) Wrap added in v0.10.0 

func (c *DestinationWithSchemaExtraction) Wrap(impl Destination) Destination

Wrap a Destination into the middleware.

type Encoder added in v0.5.0 

type Encoder interface {
	// Name returns the identifier of the encoder, used as the second half of a
	// record format name (e.g. "json").
	Name() string
	// Configure returns a new Encoder configured with the given options; it does
	// not mutate the receiver.
	Configure(options map[string]string) (Encoder, error)
	// Encode marshals the value produced by a Converter into its byte
	// representation.
	Encode(r any) ([]byte, error)
}

Encoder is a type that can encode a random struct into a byte slice. It's used in destination connectors to encode records into different formats (e.g. JSON, Avro etc.).

type GenerateDataType added in v0.3.0 

type GenerateDataType int

GenerateDataType is used in acceptance tests to control what data type will be generated. 

const (
	GenerateMixedData GenerateDataType = iota
	GenerateRawData
	GenerateStructuredData
)

type GenericRecordSerializer added in v0.10.0 

type GenericRecordSerializer struct {
	Converter
	Encoder
}

GenericRecordSerializer is a serializer that uses a Converter and Encoder to serialize a record.

func (GenericRecordSerializer) Configure added in v0.10.0 

func (rf GenericRecordSerializer) Configure(optRaw string) (RecordSerializer, error)

func (GenericRecordSerializer) Name added in v0.10.0 

func (rf GenericRecordSerializer) Name() string

Name returns the name of the record serializer combined from the converter name and encoder name.

func (GenericRecordSerializer) Serialize added in v0.10.0 

func (rf GenericRecordSerializer) Serialize(r opencdc.Record) ([]byte, error)

Serialize converts and encodes record into a byte array.

type IdempotencyClaim added in v0.16.0 

type IdempotencyClaim string

IdempotencyClaim is what a destination states about writing the same record twice. It is a recorded value rather than a flag so that "not idempotent" is a statement a registry can read, not the absence of one. 

const (
	// IdempotenceUnspecified means the connector said nothing. The baseline
	// cases run (at-least-once), a registry should not certify the connector.
	IdempotenceUnspecified IdempotencyClaim = ""
	// IdempotenceYes claims that writing the same records again (same position
	// and key) after a crash does not create duplicates. DST-05 verifies it.
	IdempotenceYes IdempotencyClaim = "idempotent"
	// IdempotenceNo declares that replayed records may be stored twice
	// (at-least-once). DST-04 verifies that nothing is lost.
	IdempotenceNo IdempotencyClaim = "at_least_once"
)

type JSONEncoder added in v0.5.0 

type JSONEncoder struct{}

JSONEncoder is an Encoder that outputs JSON.

func (JSONEncoder) Configure added in v0.5.0 

func (e JSONEncoder) Configure(map[string]string) (Encoder, error)

func (JSONEncoder) Encode added in v0.5.0 

func (e JSONEncoder) Encode(v any) ([]byte, error)

func (JSONEncoder) Name added in v0.5.0 

func (e JSONEncoder) Name() string

type OpenCDCConverter added in v0.5.0 

type OpenCDCConverter struct{}

OpenCDCConverter outputs an OpenCDC record (it does not change the structure of the record).

func (OpenCDCConverter) Configure added in v0.5.0 

func (c OpenCDCConverter) Configure(map[string]string) (Converter, error)

func (OpenCDCConverter) Convert added in v0.5.0 

func (c OpenCDCConverter) Convert(r opencdc.Record) (any, error)

func (OpenCDCConverter) Name added in v0.5.0 

func (c OpenCDCConverter) Name() string

type RecordSerializer added in v0.10.0 

type RecordSerializer interface {
	// Name returns the identifier of the serializer as it appears in the record
	// format configuration (e.g. "opencdc/json").
	Name() string
	// Configure parses the raw options string and returns a new, configured
	// RecordSerializer. It does not mutate the receiver, so the zero value can be
	// used as a template from which configured instances are derived.
	Configure(string) (RecordSerializer, error)
	opencdc.RecordSerializer
}

RecordSerializer is a type that can format a record to bytes. It's used in destination connectors to change the output structure and format.

type SchemaEvolutionPolicy added in v0.16.0 

type SchemaEvolutionPolicy string

SchemaEvolutionPolicy is how a connector handles records whose structure changes while it runs (e.g. a new field appears). The connector declares one policy; the suite verifies that exactly that policy is followed. Silent coercion or truncation is never a valid policy (data integrity invariant 6). 

const (
	// SchemaEvolutionNone means the connector makes no claim. The schema
	// evolution test is skipped.
	SchemaEvolutionNone SchemaEvolutionPolicy = ""
	// SchemaEvolutionEvolve means the connector adapts to new fields: records
	// with the new field are written/read with all fields intact.
	SchemaEvolutionEvolve SchemaEvolutionPolicy = "evolve"
	// SchemaEvolutionReject means the connector refuses records it cannot
	// represent: a destination nacks them with an error, and the data it
	// accepted earlier stays intact.
	SchemaEvolutionReject SchemaEvolutionPolicy = "reject"
)

type Source 

type Source interface {
	// Config returns the configuration that the source expects. It should return
	// a pointer to a struct that contains all the configuration keys that the
	// source expects. The struct should be annotated with the necessary
	// validation tags. The value should be a pointer to allow the SDK to
	// populate it using the values from the configuration.
	//
	// The returned SourceConfig should contain all the configuration keys that
	// the source expects, including middleware fields (see
	// [DefaultSourceMiddleware]).
	Config() SourceConfig

	// Open is called after Configure to signal the plugin it can prepare to
	// start producing records. If needed, the plugin should open connections in
	// this function. The position parameter will contain the position of the
	// last record that was successfully processed, Source should therefore
	// start producing records after this position. The context passed to Open
	// will be cancelled once the plugin receives a stop signal from Conduit.
	Open(context.Context, opencdc.Position) error

	// Read returns a new Record and is supposed to block until there is either
	// a new record or the context gets cancelled. It can also return the error
	// ErrBackoffRetry to signal to the SDK it should call Read again with a
	// backoff retry.
	// If Read receives a cancelled context or the context is cancelled while
	// Read is running it must stop retrieving new records from the source
	// system and start returning records that have already been buffered. If
	// there are no buffered records left Read must return the context error to
	// signal a graceful stop. If Read returns ErrBackoffRetry while the context
	// is cancelled it will also signal that there are no records left and Read
	// won't be called again.
	// After Read returns an error the function won't be called again (except if
	// the error is ErrBackoffRetry, as mentioned above).
	// Read can be called concurrently with Ack.
	Read(context.Context) (opencdc.Record, error)

	// ReadN is the same as Read, but returns a batch of records. The connector
	// is expected to return at most n records. If there are fewer records
	// available, it should return all of them. If there are no records available
	// it should block until there are records available or the context is
	// cancelled. If the context is cancelled while ReadN is running, it should
	// return the context error.
	//
	// Ownership: once ReadN returns, the returned records belong to the SDK.
	// The connector must not modify them, or the memory they reference
	// (positions, metadata maps, key and payload bytes), in a later call or
	// from another goroutine. The SDK may still be sending or processing a
	// batch while it calls ReadN again, for example when batching is enabled
	// (sdk.batch.size, sdk.batch.delay). The SDK copies the returned slice, so
	// reusing the slice itself is safe, but it does not copy what the records
	// point to: allocate a new position, metadata map and data buffers for
	// every record instead of reusing them. The same applies to Read.
	ReadN(context.Context, int) ([]opencdc.Record, error)

	// Ack signals to the implementation that the record with the supplied
	// position was successfully processed. This method might be called after
	// the context of Read is already cancelled, since there might be
	// outstanding acks that need to be delivered. When Teardown is called it is
	// guaranteed there won't be any more calls to Ack.
	// Ack can be called concurrently with Read.
	Ack(context.Context, opencdc.Position) error

	// Teardown signals to the plugin that there will be no more calls to any
	// other function. After Teardown returns, the plugin should be ready for a
	// graceful shutdown.
	Teardown(context.Context) error

	// LifecycleOnCreated is called after Configure and before Open when the
	// connector is run for the first time. This call will be skipped if the
	// connector was already started before. This method can be used to do some
	// initialization that needs to happen only once in the lifetime of a
	// connector (e.g. create a logical replication slot). Anything that the
	// connector creates in this method is considered to be owned by this
	// connector and should be cleaned up in LifecycleOnDeleted.
	LifecycleOnCreated(ctx context.Context, config config.Config) error
	// LifecycleOnUpdated is called after Configure and before Open when the
	// connector configuration has changed since the last run. This call will be
	// skipped if the connector configuration did not change. It can be used to
	// update anything that was initialized in LifecycleOnCreated, in case the
	// configuration change affects it.
	LifecycleOnUpdated(ctx context.Context, configBefore, configAfter config.Config) error
	// LifecycleOnDeleted is called when the connector was deleted.
	// This method can be used to clean up anything that was initialized in LifecycleOnCreated.
	// It will be the only method that is called in that case, i.e. Configure() and Open()
	// won't be called. The implementation needs to manually parse the configuration if needed.
	LifecycleOnDeleted(ctx context.Context, config config.Config) error
	// contains filtered or unexported methods
}

Source fetches records from 3rd party resources and sends them to Conduit. All implementations must embed UnimplementedSource for forward compatibility.

func SourceWithMiddleware added in v0.3.0 

func SourceWithMiddleware(s Source) Source

SourceWithMiddleware wraps the source into the middleware defined in the config.

type SourceCapabilities added in v0.16.0 

type SourceCapabilities struct {
	// OrdersByKey claims that records with the same key are produced in the
	// order they were written to the 3rd party system.
	OrdersByKey bool `json:"orders_by_key,omitempty"`
	// ExactResume claims that after a restart from the last acked position the
	// source delivers no record that was already acked. Without the claim the
	// suite tolerates [AcceptanceCapabilities.Source.MaxRedelivery] repeated
	// records per restart (siblings of the last position, a coarse watermark),
	// but a source that restarts from the beginning fails either way.
	ExactResume bool `json:"exact_resume,omitempty"`
	// MaxRedelivery is the number of already acked records a source without
	// ExactResume may deliver again per restart. Zero means
	// [DefaultMaxRedelivery]; values above [MaxAcceptedRedelivery] fail SRC-03.
	// It is ignored if ExactResume is claimed. SRC-03 scales its data set so
	// that the records acked before the last restart are at least five times
	// the bound, which means a source that starts over fails even with a large
	// bound, but the bound still weakens the case: the report flags any value
	// above the default and the certification floor is not met.
	MaxRedelivery int `json:"max_redelivery,omitempty"`
	// Synthetic declares that the source produces records on its own (a
	// generator) and cannot be fed through the 3rd party system. Tests that
	// compare what was written with what was read are skipped, the tests that
	// only need records to flow (graceful stop, cancellation, throughput) run.
	// This removes most of the source checks, so a registry must treat it as a
	// per-connector exception that maintainers approve and list.
	Synthetic bool `json:"synthetic,omitempty"`
	// SchemaEvolution is the policy for fields that appear while the source
	// is running. Only [SchemaEvolutionEvolve] can be verified on a source.
	SchemaEvolution SchemaEvolutionPolicy `json:"schema_evolution,omitempty"`
}

SourceCapabilities are the optional claims of a source connector.

type SourceConfig added in v0.13.0 

type SourceConfig interface {
	Validatable
	// contains filtered or unexported methods
}

SourceConfig represents the configuration containing all configuration keys that a source expects. The type needs to implement Validatable, which will be used to automatically validate the config when configuring the connector.

type SourceMiddleware added in v0.3.0 

type SourceMiddleware interface {
	Wrap(Source) Source
}

SourceMiddleware wraps a Source and adds functionality to it.

type SourceUtil added in v0.3.0 

type SourceUtil struct{}

SourceUtil provides utility methods for implementing a source. Use it by calling Util.Source.*.

func (SourceUtil) NewRecordCreate added in v0.3.0 

func (SourceUtil) NewRecordCreate(
	position opencdc.Position,
	metadata opencdc.Metadata,
	key opencdc.Data,
	payload opencdc.Data,
) opencdc.Record

NewRecordCreate can be used to instantiate a record with OperationCreate.

func (SourceUtil) NewRecordDelete added in v0.3.0 

func (SourceUtil) NewRecordDelete(
	position opencdc.Position,
	metadata opencdc.Metadata,
	key opencdc.Data,
	payloadBefore opencdc.Data,
) opencdc.Record

NewRecordDelete can be used to instantiate a record with OperationDelete.

func (SourceUtil) NewRecordSnapshot added in v0.3.0 

func (SourceUtil) NewRecordSnapshot(
	position opencdc.Position,
	metadata opencdc.Metadata,
	key opencdc.Data,
	payload opencdc.Data,
) opencdc.Record

NewRecordSnapshot can be used to instantiate a record with OperationSnapshot.

func (SourceUtil) NewRecordUpdate added in v0.3.0 

func (SourceUtil) NewRecordUpdate(
	position opencdc.Position,
	metadata opencdc.Metadata,
	key opencdc.Data,
	payloadBefore opencdc.Data,
	payloadAfter opencdc.Data,
) opencdc.Record

NewRecordUpdate can be used to instantiate a record with OperationUpdate.

type SourceWithBatch added in v0.11.1 

type SourceWithBatch struct {
	// Maximum size of batch before it gets read from the source.
	BatchSize *int `json:"sdk.batch.size" default:"0" validate:"gt=-1"`
	// Maximum delay before an incomplete batch is read from the source.
	BatchDelay *time.Duration `json:"sdk.batch.delay" default:"0"`
}

SourceWithBatch adds support for batching on the source.

func (*SourceWithBatch) Wrap added in v0.11.1 

func (c *SourceWithBatch) Wrap(impl Source) Source

Wrap a Source into the middleware.

type SourceWithEncoding added in v0.11.1 

type SourceWithEncoding struct{}

SourceWithEncoding is a middleware that encodes the record payload and key with the provided schema. The schema is registered with the schema service and the schema subject is attached to the record metadata.

func (SourceWithEncoding) Wrap added in v0.11.1 

func (s SourceWithEncoding) Wrap(impl Source) Source

type SourceWithSchemaContext added in v0.10.0 

type SourceWithSchemaContext struct {
	// Specifies whether to use a schema context name. If set to false, no schema context name will
	// be used, and schemas will be saved with the subject name specified in the connector
	// (not safe because of name conflicts).
	Enabled *bool `json:"sdk.schema.context.enabled" default:"true"`
	// Schema context name to be used. Used as a prefix for all schema subject names.
	// If empty, defaults to the connector ID.
	Name *string `json:"sdk.schema.context.name" default:""`
}

SourceWithSchemaContext is a middleware that makes it possible to configure the schema context for records read by a source.

func (*SourceWithSchemaContext) Wrap added in v0.10.0 

func (c *SourceWithSchemaContext) Wrap(s Source) Source

Wrap a Source into the middleware.

type SourceWithSchemaExtraction added in v0.10.0 

type SourceWithSchemaExtraction struct {
	// The type of the payload schema.
	SchemaTypeStr string `json:"sdk.schema.extract.type" validate:"inclusion=avro" default:"avro"`
	// Whether to extract and encode the record payload with a schema.
	PayloadEnabled *bool `json:"sdk.schema.extract.payload.enabled" default:"true"`
	// The subject of the payload schema. If the record metadata contains the
	// field "opencdc.collection" it is prepended to the subject name and
	// separated with a dot.
	PayloadSubject *string `json:"sdk.schema.extract.payload.subject" default:"payload"`
	// Whether to extract and encode the record key with a schema.
	KeyEnabled *bool `json:"sdk.schema.extract.key.enabled" default:"true"`
	// The subject of the key schema. If the record metadata contains the field
	// "opencdc.collection" it is prepended to the subject name and separated
	// with a dot.
	KeySubject *string `json:"sdk.schema.extract.key.subject" default:"key"`
}

SourceWithSchemaExtraction is a middleware that extracts a record's payload and key schemas. The schema is extracted from the record data for each record produced by the source. The schema is registered with the schema service and the schema subject is attached to the record metadata.

Records with the same shape (same field names and value types) are given the schema registered for the first of them, the schema is not inferred and registered again for every record. The remembered registrations expire after 15 minutes, on top of the schema service client's own cache of the same age.

func (*SourceWithSchemaExtraction) SchemaType added in v0.13.0 

func (c *SourceWithSchemaExtraction) SchemaType() schema.Type

SchemaType returns the typed schema type (and not the string value as it's in the configuration itself). todo: use https://github.com/ConduitIO/conduit-commons/issues/142 to parse the string value into schema.Type directly.

func (*SourceWithSchemaExtraction) Wrap added in v0.10.0 

func (c *SourceWithSchemaExtraction) Wrap(impl Source) Source

Wrap a Source into the middleware.

type Specification 

type Specification = pconnector.Specification

Specification contains general information regarding the plugin like its name and what it does.

func ParseYAMLSpecification added in v0.13.0 

func ParseYAMLSpecification(ctx context.Context, rawYaml, version string) (Specification, error)

ParseYAMLSpecification parses a Specification from the given YAML file, optionally overriding the connector version with the version parameter (pass an empty string to keep the version from the YAML). It is the error-returning counterpart to YAMLSpecification, which panics on failure; prefer this one when the YAML is not known to be valid at build time.

type TemplateRecordSerializer added in v0.10.0 

type TemplateRecordSerializer struct {
	// contains filtered or unexported fields
}

TemplateRecordSerializer is a RecordSerializer that serializes a record using a Go template.

func (TemplateRecordSerializer) Configure added in v0.10.0 

func (e TemplateRecordSerializer) Configure(tmpl string) (RecordSerializer, error)

func (TemplateRecordSerializer) Name added in v0.10.0 

func (e TemplateRecordSerializer) Name() string

func (TemplateRecordSerializer) Serialize added in v0.10.0 

func (e TemplateRecordSerializer) Serialize(r opencdc.Record) ([]byte, error)

type UnimplementedDestination 

type UnimplementedDestination struct{}

UnimplementedDestination should be embedded to have forward compatible implementations.

func (UnimplementedDestination) Config added in v0.13.0 

func (UnimplementedDestination) Config() DestinationConfig

Config needs to be overridden in the actual implementation.

func (UnimplementedDestination) LifecycleOnCreated added in v0.6.0 

func (UnimplementedDestination) LifecycleOnCreated(context.Context, config.Config) error

LifecycleOnCreated won't do anything by default.

func (UnimplementedDestination) LifecycleOnDeleted added in v0.6.0 

func (UnimplementedDestination) LifecycleOnDeleted(context.Context, config.Config) error

LifecycleOnDeleted won't do anything by default.

func (UnimplementedDestination) LifecycleOnUpdated added in v0.6.0 

func (UnimplementedDestination) LifecycleOnUpdated(context.Context, config.Config, config.Config) error

LifecycleOnUpdated won't do anything by default.

func (UnimplementedDestination) Open 

func (UnimplementedDestination) Open(context.Context) error

Open needs to be overridden in the actual implementation.

func (UnimplementedDestination) Teardown 

func (UnimplementedDestination) Teardown(context.Context) error

Teardown needs to be overridden in the actual implementation.

func (UnimplementedDestination) Write 

func (UnimplementedDestination) Write(context.Context, []opencdc.Record) (int, error)

Write needs to be overridden in the actual implementation.

type UnimplementedDestinationConfig added in v0.13.0 

type UnimplementedDestinationConfig struct{}

func (UnimplementedDestinationConfig) Validate added in v0.13.0 

func (UnimplementedDestinationConfig) Validate(context.Context) error

type UnimplementedSource 

type UnimplementedSource struct{}

UnimplementedSource should be embedded to have forward compatible implementations.

func (UnimplementedSource) Ack 

func (UnimplementedSource) Ack(context.Context, opencdc.Position) error

Ack should be overridden if acks need to be forwarded to the source, otherwise it is optional.

func (UnimplementedSource) Config added in v0.13.0 

func (UnimplementedSource) Config() SourceConfig

Config needs to be overridden in the actual implementation.

func (UnimplementedSource) LifecycleOnCreated added in v0.6.0 

func (UnimplementedSource) LifecycleOnCreated(context.Context, config.Config) error

LifecycleOnCreated won't do anything by default.

func (UnimplementedSource) LifecycleOnDeleted added in v0.6.0 

func (UnimplementedSource) LifecycleOnDeleted(context.Context, config.Config) error

LifecycleOnDeleted won't do anything by default.

func (UnimplementedSource) LifecycleOnUpdated added in v0.6.0 

func (UnimplementedSource) LifecycleOnUpdated(context.Context, config.Config, config.Config) error

LifecycleOnUpdated won't do anything by default.

func (UnimplementedSource) Open 

func (UnimplementedSource) Open(context.Context, opencdc.Position) error

Open needs to be overridden in the actual implementation.

func (UnimplementedSource) Read 

func (UnimplementedSource) Read(context.Context) (opencdc.Record, error)

Read needs to be overridden in the actual implementation.

func (UnimplementedSource) ReadN added in v0.11.1 

func (s UnimplementedSource) ReadN(context.Context, int) ([]opencdc.Record, error)

ReadN can be overridden. If it's not implemented, Read will be used as a fallback.

func (UnimplementedSource) Teardown 

func (UnimplementedSource) Teardown(context.Context) error

Teardown needs to be overridden in the actual implementation.

type UnimplementedSourceConfig added in v0.13.0 

type UnimplementedSourceConfig struct{}

func (UnimplementedSourceConfig) Validate added in v0.13.0 

func (UnimplementedSourceConfig) Validate(context.Context) error

type Validatable added in v0.13.0 

type Validatable interface {
	// Validate executes any custom validations on the configuration and returns
	// an error if it is invalid.
	Validate(context.Context) error
}

Validatable can be implemented by a SourceConfig or DestinationConfig or any embedded struct, to provide custom validation logic. Validate will be triggered automatically by the SDK after parsing the config. If it returns an error, the configuration is considered invalid and the connector won't be opened.

Source Files

View all Source files

Directories

Path Synopsis
readmegen
specgen
specgen/model/v1
Package internal holds implementation details shared by the SDK that are not part of the connector-author-facing API.
 Package internal holds implementation details shared by the SDK that are not part of the connector-author-facing API.
refconnector
Package refconnector is a minimal in-memory connector that the SDK runs its own acceptance suite against (see acceptance_test.go).
 Package refconnector is a minimal in-memory connector that the SDK runs its own acceptance suite against (see acceptance_test.go).
Package kafkaconnect contains utility functions and structures for processing Kafka Connect compatible data.
 Package kafkaconnect contains utility functions and structures for processing Kafka Connect compatible data.
Package schema gives connectors access to Conduit's schema registry.
 Package schema gives connectors access to Conduit's schema registry.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.