Strimzi - 0.24.0 - KafkaConnect Cluster deployment on AWS EKS for AWS MSK with AWS IAM Role

[ashemachandan]

We have a use case to connect AWS MSK with MongoDB using KafkaConnect sink connector on AWS EKS (Strimzi Operator) with AWS IAM role authentication and we are using below environment with KafkaConnect deployment yaml declarative for Service Account creation to link with AWS IAM Role.

Environment (please complete the following information):

Strimzi version: 0.24.0
Installation method: YAML files
Infrastructure: Amazon EKS
Kubernetes versionInfo: 1.19
Platform versionInfo: eks.5
AWS MSK: Apache Kafka version 2.7.0

KafkaConnect.yaml

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnect
metadata:
name: kafka-connect-cluster
annotations:
strimzi.io/use-connector-resources: "true"
spec:
image: 238514202959.dkr.ecr.eu-west-1.amazonaws.com/demo-python-microservice:strimzi-kafkaconnect-mongodb
version: 2.7.0
replicas: 1
bootstrapServers: b-2..kafka.eu-west-1.amazonaws.com:9098,b-3.b-2..kafka.eu-west-1.amazonaws.com:9098,b-1.b-2..kafka.eu-west-1.amazonaws.com:9098
template:
serviceAccount:
metadata:
labels:
name: kafka-connect-cluster
annotations:
eks.amazonaws.com/role-arn: "arn:aws:iam::XXXXXXXXXX:role/AWS-MSK-strimzi-kafka-connect"
config:
group.id: kafka-connect-cluster
offset.storage.topic: kafka-connect-cluster-offsets
config.storage.topic: kafka-connect-cluster-configs
status.storage.topic: kafka-connect-cluster-status
config.storage.replication.factor: 3
offset.storage.replication.factor: 3
status.storage.replication.factor: 3

After deployment, we found that ServiceAccount got created and KafkaConnect pods tried make initial connect with AWS MSK bootstrap on port 9098 but Pods are crashing due to readiness failure and cloud watch as well as pods logs captured the below

ServiceAccount-
Name: kafka-connect-cluster-connect
Namespace: kafka
Labels: app.kubernetes.io/instance=kafka-connect-cluster
app.kubernetes.io/managed-by=strimzi-cluster-operator
app.kubernetes.io/name=kafka-connect
app.kubernetes.io/part-of=strimzi-kafka-connect-cluster
name=kafka-connect-cluster
strimzi.io/cluster=kafka-connect-cluster
strimzi.io/kind=KafkaConnect
strimzi.io/name=kafka-connect-cluster-connect
Annotations: eks.amazonaws.com/role-arn: arn:aws:iam::XXXXXXX:role/AWS-MSK-strimzi-kafka-connect
Image pull secrets:
Mountable secrets: kafka-connect-cluster-connect-token-s999j
Tokens: kafka-connect-cluster-connect-token-s999j
Events:

Cloud Watch -
[SocketServer brokerId=3] Failed authentication with ip-10-254-192-74.eu-west-1.compute.internal/INTERNAL_IP (SSL handshake failed) (org.apache.kafka.common.network.Selector)

Pods Logs -
Warning Unhealthy 16m (x165 over 18h) kubelet Readiness probe failed: Get "http://10.254.192.74:8083/": dial tcp 10.254.192.74:8083: connect: connection refused

Since i couldn't able to make this SerivceAccount which has required AWS IAM roles for AWS MSK in KafkaConnect yaml to work with this bootstrap ports on 9098 so required more details or steps or configuration to make this working.

[scholzj]

If you could format the YAML examples and logs as code, it would be much more readable.

I'm not sure what do you expect to achieve with the service account and with AWS IAM TBH. The way you seem to have it configured is that it should connect to the bootstrap servers on port 9098 (I assume the .. part in the address is because you deleted some part of the address I guess and the real address is valid?). But you did not configures any TLS encryption or any authentication. Is that really how your Kafka cluster is configured? Assuming the SSL handshake failed error is from the MSK cluster, it suggests you need to configure at least TLS. You should check out the docs and add the options required to match your brokers configuration: https://strimzi.io/docs/operators/latest/full/using.html#proc-kafka-connect-config-str

[ashemachandan]

Thanks, scholzj for your response, you're right our AWS MSK Cluster is enabled with TLS on port 9098 which we are trying to connect using strimzi Kafka so we are looking for steps to create ca-certificates for Kafka connect cluster secrets for SSL handshake with MSK cluster, given that kafka cluster is external and strmizi doesn't provide this default.

[scholzj]

IIRC, MSK is using signed certificates by a public CA. So if it is just TLS encryption, you can just leave the list empty: https://strimzi.io/docs/operators/latest/full/using.html#con-common-configuration-trusted-certificates-reference

[ashemachandan]

Port 9092 and 9094 are working after adding trustedCertificates: [] but port 9098 with IAM enabled is failing to fetch metadata from MSK cluster like below -

KafkaConnect Logs

