kafka-connect-leboncoin-config-example-full.json

{
  "connector.class": "io.confluent.connect.s3.S3SinkConnector",
  "errors.tolerance": "none",
  "errors.logs.enable": "true",
  "errors.log.include.messages": "true",
  "topics": "leboncoin_staging_ads_ad-publications_public_avro",
  "s3.bucket.name": "leboncoin_kafka_events_bucket",
  "topics.dir": "staging/topics/ads/raw/parquet",
  "storage.class": "io.confluent.connect.s3.storage.S3Storage",
  "format.class": "io.confluent.connect.s3.format.parquet.ParquetFormat",
  "timestamp.extractor": "fr.leboncoin.data.archiver.parser.LeboncoinTimestampExtractor",
  "transforms": "tombstoneHandler,insertFields",
  "transforms.tombstoneHandler.type": "io.confluent.connect.transforms.TombstoneHandler",
  "transforms.tombstoneHandler.behavior": "ignore",
  "transforms.insertFields.type": "org.apache.kafka.connect.transforms.InsertField$Value",
  "transforms.insertFields.partition.field": "_kafka_partition",
  "transforms.insertFields.offset.field": "_kafka_offset",
  "transforms.insertFields.timestamp.field": "_kafka_timestamp",
  "rotate.interval.ms": "3600000",
  "rotate.schedule.interval.ms": "1200000",
  "flush.size": "500000",
  "path.format": "'schema_version'=VV/'event_date'=YYYY-MM-dd/'event_hour'=HH",
  "partitioner.class": "fr.leboncoin.data.archiver.partitioner.CustomTimePartitioner"
}