Run Debezium on AWS with MSK

[Shlomi Lanton]

Hey all,

I'm trying to run Debezium on AWS, I want to use MSK and ECS, is there a good tutorial for that?

Up to now, my setup includes MSK (Kafka V. 2.7.0) with "Only TLS encrypted traffic allowed" and an ECS cluster with a task that uses the Debezium docker, in the task I set up the following environment variables:

BOOTSTRAP_SERVERS (comma separated list of MSK hosts and ports)  - b-2.ssadjhsadjk.ur6chx.c1.kafka.eu-west-1.amazonaws.com:9094, b-2.ssadjhsadjk.ur6chg.c1.kafka.eu-west-1.amazonaws.com:9094

CONFIG_STORAGE_TOPIC my_connect_configs

GROUP_ID 1

OFFSET_STORAGE_TOPIC my_connect_offsets

STATUS_STORAGE_TOPIC my_connect_statuses

Those are the last lines from the Debezium log:

-02-17 15:28:27,784 INFO || [AdminClient clientId=adminclient-1] Metadata update failed [org.apache.kafka.clients.admin.internals.AdminMetadataManager]

org.apache.kafka.common.errors.TimeoutException: Call(callName=fetchMetadata, deadlineMs=1613575707783, tries=1, nextAllowedTryMs=1613575707884) timed out at 1613575707784 after 1 attempt(s)

Caused by: org.apache.kafka.common.errors.TimeoutException: Timed out waiting to send the call.

2021-02-17 15:28:57,784 INFO || [AdminClient clientId=adminclient-1] Metadata update failed [org.apache.kafka.clients.admin.internals.AdminMetadataManager]

org.apache.kafka.common.errors.TimeoutException: Call(callName=fetchMetadata, deadlineMs=1613575737784, tries=1, nextAllowedTryMs=-9223372036854775709) timed out at 9223372036854775807 after 1 attempt(s)

Caused by: org.apache.kafka.common.errors.TimeoutException: Timed out waiting to send the call.

2021-02-17 15:28:57,808 ERROR || Stopping due to error [org.apache.kafka.connect.cli.ConnectDistributed]

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=1613575737782, tries=1, nextAllowedTryMs=1613575737883) timed out at 1613575737783 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=1613575737782, tries=1, nextAllowedTryMs=1613575737883) timed out at 1613575737783 after 1 attempt(s)

Caused by: org.apache.kafka.common.errors.TimeoutException: Timed out waiting for a node assignment.

Any ideas what I'm missing? in "BOOTSTRAP_SERVERS" do I need to list the ZooKeeper hosts as well? do I need to set "security.protocol" with "SSL"? I have tried adding it as an environment variable but it didn't help.

Thanks

[Gunnar Morling]

Hi Shlomi,

You'd definitely have to set your Kafka Connect worker's configuration so it works with SSL. This answer on SO may be a good starting point:

    https://stackoverflow.com/questions/54258217/kafka-connect-failing-to-read-from-kafka-topics-over-ssl

--Gunnar