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
}