Bug: Double-wrapped OpenCDC records

[raulb]

Bug description

While the source pipeline produces a valid OpenCDC record, the destination will consume this from Kafka, assume it is raw data and create a new OpenCDC record with the source record set as rawdata in the Payload.After field. It ends up being re-wrapped in another OpenCDC record, making it double-wrapped.

Suggestions:

• Add an OpenCDC unwrap processor
• When producing to kafka topics on the platform, only emit the contents of payload.After

Steps to reproduce

1. Create a pipeline that uses a destination builtin conduit connector. example:

version: 1.0
pipelines:
  gen-to-kafka:
    status: running
    description: >
     generator to kafka
    connectors:
      source-generator:
        type: source
        plugin: builtin:generator
        settings:
          format.options: event_id:int,pg_generator:bool,sensor_id:int,msg:string,triggered:bool
          format.type: structured
          recordCount: 100
          readTime: 1000ms
      dest-kafka:
        type: destination
        plugin: standalone:kafka
        settings:
          insecureSkipVerify: true
          servers: <put local kafka brokers here>
          topic: demo-123

2. Make sure you produce a record that can be produced by the destination connector. Consuming that record you'll notice it double wrapped. Example:

{
  "key": "MTc3NzQ5NDEtNTdhMi00MmZhLWI0MzAtODkxMmE5NDI0YjNh",
  "metadata": {
    "conduit.source.connector.id": "dest-log-78lpnchx7tzpyqz:source-kafka",
    "kafka.topic": "stream-78lpnchx7tzpyqz-generator",
    "opencdc.createdAt": "1706028953595000000",
    "opencdc.readAt": "1706028953606997000",
    "opencdc.version": "v1"
  },
  "operation": "create",
  "payload": {
    "after": {
        "position": "NzgyNjJmODUtODNmMS00ZGQwLWEyZDAtNTRmNjA1ZjkyYTg0",
        "operation": "create",
        "metadata": {
          "conduit.source.connector.id": "source-generator-78lpnchx7tzpyqz:source",
          "opencdc.readAt": "1706028953595546000",
          "opencdc.version": "v1"
        },
        "key": "MTc3NzQ5NDEtNTdhMi00MmZhLWI0MzAtODkxMmE5NDI0YjNh",
        "payload": {
          "before": null,
          "after": {
            "event_id": 1747353650,
            "msg": "string 0e8955b3-7fb5-4dda-8064-e10dc007f00d",
            "pg_generator": false,
            "sensor_id": 1250383582,
            "triggered": false
          }
        }
      },
    "before": null
  },
  "position": "eyJHcm91cElEIjoiNGQ2ZTBhMjktNzAwZi00Yjk4LWEzY2MtZWUyNzZhZTc4MjVjIiwiVG9waWMiOiJzdHJlYW0tNzhscG5jaHg3dHpweXF6LWdlbmVyYXRvciIsIlBhcnRpdGlvbiI6MCwiT2Zmc2V0IjoyMjF9"
}

Version

v0.8.0

[hariso]

@raulb This looks to be an issue when Kafka is used as an intermediary? For example, source foo -> kafka -> destination bar?

[raulb]

@hariso correct.