使用Kafka协议连接Camel和EventHubs

Question

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

         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.

会出什么问题呢？Camel版本为2.21.1。camel不支持SASL_SSL安全协议吗？

Answer 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);