In S3 source connector, how to fetch only new files ?

[akhileshacc]

I am using S3 source connector tp fetch files from S3. It works fine and gets back all records.
But on adding new files, it again gives all reacords.

Here is my configuration

{
    "name": "s3-source",
    "config": {
        "name": "s3-source",
        "connector.class": "org.apache.camel.kafkaconnector.awss3.CamelAwss3SourceConnector",
        "key.converter": "org.apache.kafka.connect.storage.StringConverter",
        "value.converter": "org.apache.camel.kafkaconnector.awss3.converters.S3ObjectConverter",
        "topics": "mytopic",
        "camel.source.path.bucketNameOrArn": "checkoutkafka",
        "camel.component.aws-s3.accessKey": "accesskey",
        "camel.component.aws-s3.secretKey": "secretkey",
        "camel.component.aws-s3.region": "AP_SOUTH_1",
        "camel.component.aws-s3.deleteAfterRead": false,
        "camel.component.aws-s3.autocloseBody": false
    }
}

How do i set this configuration ?

[oscerd]

You need to set the deleteAfterRead to true and the file will be deleted once consumed.

[oscerd]

Explained.

[akhileshacc]

What if I don't want to remove content from s3 ?

[oscerd]

You can use the aws2-s3 component and use the moveAfterRead option

[oscerd]

I'll add an example in camel-kafka-connector-examples repository

[akhileshacc]

Thanks for the help. Using moveAfterRead expects a destination bucket. What if i don't want to use new bucket.
Is it possible currently ?

[oscerd]

No, it's not. The connector is generated starting from the camel component, so you need to move the file somewhere else, actually only a different bucket is supported.

[akhileshacc]

Do we plan to add this functionality down the line ?

[oscerd]

No, we don't. It doesn't make sense. If you use Camel, you can add an idempotent repository to the consumer, so you would be able to consume each file just one time and at the same time we cannot change the Camel component behavior for a particular use case. Moving stuff to a different bucket it's already a good tradeoff.

[akhileshacc]

Got it thanks.

you can add an idempotent repository to the consumer

What do you mean by this ?

[oscerd]

In Apache Camel you can use https://camel.apache.org/components/latest/eips/idempotentConsumer-eip.html

So if you really need to avoid a situation where you have to move the files on a different bucket, you can avoid using the camel-aws-s3 connector and pass your application to a pure camel route, by using the camel-aws-s3 component in combination with the camel-kafka component.

[akhileshacc]

Sure, trying it out.

A question regarding idempotent.
Will messages coming from s3 to kafka be filtered after they reach kafka-connect ?
If yes then won't we be having traffic overhead as files data will continue to move only not get added to topic ?

or this mechanism filters file list and sends only new one ?

[oscerd]

No, they will be filtered before reaching kafka. If you use the idempotentRepository don't forget to set the deleteAfterRead=false, so the file won't be deleted