MSK - kafka-console-producer error with Kafka client version 3.4.0

I have been working with MSK Serverless/Provisioned and kafka-clients 3.4.0. Everything worked as it should when I use * for a topic name.

arn:aws:kafka:region:account-id:topic/cluster-name/cluster-uuid/*

But, when I had specified a topic name and tried to produce to it: arn:aws:kafka:region:account-id:topic/cluster-name/cluster-uuid/private-topic-1

I got the following error: Caused by: org.apache.kafka.common.errors.ClusterAuthorizationException: Cluster authorization failed

So, I decided to use two terminals in parallel: 1: kafka-3.0.0 2. kafka-3.4.0

I made the same configuration for both versions. The same command(kafka-console-producer) works on the version 3.0.0, but failed on the version 3.4.0 (Cluster authorization failed).

I decided to try multiple versions, and conclusion is that everything works as it should up to kafka 3.0.1 version. Producer start failing with the 3.0.1 version.

Consumer works with the specified topic on the version 3.4.0. So the problem is in the producer.

So, whatever version of kafka (MSK) I choose, kafka-client doesn't work if the version is above 3.0.0. Has anyone had a similar experience? How to solve that issue?

I found a solution. If we are working with kafka-client version above 3.0.0 and MSK Serverless/Provisioned, we can specify IAM auth and allow only topic with a specific name, but we have to:

1. Specify kafka-cluster:WriteDataIdempotently

2. Define cluster, topic and groups in the same statement block.

    {
        "Version": "2012-10-17",
        "Statement": [
            {
                "Effect": "Allow",
                "Action": [
                    "kafka-cluster:Connect",
                    "kafka-cluster:DescribeCluster",
                    "kafka-cluster:DescribeTopic",
                    "kafka-cluster:CreateTopic",
                    "kafka-cluster:DeleteTopic",
                    "kafka-cluster:ReadData",
                    "kafka-cluster:WriteData",
                    "kafka-cluster:WriteDataIdempotently",
                    "kafka-cluster:AlterGroup",
                    "kafka-cluster:DescribeGroup"
                ],
                "Resource": [
                    "arn:aws:kafka:region:account-id:cluster/cluster-name/cluster-uuid",
                    "arn:aws:kafka:region:account-id:topic/cluster-name/cluster-uuid/dummyTopic",
                    "arn:aws:kafka:region:account-id:topic/cluster-name/cluster-uuid/private-topic",
                    "arn:aws:kafka:region:account-id:group/cluster-name/cluster-uuid/alice-group"
                ]
            }
        ]
    }