Questions tagged with Amazon Managed Streaming for Apache Kafka (Amazon MSK)

I have some IoT devices that send raw data which I am planning to store in Kafka and then process it with the consumer and save it to a database. Scenario: IoT devices send sensor data in form of packets (1 packet per sec) and you need 10 raw packets to decode its value (I have a C binary that takes 10 packets, I can enforce in IoT to send 10 packets combined as a single packet but I want IoT to focus on single packet only so that in future I can modify the algorithm to work for 20, 30 packets on the cloud). If I use the same topic for all IoT then consumers have to wait until it has received 10 packets to start processing, which I am thinking can cause some problems. (Problem of filtering packet of IoT, as single packets, can mix up in single topic) Issues: Initially, I thought of creating a single topic with partitions for each IoT device, but Kafka doesn't support dynamic partitions. (More IoT devices can be added or removed during runtime.) Another solution can be to create a topic for each IoT device with 1 partition. Also, the above approach of 1 topic for each IoT device may enforce to use of one consumer for each IoT device added (can this be improved?) I am looking for a better and more efficient implementation if possible for this problem. The number of IoT devices is dynamic and can go up to 100+.

I'm trying to understand how to code my Terraform for MSK, and noticing that the "current_version" changed after my changes were applied. That makes sense since it did change the PCA configuration... but which kinds of changes actually cause that property to change? Does every `aws kafka update-cluster-configuration` change it?

We have a usecase to read Oracle table and publish the records into AWS MSK topic. For that purpose we are using MSKConnect and trying to deploy Confluent JDBCSourceConnector. We are using AWS Glue schema registry for schema management. We have used below configuration in our connector but its just giving the error and connector always goes to failed status. ``` key.converter= org.apache.kafka.connect.storage.StringConverter key.converter.schemas.enable= false key.converter.avroRecordType= GENERIC_RECORD key.converter.region= us-east-1 key.converter.registry.name= ebx-control-tbl-registryE key.converter.schemaAutoRegistrationEnabled= true value.converter= com.amazonaws.services.schemaregistry.kafkaconnect.AWSKafkaAvroConverter value.converter.schemas.enable= true value.converter.avroRecordType= GENERIC_RECORD value.converter.region= us-east-1 value.converter.registry.name= ebx-control-tbl-registry value.converter.schemaAutoRegistrationEnabled= true ``` Its giving below error. ``` [Worker-0dc06f886ba9272ef] Caused by: org.apache.kafka.connect.errors.DataException: Converting Kafka Connect data to byte[] failed due to serialization error: ``` Has anyone successfully used Confluent JDBCSourceConnector with MSK connect and AWS Glue Schema registry?

