Developing a Destination Connector

A Destination is responsible for writing an OpenCDC Record to third party systems.

You need to implement the functions required by Destination and provide your own implementations. Information about individual functions are listed below. The * destination.go* file is the main file where the functionality of your Destination Connector is implemented.

Destination struct

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 backward compatible with existing Destination implementations. This struct can be modified to add additional fields that can be accessed throughout the lifecycle of the Connector.

type Destination struct {
    sdk.UnimplementedDestination

    config DestinationConfig
}

Destination Connector Lifecycle Functions

NewDestination()

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 be used to wrap your Destination in the default middleware. You can add additional middleware, but unless you have a very good reason, you should always include the default middleware.

func NewDestination() sdk.Destination {
    // Create Destination and wrap it in the default middleware.
    return sdk.DestinationWithMiddleware(
      &Destination{}, 
      sdk.DefaultDestinationMiddleware()...,
    )
}

Additional options via DestinationMiddlewareOption:

Currently, the available destination middleware options can be found here.

Parameters()

A map of named Parameters that describe how to configure the connector. This map is typically generated using paramgen.

func (d *Destination) Parameters() config.Parameters {
    return d.config.Parameters()
}

Configure()

Validates and stores configuration data for the connector. Any complex validation logic should be implemented here.

func (d *Destination) Configure(ctx context.Context, cfg config.Config) error {
    err := sdk.Util.ParseConfig(ctx, cfg, &d.config, NewDestination().Parameters())
    if err != nil {
        return err
    }
    // add custom validations here
    return nil
}

Open()

Prepares the connector to start writing records. If needed, the connector should open connections in this function.

func (d *Destination) Open(context.Context) error {
    // opens or creates a file at the given path
    file, err := d.openOrCreate(d.config.Path)
    if err != nil {
        return err
    }

    d.file = file
    return nil
}

Write()

Writes len(records) from a slice of opencdc.Records received from the Conduit pipeline to the destination right away without caching. It should return the number of records written from the slice and any error encountered that caused the write to stop early.

func (d *Destination) Write(_ context.Context, recs []opencdc.Record) (int, error) {
    for i, r := range recs {
        _, err := d.file.Write(append(r.Bytes(), '\n'))
        if err != nil {
            return i, err
        }
    }
    return len(recs), nil
}

Teardown()

Teardown signals to the connector that there will be no more calls to any other function. Any connections that were created in the Open() function should be closed here.

func (d *Destination) Teardown(context.Context) error {
    if d.file != nil {
        return d.file.Close()
    }
    return nil
}

Schema

By default, the sdk destination middleware will use the record schema fields to decode its key and payload. This isn't as problematic as the sdk source middleware, as the schema should be already created.

If the destination connector is writing unstructured data, or you want to use your own schema writing logic, you can disable the middleware like this, as mentioned before:

func NewDestination() sdk.Destination {
    return sdk.DestinationWithMiddleware(
        &Destination{},
        sdk.DefaultDestinationMiddleware(
            sdk.DestinationWithSchemaExtractionConfig{
                PayloadEnabled: lang.Ptr(false),
                KeyEnabled:     lang.Ptr(false),
            }
        )...,
    )
}

Otherwise you can use the schema fetched from schema.Get to manually decode records:

	var rec opencdc.Record
	sub, err := rec.Metadata.GetPayloadSchemaSubject()
	v, err := rec.Metadata.GetPayloadSchemaVersion()

	s, err := schema.Get(ctx, sub, v)
	err = s.Unmarshal(rec.Payload.After.Bytes(), &someStruct)

	// ... write the struct in any way