Grab topics using regex and put in S3 by topic name

[cristianburca]

Hello,

I would like to know if possible to put grabbed topics with regex into particular folder in s3 by topic name.

let's say we have:

test.topic1.error
test.topic2.error
test.topic3.error
test.topic1.raw
test.topic2.raw
test.topic3.raw

And my sink s3 connector using a regex looks like:

{
 "name": "s3-camel-pattern",
 "config": {
    "connector.class": "org.apache.camel.kafkaconnector.awss3.CamelAwss3SinkConnector",
	"tasks.max": "1",
	"key.converter": "org.apache.kafka.connect.converters.ByteArrayConverter",
    "value.converter": "org.apache.kafka.connect.converters.ByteArrayConverter",
	"topics.regex": ".*(error|raw)$",
    "camel.sink.path.bucketNameOrArn": "camelecareconnectortestreg",
	"camel.sink.endpoint.keyName": "${date:now:yyyyMMdd-HHmmssSSS}-${exchangeId}",
	"camel.component.aws-s3.region": "xxx",
        "camel.component.aws-s3.accessKey": "xxx",
        "camel.component.aws-s3.secretKey": "xxx"
   }
}

Right now all the topics messages are bulk put directly into root folder of s3 bucket.

I was thinking if we can split by folders using the topic name.
eg adding something like:
"camel.sink.endpoint.keyName": "$TOPIC_NAME/${date:now:yyyyMMdd-HHmmssSSS}-${exchangeId}"
where TOPIC_NAME is automatically retrieved as test.topic1.error and the other names.

Then the idea is to use a source s3 connector and put the messages from a specific s3 prefix into a particular topic (eg test.topic1.error)

{
 "name": "s3-camel-ecare-source-for-topic1",
 "config": {
    "connector.class": "org.apache.camel.kafkaconnector.awss3.CamelAwss3SourceConnector",
	"tasks.max": "1",
	"key.converter": "org.apache.kafka.connect.converters.ByteArrayConverter",
        "value.converter": "org.apache.kafka.connect.converters.ByteArrayConverter",
	"topics": "s3.test.topic1.error",
        "camel.source.path.bucketNameOrArn": "camelecareconnectortestreg?autocloseBody=false",
        "camel.source.endpoint.prefix" : "test.topic1.error/", 
	"camel.source.endpoint.includeBody": "true",
        "camel.component.aws-s3.deleteAfterRead": "false",
	"camel.component.aws-s3.region": "xxx",
        "camel.component.aws-s3.accessKey": "xxx",
        "camel.component.aws-s3.secretKey": "xxx"
   }
}

[oscerd]

The expression you see in the filename is coming from Camel, while the topic name should be retrieved by the connector configuration, so no, at the moment is not possible. You'll have to define n connectors one for each topic, if you want to achieve this.

[dhope-nagesh]

@oscerd Any timeline for this feature, if your are going to consider it in future?

[oscerd]

We are actually not considering this. I think I'll close this one.

[oscerd]

Thanks for pinging.