Support for S3 Alternatives when using KafkaConnect S3 Sink

[JimFawkes]

I would like to test the KafkaConnect S3 Sink locally. Therefore I would like to setup the connector to write to localstack or minio. Unfortunately I continuously get an AWS Credentials Error.

Questions:

1. Has anyone tried this and succeeded before?
2. Is there an alternative configuration option?
3. Is the store.url configuration option supported (as for the confluent sink, originally from this GH Issue)?

Setup

Running everything locally via docker-compose based on this documentation. The images are:

• redpandadata/connectors:1.0.0-dev-bc9a905
• redpandadata/console:v2.2.0
• redpandadata/redpanda:v22.3.11

Connector configuration:

{
    "aws.access.key.id": "test",
    "aws.s3.bucket.name": "my-test-bucket",
    "aws.secret.access.key": "test",
    "connector.class": "com.redpanda.kafka.connect.s3.S3SinkConnector",
    "format.output.type": "json",
    "name": "s3-test-connector",
    "store.url": "http://minio:9000",
    "topics": "my-topic"
}

Error Taceback:

org.apache.kafka.common.config.ConfigException: Cannot connect to 'my-test-bucket' S3 bucket due to: The AWS Access Key Id you provided does not exist in our records.
	at com.redpanda.kafka.connect.s3.config.AwsConfigValidator.validate(AwsConfigValidator.java:57)
	at com.redpanda.kafka.connect.s3.S3SinkConnector.start(S3SinkConnector.java:71)
	at org.apache.kafka.connect.runtime.WorkerConnector.doStart(WorkerConnector.java:190)
	at org.apache.kafka.connect.runtime.WorkerConnector.start(WorkerConnector.java:215)
	at org.apache.kafka.connect.runtime.WorkerConnector.doTransitionTo(WorkerConnector.java:360)
	at org.apache.kafka.connect.runtime.WorkerConnector.doTransitionTo(WorkerConnector.java:343)
	at org.apache.kafka.connect.runtime.WorkerConnector.doRun(WorkerConnector.java:143)
	at org.apache.kafka.connect.runtime.WorkerConnector.run(WorkerConnector.java:121)
	at org.apache.kafka.connect.runtime.isolation.Plugins.lambda$withClassLoader$1(Plugins.java:177)
	at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:539)
	at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136)
	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635)
	at java.base/java.lang.Thread.run(Thread.java:833)

[dynamike2010]

Perhaps way too late but this is what works fine for me with minio:

{
"aws.access.key.id": "key",
"aws.secret.access.key": "access",
"connector.class": "io.confluent.connect.s3.S3SinkConnector",
"flush.size": "1", // just to see results after each message you produce
"format.class": "io.confluent.connect.s3.format.json.JsonFormat",
"key.converter": "org.apache.kafka.connect.json.JsonConverter",
"key.converter.schemas.enable": "false",
"locale": "US",
"name": "s3-minio-sink",
"partitioner.class": "io.confluent.connect.storage.partitioner.DailyPartitioner",
"path.format": "'year'=YYYY/'month'=MM/'day'=dd",
"s3.bucket.name": "warehouse",
"s3.compression.type": "gzip",
"s3.part.size": "5242880",
"schema.compatibility": "NONE",
"schemas.enable": "false",
"storage.class": "io.confluent.connect.s3.storage.S3Storage",
"store.url": "http://minio:9000",
"tasks.max": "1",
"timezone": "UTC",
"topics": "events",
"value.converter": "org.apache.kafka.connect.json.JsonConverter",
"value.converter.schemas.enable": "false"
}

I am using the following components:
services:
redpanda:
image: vectorized/redpanda:v23.3.12
connect:
image: confluentinc/cp-kafka-connect-base:7.6.1
minio:
image: minio/minio

After this I got files in the following path: warehouse/topics/events/year=2024/month=10/day=11
Hope this helps ;-)

[rockwotj]

Also checkout Redpanda Connect if you want a more modern replacement: https://www.redpanda.com/connect