无法通过 MQTT Broker 在 Lenses Kafka 上接收数据

Question

我对 Lenses Kafka 的使用有一个特殊的问题。没有收到传感器数据。

也就是说，我尝试使用 Docker 容器创建网络连接，其中每个容器都代表服务。 服务如下：

1. 第一个是使用 MQTT 协议发送数据的传感器（在本例中是连接传感器的 Raspberry Pi）
2. MQTT 代理 - 在本例中为 Mosquitto，用于获取消息并与 Lenses 交互
3. Lenses Kafka - 在这种情况下用于 Mosquitto 与其他服务（如另一端的 InfluxDB）之间交互的系统

我可以肯定地说，发送的数据已成功接收，因为我使用 Apache JMeter 检查从 Mosquitto 站点接收到的数据。问题出现在镜头端，连接被识别，但没有接收到数据，并显示以下错误：

MqttException (0) - java.net.NoRouteToHostException: Host is unreachable (Host unreachable) 
 at org.eclipse.paho.client.mqttv3.internal.ExceptionHelper.createMqttException(ExceptionHelper.java:38)
 at org.eclipse.paho.client.mqttv3.internal.ClientComms$ConnectBG.run(ClientComms.java:715)
 at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) 
 at java.util.concurrent.FutureTask.run(FutureTask.java:266)
 at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180)
 at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293) 
 at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
 at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
 at java.lang.Thread.run(Thread.java:748) 
Caused by: java.net.NoRouteToHostException: Host is unreachable (Host unreachable)
 at java.net.PlainSocketImpl.socketConnect(Native Method)
 at java.net.AbstractPlainSocketImpl.doConnect(AbstractPlainSocketImpl.java:350)
 at java.net.AbstractPlainSocketImpl.connectToAddress(AbstractPlainSocketImpl.java:206)
 at java.net.AbstractPlainSocketImpl.connect(AbstractPlainSocketImpl.java:188) 
 at java.net.SocksSocketImpl.connect(SocksSocketImpl.java:392)
 at java.net.Socket.connect(Socket.java:589) 
 at org.eclipse.paho.client.mqttv3.internal.TCPNetworkModule.start(TCPNetworkModule.java:84)
 at org.eclipse.paho.client.mqttv3.internal.ClientComms$ConnectBG.run(ClientComms.java:701) 
... 7 more

Lenses端读取数据的配置代码为：

connector.class=com.datamountaineer.streamreactor.connect.mqtt.source.MqttSourceConnector
connect.mqtt.clean=true
connect.mqtt.timeout=1000
tasks.max=1
connect.mqtt.keep.alive=1000
name=Mosquitto
connect.mqtt.kcql=INSERT INTO kafka_sensor_data SELECT * FROM /sensor_data WITHCONVERTER=`com.datamountaineer.streamreactor.connect.converters.source.JsonSimpleConverter`
connect.mqtt.service.quality=1
connect.mqtt.client.id=dm_source_id
connect.mqtt.converter.throw.on.error=true
connect.mqtt.hosts=tcp://XXX:1883

（XXX只是为了安全起见，已经有经过测试的IP地址）。

这种方法实际上直到最近才有效，但突然之间我现在变成了这个错误。我检查了连接和端口连接，但一切正常。真的不知道在哪里检查了。

Answer 1

一段时间后，我设法找到了问题。

这是通过使用 docker 容器。虽然该服务设法保持端口和地址对外部网络开放和组织，但内部存在问题，因为 Lenses 无法识别 MQTT 代理。

在配置我自己的网络而不是让它默认创建之后，我设法解决了这个问题。