I have uploaded some avro messages from a topic, let's say my.topic, to an amazon s3 bucket, let's say s3-bucket, using the kafka-connect s3 sink connector. The config for the sink connector was as follows:
{
"connector.class": "io.confluent.connect.s3.S3SinkConnector",
"key.converter": "org.apache.kafka.connect.converters.LongConverter",
"value.converter": "io.confluent.connect.avro.AvroConverter",
"value.converter.schema.registry.url": "http://schemaregistry:8099",
"value.converter.value.subject.name.strategy": "io.confluent.kafka.serializers.subject.TopicRecordNameStrategy",
"tasks.max": "1",
"topics": "my.topic",
"s3.region": "eu-west-2",
"s3.bucket.name": "s3-bucket",
"flush.size": "5",
"storage.class": "io.confluent.connect.s3.storage.S3Storage",
"format.class": "io.confluent.connect.s3.format.avro.AvroFormat",
"schema.generator.class": "io.confluent.connect.storage.hive.schema.DefaultSchemaGenerator",
"schema.compatibility": "NONE",
"partitioner.class": "io.confluent.connect.storage.partitioner.DefaultPartitioner"
}
This worked as expected, all the messages were the same record with the same schema version, I write 5 to the topic and see one s3 object in my bucket with path
/topics/my.topic/partition=0/my.topic+0+0000000000.avro
Now I want to put these stored messages onto another empty topic. I start up the s3 source connector with the following config:
{
"confluent.topic.bootstrap.servers": "kafka:9092",
"confluent.topic.replication.factor": 1,
"connector.class": "io.confluent.connect.s3.source.S3SourceConnector",
"s3.region": "eu-west-2",
"s3.bucket.name": "s3-bucket",
"format.class": "io.confluent.connect.s3.format.avro.AvroFormat",
"partitioner.class": "io.confluent.connect.storage.partitioner.DefaultPartitioner",
"transforms": "AddPrefix",
"transforms.AddPrefix.type": "org.apache.kafka.connect.transforms.RegexRouter",
"transforms.AddPrefix.regex": ".*",
"transforms.AddPrefix.replacement": "recovery_$0"
}
When I look through the logs produced by kafka-connect (running inside a docker container) it seems happy, there are no errors and it correctly identifies my bucket and the directory path inside it is assigned to be watched
/topics/my.topic/partition=0/
however it never detects the file inside and never writes anything to the expected recovery_my.topic topic. It repeatedly logs
kafka-connect | [2020-07-05 15:31:46,311] INFO PartitionCheckingTask - Checking if Partitions have changed. (io.confluent.connect.cloud.storage.source.util.PartitionCheckingTask)
kafka-connect | [2020-07-05 15:31:47,963] INFO WorkerSourceTask{id=tx-s3-restore-0} Committing offsets (org.apache.kafka.connect.runtime.WorkerSourceTask)
kafka-connect | [2020-07-05 15:31:47,964] INFO WorkerSourceTask{id=tx-s3-restore-0} flushing 0 outstanding messages for offset commit (org.apache.kafka.connect.runtime.WorkerSourceTask)
kafka-connect | [2020-07-05 15:31:50,483] INFO AvroDataConfig values:
kafka-connect | schemas.cache.config = 50
kafka-connect | enhanced.avro.schema.support = false
kafka-connect | connect.meta.data = true
kafka-connect | (io.confluent.connect.avro.AvroDataConfig)
kafka-connect | [2020-07-05 15:31:50,483] INFO AvroDataConfig values:
kafka-connect | schemas.cache.config = 50
kafka-connect | enhanced.avro.schema.support = false
kafka-connect | connect.meta.data = true
kafka-connect | (io.confluent.connect.avro.AvroDataConfig)
kafka-connect | [2020-07-05 15:31:50,537] INFO AvroDataConfig values:
kafka-connect | schemas.cache.config = 50
kafka-connect | enhanced.avro.schema.support = false
kafka-connect | connect.meta.data = true
kafka-connect | (io.confluent.connect.avro.AvroDataConfig)
kafka-connect | [2020-07-05 15:31:50,589] INFO No new files ready after scan task assigned folders (io.confluent.connect.cloud.storage.source.StorageSourceTask)
which indicates to me that it is ignoring the file for some reason? Here is the full s3 source connector config pulled from the logs
kafka-connect | [2020-07-05 15:10:49,427] INFO S3SourceConnectorConfig values:
kafka-connect | behavior.on.error = fail
kafka-connect | confluent.license =
kafka-connect | confluent.topic = _confluent-command
kafka-connect | confluent.topic.bootstrap.servers = [kafka:9092]
kafka-connect | confluent.topic.replication.factor = 1
kafka-connect | directory.delim = /
kafka-connect | filename.regex = (.+)\+(\d+)\+.+$
kafka-connect | folders = [topics/my.topic/partition=0/]
kafka-connect | format.bytearray.extension = .bin
kafka-connect | format.bytearray.separator =
kafka-connect | format.class = class io.confluent.connect.s3.format.avro.AvroFormat
kafka-connect | partition.field.name = []
kafka-connect | partitioner.class = class io.confluent.connect.storage.partitioner.DefaultPartitioner
kafka-connect | path.format =
kafka-connect | record.batch.max.size = 200
kafka-connect | s3.bucket.name = s3-bucket
kafka-connect | s3.credentials.provider.class = class com.amazonaws.auth.DefaultAWSCredentialsProviderChain
kafka-connect | s3.http.send.expect.continue = true
kafka-connect | s3.part.retries = 3
kafka-connect | s3.poll.interval.ms = 60000
kafka-connect | s3.proxy.password = [hidden]
kafka-connect | s3.proxy.url =
kafka-connect | s3.proxy.user = null
kafka-connect | s3.region = eu-west-2
kafka-connect | s3.retry.backoff.ms = 200
kafka-connect | s3.sse.customer.key = [hidden]
kafka-connect | s3.ssea.name =
kafka-connect | s3.wan.mode = false
kafka-connect | schema.cache.size = 50
kafka-connect | store.url = null
kafka-connect | topics.dir = topics
kafka-connect | (io.confluent.connect.s3.source.S3SourceConnectorConfig)
kafka-connect | [2020-07-05 15:10:49,428] INFO [Producer clientId=connector-producer-tx-s3-restore-0] Cluster ID: nlQYzBVYRbWozKk54-Qx_A (org.apache.kafka.clients.Metadata)
kafka-connect | [2020-07-05 15:10:49,432] INFO AvroDataConfig values:
kafka-connect | schemas.cache.config = 50
kafka-connect | enhanced.avro.schema.support = false
kafka-connect | connect.meta.data = true
kafka-connect | (io.confluent.connect.avro.AvroDataConfig)
kafka-connect | [2020-07-05 15:10:49,434] INFO Starting source connector task with assigned folders [topics/my.topic/partition=0/] using partitioner io.confluent.connect.storage.partitioner.DefaultPartitioner (io.confluent.connect.cloud.storage.source.StorageSourceTask)
If anyone has ideas about why my file is ignored I would be very grateful.