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

I have an on-prem data pipeline with MQTT + Kafka, each containerized locally. Now, I want to enable the upstream connection to the Cloud/Internet with AWS Kinesis, but I need a Kafka/Kinesis connector. ``` version: '3' services: nodered: container_name: nodered image: nodered/node-red ports: - "1880:1880" volumes: - ./nodered:/data depends_on: - mosquitto environment: - TZ=America/Toronto - NODE_RED_ENABLE_PROJECTS=true restart: always mosquitto: image: eclipse-mosquitto container_name: mqtt restart: always ports: - "1883:1883" volumes: - "./mosquitto/config:/mosquitto/config" - "./mosquitto/data:/mosquitto/data" - "./mosquitto/log:/mosquitto/log" environment: - TZ=America/Toronto user: "${PUID}:${PGID}" portainer: ports: - "9000:9000" container_name: portainer restart: always volumes: - "/var/run/docker.sock:/var/run/docker.sock" - "./portainer/portainer_data:/data" image: portainer/portainer-ce zookeeper: image: zookeeper:3.4 container_name: zookeeper ports: - "2181:2181" volumes: - "zookeeper_data:/data" kafka: image: wurstmeister/kafka:1.0.0 container_name: kafka ports: - "9092:9092" - "9093:9093" volumes: - "kafka_data:/data" environment: - KAFKA_ZOOKEEPER_CONNECT=10.0.0.129:2181 - KAFKA_ADVERTISED_HOST_NAME=10.0.0.129 - JMX_PORT=9093 - KAFKA_ADVERTISED_PORT=9092 - KAFKA_LOG_RETENTION_HOURS=1 - KAFKA_MESSAGE_MAX_BYTES=10000000 - KAFKA_REPLICA_FETCH_MAX_BYTES=10000000 - KAFKA_GROUP_MAX_SESSION_TIMEOUT_MS=60000 - KAFKA_NUM_PARTITIONS=2 - KAFKA_DELETE_RETENTION_MS=1000 depends_on: - zookeeper restart: on-failure cmak: image: hlebalbau/kafka-manager:1.3.3.16 container_name: kafka-manager restart: always depends_on: - kafka - zookeeper ports: - "9080:9080" environment: - ZK_HOSTS=10.0.0.129 - APPLICATION_SECRET=letmein command: -Dconfig.file=/kafka-manager/conf/application.conf -Dapplication.home=/kafkamanager -Dhttp.port=9080 volumes: zookeeper_data: driver: local kafka_data: driver: local ``` I found this one from your labs: https://github.com/awslabs/kinesis-kafka-connector Again, I run everything from a docker-compose and that works, but now I'm not sure if there's either an existing image or documentation that can help me figure out how to containerize this connector. Will I have to create my own custom image via a DockerFile? Any examples? Thank you.lg...

Hi, we have a use case where some files will be placed into s3 and an event will be placed to MSK to trigger a lambda event to process the file. Each file will correspond to a client identifier and we can have multiple files for one client identifier with a different processing timestamp in it. As lambda will be triggered for each file, is there a mechanism to check if any lambda started processing for the same client identifier file and make sure to process only latest files based on processing timestamp check? Process Locking: If we can create a locking mechanism with a key so that while processing any file, a check will happen whether any other file with same client identifier picked up and is in process so that until that process is completed, latest process should wait. Please let me know if you need more information on itlg...

Hi, We are exploring with KDA for our streaming data ingestion pipeline for ML purpose. From https://aws.amazon.com/blogs/machine-learning/using-streaming-ingestion-with-amazon-sagemaker-feature-store-to-make-ml-backed-decisions-in-near-real-time/ it seems possible to connect KDA with MSK and use Lambda as destination to ingest data to Sagemaker Feature Store (or any custom location). However checking with KDA documentation it seems connecting with MSK is only possible with KDA studio or as a Flink app which does not provide a Lambda destination directly - instead would require another Kinesis stream in between, whereas KDA SQL application does support connecting with Lambda but not support reading from MSK. Our use case is very similar to the blog post above and we prefer to use Python as much as possible. What can be a viable path here? Thanks, Roblg...

