S3SinkConnector: mismatch message count on S3

Kafka Connect
1

Hi, I’ve ~130M messages (280 partitions) on Kafka. Using below config writing them to S3. Seeing lag is ~100 for my consumer. when I check S3 it has ~30M messages.
what could be a problem. There is no error on logs.

“config”: {

"connector.class": "S3SinkConnector",

"topics": "xxxx",

"s3.bucket.name": "xxx",

"format.class": "io.confluent.connect.s3.format.parquet.ParquetFormat",

"parquet.codec": "snappy",

"storage.class": "io.confluent.connect.s3.storage.S3Storage",

"flush.size": "2147483647",

"s3.region": "xxx,

"tasks.max": "3",

"partition.duration.ms": "3600000",

"partitioner.class": "io.confluent.connect.storage.partitioner.TimeBasedPartitioner",

"locale": "en-US",

"timezone": "UTC",

"rotate.schedule.interval.ms": "3600000",

"rotate.interval.ms": "3600000",

"path.format": "YYYY-MM-dd",

"timestamp.extractor": "Record",

"transforms": "ReplaceField",

"transforms.ReplaceField.type": "org.apache.kafka.connect.transforms.ReplaceField$Value",

"transforms.ReplaceField.blacklist": "xxxxx"

}

2

How are you checking the files? You will have records distributed over multiple files, across multiple YYYY-mm-dd prefixes… And 3600000ms is not equal to one day for your partitioning scheme.

Note: You may not want both rotate.schedule.interval.ms and rotate.interval.ms.

3

I just setup for hour.
I’ve talked with the owner of topic. it seems they just delete messages but offset info still show 130M records. so false alarm!

4

This topic was automatically closed 30 days after the last reply. New replies are no longer allowed.