CODEWITHSUNDEEP
apache-kafkaapache-kafka-connect
ajkush
ajkush

Reputation: 589

Kafka Connect - Transformes rename field only if it exist

I have an S3 sink connector for multiple topics (topic_a, topic_b, topic_c) and topic_a have field created_date and topic_b, topic_c have creation_date . I have used the below transforms.RenameField.renames to rename the field (created_date:creation_date) but since the only topic_a have created_date and others don't, the connector is failing.

I want to move all the messages (from all topics with single connector) into s3 with creation_date (and rename created_date to creation_date if exist) but I am not able to figure out the regex or transformer to rename the field (if it exists) for the specific topic.

   "config":{
      "connector.class":"io.confluent.connect.s3.S3SinkConnector",
      "errors.log.include.messages":"true",
      "s3.region":"eu-west-1",
      "topics.dir":"dir",
      "flush.size":"5",
      "tasks.max":"2",
      "s3.part.size":"5242880",
      "timezone":"UTC",
      "locale":"en",
      "format.class":"io.confluent.connect.s3.format.json.JsonFormat",
      "errors.log.enable":"true",
      "s3.bucket.name":"bucket",
      "topics": "topic_a, topic_b, topic_c",
      "s3.compression.type":"gzip",
      "partitioner.class":"io.confluent.connect.storage.partitioner.DailyPartitioner",
      "name":"NAME",
      "storage.class":"io.confluent.connect.s3.storage.S3Storage",
      "key.converter.schemas.enable":"true",
      "key.converter":"org.apache.kafka.connect.storage.StringConverter",
      "value.converter.schemas.enable":"true",
      "value.converter":"io.confluent.connect.avro.AvroConverter",
      "value.converter.schema.registry.url":"https://schemaregistry.com",
      "enhanced.avro.schema.support": "true",
      "transforms": "RenameField",
      "transforms.RenameField.type": "org.apache.kafka.connect.transforms.ReplaceField$Value",
      "transforms.RenameField.renames": "created_date:creation_date"
   }

Upvotes: 0

Views: 1047

Answers (1)

OneCricketeer
OneCricketeer

Reputation: 191733

only topic_a have created_date and others don't,

Then you would use separate Connectors. One with the Transform and all topics with the field, then another without the transform.

from all topics with single connector

This doesn't scale very well. You're making limited consumer threads and one consumer group to read from many topics at once. Multiple connectors would be better to distribute the load.

Upvotes: 1

Related Questions