Current supported versions (https://docs.aws.amazon.com/msk/latest/developerguide/supported-kafka-versions.html) suffer from https://issues.apache.org/jira/browse/KAFKA-13636 which is a serious issue. It is fixed in 2.8.2 although the Kafka team currently have no intention of releasing that version. So the only released version with it fixed is 3.0.1. Can we expect MSK to support 3.0.1 or build and support 2.8.2 any time soon?

I use AWS MSK cluster with brokers logging turned on to CloudWatch. I can see brokers logs. We have some topics with cleanup.policy=compact and some with cleanup.policy=delete. The system is running on the new cluster for about 2 weeks now. From my research (e.g. https://zendesk.engineering/an-investigation-into-kafka-log-compaction-5e520f4291f0) I see that kafka should run log cleaner (obviously) and there should be some traces in logs of this activity. However in my CloudWatch log group I cannot find a word "cleaner" or "cleaned" and I cannot find any trace of log cleaner running. Is it something wrong with my logging settings? Or is log cleaner not running at all (this is suspicious, because we have a lot of messages eligible for cleanup but still not cleaned)?

We added our own Jar with transformation classes to MSK Connect plugin zip file. We then specified the name of the connector in a transform property . Unfortunately we are getting 'Class not found' error. We verified and jar & class exist in ZIP & plugin configuration is correct. Is it possible to add our custom transform classes to MSK connect (plugin.path?) ?

I have created consumer using MSK connect but when I try to reset the offset to the beginning, it keep throwing this error `Error: Assignments can only be reset if the group 'connect-opensearch-connector-v1' is inactive, but the current state is Stable. ` if I were to delete the connector, would I need to re-create it with the same name so that it subscribe to the same consumer group? and pull the records based on the offset which was set post deleting it?

Hi there, We had about 10 mins downtime today while MSK was in a HEALING state. But according to the doc available online. A healing state should not affect the cluster to produce or consume. Is there a reason why we had downtime? Below is our cluster configuration ``` auto.create.topics.enable=true default.replication.factor=3 min.insync.replicas=2 num.io.threads=8 num.network.threads=5 num.partitions=1 num.replica.fetchers=2 replica.lag.time.max.ms=30000 socket.receive.buffer.bytes=102400 socket.request.max.bytes=104857600 socket.send.buffer.bytes=102400 unclean.leader.election.enable=true zookeeper.session.timeout.ms=18000 log.retention.hours=-1 ``` I found these logs on the consumer side. ``` "error":"Messages are rejected since there are fewer in-sync replicas than required","correlationId":5,"size":57} ``` Kindly advise if anything is wrong with the configuration. Thanks

Since IAM authentication on MSK requires monkey-patching the client libraries's classpath in order to work, it's unsuitable for the vast majority of use cases, such as: * Usage with any non-JVM Kafka tools or libraries * Lambda code written in any non-JVM language (most 'serverless' code is NOT written for JVM) * Scenarios where modifying a packaged JVM client library would void support contracts * Scenarios where maintaining modifications to packaged JVM clients every time they're updated is not realistic I'm trying to think of realistic scenarios where a development team would want the simplicity and lack of maintenance of a serverless kafka cluster, but also are willing to commit to throwing out the majority of the available Kafka tools and libraries out there, while also committing to maintaining monkey-patched versions of all of the remaining tools. It's pretty difficult for me to imagine. Given the above, are there any plans for the future to support any security mechanisms on MSK Serverless other than IAM? If not, given the enormous compromises required in order to support IAM usage on MSK, who is the MSK Serverless offering actually targeted at?

Is Burrow still useful for MSK clusters? In the past I have used Burrow to monitor consumer lag. Various blogs have instructions to install Burrow for MSK, but any references to that seem to have disappeared lately. Have the client lag metrics in recent MSK versions mostly obviated the need for tools such as Burrow?

Hi, I've setup a Kakfa cluster on MSK, and it has IAM role-based authentication and Public Access enabled. I'm able to connect from my local environment so no issues there. However, I can no longer update the security settings on the cluster. I'm trying to enable SASL/SCRAM authentication, however I get the following error in the console: > Error updating security settings Amazon MSK wasn’t able to update the security settings. Wait a few minutes and then try again. API response Configuration ARN does not exist. I've re-tried the request a number of times, trying a few different settings, however, it keeps returning this error. Is there anything I could be missing here?

Hi there, I setup MSK connect to push topics from my MSK cluster to the s3 bucket with lenses Plugin. The connector is running at the moment but I got the below error on cloudwatch logs. May I ask what the possible cause is? `Resetting offset for partition telemetry-dev-2 to position FetchPosition{offset=736334, offsetEpoch=Optional.empty, currentLeader=LeaderAndEpoch{leader=Optional[b-2.non-prod-cluster.xjgva0.c3.kafka.ap-southeast-1.amazonaws.com:9092 (id: 2 rack: apse1-az2)], epoch=6}}. (org.apache.kafka.clients.consumer.internals.SubscriptionState:396) [Worker-0fb7a8004d7427620] [2022-03-28 11:25:07,748] INFO [lenseio-msk-non-prod-trans|task-1] [Consumer clientId=connector-consumer-lenseio-msk-non-prod-trans-1, groupId=connect-lenseio-msk-non-prod-trans] Resetting offset for partition telemetry-dev-2 to position FetchPosition{offset=736334, offsetEpoch=Optional.empty, currentLeader=LeaderAndEpoch{leader=Optional[b-2.non-prod-cluster.xjgva0.c3.kafka.ap-southeast-1.amazonaws.com:9092 (id: 2 rack: apse1-az2)], epoch=6}}. (org.apache.kafka.clients.consumer.internals.SubscriptionState:396)` followed by another log as follows `sending LeaveGroup request to coordinator *******.kafka.ap-southeast-1.amazonaws.com:9092 (id: 2147483646 rack: null) due to consumer poll timeout has expired.` And below is the cluster configuration ``` auto.create.topics.enable=true default.replication.factor=2 min.insync.replicas=1 num.io.threads=8 num.network.threads=5 num.partitions=2 num.replica.fetchers=2 replica.lag.time.max.ms=30000 socket.receive.buffer.bytes=102400 socket.request.max.bytes=104857600 socket.send.buffer.bytes=102400 unclean.leader.election.enable=true zookeeper.session.timeout.ms=18000 log.retention.hours=-1```

Hi AWS Users, I am trying to spin up a MSK cluster with a custom MSK configuration using my serverless app. I wrote the cloudformation template for the generation of the MSK Cluster and was able to successfully bring it up. I recently saw that AWS added cloudformation template of `AWS::MSK::Configuration`. [1] I was trying that out to create a custom configuration. The Configuration requires a `ServerProperties`key that is usually a PlainText in AWS console. An example of Server Properties: ``` auto.create.topics.enable=true default.replication.factor=2 min.insync.replicas=2 num.io.threads=8 num.network.threads=5 num.partitions=10 num.replica.fetchers=2 replica.lag.time.max.ms=30000 socket.receive.buffer.bytes=102400 socket.request.max.bytes=104857600 socket.send.buffer.bytes=102400 unclean.leader.election.enable=true zookeeper.session.timeout.ms=18000 ``` `AWS::MSK::Configuration` accepts base64 (api functionality) and I have been trying to implement this. I am using the cloudformation `Fn::Base64` functionality. e.g: ``` Resources: ServerlessMSKConfiguration: Type: AWS::MSK::Configuration Properties: ServerProperties: Fn::Base64: auto.create.topics.enable=true ``` This gives me back a 400 error during deploy. ``` Resource handler returned message: "[ClientRequestToken: xxxxx] Invalid request body (Service: Kafka, Status Code: 400, Request ID: 1139d840-c02d-4fdb-b68c-cee93673d89d, Extended Request ID: null)" (RequestToken: xxxx HandlerErrorCode: InvalidRequest) ``` Can someone please help me format this ServerProperties properly, not sure how to give the proper base64 string in the template. Any help is much appreciated. [1] - [https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-msk-configuration.html](MSK::Configuration)

For example, if I change partion num in my cluster configuration and apply to the cluster, will it also help me update the existing topic partition number? If not, is there a list of configuration that will not be overriden by cluster configuration change after the resource has been initialized?

We have an AWS MSK Cluster setup with IAM Authentication in Account A. We are able to setup an IAM Role in Account A, and allow that role to be assumed by a user in Account B to allow a user cross-account access to the cluster. If we want to run something like AWS Glue for example in Account B that needs to run as an IAM Role in Account B, how can we setup cross-account access to the Cluster in Account A? For other services we would configure a service policy that allows the cross-account trust relationship. I do not see anything like this on the MSK Cluster resource. The only thing I can think of is to use SCRAM authentication with pre-shared user credentials in a secret. However, we really need to use IAM authentication for compliance.

Hello, is MSK Connect available for MSK Serverless? My customer is interested on using a MongoDB connector with MSK Serverless. Thanks

Hi, We have enabled kafka cluster for public access. we are getting below error when we are trying to connect from virtual machines. SSL_connect:SSLv3/TLS write client hello read from 0x55e272aa0740 [0x55e272ab5f33] (5 bytes => -1 (0xFFFFFFFFFFFFFFFF)) SSL_connect:error in SSLv3/TLS write client hello write:errno=104 --- no peer certificate available --- No client certificate CA names sent --- SSL handshake has read 0 bytes and written 359 bytes Verification: OK --- New, (NONE), Cipher is (NONE) Secure Renegotiation IS NOT supported Compression: NONE Expansion: NONE No ALPN negotiated Early data was not sent Verify return code: 0 (ok)

I have an s3 connector deployed on MSK Connect, and a repository on github with the json connector configuration file. I'd like to update the connectors configuration on demand via MSK's REST API. I've checked the API documentatio, but it seems like the UpdateConnector API only allows to modify the capacity configuration. The CreateConnector API does allow to provide connector configuration, but it returns an error if the connector already exists. I could delete and then recreate the connector, but this doesn't seem like a good approach. Is there another way to update a running connector configuration?

Hi there, We recently got a notification about the Kafka MSK security patches update. We had brokers spread across 3 AZs to avoid data loss according to AWS best practices. Upon the patches completion, we realised the Kafka was shut down during the patches, See the Cloudwatch log below. [](https://postimg.cc/yJqW7Fk3) here is the Cluster configuration. Please lemme know if I'm missing anything ``` auto.create.topics.enable=true default.replication.factor=2 min.insync.replicas=2 num.io.threads=8 num.network.threads=5 num.partitions=1 num.replica.fetchers=2 replica.lag.time.max.ms=30000 socket.receive.buffer.bytes=102400 socket.request.max.bytes=104857600 socket.send.buffer.bytes=102400 unclean.leader.election.enable=true zookeeper.session.timeout.ms=18000 ``` Thanks

We are trying to integrate change data capture into our MSK cluster using the Debezium SQL Server plugin with AVRO serialization and AWS Glue Schema Registry. The connector appears to be working correctly as we can see the messages flowing into the Kafka topic and the AVRO schema is registered into our schema registry. However, when we attempt to deserialize the message from Kafka in a lambda function (MSK trigger), we are getting an error saying that the Schema cannot be found. We are using [this](https://github.com/awslabs/aws-glue-schema-registry) library to do the AVRO serializaton/deserialization in both the connector and our lambda function. We've added some additional logging to the library and have found that the schema version id that gets written into the message during serialization is different than the one that gets pulled out of the message during deserialization. Has anyone successfully used Debezium with MSK connect and AWS Glue Schema registry? Pertinent configuration for our MSK connector is provided below: ``` "key.converter": "org.apache.kafka.connect.storage.StringConverter", "key.converter.schemas.enable": "false", "key.converter.avroRecordType": "GENERIC_RECORD", "key.converter.region": "AWS_REGION", "key.converter.registry.name": "SCHEMA_REGISTRY_NAME", "key.converter.schemaAutoRegistrationEnabled": "true", "value.converter": "com.amazonaws.services.schemaregistry.kafkaconnect.AWSKafkaAvroConverter", "value.converter.schemas.enable": "true", "value.converter.avroRecordType": "GENERIC_RECORD", "value.converter.region": "AWS_REGION", "value.converter.registry.name": "SCHEMA_REGISTRY_NAME", "value.converter.schemaAutoRegistrationEnabled": "true", ```

We're trying to route the messages from debezium connector using regex transform(org.apache.kafka.connect.transforms.RegexRouter). But it's working fine in Apache Kafka connect but not on MSK connect. Is MSK connect supports Regex Transforms.

From the MSK connect documentation we could see only the ROLE based access to the VPC. But How to assign the MSK permission to an IAM user(aws credentials file) and add it to the MSK connect for Authentication.

Not able to find a proper documentation on 3rd party openid connect integration with msk iam role identity provider. It would be helpful if someone point to right documents or example. We are registering 3rd party on prem identity provider, but not sure how to connect to msk cluster with these options. below project doesn't have proper documentation. https://github.com/aws/aws-msk-iam-auth

authentication is failing from outside of aws using AIM

We use SASL OAUTHBEARER for our on-prem authentication and looks like Amazon MSK currently does not support that. It would be great if anyone from Amazon MSK team let us know if the support the same is planned in near future. similar feature is avaiable on confluent. https://docs.confluent.io/platform/current/kafka/authentication_sasl/authentication_sasl_oauth.html#

If you look at the supported version of MSK (https://docs.aws.amazon.com/msk/latest/developerguide/supported-kafka-versions.html) you'll notice that there is a supported version (currently 2.6.2). Anyone an idea why? What's the drawback with other/newer versions? Less tested? Less stable? ...?

We're planning to use Amazon MSK connect for Debezium Postgresql Connector to streaming the change data log from a RDS database. We want to utilize the JMX metrics Provided by the Debezium to monitor the connector status. But in MSK connect we don't see the options/configurations to enable these metrics. Does MSK connect allows exposing the connect specific metrics as we do it in Apache Kafka connect?

We have a Kafka cluster in Amazon MSK that has 3 brokers in different availability zones of the same region. I want to set up a Kafka Connect connector that backs up all data from our Kafka brokers to Amazon S3, and I'm trying to do it with MSK Connect. I set up Confluent's S3 Sink Connector on MSK Connect and it works - everything is uploaded to S3 as expected. The problem is that it costs a fortune in data transfer charges - our AWS bills for MSK nearly double whenever the connector is running, with `EU-DataTransfer-Regional-Bytes` accounting for the entire increase. It seems that the connector is pulling messages from all three of our brokers, i.e. from three different AZs, and so we're getting billed for inter-AZ data transfer. This makes sense because by default it will read a partition from that partition's leader, which could be any of the three brokers. But if we were creating a normal consumer, not a connector, it would be possible to restrict the consumer to read all partitions from a specific broker: ``` "client.rack" : "euw1-az3" ``` ☝️ For a consumer in the euw1-az3 AZ, this setting makes the consumer read all partitions from the local broker, regardless of the partitions' leader - which avoids the need for inter-AZ data transfer and brings the bills down massively. My question is, is it possible to do something similar for a Kafka Connector? What config setting do I have to pass to the connector, or the worker, to make it only read from one specific broker/AZ? Is this possible with MSK Connect?

I have created many connectors already using very similar configuration. Since yesterday (25th Jan 2022) I am unable to create a kafka connector using the AWSServiceRoleForKafkaConnect role. The existing connectors that have already been created are still working fine. Here is the error I get when clicking Create Connector on the last page in the form: `Error creating connector There was a problem creating a connector. If the problem persists, contact AWS Support. API response Invalid parameter serviceExecutionRoleArn: A service linked role ARN cannot be provided as service execution role ARN.` I have tried to create a connector with the same configuration that has already worked, only now I'm receiving the error above. Has something been updated around this? Do I need to create a new service role? Other Details: Using small MSK cluster with Authenticate=None using camel connector jar file (that is currently working with other connectors)

To use mTLS for authentication to AWS managed kafka (MSK) you need to use an AWS private certificate authority to generate the client certificates as per this document https://docs.aws.amazon.com/msk/latest/developerguide/msk-authentication.html Is it possible to generate a subordinate certificate authority from the Private CA that MSK trusts and generate the client certificates from that subordinate CA from another tool ?

Hi there, I would like to update an active MSK configuration from 1 to 2. Would that cause any downtime on the current partition as the current data is crucial?

I created and deleted topics on my Amazon MSK for weeks. But today when I use the same zookeeper connection string I get the error: java.nio.channels.UnresolvedAddressException The only difference is I upgraded the java version on my linux instance to Java 17. Can that cause this issue? Nothing else has changed

Hi All, I have an MSK Cluster set with 1 broker per 3 AZ's on m5.large and looking to switch our ECS Containers I have running Kafka Connect to using MSK Kafka Connect. I have a working ECS configuration that runs Kafka Connect with a Snowflake Connector. This user's private key, username, and password per the Snowflake Documentation. When I take the same working configuration and run it with MSK - I get the following error message: ```2022-01-13T07:52:08.000-06:00 [Worker-0576cf5f4676783e1] [SF_KAFKA_CONNECTOR] Exception: Invalid encrypted private key or passphrase 2022-01-13T07:52:08.000-06:00 [Worker-0576cf5f4676783e1] [SF_KAFKA_CONNECTOR] Error Code: 0018 2022-01-13T07:52:08.000-06:00 [Worker-0576cf5f4676783e1] [SF_KAFKA_CONNECTOR] Detail: failed to decrypt private key. Please verify input private key and passphrase. ``` Since my password contains a special character e.g. `!` and the private key contains `/` etc. I am assuming this might need to be escaped? Appreciate helps. Thanks Wayne

If my cluster is setup as follows: Brokers: 3 AZ: 3 RF: 3 MinISR: 1 Ack: all Q1: If a broker is being upgraded, Kafka will reassign the leadership of some partitions. After the upgrade will the leaderships get reassigned again so that all brokers are being used as before? Q2: If 1 AZ (AZ1) goes down, I understand that Kafka will automatically reassign the partitions to the other brokers in the two AZs without impacting the producers and consumers. When AZ1 comes back will MSK **automatically** create/restart the failed broker and redistribute the partitions?

Hi, We have an existing AWS managed Kafka cluster which is setup on 2 AZs. How can we add additional AZ to the this cluster without any downtime or recreating the cluster? I couldn't find any documenation about it. Much appreciated, jon

I have setup DMS to connect to MySQL database as the source and Kafka as target. I tested the source and target flow without TLS authentication works fine.Now I want to setup DMS with Kafka endpoint using TLS authentication. I tried out this [link](https://docs.aws.amazon.com/msk/latest/developerguide/msk-authentication.html) to check if I am able to connect using TLS authentication to Kafka cluster using the certificates generated, which worked. I used this with Glue as well and the connectivity works. Now for DMS, I followed the steps provided [here](https://docs.aws.amazon.com/dms/latest/userguide/CHAP_Target.Kafka.html) ( Section: Using SSL authentication ). But when I test the endpoint I get the error: `Test failed due to KMS key Exception : No permission to access Key 'arn:aws:kms:ap-southeast-2:xxxxxxxxxxxx:key/xxxxxxxxxxxxx'`. The KMS key here is the default key that DMS creates when we launch a replication instance. I tried using a CMK for DMS and created the endpoint using this KMS key. But it did not work. In this case, the endpoint errored out saying it could not connect to Kafka. Not really sure what the issue is. Has anyone come across this issue?

What's the easiest way to have a super user on MSK using SASL/SCRAM

What is the Authn and Authz mechanism when both IAM as well as SASL/SCRAM are enabled? | **Authn & Authz mech** | MSK authn | MSK authz | Kafka client authn | Kafka client authz | Kafka ACL behaviour | Property allow.everyone.if.no.acl.found | | --- | --- | --- | --- | --- | --- | | SASL/IAM clients | IAM | IAM | SASL/IAM | IAM | No effect | No effect | | SASL/SCRAM clients | IAM | IAM | SASL/SCRAM | **?** | **?** | **?** | Can someone please clarify about the question marks in the table above Amazon MSK allows using both IAM as well as SASL/SCRAM together in the same cluster. The behaviour of Authn and Authz for Kafka API is not clear when both IAM role-based and SASL/SCRAM are enabled. 1. For SASL/SCRAM clients, Does that mean we can we use SASL/SCRAM for authn and IAM for authorisations at topic level? (The documentation says Kafka ACLS have no effect when IAM authentication is enabled for an MSK cluster) 2. Or does that mean (that for SASL/SCRAM clients) we use SASL/SCRAM for authn but there is no authz at topic level (because Kafka ACLs do not work)? 3. Is it a bug that aws-msk allows both mechanisms and should really allow only one of them in a cluster? These are not clear from the documentation unfortunately.

Hello, we are encountering the same issues as e.g. https://github.com/aws/aws-msk-iam-auth/issues/28 regarding the `SaslAuthenticationException` while using MSK Connect with a kafka.t3.small instance. Setting `reconnect.backoff.ms` to e.g. 10000 ms will not resolve the issue, since the exception that is being thrown (`SaslAuthenticationException`) is not retryable (see https://github.com/apache/kafka/blob/trunk/clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java#L808) and ultimately leads to a new creation of a client, not a reconnect. When would the reconnect take place? As I went through the implementation, what I see is: 1. that `startConnect()` in `ConnectDistributed` is calling the constructor of `Worker` 2. the constructor of `Worker` calls `ConnectUtils.lookupKafkaClusterId(config)` 3. that method calls `Admin.create(config.originals())` - which opens up a new connection 4. if you follow the calls from there, you will see that you end up not retrying upon obtaining `SaslAuthenticationException` (https://github.com/apache/kafka/blob/trunk/clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java#L808) Even if the retry would work, several AdminClients are created, which all connect to the MSK cluster. Since this is not a reconnect, `reconnect.backoff.ms` settings cannot work for remediation. There is no mechanism in the Kafka code that would globally allow restricting these connections to happen only every x seconds. Unless I oversee something, MSK Connect should only work by chance with t3.small instances. This forces us to either: * not use IAM and go for SASL/SCRAM * use a kafka.m5.large instance and go from about 32 USD/Month to 151 USD/Month per instance - meaning 90 USD vs 450 USD in our case The limitation on the t3.small instance really limits what we want to achieve. The workaround presented [here](https://aws.amazon.com/premiumsupport/knowledge-center/msk-connector-connect-errors/) is not working and thus forces us to buy the larger instance. We have no need for a large instance and we don't want to have additional costs for simply using IAM for MSK Connect. **Can AWS remove the limit on the t3.small instance or present a different workaround? That would be great :) ** I cannot open a support case for this, since we don't have the required subscription and I believe that this could be of general interest. See parts of our logs using AWS MSK Connect: ```` [Worker-05ea3408948fa0a4c] [2022-01-01 22:41:53,059] INFO Creating Kafka admin client (org.apache.kafka.connect.util.ConnectUtils:49) [Worker-05ea3408948fa0a4c] [2022-01-01 22:41:53,061] INFO AdminClientConfig values: ... [Worker-05ea3408948fa0a4c] reconnect.backoff.max.ms = 10000 [Worker-05ea3408948fa0a4c] reconnect.backoff.ms = 10000 [Worker-05ea3408948fa0a4c] request.timeout.ms = 30000 [Worker-05ea3408948fa0a4c] retries = 2147483647 [Worker-05ea3408948fa0a4c] retry.backoff.ms = 10000 ... [Worker-05ea3408948fa0a4c] [2022-01-01 22:41:54,269] ERROR Stopping due to error (org.apache.kafka.connect.cli.ConnectDistributed:86) [Worker-05ea3408948fa0a4c] org.apache.kafka.connect.errors.ConnectException: Failed to connect to and describe Kafka cluster. Check worker's broker connection and security properties. [Worker-05ea3408948fa0a4c] at org.apache.kafka.connect.util.ConnectUtils.lookupKafkaClusterId(ConnectUtils.java:70) [Worker-05ea3408948fa0a4c] at org.apache.kafka.connect.util.ConnectUtils.lookupKafkaClusterId(ConnectUtils.java:51) [Worker-05ea3408948fa0a4c] at org.apache.kafka.connect.runtime.Worker.<init>(Worker.java:140) [Worker-05ea3408948fa0a4c] at org.apache.kafka.connect.runtime.Worker.<init>(Worker.java:127) [Worker-05ea3408948fa0a4c] at org.apache.kafka.connect.cli.ConnectDistributed.startConnect(ConnectDistributed.java:118) [Worker-05ea3408948fa0a4c] at org.apache.kafka.connect.cli.ConnectDistributed.main(ConnectDistributed.java:80) [Worker-05ea3408948fa0a4c] Caused by: java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.SaslAuthenticationException: [e4afe53f-73b5-4b94-9ac3-30d737071e56]: Too many connects [Worker-05ea3408948fa0a4c] at org.apache.kafka.common.internals.KafkaFutureImpl.wrapAndThrow(KafkaFutureImpl.java:45) [Worker-05ea3408948fa0a4c] at org.apache.kafka.common.internals.KafkaFutureImpl.access$000(KafkaFutureImpl.java:32) [Worker-05ea3408948fa0a4c] at org.apache.kafka.common.internals.KafkaFutureImpl$SingleWaiter.await(KafkaFutureImpl.java:89) [Worker-05ea3408948fa0a4c] at org.apache.kafka.common.internals.KafkaFutureImpl.get(KafkaFutureImpl.java:260) [Worker-05ea3408948fa0a4c] at org.apache.kafka.connect.util.ConnectUtils.lookupKafkaClusterId(ConnectUtils.java:64) [Worker-05ea3408948fa0a4c] ... 5 more [Worker-05ea3408948fa0a4c] Caused by: org.apache.kafka.common.errors.SaslAuthenticationException: [e4afe53f-73b5-4b94-9ac3-30d737071e56]: Too many connects [Worker-05ea3408948fa0a4c] [2022-01-01 22:41:54,281] INFO Stopped http_0.0.0.08083@68631b1d{HTTP/1.1, (http/1.1)}{0.0.0.0:8083} (org.eclipse.jetty.server.AbstractConnector:381) [Worker-05ea3408948fa0a4c] [2022-01-01 22:41:54,283] INFO Stopped https_0.0.0.08443@611d0763{SSL, (ssl, http/1.1)}{0.0.0.0:8443} (org.eclipse.jetty.server.AbstractConnector:381) [Worker-05ea3408948fa0a4c] MSK Connect encountered errors and failed. ````

Hello world, Let's say I have 3 sets of users consuming 2 topics and I want: User 1 to have full control to both topics; User 2 can have full control to topic 1; User 3 can subscribe to topic 2. What is the best way of implementing such? Thank you

How to enroll for MSK Serverless

Hello! I think the question is clear. I searched both AWS CLI and Web UI but could not find any solution. I also deleted the plugin object from S3 but it still seems active in MSK Connect custom plugins tab. How can I delete custom plugins from MSK Connect?

When setting up a Glue Connection for our MSK cluster configured woth IAM authentication, the broker disconnects the connection from Glue jobs using the connection with a message that says SASL failed. We have given ghe role used on the Glue job full permission. Is it currently possible to connect using IAM authentication?

Hi there, I'm attempting to deploy an MSK Connector through the AWS CLI. This connector needs to refer to 2 plugins, 1 which is the Elasticsearch sink connector and another which is the Avro converter as the data in the topic is Avro-encoded (and the Elasticsearch zip archive doesn't include the Confluent Avro converter) I have registered the 2x zip packages in MSK Connect Plugins fine, but when I come to create the connector & refer to both plugins, I get the following error: _An error occurred (BadRequestException) when calling the CreateConnector operation: Invalid parameter plugins: The number of plugins per connector must be 1._ Which is misleading as the "plugins" section of the CLI create-connector command, refers to an **array** of plugins for a connector: https://docs.aws.amazon.com/cli/latest/reference/kafkaconnect/create-connector.html (see --plugins) Am I missing something, should we be able to pass multiple plugins in when creating an MSK connector? Thanks

I am trying to use the Kafka Connect source connector for copying data from IBM MQ into Amazon MSK. I was able to create the custom plugin in MSK, but when creating the connector it is transitioning to failed state. The custom plugin i build from. https://github.com/ibm-messaging/kafka-connect-mq-source For the connector configuration i am using connector.class=com.ibm.eventstreams.connect.mqsource.MQSourceConnector mq.connection.name.list=<host>(1414) mq.channel=<channel name> tasks.max=1 mq.queue.manager=<name> mq.queue=<queue name> mq.channel.name=<channel name> kafka.topic=<topic name> value.converter=org.apache.kafka.connect.storage.StringConverter key.converter=org.apache.kafka.connect.storage.StringConverter mq.connection.mode=client The configuration works and i am able to copy messages from IBM MQ to Apache Kafka installed on EC2. It is not working from IBM MQ to Amazon MSK. Are there any limitations on what source connectors you use with Amazon MSK?

AWS MSK allows the number of brokers only as multiple of the number of availability zones selected. It is recommended for a well-balanced infra, but why is has this been enforced? I can't have 4/5 brokers in my 3 AZs & will therefore have to bear x2 cost instead.

The endpoint for MSK is kafka. It is https. I checked endpoint for the MSK connect. I couldn't. Is it kafka or is there something like kafkaconnect endpoint. Other than this doc: https://docs.aws.amazon.com/general/latest/gr/aws-service-information.html, is there a way to get endpoint for a service. I have looked into Cloudtrail but I am not certain about it because I don't understand the behavior of this service. It does not log every API call.

Hi, my cluster has 3 broker and Prometheus monitoring is enabled. Everyhing was fine and this afternoon 2 of 3 brokers stopped publishing metrics. **curl -s b-1.XXX.kafka.eu-central-1.amazonaws.com:11001/metrics | grep kafka_consumer_group_ConsumerLagMetrics_Value** Any ideas why this metric for example is missing on 2 brokers?

I have been experimenting with a cluster that has IAM Authentication, and I cannot seem to get it working. -I have a security group in the cluster that allows in-bound traffic from the ec2 instance I am testing from. I can even do zookeeper interactions like list topics just fine. -My ec2 instance has an IAM role with a policy that specifically allows for all kafka interactions on all resources -I also tried an aws local profile that has the same attached policy. -I am using the following command to attempt a consumer interaction bin/kafka-console-consumer.sh --bootstrap-server b-1.examplename.kafka.us-east-1.amazonaws.com:9098 --topic exampleTopic --consumer.config config/consumer.properties consumer.properties has the below properties security.protocol=SASL_SSL sasl.mechanism=AWS_MSK_IAM sasl.jaas.config=software.amazon.msk.auth.iam.IAMLoginModule required; sasl.client.callback.handler.class=software.amazon.msk.auth.iam.IAMClientCallbackHandler Am I missing anything?

I have created an MSK cluster with IAM Access Control as a mechanism for authentication and authorization. I can see the broker nodes for my cluster on MSK Console but when I try to write a message to a topic from my kafka client, it gives me error "Broker may not be available". When I used the cli command i.e. get-bootstrap-brokers, it shows no value for the "BootstrapBrokerStringSaslIam". It gives me nothing at all. Not even a message telling the status of brokers. I have followed the official documentation for it https://docs.aws.amazon.com/msk/latest/developerguide/iam-access-control.html However, I can get the ZooKeeper string for my IAM Access Control cluster and can create kafka topics from my kafka client. I don't know what's happened to my brokers. Can anyone tell me what should I do?

MSK has released a new feature. Basically, kafka client can authenticate and authorize using IAM. Link is below: https://aws.amazon.com/about-aws/whats-new/2021/05/introducing-iam-access-control-amazon-msk/ I tried out this by following the official documentation, https://docs.aws.amazon.com/msk/latest/developerguide/iam-access-control.html#configure-clients-for-iam-access-control But when I try to create policy in step 3 i.e. Create Authorization policies, I get the following error _Invalid Service In Action: The service kafka-cluster:Connect specified in the action does not exist._ When I checked the IAM actions, resources and conditions for MSK from https://docs.aws.amazon.com/service-authorization/latest/reference/list_amazonmanagedstreamingforapachekafka.html, I could not see any actions for kafka-cluster. How can I use this new feature to authenticate and authorize kafka client using IAM Access Control? What am I doing wrong here? Any help would be highly appreciated

A customer has the following questions related to MSK service. 1. What is the good way to manage topics and their settings? (command line tool / UI) 2. Can we adjust settings for existing topic? Can we change replication factor or number of partitions? 3. How to determine the numbers of partitions for the topic? Does it depend on how many brokers do we have? 4. What will happen if we increase the number of brokers? Is it even possible without recreating cluster? 5. We fill-up the Timestamp field in kafka message with timestamps from our devices. Is it ok that these timestamps will not be ordered in time ( we can get timestamps from the past)? 6. Is using protobuf for encode/decode messages a good choice? 7. What do you think about segmentio/kafka-go golang client? Can you advise us on choosing the "best" client? 8. Is it possible to manage commits ourself while using consumer groups? 9. Is there a way to add more disks to a broker instance?

My msk service are running well until recently kafka brokers cannot be connected but zookeepers are still connected successfully. I also double checked VPC, It's same. Eventually restart all brokers I tried these commands below ``` telnet broker_dns_1 9092 #=> Trying ip... telnet: Unable to connect to remote host: Connection refused telnet zookeeper_dns_1 2181 #=> Trying ip... connected to zookeeper_dns ``` Because response of telnet broker_dns command are so fast, I guess they can be blocked by some rules. have you guys encountered this issue or known the root cause ? Edited by: taisytran on Jan 17, 2021 8:50 PM Edited by: taisytran on Jan 17, 2021 8:50 PM

I have a MSK cluster I am unable to delete, it just gets to a failed state over and over. The only error I get is - There is an issue with your cluster Code: InternalServerError.Unknown Message: The last operation failed because of a service issue. Retry the operation. And retrying doesn't seem to change the outcome, does anyone have any ideas?

I have set up an MSK cluster using the defaults, added NAT Gateways to the two subnets MSK is using, and set up all of the permissions on my lambda role described in this article about using MSK as an event source: https://aws.amazon.com/blogs/compute/using-amazon-msk-as-an-event-source-for-aws-lambda/ I know my cluster works with the console tools, so I don't understand what I'm doing wrong. The lambda trigger interface in the UI eventually just says: Last processing result: PROBLEM: Connection error. Please check your event source connection configuration. I don't know what to try next to even troubleshoot this, but it's taking a lot of time and is frustrating.

I am using the MSK and it seems to be working fine as data is getting published and consumed correctly. Now the requirement is to have another MSK cluster and destroy current one, so I want to utilize the current data stored in ebs and attach it to new MSK cluster. How can we do it, can we take snapshot of current MSK volume.?

A customer is looking to migrate their kafka workload to MSK. they have a question on how to mange the broker strings when a scaling event occurs in the cluster. for example: the producer has a config which has a property: `bootstrap.servers`={comma separated string of 3 broker urls} now a scaling event occurs and now you have another broker. Now the ugly way is to update the brokerstring url section manually. the other option is to use a round robin DNS or an ALB and all the clients(producers/consumers) talk to the endpoint and behind the scenes it maintains the broker strings. though i doubt if second option is doable since MSK is managed. or may be i am understanding it incorrectly? Looking for some guidance on how this scenario is solved for other customers.

We're running AWS MSK in production and for us, it's mission critical. Currently we're using Apache Kafka 2.2.1, but we'd like to upgrade to Kafka 2.4.1. Is there any risk for data loss or downtime when we press the "Upgrade the Apache Kafka version" button and upgrade to Kafka 2.4.1?

I am trying out AWS MSK and created a cluster following the tutorial. What I want to know is, how can I connect my on-premises web sever to the created cluster as a producer. The tutorial only mention how to use AWS MSK with ec2 instance. Is there any other tutorial or blog post that I can refer to?

Does **CPU usage by broker** in the Kafka Monitoring section of a cluster correspond the **CPU System** or **CPU User** metrics available from the Kafka cluster ? I have a dashboard with these two metrics however they don't correlate to what I see in the **CPU usage by broker**. (0.2% in Kafka CPU Usage by Broker VS 8% in the metrics) Thanks Jason

I reported this before: We try to migrate to MSK, but at the point where I have some low load on the cluster representing the _idle_ state of our application with just a few messages per _minute_ total coming from about 30 clients the MSK brokers just start to fail with timeouts. ARN of the cluster: arn:aws:kafka:eu-west-1:499577160181:cluster/eks-20190111-production-201902-20190708/6e3e6674-9c9b-40bb-98b4-506edf76a6b2-4 Output from `kafka-broker-api-versions`: ``` b-1.....kafka.eu-west-1.amazonaws.com:9092 (id: 1 rack: subnet-0e2b1a418025f9b10) -> ERROR: org.apache.kafka.common.errors.DisconnectException b-3.....kafka.eu-west-1.amazonaws.com:9092 (id: 3 rack: subnet-000880b697986a0d8) -> ERROR: org.apache.kafka.common.errors.DisconnectException b-2.....kafka.eu-west-1.amazonaws.com:9092 (id: 2 rack: subnet-0f50b40038ca2bf62) -> ( Produce(0): 0 to 7 [usable: 7], Fetch(1): 0 to 10 [usable: 10], ListOffsets(2): 0 to 4 [usable: 4], Metadata(3): 0 to 7 [usable: 7], LeaderAndIsr(4): 0 to 1 [usable: 1], StopReplica(5): 0 [usable: 0], UpdateMetadata(6): 0 to 4 [usable: 4], ControlledShutdown(7): 0 to 1 [usable: 1], OffsetCommit(8): 0 to 6 [usable: 6], OffsetFetch(9): 0 to 5 [usable: 5], FindCoordinator(10): 0 to 2 [usable: 2], JoinGroup(11): 0 to 3 [usable: 3], Heartbeat(12): 0 to 2 [usable: 2], LeaveGroup(13): 0 to 2 [usable: 2], SyncGroup(14): 0 to 2 [usable: 2], DescribeGroups(15): 0 to 2 [usable: 2], ListGroups(16): 0 to 2 [usable: 2], SaslHandshake(17): 0 to 1 [usable: 1], ApiVersions(18): 0 to 2 [usable: 2], CreateTopics(19): 0 to 3 [usable: 3], DeleteTopics(20): 0 to 3 [usable: 3], DeleteRecords(21): 0 to 1 [usable: 1], InitProducerId(22): 0 to 1 [usable: 1], OffsetForLeaderEpoch(23): 0 to 2 [usable: 2], AddPartitionsToTxn(24): 0 to 1 [usable: 1], AddOffsetsToTxn(25): 0 to 1 [usable: 1], EndTxn(26): 0 to 1 [usable: 1], WriteTxnMarkers(27): 0 [usable: 0], TxnOffsetCommit(28): 0 to 2 [usable: 2], DescribeAcls(29): 0 to 1 [usable: 1], CreateAcls(30): 0 to 1 [usable: 1], DeleteAcls(31): 0 to 1 [usable: 1], DescribeConfigs(32): 0 to 2 [usable: 2], AlterConfigs(33): 0 to 1 [usable: 1], AlterReplicaLogDirs(34): 0 to 1 [usable: 1], DescribeLogDirs(35): 0 to 1 [usable: 1], SaslAuthenticate(36): 0 [usable: 0], CreatePartitions(37): 0 to 1 [usable: 1], CreateDelegationToken(38): 0 to 1 [usable: 1], RenewDelegationToken(39): 0 to 1 [usable: 1], ExpireDelegationToken(40): 0 to 1 [usable: 1], DescribeDelegationToken(41): 0 to 1 [usable: 1], DeleteGroups(42): 0 to 1 [usable: 1] ) ``` I do not see anything particular in the CloudWatch metrics, except that network traffic went up when I started my services, and went down when presumably the timeouts started.

I'm trying to create MSK clusters, and since yesterday somewhen early afternoon in CEST a newly created cluster no longer has a "BootstrapBrokerString", but rather `aws kafka get-bootstrap-brokers` returns a response with only "BootstrapBrokerStringTls". This is clearly an unexpected API change, in a GA product. I would have expected that any move towards TLS-support (yeah! great! awesome!) would be announced, and would not affect existing documented things. Switching to TLS will be quite some work, so right now I rather would _not_ want to do that. How can I get back to the previous behavior? EDIT: I'm also looking at the AWS console now. The cluste says "TLS client authentication" is "Disabled", and enryption in transit between clients and brokers is "Only TLS encrypted traffic allowed". So I guess that makes sense that the client information only returns "TLS" entries. I looked at bit further, and it seems that the API behavior indeed changed, and the TLS options appeared. The default is said to be "TLS/Plaintext" at https://aws.amazon.com/msk/faqs/ (which probably would still produce a `BootstrapBrokerStringTls`), the actual default I saw looks like "TLS" though. I'm now trying to adapt at least my creation scripts to explicitly configure ClientBroker encryption as 'PLAINTEXT', and then will have to work out how to move towards a "both" situation. Edited by: ankon on Jun 21, 2019 11:48 AM: Added information for console output, and my next steps. Edited by: ankon on Jun 21, 2019 12:11 PM: Updated with more information from documentation where I could find references to the change.

Is it possible to scale a cluster down ? We are currently set up with 6 brokers (2 per AZ), and I would like to scale that down to 3 brokers. Is this possible without having to stand up a new cluster and switch DNS names to point to the new Zookeepers and Brokers ? Thanks, Jason

Hello! I've managed to create a cluster configuration through the CLI (https://docs.aws.amazon.com/cli/latest/reference/kafka/create-configuration.html), and created a cluster that uses that configuration. How would I go about updating a configuration ("create a new revision?"), and what impact would that have on a cluster? I think I won't have to update the configuration often, and am mostly looking for the ability to react to operational changes (tune settings for performance, allow/disallow client behaviors such as deleting topics). I'm not "bound" to using the CLI, so if there is a REST API endpoint I'd be fine with that.

Howdy - can someone clarify the MSK data transfer charges within a region or AZ? The MSK pricing page says "You will pay standard AWS data transfer charges for data transferred in and out of Amazon MSK clusters" with a link to the EC2 on-demand pricing page. The "Data Transfer within the same AWS Region" section of that page gives the pricing for different combinations of EC2 and various AWS services. Some combinations have free data transfer within a region. Some combos have free data transfer within an AZ. But MSK is not listed there so I can't tell what the pricing is. Can someone clarify how the pricing between MSK and EC2 works within a region? I'm particularly interesting in learning about steps I can take to achieve free data transfer between EC2 and the MSK cluster :-) Thanks! JP

Do the hostnames for the brokers follow any convention ? b-3.CLUSTER NAME.??????.?????.kafka.us-east-2.amazonaws.com Thanks, Jason Edited by: jcoelho on May 9, 2019 2:17 PM Edited by: jcoelho on May 9, 2019 2:17 PM

How do I change the monitoring level after a MSK cluster has been created ? Is this something that is possible on the CLI ? I did not see any documentation around this. Thanks, Jason

MSK is a great service, which considerably reduces operational overhead. One remaining piece is backing up Kafka topics. In particular, when connecting microservices via Kafka this is quite important, i.e. there is high screw up potential if it fails. Hence, do you have plans to provide backups for MSK, e.g. stream Kafka topics to S3 via Firehose?

In MSK, is it possible to customize broker properties like `authorizer.class.name` to enable the built-in kafka.security.auth.SimpleAclAuthorizer? And, add a listener with the SASL_PLAINTEXT protocol? If so, where do I set these? Edited by: kinghuang970 on May 6, 2019 2:34 PM

I created a Kafka cluster with 2 Brokers per AZ, and when I click on the **View client information** button, it only lists 3 zookeeper and broker hosts. Isn't it supposed to have 6 in total ? Here are the host names of the brokers ``` b-6.********.kafka.us-east-2.amazonaws.com:9092 b-1.**********.kafka.us-east-2.amazonaws.com:9092, b-4.**********.kafka.us-east-2.amazonaws.com:9092 ``` I am assuming it is missing b-2,b-3 and b-5 Thanks, Jason Edited by: jcoelho on May 2, 2019 6:55 AM

Hey everyone. I'm getting a timeout when trying to send messages: ERROR Error when sending message to topic test with key: null, value: 4 bytes with error: (org.apache.kafka.clients.producer.internals.ErrorLoggingCallback) org.apache.kafka.common.errors.TimeoutException: Failed to update metadata after 60000 ms. I have properly setup the cluster (which has ACTIVE state), and created a couple of topics (which I can list properly as well) using a client machine created using the guide with its security group added to the security group of the cluster nodes. Somehow though, I can't send messages. Can anyone please help me troubleshoot my issue? Edited by: joaquinscript on Feb 27, 2019 2:35 PM