An error occurs after removing data from the Zendesk Source table when transferring data from the Zendesk Source to the Oracle Destination

[SerhiiSpivak]

Bug description

The user creates the Zendesk Sorce connector and the Oracle Destination connector -> data is transferred to the Oracle Destination; then the user removes data from the Zendesk Source -> the system returns an error

2022-09-29T10:18:12+00:00 ERR error writing record error="structurize key: unmarshal \"70\" into structured data: json: cannot unmarshal number into Go value of type sdk.StructuredData" component=plugin connector_id=4fc3a3c9-4f1e-4254-863b-19af6d78503d plugin_name=conduit-connector-oracle record_position="{\"mode\":\"cdc\",\"last_modified_time\":\"2022-09-29T07:17:00Z\",\"id\":70}" 2022-09-29T10:18:12+00:00 ERR destination acker node stopped before processing all messages error="nacked 1 messages when stopping destination acker node, some nacks failed: no DLQ handler configured" component=DestinationAckerNode node_id=4fc3a3c9-4f1e-4254-863b-19af6d78503d-acker pipeline_id=9b8daf3c-a063-42ff-8d1d-9b79279ebf4e stack=[{"file":"/home/serhiispivak/go/src/github.com/conduit/pkg/pipeline/stream/destination_acker.go","func":"[github.com/conduitio/conduit/pkg/pipeline/stream.(*DestinationAckerNode](http://github.com/conduitio/conduit/pkg/pipeline/stream.(*DestinationAckerNode)).teardown","line":182},{"file":"/home/serhiispivak/go/src/github.com/conduit/pkg/pipeline/stream/source_acker.go","func":"[github.com/conduitio/conduit/pkg/pipeline/stream.(*SourceAckerNode](http://github.com/conduitio/conduit/pkg/pipeline/stream.(*SourceAckerNode)).registerNackHandler.func1","line":118}] 2022-09-29T10:18:12+00:00 ERR node stopped error="node 4fc3a3c9-4f1e-4254-863b-19af6d78503d-acker stopped with error: failed to receive ack: error receiving ack: structurize key: unmarshal \"70\" into structured data: json: cannot unmarshal number into Go value of type sdk.StructuredData" component=pipeline.Service node_id=4fc3a3c9-4f1e-4254-863b-19af6d78503d-acker stack=[{"file":"/home/serhiispivak/go/src/github.com/conduit/pkg/pipeline/lifecycle.go","func":"[github.com/conduitio/conduit/pkg/pipeline.(*Service](http://github.com/conduitio/conduit/pkg/pipeline.(*Service)).runPipeline.func1","line":411},{"file":"/home/serhiispivak/go/src/github.com/conduit/pkg/pipeline/stream/destination_acker.go","func":"[github.com/conduitio/conduit/pkg/pipeline/stream.(*DestinationAckerNode](http://github.com/conduitio/conduit/pkg/pipeline/stream.(*DestinationAckerNode)).worker","line":131},{"file":"/home/serhiispivak/go/src/github.com/conduit/pkg/connector/destination.go","func":"[github.com/conduitio/conduit/pkg/connector.(*destination](http://github.com/conduitio/conduit/pkg/connector.(*destination)).Ack","line":247},{"file":"/home/serhiispivak/go/src/github.com/conduit/pkg/plugin/standalone/v1/destination.go","func":"[github.com/conduitio/conduit/pkg/plugin/standalone/v1.(*destinationPluginClient](http://github.com/conduitio/conduit/pkg/plugin/standalone/v1.(*destinationPluginClient)).Ack","line":119}]

Steps to reproduce

1. Create the pipeline
2. Create Zendesk source connector and Oracle destination connector (see below config):

version: 1.0                  