Hi, Are there any Open Source tools for for UI for Serverless Amazon MSK? 1) **Yahoo CMK** (https://github.com/yahoo/CMAK) No mentions, and 1 open issue with a question Seems no support 2) **Kafka Drop** https://github.com/obsidiandynamics/kafdrop Only has a PR for IAM, but probably will not support the Amazon MSK Serverless Seems no support 3) **Kafka UI (not yet) :** https://github.com/provectus/kafka-ui/issues/2185 No support, atleast until v0.5 is released Due to: Amazon MSK Serverless doesn't support fetching broker or broker-logger configurations.lg...

Hi I cant seem to find a full example of Connector configuration for msk (sink configuration) with document db as target. Especially,Im not sure how to set the connection.uri,as it gets the default value, hence the connector creation fails Please advise Thanks, Rudilg...

Hi , We have a msk cluster provisioned and try to create custom connector datagen source for s3, but its failing without errors, we can only see below logs in broker logs: [2022-08-26 12:52:00,713] ERROR [ReplicaFetcher replicaId=3, leaderId=1, fetcherId=1] Error for partition __amazon_msk_connect_offsets_S3-Sink-Orders_ae585cdd-321c-4e70-8de0-a115ed0e127b-4-13 at offset 0 (kafka.server.ReplicaFetcherThread) org.apache.kafka.common.errors.UnknownTopicOrPartitionException: This server does not host this topic-partition. [2022-08-26 12:52:00,713] ERROR [ReplicaFetcher replicaId=3, leaderId=1, fetcherId=1] Error for partition __amazon_msk_connect_offsets_S3-Sink-Orders_ae585cdd-321c-4e70-8de0-a115ed0e127b-4-19 at offset 0 (kafka.server.ReplicaFetcherThread) org.apache.kafka.common.errors.UnknownTopicOrPartitionException: This server does not host this topic-partition. [2022-08-26 12:52:00,713] ERROR [ReplicaFetcher replicaId=3, leaderId=1, fetcherId=1] Error for partition __amazon_msk_connect_offsets_S3-Sink-Orders_ae585cdd-321c-4e70-8de0-a115ed0e127b-4-7 at offset 0 (kafka.server.ReplicaFetcherThread) org.apache.kafka.common.errors.UnknownTopicOrPartitionException: This server does not host this topic-partition. [2022-08-26 12:52:00,713] ERROR [ReplicaFetcher replicaId=3, leaderId=1, fetcherId=1] Error for partition __amazon_msk_connect_offsets_S3-Sink-Orders_ae585cdd-321c-4e70-8de0-a115ed0e127b-4-1 at offset 0 (kafka.server.ReplicaFetcherThread) org.apache.kafka.common.errors.UnknownTopicOrPartitionException: This server does not host this topic-partition.lg...

Hi, I need to setup a msk connect service on top of my msk service. The purpose is to connect the msk service to my document db service .Eventually,there will be 2 connectors : source & sink. I understand that I need to create a role and assign permissions . However,I cant figure what should be the permissions which I need to assign to the role and I cant find a working example. Please advise Rudilg...

My question: If MSK doesn't allow you to specify Super Users, is there a fault tolerant way to create ACLs to "mimic" a super user like the one specified in: https://repost.aws/questions/QUPLKdkgDPQ_uH8LaLhB8bew/how-to-create-a-super-user-on-msk It seems fraught with accidental peril. If you start giving a SASL user access to cluster operations one at a time, you'll accidentally lock the Cluster's ACLs entirely. ``` kafka-acls.sh --bootstrap-server [broker hosts] --add --add-principal="User:my_super_user" --cluster --operation Describe ``` Oops! You'll violate the `allow.everyone.if.no.acl.found` rule. An ACL for the cluster _will_ be found. It'll just only have the one operation. And since the `super.users` configuration property isn't supported. You're hosed. No matter how many SASL/SCRAM users you make, you've lost any ability to set or change any more ACLs... This happened to me, because I wasn't sure if I specified `ALL` for the operation, I would suddenly run into trouble with blocking brokers (as that linked question answer indicates). Is the only solution, in this case, to enable IAM, create an IAM policy that allows you to "do anything", update the ACLs, deactivate IAM? R.I.P super users property if using SASL/SCRAMlg...

I have problem using MSK Postgres Debezium connector with Glue Schema Registry avro serialisation, and getting "connect timed out" to GSR. Logs are following: ``` [Worker-051272e114b69c525] [2022-08-17 08:47:55,387] ERROR [route-events-connector|task-0] WorkerSourceTask{id=route-events-connector-0} Task threw an uncaught and unrecoverable exception. Task is being killed and will not recover until manually restarted (org.apache.kafka.connect.runtime.WorkerTask:191) ... com.amazonaws.services.schemaregistry.kafkaconnect.AWSKafkaAvroConverter.fromConnectData(AWSKafkaAvroConverter.java:97) [Worker-051272e114b69c525] at org.apache.kafka.connect.storage.Converter.fromConnectData(Converter.java:63) [Worker-051272e114b69c525] at org.apache.kafka.connect.runtime.WorkerSourceTask.lambda$convertTransformedRecord$2(WorkerSourceTask.java:313) [Worker-051272e114b69c525] at com.amazonaws.services.schemaregistry.kafkaconnect.AWSKafkaAvroConverter.fromConnectData(AWSKafkaAvroConverter.java:95) [Worker-051272e114b69c525] ... 15 more [Worker-051272e114b69c525] Caused by: com.amazonaws.services.schemaregistry.exception.AWSSchemaRegistryException: Failed to get schemaVersionId by schema definition for schema name = key-schema com.amazonaws.services.schemaregistry.common.AWSSchemaRegistryClient.getSchemaVersionIdByDefinition(AWSSchemaRegistryClient.java:144) [Worker-051272e114b69c525] ... 28 more [Worker-051272e114b69c525] Caused by: java.net.SocketTimeoutException: connect timed out [Worker-051272e114b69c525] at java.base/java.net.PlainSocketImpl.socketConnect(Native Method) ``` Related to GSR connector config: ``` ... key.converter.region=eu-central-1 key.converter.registry.name=my-schema-registry key.converter.schemaAutoRegistrationEnabled=true key.converter.schemaName=key-schema key.converter.avroRecordType=GENERIC_RECORD key.converter=com.amazonaws.services.schemaregistry.kafkaconnect.AWSKafkaAvroConverter value.converter.region=eu-central-1 value.converter.registry.name=my-schema-registry value.converter.schemaAutoRegistrationEnabled=true value.converter.schemaName=value-schema value.converter.avroRecordType=GENERIC_RECORD value.converter=com.amazonaws.services.schemaregistry.kafkaconnect.AWSKafkaAvroConverter ``` We have already configured MSK connect json Postgres connectors which are working fine and publishing data to MSK topics. Has anyone successfully configured MSK Connect with Glue Schema Registry for avro serialization? Thanks.lg...

I have an use case where we are using Camel Kafka client and need access to Zookeeper. We also were planning to use MSK severless so that we don't have to worry about Broker nodes going down or rebalancing the partitions. I could not find a way to access zookeeper information with MSK server less. Any help would be greatly appreciated.lg...

Is librdkafka 1.6.0(https://github.com/edenhill/librdkafka) compatible with AWS MSK(Kafka) ? If NOT compatible could you please mention the 1. Compatible librdkafka version 2. Other C++ Kafka client Libraries which we can use with MSK.lg...

Hi. In kafka standalone I have kafka-authorizer.log whe I can find information about blocked requests to cluster. IE: [2022-07-18 12:23:12,131] INFO Principal = User:myuser-dev is Denied Operation = Describe from host = X.X.X.X on resource = Topic:LITERAL:mytopicname for request = Metadata with resourceRefCount = 1 (kafka.authorizer.logger) Do you know how I can fing this logs in AWS MSK or how can I configure it? Regardslg...