将Mulesoft连接到Heroku举办的Kafka

Question

社区！

我正在尝试将我的Mulesoft应用程序连接到在赫洛库托管的Kafka服务器。任意点工作室 7 （骡子 4） 有一个Kafka连接器， 有几个选项可以连接 （基本， Kerberos， Kerberos SSL 和 SSL）：

根据Heroku的文档，它“支持”SSL，但没有提到这是必需的。有人能确认吗？

我看到的Heroku留档[https://devcenter.heroku.com/articles/kafka-on-heroku]

当我在 Heroku 上设置应用程序，添加 Kafka AddOn，创建主题时，我通过运行命令 heroku config：get KAFKA_URL获得引导服务器，如果我尝试进行粗略测试，我得到以下错误：

错误状态org.mule.runtime.api.connection.ConnectionException：无效连接！org.mule.runtime.api.connection.ConnectionException：无效连接！原因：org.apache.kafka.common.errors.TimeoutException：等待节点分配超时。

我试图构建一个有两个流的PoC…一个流向Topic生成消息，另一个流从Topic消费。

关于如何设置连接器和Heroku环境的任何帮助都将是最受欢迎的

更新:

我发现，当你在Heroku创建应用程序并包含KAFKA插件时，你会得到以下变量:

KAFKA_URL：组成集群的 Kafka 代理的 SSL URL 的逗号分隔列表。例：

kafka ssl：//ec2-3-*****- 100.compute-1.amazonaws.com:9096，kafka ssl：//ec2-3-******- 127.compute-1.amazonaws.com:9096

KAFKA_TRUSTED_CERT:代理的SSL证书(PEM格式)，用于检查您是否连接到了正确的服务器。示例:

-BEGIN CERTIFICATE-miidfzcamegwabibadabgkqhkig 9 w 0 baqsfadaymtawlgydvqddcdjys 1 l zjqwowuzny 00 njhltrimgetogvkoc 0 wzwyxmmrhyjkyzwewhhcnmtkxmjeymtux nzu 5 whcnmjkxmjeymtuxnzu 5 wymtawlgydvqddcdjys 1 ljwuzjwuzzn y 00 nhh

KAFKA_CLIENT_CERT：根据代理对客户进行身份验证所需的客户端证书（PEM格式）。示例：

-BEGIN CERTIFICATE-miidqzcccaiugawibagibadadanbgkqhkig 9 w 0 baqsfadaymtawlgydvqddcdjys 1 l zjqwowuzny 00 jhltrimgetogvkoc 0 wzwxmmrhyjkyzwewhhcnmjawmte 1 MTU 1 mju 2 whcnmzawmte 1 MTU 1 mju 2 wjazmrcwfqydvqdda 51 nztywvzm 2c wznmyzjc

KAFKA_CLIENT_CERT_KEY：根据代理对客户端进行身份验证所需的客户端证书密钥（PEM格式）。Kafka集群需要使用提供的客户端证书进行身份验证。任何不使用客户端证书的请求都将被拒绝。示例：

- BEGIN RSA私钥-MIIEpAIBAAKCAQEAmmu j 9 dulvnqwiot 02 6 ehw 9mg 7 kaocdigjodvvipc5 ayk iahsdnvh 9 bgjqajsfstiv/6o 1 mclmjus/ymyieegmbvatfxcldpgbgcppvsv 2r 3 q

我假设我需要将它们“转换”到连接器SSL配置的KeyStore和TrustStore中。有人能确认吗？因为我从文档中看到，SSL是必需的

另一个更新：

我下载了上面描述的证书，并使用工具“Key explorer”创建了一个密钥库。jks并将KAFKA_CLIENT_CERT和KAFQA_CLIENT_CERT_KEY导入其中，然后我创建了一个信任库。jsk并将文件KAFKA_TRUSTED_CERT导入其中。在这两种情况下，我都设置了一个基本密码……看起来不错，但我得到一个错误：

org.mule.runtime.api.connection.ConnectionException: invalid connection!
  org.mule.runtime.api.connection.ConnectionException: invalid connection!
  Caused by: org.apache.kafka.common.errors.SslAuthenticationException: SSL handshake failed
  Caused by: javax.net.ssl.SSLHandshakeException: General SSLEngine problem
    at sun.security.ssl.Handshaker.checkThrown(Handshaker.java:1521)
    at sun.security.ssl.SSLEngineImpl.checkTaskThrown(SSLEngineImpl.java:528)
    at sun.security.ssl.SSLEngineImpl.writeAppRecord(SSLEngineImpl.java:1197)
    at sun.security.ssl.SSLEngineImpl.wrap(SSLEngineImpl.java:1165)
    at javax.net.ssl.SSLEngine.wrap(SSLEngine.java:469)
    at org.apache.kafka.common.network.SslTransportLayer.handshakeWrap(SslTransportLayer.java:448)
    at org.apache.kafka.common.network.SslTransportLayer.doHandshake(SslTransportLayer.java:313)
    at org.apache.kafka.common.network.SslTransportLayer.handshake(SslTransportLayer.java:265)
    at org.apache.kafka.common.network.KafkaChannel.prepare(KafkaChannel.java:170)
    at org.apache.kafka.common.network.Selector.pollSelectionKeys(Selector.java:547)
    at org.apache.kafka.common.network.Selector.poll(Selector.java:483)
    at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:540)
    at org.apache.kafka.clients.admin.KafkaAdminClient$AdminClientRunnable.run(KafkaAdminClient.java:1196)
    at java.lang.Thread.run(Thread.java:748)
  Caused by: javax.net.ssl.SSLHandshakeException: General SSLEngine problem
    at sun.security.ssl.Alerts.getSSLException(Alerts.java:192)
    at sun.security.ssl.SSLEngineImpl.fatal(SSLEngineImpl.java:1709)
    at sun.security.ssl.Handshaker.fatalSE(Handshaker.java:318)
    at sun.security.ssl.Handshaker.fatalSE(Handshaker.java:310)
    at sun.security.ssl.ClientHandshaker.serverCertificate(ClientHandshaker.java:1639)
    at sun.security.ssl.ClientHandshaker.processMessage(ClientHandshaker.java:223)
    at sun.security.ssl.Handshaker.processLoop(Handshaker.java:1037)
    at sun.security.ssl.Handshaker$1.run(Handshaker.java:970)
    at sun.security.ssl.Handshaker$1.run(Handshaker.java:967)
    at java.security.AccessController.doPrivileged(Native Method)
    at sun.security.ssl.Handshaker$DelegatedTask.run(Handshaker.java:1459)
    at org.apache.kafka.common.network.SslTransportLayer.runDelegatedTasks(SslTransportLayer.java:402)
    at org.apache.kafka.common.network.SslTransportLayer.handshakeUnwrap(SslTransportLayer.java:484)
    at org.apache.kafka.common.network.SslTransportLayer.doHandshake(SslTransportLayer.java:340)
    ... 7 more
  Caused by: java.security.cert.CertificateException: No name matching ec2-3-220-121-33.compute-1.amazonaws.com found
    at sun.security.util.HostnameChecker.matchDNS(HostnameChecker.java:231)
    at sun.security.util.HostnameChecker.match(HostnameChecker.java:96)
    at sun.security.ssl.X509TrustManagerImpl.checkIdentity(X509TrustManagerImpl.java:455)
    at sun.security.ssl.X509TrustManagerImpl.checkIdentity(X509TrustManagerImpl.java:436)
    at sun.security.ssl.X509TrustManagerImpl.checkTrusted(X509TrustManagerImpl.java:252)
    at sun.security.ssl.X509TrustManagerImpl.checkServerTrusted(X509TrustManagerImpl.java:136)
    at sun.security.ssl.ClientHandshaker.serverCertificate(ClientHandshaker.java:1626)
    ... 16 more