pipelines:                    
  pipeline1:                   
    status: stopped             
    name: pipeline1             
    description: desc           
    connectors:                 
      con1:                 
        type: source            
        plugin: standalone:zendesk    
        name: zendesk-source             
        settings:              
          zendesk.domain: <domain> 
          zendesk.userName: <user name in zendesk>
          zendesk.apiToken: <zendesk api token>
          pollingPeriod:
      con2:
        type: destination
        plugin: standalone:oracle
        name: Oracle-destination
        settings:
           url: <username/password@path:1521/my.domain.com> 
           table: test_table
           keyColumn: <column name >

3. Strat Pipeline -> data is tranfered
4. Remove data from Zendesk Source
5. Get an error

2022-09-29T10:18:12+00:00 ERR error writing record error="structurize key: unmarshal \"70\" into structured data: json: cannot unmarshal number into Go value of type sdk.StructuredData" component=plugin connector_id=4fc3a3c9-4f1e-4254-863b-19af6d78503d plugin_name=conduit-connector-oracle record_position="{\"mode\":\"cdc\",\"last_modified_time\":\"2022-09-29T07:17:00Z\",\"id\":70}" 2022-09-29T10:18:12+00:00 ERR destination acker node stopped before processing all messages error="nacked 1 messages when stopping destination acker node, some nacks failed: no DLQ handler configured" component=DestinationAckerNode node_id=4fc3a3c9-4f1e-4254-863b-19af6d78503d-acker pipeline_id=9b8daf3c-a063-42ff-8d1d-9b79279ebf4e stack=[{"file":"/home/serhiispivak/go/src/github.com/conduit/pkg/pipeline/stream/destination_acker.go","func":"[github.com/conduitio/conduit/pkg/pipeline/stream.(*DestinationAckerNode](http://github.com/conduitio/conduit/pkg/pipeline/stream.(*DestinationAckerNode)).teardown","line":182},{"file":"/home/serhiispivak/go/src/github.com/conduit/pkg/pipeline/stream/source_acker.go","func":"[github.com/conduitio/conduit/pkg/pipeline/stream.(*SourceAckerNode](http://github.com/conduitio/conduit/pkg/pipeline/stream.(*SourceAckerNode)).registerNackHandler.func1","line":118}] 2022-09-29T10:18:12+00:00 ERR node stopped error="node 4fc3a3c9-4f1e-4254-863b-19af6d78503d-acker stopped with error: failed to receive ack: error receiving ack: structurize key: unmarshal \"70\" into structured data: json: cannot unmarshal number into Go value of type sdk.StructuredData" component=pipeline.Service node_id=4fc3a3c9-4f1e-4254-863b-19af6d78503d-acker stack=[{"file":"/home/serhiispivak/go/src/github.com/conduit/pkg/pipeline/lifecycle.go","func":"[github.com/conduitio/conduit/pkg/pipeline.(*Service](http://github.com/conduitio/conduit/pkg/pipeline.(*Service)).runPipeline.func1","line":411},{"file":"/home/serhiispivak/go/src/github.com/conduit/pkg/pipeline/stream/destination_acker.go","func":"[github.com/conduitio/conduit/pkg/pipeline/stream.(*DestinationAckerNode](http://github.com/conduitio/conduit/pkg/pipeline/stream.(*DestinationAckerNode)).worker","line":131},{"file":"/home/serhiispivak/go/src/github.com/conduit/pkg/connector/destination.go","func":"[github.com/conduitio/conduit/pkg/connector.(*destination](http://github.com/conduitio/conduit/pkg/connector.(*destination)).Ack","line":247},{"file":"/home/serhiispivak/go/src/github.com/conduit/pkg/plugin/standalone/v1/destination.go","func":"[github.com/conduitio/conduit/pkg/plugin/standalone/v1.(*destinationPluginClient](http://github.com/conduitio/conduit/pkg/plugin/standalone/v1.(*destinationPluginClient)).Ack","line":119}]

Additional info: Data isn't removed from the Oracle Destination table and an error occurs

Version

v0.3.0-nightly.20220927