AWS MSK - InvalidInput.InvalidConnectorConfiguration issue

Does anyone have experience setting up a Redshift Connector on AWS MSK Connect offering?
I used the config from the docs

But I’m unable to get passed the same error message, which I cant decypher.

Message: The connector configuration is invalid. Message: class java.lang.String cannot be cast to class java.util.List (java.lang.String and java.util.List are in module java.base of loader 'bootstrap')```

How does your connector configuration looks like?

    "connectorName": "redshift-sink",
    "kafkaConnectVersion": "2.7.1",
    "serviceExecutionRoleArn": "<RoleARN>",
    "plugins": [
            "customPlugin": {
                "customPluginArn": "<RoleARN>",
                "revision": 1
    "capacity": {
        "provisionedCapacity": {
            "workerCount": 2,
            "mcuCount": 1
    "kafkaCluster": {
        "apacheKafkaCluster": {
            "bootstrapServers": "BOOSTRAP_SERVERS",
            "vpc": {
                "subnets": [
                "securityGroups": [
    "kafkaClusterClientAuthentication": {"authenticationType": "IAM"},
    "kafkaClusterEncryptionInTransit": {"encryptionType": "TLS"},
    "logDelivery": {
        "workerLogDelivery": {
            "cloudWatchLogs": {
                "logGroup": "/aws/msk/connectors/redshiftsink",
                "enabled": true
    "connectorConfiguration": {
        "confluent.topic.bootstrap.servers": "comma seperated list of bootstrap servers",
        "confluent.topic.replication.factor": "1",
        "connector.class": "",
        "tasks.max": "2",
        "topics": "topic_name",
        "aws.redshift.domain": "domain",
        "aws.redshift.port": "5439",
        "aws.redshift.database": "table_name",
        "aws.redshift.user": "${secretManager:<SECRET>}",
        "aws.redshift.password": "${secretManager:SECRET}",
        "pk.mode": "kafka",
        "auto.create": "true",
        "": "SASL_SSL",
        "confluent.topic.sasl.mechanism": "AWS_MSK_IAM",
        "confluent.topic.sasl.jaas.config":" required;",
        "confluent.topic.sasl.client.callback.handler.class": ""
    "workerConfiguration": {
        "workerConfigurationArn": "SECRETSTOREARN",
        "revision": 1

Hi This is the config, I’ve removed all sensitive info

What did you pass to: “confluent.topic.bootstrap.servers”: “comma seperated list of bootstrap servers”?

Did you pass the correct url to the server?

Yeah the bootstrap servers are from the kafka cluster i have running in the same vpc as the connector

The value looks like this

I did look at my config too and the only difference it was that bootstrap.servers, but I set just one server like “confluent.topic.bootstrap.servers”: “SASL_SSL://

Previously I didn’t have the SASL_SSL setting, on the boostrap.servers. I added that but still no dice

Can you send me what your config file looks like