AWS MSK with IAM - Timeout Exception

Question

We are using Aws MSK, recently they announce the AWS MSK IAM AUTH(https://github.com/aws/aws-msk-iam-auth)

When i tried to use it with the bellow configuration on my clients i got the exception:

org.apache.kafka.common.errors.TimeoutException: Call(callName=fetchMetadata, deadlineMs=1620677932733) timed out at 1620677932734 after 1 attempt(s)
Caused by: org.apache.kafka.common.errors.TimeoutException: Timed out waiting for a node assignment.

My application.yml:

sasl:
  mechanism: AWS_MSK_IAM
  jaas:
    config: software.amazon.msk.auth.iam.IAMLoginModule required
  client:
    callback:
      handler:
        class: software.amazon.msk.auth.iam.IAMClientCallbackHandler

The telnet works but it seems that something is missing, any suggestions?

Answer 1

I had this issue. AWS MSK IAM relies on communication over port 9098. In my case, I had security groups allow inbound communication over the typical ports 9092 (plaintext) and 9094 (TLS) but not 9098. Make sure your cluster is accepting traffic over that port

Answer 2

I was facing similar issue and below is the config i had used in java code. Turns out I was using old kafka-client lib and after upgrading version till 2.6.11, I was able to resolve the issue.

<dependency>
      <groupId>org.apache.kafka</groupId>
      <artifactId>kafka-clients</artifactId>
      <version>2.6.3</version>
 </dependency>


 props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, brokers);
 props.put("security.protocol", "SASL_SSL");
 props.put("sasl.mechanism", "AWS_MSK_IAM");
 props.put("sasl.jaas.config", "software.amazon.msk.auth.iam.IAMLoginModule required;");
 props.put("sasl.client.callback.handler.class", "software.amazon.msk.auth.iam.IAMClientCallbackHandler");
 props.put(ConsumerConfig.GROUP_ID_CONFIG, "Random123");
 props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
 props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringDeserializer");
 props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");