mongodb - 将 MongoDB 作为源连接到 Kafka 时出错

Question

我正在尝试将 MongoDB 作为源连接到 Kafka 服务器，但是当我运行 Debezium Mongo 源连接器时，出现错误。我不明白为什么超时？

[2019-08-22 13:28:58,194] INFO Cluster description not yet available. Waiting for 30000 ms before timing out (org.mongodb.driver.cluster:71)
[2019-08-22 13:28:58,648] INFO Exception in monitor thread while connecting to server morgan-shard-00-00-ayfai.mongodb.net:27017 (org.mongodb.driver.cluster:76)
com.mongodb.MongoSocketReadException: Prematurely reached end of stream
    at com.mongodb.internal.connection.SocketStream.read(SocketStream.java:112)
    at com.mongodb.internal.connection.InternalStreamConnection.receiveResponseBuffers(InternalStreamConnection.java:554)
    at com.mongodb.internal.connection.InternalStreamConnection.receiveMessage(InternalStreamConnection.java:425)
    at com.mongodb.internal.connection.InternalStreamConnection.receiveCommandMessageResponse(InternalStreamConnection.java:289)
    at com.mongodb.internal.connection.InternalStreamConnection.sendAndReceive(InternalStreamConnection.java:255)
    at com.mongodb.internal.connection.CommandHelper.sendAndReceive(CommandHelper.java:83)
    at com.mongodb.internal.connection.CommandHelper.executeCommand(CommandHelper.java:33)
    at com.mongodb.internal.connection.InternalStreamConnectionInitializer.initializeConnectionDescription(InternalStreamConnectionInitializer.java:106)
    at com.mongodb.internal.connection.InternalStreamConnectionInitializer.initialize(InternalStreamConnectionInitializer.java:63)
    at com.mongodb.internal.connection.InternalStreamConnection.open(InternalStreamConnection.java:127)
    at com.mongodb.internal.connection.DefaultServerMonitor$ServerMonitorRunnable.run(DefaultServerMonitor.java:117)
    at java.lang.Thread.run(Thread.java:748)
[2019-08-22 13:29:08,196] INFO Created connector mongodb-source-connector (org.apache.kafka.connect.cli.ConnectStandalone:112)
[2019-08-22 13:29:28,195] ERROR Error while reading the 'shards' collection in the 'config' database: Timed out after 30000 ms while waiting to connect. Client view of cluster state is {type=UNKNOWN, servers=[{address=morgan-shard-00-00-ayfai.mongodb.net:27017, type=UNKNOWN, state=CONNECTING, exception={com.mongodb.MongoSocketReadException: Prematurely reached end of stream}}] (io.debezium.connector.mongodb.ReplicaSetDiscovery:80)
com.mongodb.MongoTimeoutException: Timed out after 30000 ms while waiting to connect. Client view of cluster state is {type=UNKNOWN, servers=[{address=morgan-shard-00-00-ayfai.mongodb.net:27017, type=UNKNOWN, state=CONNECTING, exception={com.mongodb.MongoSocketReadException: Prematurely reached end of stream}}]
    at com.mongodb.internal.connection.BaseCluster.getDescription(BaseCluster.java:179)
    at com.mongodb.internal.connection.SingleServerCluster.getDescription(SingleServerCluster.java:41)
    at com.mongodb.client.internal.MongoClientDelegate.getConnectedClusterDescription(MongoClientDelegate.java:136)
    at com.mongodb.client.internal.MongoClientDelegate.createClientSession(MongoClientDelegate.java:94)
    at com.mongodb.client.internal.MongoClientDelegate$DelegateOperationExecutor.getClientSession(MongoClientDelegate.java:249)
    at com.mongodb.client.internal.MongoClientDelegate$DelegateOperationExecutor.execute(MongoClientDelegate.java:172)
    at com.mongodb.client.internal.MongoIterableImpl.execute(MongoIterableImpl.java:132)
    at com.mongodb.client.internal.MongoIterableImpl.iterator(MongoIterableImpl.java:86)
    at com.mongodb.client.internal.MappingIterable.iterator(MappingIterable.java:39)
    at io.debezium.connector.mongodb.MongoUtil.contains(MongoUtil.java:183)
    at io.debezium.connector.mongodb.MongoUtil.contains(MongoUtil.java:172)
    at io.debezium.connector.mongodb.MongoUtil.onDatabase(MongoUtil.java:116)
    at io.debezium.connector.mongodb.MongoUtil.onCollection(MongoUtil.java:131)
    at io.debezium.connector.mongodb.MongoUtil.onCollectionDocuments(MongoUtil.java:150)
    at io.debezium.connector.mongodb.ReplicaSetDiscovery.getReplicaSets(ReplicaSetDiscovery.java:67)
    at io.debezium.connector.mongodb.ReplicaSetMonitorThread.run(ReplicaSetMonitorThread.java:65)
    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)
[2019-08-22 13:29:28,195] INFO Cluster description not yet available. Waiting for 30000 ms before timing out (org.mongodb.driver.cluster:71)
[2019-08-22 13:29:58,196] ERROR Error while trying to get information about the replica sets (io.debezium.connector.mongodb.ReplicaSetMonitorThread:87)
com.mongodb.MongoTimeoutException: Timed out after 30000 ms while waiting to connect. Client view of cluster state is {type=UNKNOWN, servers=[{address=morgan-shard-00-00-ayfai.mongodb.net:27017, type=UNKNOWN, state=CONNECTING, exception={com.mongodb.MongoSocketReadException: Prematurely reached end of stream}}]
    at com.mongodb.internal.connection.BaseCluster.getDescription(BaseCluster.java:179)
    at com.mongodb.internal.connection.SingleServerCluster.getDescription(SingleServerCluster.java:41)
    at com.mongodb.Mongo.getClusterDescription(Mongo.java:412)
    at com.mongodb.Mongo.getReplicaSetStatus(Mongo.java:455)
    at io.debezium.connector.mongodb.ReplicaSetDiscovery.getReplicaSets(ReplicaSetDiscovery.java:85)
    at io.debezium.connector.mongodb.ReplicaSetMonitorThread.run(ReplicaSetMonitorThread.java:65)
    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)

Answer 1

如果不查看连接器配置，就不可能知道确切的问题，但根据我使用 MongoDB Atlas 和 Debezium 的经验，这可能是 SSL 的问题。

尝试启用 SSL

mongodb.ssl.enabled: true

https://debezium.io/documentation/reference/1.2/connectors/mongodb.html#mongodb-property-mongodb-ssl-enabled