2021-07-16 09:47:20,888 INFO [AdminClient clientId=adminclient-1] Metadata update failed (org.apache.kafka.clients.admin.internals.AdminMetadataManager) [kafka-admin-client-thread | adminclient-1]
org.apache.kafka.common.errors.TimeoutException: Call(callName=fetchMetadata, deadlineMs=1626428840887, tries=70, nextAllowedTryMs=1626428840988) timed out at 1626428840888 after 70 attempt(s)
Caused by: org.apache.kafka.common.errors.TimeoutException: Timed out waiting for a node assignment. Call: fetchMetadata
2021-07-16 09:47:50,887 INFO App info kafka.admin.client for adminclient-1 unregistered (org.apache.kafka.common.utils.AppInfoParser) [kafka-admin-client-thread | adminclient-1]
2021-07-16 09:47:50,887 INFO [AdminClient clientId=adminclient-1] Metadata update failed (org.apache.kafka.clients.admin.internals.AdminMetadataManager) [kafka-admin-client-thread | adminclient-1]
org.apache.kafka.common.errors.TimeoutException: Call(callName=fetchMetadata, deadlineMs=1626428870888, tries=72, nextAllowedTryMs=-9223372036854775709) timed out at 9223372036854775807 after 72 attempt(s)
Caused by: org.apache.kafka.common.errors.TimeoutException: The AdminClient thread has exited. Call: fetchMetadata
2021-07-16 09:47:50,913 INFO Metrics scheduler closed (org.apache.kafka.common.metrics.Metrics) [kafka-admin-client-thread | adminclient-1]
2021-07-16 09:47:50,913 INFO Closing reporter org.apache.kafka.common.metrics.JmxReporter (org.apache.kafka.common.metrics.Metrics) [kafka-admin-client-thread | adminclient-1]
2021-07-16 09:47:50,913 INFO Metrics reporters closed (org.apache.kafka.common.metrics.Metrics) [kafka-admin-client-thread | adminclient-1]
2021-07-16 09:47:50,913 ERROR Stopping due to error (org.apache.kafka.connect.cli.ConnectDistributed) [main]
org.apache.kafka.connect.errors.ConnectException: Failed to connect to and describe Kafka cluster. Check worker's broker connection and security properties.
        at org.apache.kafka.connect.util.ConnectUtils.lookupKafkaClusterId(ConnectUtils.java:70)
        at org.apache.kafka.connect.util.ConnectUtils.lookupKafkaClusterId(ConnectUtils.java:51)
        at org.apache.kafka.connect.cli.ConnectDistributed.startConnect(ConnectDistributed.java:95)
        at org.apache.kafka.connect.cli.ConnectDistributed.main(ConnectDistributed.java:78)
Caused by: java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.TimeoutException: Call(callName=listNodes, deadlineMs=1626428870885, tries=1, nextAllowedTryMs=1626428870986) timed out at 1626428870886 after 1 attempt(s)
        at org.apache.kafka.common.internals.KafkaFutureImpl.wrapAndThrow(KafkaFutureImpl.java:45)
        at org.apache.kafka.common.internals.KafkaFutureImpl.access$000(KafkaFutureImpl.java:32)
        at org.apache.kafka.common.internals.KafkaFutureImpl$SingleWaiter.await(KafkaFutureImpl.java:89)
        at org.apache.kafka.common.internals.KafkaFutureImpl.get(KafkaFutureImpl.java:260)
        at org.apache.kafka.connect.util.ConnectUtils.lookupKafkaClusterId(ConnectUtils.java:64)
        ... 3 more
Caused by: org.apache.kafka.common.errors.TimeoutException: Call(callName=listNodes, deadlineMs=1626428870885, tries=1, nextAllowedTryMs=1626428870986) timed out at 1626428870886 after 1 attempt(s)
Caused by: org.apache.kafka.common.errors.TimeoutException: Timed out waiting for a node assignment. Call: listNodes

AWS MSK Cloud logs

[2021-07-16 11:08:43,234] INFO [SocketServer listenerType=ZK_BROKER, nodeId=1] Failed authentication with ip-10-254-192-254.eu-west-1.compute.internal/INTERNAL_IP (Unexpected Kafka request of type METADATA during SASL handshake.) (org.apache.kafka.common.network.Selector)

We have added the below SA template in the kafkaConnect deployment so looking for any additional things like - authentication, clusterRoleBinding and buildServiceAccount to be added if yes, then please provide more details and our goal is to connect AWS MSK with client_sasl_iam_enabled port 9098 via kafkaconnect, kafkaconnectors, kafkamirrormaker, kafkabridge running on aws eks with strimizi operator.

 template:
    serviceAccount:
      metadata:
        labels: 
          name: kafka-connect-cluster
        annotations:
          eks.amazonaws.com/role-arn: "arn:aws:iam::238514202959:role/AWS-MSK-strimzi-kafka-connect"

[shubhamg7]

Hey @ashemachandan were you able to get this to work? We are running into the same issue.

[tooptoop4]

@ashemachandan did u solve?

[yxw]

Not sure if the OP had solved his issue, but we ended up solved our same issue with podTemplate with something like this.

  template:
    pod:
      metadata:
        annotations:
          iam.amazonaws.com/role: <my-arn-aws-iam>

[Rezkmike]

Hi @yxw , did you apply this changes on KafkaConnect? Is it required to configure IRSA?

[yxw]

Yes, this configuration is on the KafkaConnect.

Not sure if this is required for your case, I guess this is only for IAM role based access to S3. In my case I can't use other ways except for the IAM role based access.

[Rezkmike]

are you using "Camel Kafka Connector" or others?