Can I use single s3-sink connector to point same field name for the timeStamp field by using different type of Avro Schema's for different topics?

Question

schema for topic t1

{
  "type": "record",
  "name": "Envelope",
  "namespace": "t1",
  "fields": [
    {
      "name": "before",
      "type": [
        "null",
        {
          "type": "record",
          "name": "Value",
          "fields": [
            {
              "name": "id",
              "type": {
                "type": "long",
                "connect.default": 0
              },
              "default": 0
            },
            {
              "name": "createdAt",
              "type": [
                "null",
                {
                  "type": "string",
                  "connect.version": 1,
                  "connect.name": "io.debezium.time.ZonedTimestamp"
                }
              ],
              "default": null
            },
           
          ],
          "connect.name": "t1.Value"
        }
      ],
      "default": null
    },
    {
      "name": "after",
      "type": [
        "null",
        "Value"
      ],
      "default": null
    }
   
  ],
  "connect.name": "t1.Envelope"
}

schema for topic t2

{
    "type": "record",
    "name": "Value",
    "namespace": "t2",
    "fields": [
          {
              "name": "id",
              "type": {
                "type": "long",
                "connect.default": 0
              },
              "default": 0
          },
          {
            "name": "createdAt",
            "type": [
                "null",
                {
                    "type": "string",
                    "connect.version": 1,
                    "connect.name": "io.debezium.time.ZonedTimestamp"
                }
            ],
            "default": null
          }
    ],
    "connect.name": "t2.Value"
}

s3-sink Connector configuration

connector.class=io.confluent.connect.s3.S3SinkConnector
behavior.on.null.values=ignore
s3.region=us-west-2
partition.duration.ms=1000
flush.size=1
tasks.max=3
timezone=UTC
topics.regex=t1,t2
aws.secret.access.key=******
locale=US
format.class=io.confluent.connect.s3.format.json.JsonFormat
partitioner.class=io.confluent.connect.storage.partitioner.TimeBasedPartitioner
value.converter.schemas.enable=false
name=s3-sink-connector
aws.access.key.id=******
errors.tolerance=all
value.converter=org.apache.kafka.connect.json.JsonConverter
storage.class=io.confluent.connect.s3.storage.S3Storage
key.converter=org.apache.kafka.connect.storage.StringConverter
s3.bucket.name=s3-sink-connector-bucket
path.format=YYYY/MM/dd
timestamp.extractor=RecordField
timestamp.field=after.createdAt

By using this connector configuration I got error for t2 topic that is "createdAt field does not exist". If I set timestamp.field = createdAt then error is thrown for t1 topic "createdAt field does not exist".

How can I point "createdAt" field in both schemas at the same time by using same connector for both?

Is it possible to achieve this by using a single s3-sink connector configuration ?

If this scenario is possible then how can I do this, which properties I have to use for achieve this?

If anybody has idea about this, please suggest on this. If there is any other way to do this then please suggest that way also.

Answer 1

All topics will need the same timestamp field; there's no way to configure topic-to-field mappings.

Your t2 schema doesn't have an after field, so you need to run two separate connectors

The field is also required to be present in all records, otherwise the partitioner won't work.