# Filter topics based on topic name while using topics.regex for s3 sink connector

**URL:** <https://forum.confluent.io/t/filter-topics-based-on-topic-name-while-using-topics-regex-for-s3-sink-connector/8102>\
**Category:** Stream Processing\
**Created:** [15 June 2023 04:47 UTC](https://forum.confluent.io/t/filter-topics-based-on-topic-name-while-using-topics-regex-for-s3-sink-connector/8102 "2023-06-15T04:47:27Z")\
**Posts on this page:** 8\
**Page:** 1

<div class="post-metadata">

**Author:** ![Akshay](https://avatars.discourse-cdn.com/v4/letter/a/3ab097/32.png) [@Akshay](https://forum.confluent.io/u/Akshay)\
**Post date:** [15 June 2023 04:47 UTC](https://forum.confluent.io/t/filter-topics-based-on-topic-name-while-using-topics-regex-for-s3-sink-connector/8102/1 "2023-06-15T04:47:27Z")

</div>

i want to write records for all of my topics for a source connector using single s3 sink connector.

i m able to achieve this using topics.regex

but i m facing 2 issues -  
1 i m not able to map datasource name → {schema}/{table} using topics.dir  
2 i m not able to filter few topics e.g. - heartbeat topic data, i dont want heartbeat table records in my s3 bucket.

could you pls help with the appropriate topic router and filter syntax

thanks.

current config of the connector – its not working as per requirements.

```auto
{
    "connector.class": "io.confluent.connect.s3.S3SinkConnector",
    "s3.region": "ap-south-1",
    "topics.dir": "0/single_sink/metis/${topic}",
    "flush.size": "10000",
    "tasks.max": "1",
    "s3.part.size": "67108864",
    "timezone": "Asia/Calcutta",
    "rotate.interval.ms": "60000",
    "locale": "en_GB",
    "format.class": "io.confluent.connect.s3.format.parquet.ParquetFormat",
    "value.converter": "io.confluent.connect.avro.AvroConverter",
    "s3.bucket.name": "zeta-aws-aps1-metis-0-s3-pvt",
    "key.converter": "io.confluent.connect.avro.AvroConverter",
    "partition.duration.ms": "86400000",
    "schema.compatibility": "NONE",
    "topics.regex": "cdc_0_metis_metis_0_pgdb.*",
    "parquet.codec": "gzip",
    "connect.meta.data": "true",
    "parquet.avro.write-old-list-structure": "false",
    "value.converter.schema.registry.url": "https://schemaregistry-metis.internal.mum1-pp.zetaapps.in:443",
    "partitioner.class": "io.confluent.connect.storage.partitioner.TimeBasedPartitioner",
    "name": "cdc_0_metis_metis_0_pgdb.single_sink-aws-aps1-metis-0-s3-pvt_ap-south-1",
    "storage.class": "io.confluent.connect.s3.storage.S3Storage",
    "path.format": "'date'=YYYY-MM-dd",
    "rotate.schedule.interval.ms": "180000",
    "timestamp.extractor": "RecordField",
    "key.converter.schema.registry.url": "https://schemaregistry-metis.internal.mum1-pp.zetaapps.in:443",
    "timestamp.field": "cdc_source_ts_ms",
    "transforms": "filter",
    "transforms.filter.type": "org.apache.kafka.connect.transforms.RegexRouter",
    "transforms.filter.regex": "cdc_0_metis_metis_0_pgdb.metis.cdc_heartbeat",
    "transforms.filter.replacement": "",
    "errors.tolerance": "all",
    "partition.field.name": "topic"
}

```

---

<div class="post-metadata">

**Author:** ![Akshay](https://avatars.discourse-cdn.com/v4/letter/a/3ab097/32.png) [@Akshay](https://forum.confluent.io/u/Akshay)\
**Post date:** [15 June 2023 06:31 UTC](https://forum.confluent.io/t/filter-topics-based-on-topic-name-while-using-topics-regex-for-s3-sink-connector/8102/2 "2023-06-15T06:31:55Z")

</div>

@OneCricketeer could you pls help. I saw some similar discussions over stackoverflow.

---

<div class="post-metadata">

**Author:** ![OneCricketeer](https://sea1.discourse-cdn.com/flex019/user_avatar/forum.confluent.io/onecricketeer/32/1551_2.png) [@OneCricketeer](https://forum.confluent.io/u/OneCricketeer)\
**Post date:** [15 June 2023 07:31 UTC](https://forum.confluent.io/t/filter-topics-based-on-topic-name-while-using-topics-regex-for-s3-sink-connector/8102/3 "2023-06-15T07:31:35Z")

</div>

Regex router is for renaming, not filtering

Use a predicate for filtering

---

<div class="post-metadata">

**Author:** ![Akshay](https://avatars.discourse-cdn.com/v4/letter/a/3ab097/32.png) [@Akshay](https://forum.confluent.io/u/Akshay)\
**Post date:** [15 June 2023 08:43 UTC](https://forum.confluent.io/t/filter-topics-based-on-topic-name-while-using-topics-regex-for-s3-sink-connector/8102/4 "2023-06-15T08:43:02Z")

</div>

could you pls help to share one example. thanks.

```auto
{
    "connector.class": "io.confluent.connect.s3.S3SinkConnector",
    "s3.region": "ap-south-1",
    "topics.dir": "0/single_sink/metis/${topic}",
    "flush.size": "10000",
    "tasks.max": "1",
    "s3.part.size": "67108864",
    "timezone": "Asia/Calcutta",
    "rotate.interval.ms": "60000",
    "locale": "en_GB",
    "format.class": "io.confluent.connect.s3.format.parquet.ParquetFormat",
    "value.converter": "io.confluent.connect.avro.AvroConverter",
    "s3.bucket.name": "zeta-aws-aps1-metis-0-s3-pvt",
    "key.converter": "io.confluent.connect.avro.AvroConverter",
    "partition.duration.ms": "86400000",
    "schema.compatibility": "NONE",
    "topics.regex": "cdc_0_metis_metis_0_pgdb.*",
    "parquet.codec": "gzip",
    "connect.meta.data": "true",
    "parquet.avro.write-old-list-structure": "false",
    "value.converter.schema.registry.url": "https://schemaregistry-metis.internal.mum1-pp.zetaapps.in:443",
    "partitioner.class": "io.confluent.connect.storage.partitioner.TimeBasedPartitioner",
    "name": "cdc_0_metis_metis_0_pgdb.single_sink-aws-aps1-metis-0-s3-pvt_ap-south-1",
    "storage.class": "io.confluent.connect.s3.storage.S3Storage",
    "path.format": "'date'=YYYY-MM-dd",
    "rotate.schedule.interval.ms": "180000",
    "timestamp.extractor": "RecordField",
    "key.converter.schema.registry.url": "https://schemaregistry-metis.internal.mum1-pp.zetaapps.in:443",
    "timestamp.field": "cdc_source_ts_ms",
    "partition.field.name": "topic",
    "transforms": "filter,predicate",
    "transforms.filter.type": "org.apache.kafka.connect.transforms.Filter",
    "transforms.predicate.type": "org.apache.kafka.connect.transforms.Predicate",
    "transforms.predicate.predicate": "if (record.topic() == 'heartbeat_topic') false else true",
    "transforms.filter.condition": "if (value == null) false else true"
}

```

i tried this but its not working.

---

<div class="post-metadata">

**Author:** ![OneCricketeer](https://sea1.discourse-cdn.com/flex019/user_avatar/forum.confluent.io/onecricketeer/32/1551_2.png) [@OneCricketeer](https://forum.confluent.io/u/OneCricketeer)\
**Post date:** [15 June 2023 12:07 UTC](https://forum.confluent.io/t/filter-topics-based-on-topic-name-while-using-topics-regex-for-s3-sink-connector/8102/5 "2023-06-15T12:07:01Z")

</div>

`predicates` is it’s own config, not to be put in the `transforms` list

- [Apache Kafka](https://kafka.apache.org/documentation/#sourceconnectorconfigs_predicates)
- [Apache Kafka](https://kafka.apache.org/documentation/#connect_predicates)

I don’t have an example because I’ve not done what you’re trying to do. I’ve also never used topics.regex and only used an explicit list of topics.

---

<div class="post-metadata">

**Author:** ![Akshay](https://avatars.discourse-cdn.com/v4/letter/a/3ab097/32.png) [@Akshay](https://forum.confluent.io/u/Akshay)\
**Post date:** [15 June 2023 12:39 UTC](https://forum.confluent.io/t/filter-topics-based-on-topic-name-while-using-topics-regex-for-s3-sink-connector/8102/6 "2023-06-15T12:39:40Z")

</div>

thanks man. its working now.

```auto
{
    "connector.class": "io.confluent.connect.s3.S3SinkConnector",
    "s3.region": "ap-south-1",
    "topics.dir": "0/single_sink/metis/${topic}",
    "flush.size": "10000",
    "tasks.max": "1",
    "s3.part.size": "67108864",
    "timezone": "Asia/Calcutta",
    "rotate.interval.ms": "60000",
    "locale": "en_GB",
    "format.class": "io.confluent.connect.s3.format.parquet.ParquetFormat",
    "value.converter": "io.confluent.connect.avro.AvroConverter",
    "s3.bucket.name": "zeta-aws-aps1-metis-0-s3-pvt",
    "key.converter": "io.confluent.connect.avro.AvroConverter",
    "partition.duration.ms": "86400000",
    "schema.compatibility": "NONE",
    "topics.regex": "cdc_0_metis_metis_0_pgdb.*",
    "parquet.codec": "gzip",
    "connect.meta.data": "true",
    "parquet.avro.write-old-list-structure": "false",
    "value.converter.schema.registry.url": "https://schemaregistry-metis.internal.mum1-pp.zetaapps.in:443",
    "partitioner.class": "io.confluent.connect.storage.partitioner.TimeBasedPartitioner",
    "name": "cdc_0_metis_metis_0_pgdb.single_sink-aws-aps1-metis-0-s3-pvt_ap-south-1",
    "storage.class": "io.confluent.connect.s3.storage.S3Storage",
    "path.format": "'date'=YYYY-MM-dd",
    "rotate.schedule.interval.ms": "180000",
    "timestamp.extractor": "RecordField",
    "key.converter.schema.registry.url": "https://schemaregistry-metis.internal.mum1-pp.zetaapps.in:443",
    "timestamp.field": "cdc_source_ts_ms",
    "transforms": "Filter",
    "transforms.Filter.type": "org.apache.kafka.connect.transforms.Filter",
    "transforms.Filter.predicate": "IsFoo",
    "transforms.Filter.negate": "true",
    "predicates": "IsFoo",
    "predicates.IsFoo.type": "org.apache.kafka.connect.transforms.predicates.TopicNameMatches",
    "predicates.IsFoo.pattern": "cdc_0_metis_metis_0_pgdb.metis.*,cdc_0_metis_metis_0_pgdb.transactions"
}

```

Its able to block the unwanted topics.

---

<div class="post-metadata">

**Author:** ![Akshay](https://avatars.discourse-cdn.com/v4/letter/a/3ab097/32.png) [@Akshay](https://forum.confluent.io/u/Akshay)\
**Post date:** [15 June 2023 12:44 UTC](https://forum.confluent.io/t/filter-topics-based-on-topic-name-while-using-topics-regex-for-s3-sink-connector/8102/7 "2023-06-15T12:44:37Z")

</div>

could u pls help regarding the first requirement as well.

I want the files to be saved on a custom path – 0/single\_sink/metis/{schema}/{table}  
So I added it in topics.dir, but i am not able to get the files on the desired path.

“topics.dir”: “0/single\_sink/metis/${topic}”,

using this its writing to literally this path → 0/single\_sink/metis/${topic}

and ${topic} is not getting replaced with topicname.  
my topicname consists of {connectorName}.{schema}.{table}

is there a way to get the {schema}.{table} from the topic name, and put it into topics.dir.  
@OneCricketeer  
thanks.

---

<div class="post-metadata">

**Author:** ![OneCricketeer](https://sea1.discourse-cdn.com/flex019/user_avatar/forum.confluent.io/onecricketeer/32/1551_2.png) [@OneCricketeer](https://forum.confluent.io/u/OneCricketeer)\
**Post date:** [15 June 2023 13:26 UTC](https://forum.confluent.io/t/filter-topics-based-on-topic-name-while-using-topics-regex-for-s3-sink-connector/8102/8 "2023-06-15T13:26:50Z")

</div>

The S3 bucket path cannot be changed without forking the S3 sink code