可能与以下原因引起的行有关：

java.security.cert.CertificateException：没有与ec2-3-220-121-33.compute-1.amazonaws匹配的名称。找到com

Answer 1

使用以下命令创建JKS文件(需要HEROKU CLI)

client_key=`heroku config:get KAFKA_CLIENT_CERT_KEY --app <SET_HEROKU_APP_NAME_HERE>`
client_cert=`heroku config:get KAFKA_CLIENT_CERT --app <SET_HEROKU_APP_NAME_HERE>`
trusted_cert=`heroku config:get KAFKA_TRUSTED_CERT --app <SET_HEROKU_APP_NAME_HERE>`
# Write config vars to files.
echo "$client_key" >> keystore.pem
echo -n "$client_cert" >> keystore.pem
echo -n "$trusted_cert" > truststore.pem
# Set passwords
TRUSTSTORE_PASSWORD=<SET_PASSWORD_HERE>
KEYSTORE_PASSWORD=<SET_PASSWORD_HERE>
echo $TRUSTSTORE_PASSWORD
echo $KEYSTORE_PASSWORD
# Import cert.
keytool -importcert -file truststore.pem -keystore kafka.client.truststore.jks -deststorepass $TRUSTSTORE_PASSWORD -noprompt
# Create PKCS12 file.
openssl pkcs12 -export -in keystore.pem -out keystore.pkcs12 -password pass:$KEYSTORE_PASSWORD
# Create jks files.
keytool -importkeystore -srcstoretype PKCS12 \
    -destkeystore kafka.client.keystore.jks -deststorepass $KEYSTORE_PASSWORD \
    -srckeystore keystore.pkcs12 -srcstorepass $KEYSTORE_PASSWORD

然后使用SSL选项配置连接器并添加以下INLINE：