使用 Kafka 协议将 Camel 与 EventHubs 连接答案

【问题标题】：Connect Camel with EventHubs using Kafka protocol使用 Kafka 协议将 Camel 与 EventHubs 连接
【发布时间】：2019-11-14 16:28:55
【问题描述】：

我正在尝试将消息从 Camel 路由到 Azure EventHubs。 EventHubs 命名空间是使用启用 Kafka 的标志创建的。

         String eventHubsPassword = "org.apache.kafka.common.security.plain.PlainLoginModule " +
        "required username=\"$ConnectionString\" " + 
        "password=\"<Connection String>\";";

        String eventHubsConfig = 
        "&requestTimeoutMs=30000" +
        "&securityProtocol=SASL_SSL" + 
        "&saslMechanism=PLAIN" +
        "&saslJaasConfig=" + eventHubsPassword;

        from(component + ":queue:" + queue )
        .to("kafka:mock-topic?brokers=" + eventHubsKafkaBrokers + eventHubsConfig)

其中mock-topic 是事件中心的名称，eventHubsKafkaBrokers 类似于mynamespace.servicebus.windows.net:9093，<connection string> 是事件中心命名空间的连接字符串。

所以我得到了这个日志

2019-07-03 23:35:23 INFO  AbstractLogin:53 - Successfully logged in.
2019-07-03 23:35:23 INFO  AppInfoParser:109 - Kafka version : 1.0.0

在发送消息时我得到

2019-07-03 23:37:51 WARN  NetworkClient:241 - [Producer clientId=producer-2] Connection to node -1 could not be established. Broker may not be available.

可能出了什么问题？骆驼版是2.21.1。骆驼不支持SASL_SSL安全协议吗？

【问题讨论】：

标签： apache-kafka apache-camel azure-eventhub

【解决方案1】：

如果它对任何人有帮助，我最终会使用 EventHubs 提供的 amqp 支持。然后把路由的目的地放到定义好的amqp组件中。

AMQPComponent authorizedAmqp = AMQPComponent.amqpComponent( eventHubsNamespace, eventHubsUsername, //Shared access policy name eventHubsPassword); //Secret code for this shared access policy main.bind("amqps", authorizedAmqp); ... from('jms://source") .to("amqps" + ":queue:" + eventhubInstance);

【讨论】：