Bug: processor chain crashes after filter

[rauanmayemir]

Bug description

Any processor that executes after filter will crash with:

ERR pipeline worker stopped error="task outbox-events:source:enrich_metadata: processor didn't return any records"

As explained by @hariso, this is because a filter record is of a special type.

Steps to reproduce

Example pipeline:

version: 2.2
pipelines:
  - id: outbox-events
    status: running
    connectors:
      - id: source
        type: source
        plugin: builtin:postgres
        settings:
          url: ${PG_SOURCE_DSN}
          tables: outbox_events
          cdcMode: logrepl
          logrepl.autoCleanup: 'false'
          logrepl.publicationName: cnd_outbox
          logrepl.slotName: cnd_outbox_events
          logrepl.withAvroSchema: 'false'
          snapshot.fetchSize: '1000'
          snapshotMode: initial
        processors:
          - id: insert_only
            plugin: 'filter'
            condition: '{{ ne .Operation 1 }}'
          - id: enrich_metadata
            plugin: 'custom.javascript'
            settings:
              script: |
                function process(rec) {
                  return rec;
                }
     - id: output-log
       type: destination
       plugin: builtin:log
       settings:
         sdk.schema.extract.key.enabled: 'false'
         sdk.schema.extract.payload.enabled: 'false'

Version

v0.13.4

[lovromazgon]

That looks like an important bug, we'll try to investigate it as soon as we get some time. Thanks for reporting it!

[lovromazgon]

Note that this only happens with the new architecture (i.e. using flag --preview.pipeline-arch-v2).

[rauanmayemir]

Another issue related to this, but much more serious.

panic: runtime error: index out of range [-1]

goroutine 2139 [running]:
github.com/conduitio/conduit-connector-sdk.(*destinationPluginAdapter).Run(0xc0018da690, {0x2ef6af8?, 0xc0015e5980?}, {0x2ee4410, 0xc000d24268})
 github.com/conduitio/conduit-connector-sdk@v0.13.3/destination.go:202 +0x294
github.com/conduitio/conduit/pkg/plugin/connector/builtin.(*destinationPluginAdapter).Run.func1()
 github.com/conduitio/conduit@v0.13.4/pkg/plugin/connector/builtin/destination.go:84 +0xf0
created by github.com/conduitio/conduit/pkg/plugin/connector/builtin.(*destinationPluginAdapter).Run in goroutine 1
 github.com/conduitio/conduit@v0.13.4/pkg/plugin/connector/builtin/destination.go:83 +0x29b

This happens when destination is not receiving any records because they've all been filtered out.

[lovromazgon]

Yeah, I noticed this when testing it myself. A fix is on the way, but it might take a while, as it's not straightforward.