AWS MSK with IAM - Timeout Exception

user_825045 0
Viewed 2017

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?

aws-msk spring-boot kotlin amazon-web-services java
4 Answers

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

user_47313120

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");
user_40236010

this issue is seems to relate to networking.

make sure you expose the correct ports in your MSK security group ingress

in case you use cloudformation to generate your security group, you can use the ingress that suits you (more than one option can be appplied)

example:

  MskSecurityGroup:
    Type: "AWS::EC2::SecurityGroup"
    Properties:
      GroupDescription: MSK security group
      GroupName: !Sub "${AWS::StackName}-securitygroup"
      SecurityGroupEgress:
        - Description: No restrictions
          CidrIp: "0.0.0.0/0"
          FromPort: 0
          IpProtocol: tcp
          ToPort: 65535
      SecurityGroupIngress:

        # bootstrap servers IAM auth
        - CidrIp: !Ref CidrIp
          FromPort: 9098
          IpProtocol: tcp
          ToPort: 9098

        # bootstrap servers TLS auth
        - CidrIp: !Ref CidrIp
          FromPort: 9094
          IpProtocol: tcp
          ToPort: 9094

        # bootstrap servers PLAINTEXT
        - CidrIp: !Ref CidrIp
          FromPort: 9092
          IpProtocol: tcp
          ToPort: 9092

        # zookeeper TLS
        - CidrIp: !Ref CidrIp
          FromPort: 2182
          IpProtocol: tcp
          ToPort: 2182

        # zookeeper PLAINTEXT
        - CidrIp: !Ref CidrIp
          FromPort: 2181
          IpProtocol: tcp
          ToPort: 2181
      VpcId: !Ref myVpcId
user_26936080 
@Bean
public Map<String, Object> consumerConfigs() {
    Map<String, Object> props = new HashMap<>();
    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, url);
    props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_SSL");
    props.put(SaslConfigs.SASL_MECHANISM, "AWS_MSK_IAM");
    props.put(SaslConfigs.SASL_JAAS_CONFIG, "software.amazon.msk.auth.iam.IAMLoginModule required;");
    props.put(SaslConfigs.SASL_CLIENT_CALLBACK_HANDLER_CLASS, "software.amazon.msk.auth.iam.IAMClientCallbackHandler");
    return props;
}

@Bean
public ConsumerFactory<String, String> consumerFactory() {
    return new DefaultKafkaConsumerFactory<>(consumerConfigs());
}

@Bean
public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> kafkaListenerContainerFactory() {
    ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(consumerFactory());
    return factory;
}

  1. Make sure that MSK cluster's security group opened access for current EC2.

  2. Make sure current EC2 has IAM role and that role should be configured with policies that allow or deny actions. If its serverless cluster you can see "Configure IAM policy" button, there you will see a json which you should copy then edit as described there and insert into your EC2 IAM role's policy and you will have access.

  3. Also you must have current dependency for maven

    <dependency>
   <groupId>software.amazon.msk</groupId>
   <artifactId>aws-msk-iam-auth</artifactId>
   <version>1.1.4</version>
</dependency>

which will do the client work part for you. Be informed that sometimes it gives conflict with "aws-java-sdk" dependency.

user_73220390
Related