This message was deleted.

Please share any errors from the connector log. Does the mongoURI contain valid user credentials, etc?

@David K

WARNING: sun.reflect.Reflection.getCallerClass is not supported. This will impact performance.
Starting function instance...
2023-12-03T22:52:55,285+0000 [main] INFO  org.apache.pulsar.common.nar.FileUtils - Jar file /pulsar/download/pulsar_functions/pulsar-io-mongo-3.0.1.nar contains META-INF/bundled-dependencies, it may be a NAR file
2023-12-03T22:52:55,287+0000 [main] INFO  org.apache.pulsar.functions.runtime.thread.ThreadRuntime - Trying Loading file as NAR file: /pulsar/download/pulsar_functions/pulsar-io-mongo-3.0.1.nar
2023-12-03T22:52:55,289+0000 [main] INFO  org.apache.pulsar.common.nar.NarUnpacker - Created directory /tmp/pulsar-nar/pulsar-io-mongo-3.0.1.nar-unpacked
2023-12-03T22:52:55,567+0000 [main] INFO  org.apache.pulsar.common.nar.NarUnpacker - Extracting /pulsar/download/pulsar_functions/pulsar-io-mongo-3.0.1.nar to /tmp/pulsar-nar/pulsar-io-mongo-3.0.1.nar-unpacked/qbPXOMCHZge4vdkxkExDpA
2023-12-03T22:52:55,675+0000 [main] INFO  org.apache.pulsar.functions.runtime.thread.ThreadRuntime - Initialize function class loader for function mongodb at function cache manager, functionClassLoader: org.apache.pulsar.common.nar.NarClassLoader[/tmp/pulsar-nar/pulsar-io-mongo-3.0.1.nar-unpacked/qbPXOMCHZge4vdkxkExDpA]
2023-12-03T22:52:56,872+0000 [main] INFO  org.apache.pulsar.functions.runtime.JavaInstanceStarter - JavaInstance Server started, listening on 9093
2023-12-03T22:52:56,872+0000 [main] INFO  org.apache.pulsar.functions.runtime.JavaInstanceStarter - Starting runtimeSpawner
2023-12-03T22:52:56,873+0000 [main] INFO  org.apache.pulsar.functions.runtime.RuntimeSpawner - public/default/mongodb-0 RuntimeSpawner starting function
2023-12-03T22:52:56,874+0000 [main] INFO  org.apache.pulsar.common.nar.FileUtils - Jar file /pulsar/download/pulsar_functions/pulsar-io-mongo-3.0.1.nar contains META-INF/bundled-dependencies, it may be a NAR file
2023-12-03T22:52:56,875+0000 [main] INFO  org.apache.pulsar.functions.runtime.thread.ThreadRuntime - Trying Loading file as NAR file: /pulsar/download/pulsar_functions/pulsar-io-mongo-3.0.1.nar
2023-12-03T22:52:56,875+0000 [main] INFO  org.apache.pulsar.functions.runtime.thread.ThreadRuntime - Initialize function class loader for function mongodb at function cache manager, functionClassLoader: org.apache.pulsar.common.nar.NarClassLoader[/tmp/pulsar-nar/pulsar-io-mongo-3.0.1.nar-unpacked/qbPXOMCHZge4vdkxkExDpA]
2023-12-03T22:52:56,881+0000 [main] INFO  org.apache.pulsar.functions.runtime.thread.ThreadRuntime - ThreadContainer starting function with instanceId 0 functionId 133bc9e6-1b92-411b-bf43-973365571ef0 namespace default
2023-12-03T22:52:56,882+0000 [main] INFO  org.apache.pulsar.functions.runtime.JavaInstanceStarter - Starting metrics server on port 9094
2023-12-03T22:52:56,965+0000 [public/default/mongodb-0] INFO  org.apache.pulsar.functions.instance.JavaInstanceRunnable - Starting Java Instance mongodb : 
 Details = tenant: "public"
namespace: "default"
name: "mongodb"
className: "org.apache.pulsar.functions.api.utils.IdentityFunction"
autoAck: true
parallelism: 1
source {
  typeClassName: "[B"
  inputSpecs {
    key: "<persistent://public/default/msg2>"
    value {
    }
  }
  cleanupSubscription: true
}
sink {
  className: "org.apache.pulsar.io.mongodb.MongoSink"
  configs: "{\"mongoUri\":\"<my_mongo_db_connectionstring>",\"database\":\"pulsar\",\"collection\":\"messages\",\"batchSize\":\"2\",\"batchTimeMs\":\"500\"}"
  typeClassName: "[B"
  builtin: "mongo"
}
resources {
  cpu: 1.0
  ram: 1073741824
  disk: 10737418240
}
componentType: SINK

2023-12-03T22:52:57,397+0000 [public/default/mongodb-0] INFO  org.apache.pulsar.io.mongodb.MongoSink - Open MongoDB Sink
2023-12-03T22:52:57,671+0000 [public/default/mongodb-0] INFO  org.mongodb.driver.cluster - Cluster created with settings {hosts=[127.0.0.1:27017], srvHost=<http://ojoxdan.0swgc0s.mongodb.net|ojoxdan.0swgc0s.mongodb.net>, mode=MULTIPLE, requiredClusterType=REPLICA_SET, serverSelectionTimeout='30000 ms', requiredReplicaSetName='atlas-u82gfk-shard-0'}
2023-12-03T22:52:57,681+0000 [cluster-ClusterId{value='656d06c956e567163837aa99', description='null'}-<http://srv-ojoxdan.0swgc0s.mongodb.net]|srv-ojoxdan.0swgc0s.mongodb.net]> INFO  org.mongodb.driver.cluster - Adding discovered server <http://ac-sw94ucp-shard-00-02.0swgc0s.mongodb.net:27017|ac-sw94ucp-shard-00-02.0swgc0s.mongodb.net:27017> to client view of cluster
2023-12-03T22:52:57,768+0000 [cluster-ClusterId{value='656d06c956e567163837aa99', description='null'}-<http://srv-ojoxdan.0swgc0s.mongodb.net]|srv-ojoxdan.0swgc0s.mongodb.net]> INFO  org.mongodb.driver.cluster - Adding discovered server <http://ac-sw94ucp-shard-00-01.0swgc0s.mongodb.net:27017|ac-sw94ucp-shard-00-01.0swgc0s.mongodb.net:27017> to client view of cluster
2023-12-03T22:52:57,770+0000 [cluster-ClusterId{value='656d06c956e567163837aa99', description='null'}-<http://srv-ojoxdan.0swgc0s.mongodb.net]|srv-ojoxdan.0swgc0s.mongodb.net]> INFO  org.mongodb.driver.cluster - Adding discovered server <http://ac-sw94ucp-shard-00-00.0swgc0s.mongodb.net:27017|ac-sw94ucp-shard-00-00.0swgc0s.mongodb.net:27017> to client view of cluster
2023-12-03T22:52:58,291+0000 [public/default/mongodb-0] INFO  org.apache.pulsar.functions.source.SingleConsumerPulsarSource - Opening pulsar source with config: SingleConsumerPulsarSourceConfig(topic=<persistent://public/default/msg2>, consumerConfig=ConsumerConfig(schemaType=null, serdeClassName=null, isRegexPattern=false, schemaProperties={}, consumerProperties={}, receiverQueueSize=null, cryptoConfig=null, poolMessages=false))
2023-12-03T22:52:58,469+0000 [public/default/mongodb-0] INFO  org.apache.pulsar.functions.source.SingleConsumerPulsarSource - Creating consumer for topic : <persistent://public/default/msg2>, schema : org.apache.pulsar.client.impl.schema.BytesSchema@82c1480, schemaInfo: {
  "name": "Bytes",
  "schema": "",
  "type": "BYTES",
  "timestamp": 0,
  "properties": {}
}
2023-12-03T22:52:58,895+0000 [pulsar-client-io-1-1] INFO  org.apache.pulsar.client.impl.ConnectionPool - [[id: 0x75f57cfc, L:/192.168.140.69:47502 - R:pulsar-init-broker.pulsar.svc.cluster.local/192.168.140.94:6650]] Connected to server
2023-12-03T22:52:59,178+0000 [cluster-ClusterId{value='656d06c956e567163837aa99', description='null'}-<http://ac-sw94ucp-shard-00-02.0swgc0s.mongodb.net:27017]|ac-sw94ucp-shard-00-02.0swgc0s.mongodb.net:27017]> INFO  org.mongodb.driver.connection - Opened connection [connectionId{localValue:2, serverValue:6982}] to <http://ac-sw94ucp-shard-00-02.0swgc0s.mongodb.net:27017|ac-sw94ucp-shard-00-02.0swgc0s.mongodb.net:27017>
2023-12-03T22:52:59,179+0000 [cluster-ClusterId{value='656d06c956e567163837aa99', description='null'}-<http://ac-sw94ucp-shard-00-00.0swgc0s.mongodb.net:27017]|ac-sw94ucp-shard-00-00.0swgc0s.mongodb.net:27017]> INFO  org.mongodb.driver.connection - Opened connection [connectionId{localValue:5, serverValue:6976}] to <http://ac-sw94ucp-shard-00-00.0swgc0s.mongodb.net:27017|ac-sw94ucp-shard-00-00.0swgc0s.mongodb.net:27017>
2023-12-03T22:52:59,179+0000 [cluster-rtt-ClusterId{value='656d06c956e567163837aa99', description='null'}-<http://ac-sw94ucp-shard-00-00.0swgc0s.mongodb.net:27017]|ac-sw94ucp-shard-00-00.0swgc0s.mongodb.net:27017]> INFO  org.mongodb.driver.connection - Opened connection [connectionId{localValue:1, serverValue:6976}] to <http://ac-sw94ucp-shard-00-00.0swgc0s.mongodb.net:27017|ac-sw94ucp-shard-00-00.0swgc0s.mongodb.net:27017>
2023-12-03T22:52:59,179+0000 [cluster-rtt-ClusterId{value='656d06c956e567163837aa99', description='null'}-<http://ac-sw94ucp-shard-00-02.0swgc0s.mongodb.net:27017]|ac-sw94ucp-shard-00-02.0swgc0s.mongodb.net:27017]> INFO  org.mongodb.driver.connection - Opened connection [connectionId{localValue:4, serverValue:6945}] to <http://ac-sw94ucp-shard-00-02.0swgc0s.mongodb.net:27017|ac-sw94ucp-shard-00-02.0swgc0s.mongodb.net:27017>
2023-12-03T22:52:59,179+0000 [cluster-ClusterId{value='656d06c956e567163837aa99', description='null'}-<http://ac-sw94ucp-shard-00-01.0swgc0s.mongodb.net:27017]|ac-sw94ucp-shard-00-01.0swgc0s.mongodb.net:27017]> INFO  org.mongodb.driver.connection - Opened connection [connectionId{localValue:3, serverValue:6166}] to <http://ac-sw94ucp-shard-00-01.0swgc0s.mongodb.net:27017|ac-sw94ucp-shard-00-01.0swgc0s.mongodb.net:27017>
2023-12-03T22:52:59,179+0000 [cluster-rtt-ClusterId{value='656d06c956e567163837aa99', description='null'}-<http://ac-sw94ucp-shard-00-01.0swgc0s.mongodb.net:27017]|ac-sw94ucp-shard-00-01.0swgc0s.mongodb.net:27017]> INFO  org.mongodb.driver.connection - Opened connection [connectionId{localValue:6, serverValue:6166}] to <http://ac-sw94ucp-shard-00-01.0swgc0s.mongodb.net:27017|ac-sw94ucp-shard-00-01.0swgc0s.mongodb.net:27017>
2023-12-03T22:52:59,272+0000 [cluster-ClusterId{value='656d06c956e567163837aa99', description='null'}-<http://ac-sw94ucp-shard-00-02.0swgc0s.mongodb.net:27017]|ac-sw94ucp-shard-00-02.0swgc0s.mongodb.net:27017]> INFO  org.mongodb.driver.cluster - Monitor thread successfully connected to server with description ServerDescription{address=<http://ac-sw94ucp-shard-00-02.0swgc0s.mongodb.net:27017|ac-sw94ucp-shard-00-02.0swgc0s.mongodb.net:27017>, type=REPLICA_SET_PRIMARY, state=CONNECTED, ok=true, minWireVersion=0, maxWireVersion=17, maxDocumentSize=16777216, logicalSessionTimeoutMinutes=30, roundTripTimeNanos=806597035, setName='atlas-u82gfk-shard-0', canonicalAddress=<http://ac-sw94ucp-shard-00-02.0swgc0s.mongodb.net:27017|ac-sw94ucp-shard-00-02.0swgc0s.mongodb.net:27017>, hosts=[<http://ac-sw94ucp-shard-00-02.0swgc0s.mongodb.net:27017|ac-sw94ucp-shard-00-02.0swgc0s.mongodb.net:27017>, <http://ac-sw94ucp-shard-00-00.0swgc0s.mongodb.net:27017|ac-sw94ucp-shard-00-00.0swgc0s.mongodb.net:27017>, <http://ac-sw94ucp-shard-00-01.0swgc0s.mongodb.net:27017]|ac-sw94ucp-shard-00-01.0swgc0s.mongodb.net:27017]>, passives=[], arbiters=[], primary='<http://ac-sw94ucp-shard-00-02.0swgc0s.mongodb.net:27017|ac-sw94ucp-shard-00-02.0swgc0s.mongodb.net:27017>', tagSet=TagSet{[Tag{name='availabilityZone', value='use1-az4'}, Tag{name='nodeType', value='ELECTABLE'}, Tag{name='provider', value='AWS'}, Tag{name='region', value='US_EAST_1'}, Tag{name='workloadType', value='OPERATIONAL'}]}, electionId=7fffffff000000000000016b, setVersion=9, topologyVersion=TopologyVersion{processId=656cd1d1a6bf2b5e83945d0c, counter=6}, lastWriteDate=Sun Dec 03 22:52:59 UTC 2023, lastUpdateTimeNanos=191066711802574}
2023-12-03T22:52:59,272+0000 [cluster-ClusterId{value='656d06c956e567163837aa99', description='null'}-<http://ac-sw94ucp-shard-00-01.0swgc0s.mongodb.net:27017]|ac-sw94ucp-shard-00-01.0swgc0s.mongodb.net:27017]> INFO  org.mongodb.driver.cluster - Monitor thread successfully connected to server with description ServerDescription{address=<http://ac-sw94ucp-shard-00-01.0swgc0s.mongodb.net:27017|ac-sw94ucp-shard-00-01.0swgc0s.mongodb.net:27017>, type=REPLICA_SET_SECONDARY, state=CONNECTED, ok=true, minWireVersion=0, maxWireVersion=17, maxDocumentSize=16777216, logicalSessionTimeoutMinutes=30, roundTripTimeNanos=804724549, setName='atlas-u82gfk-shard-0', canonicalAddress=<http://ac-sw94ucp-shard-00-01.0swgc0s.mongodb.net:27017|ac-sw94ucp-shard-00-01.0swgc0s.mongodb.net:27017>, hosts=[<http://ac-sw94ucp-shard-00-02.0swgc0s.mongodb.net:27017|ac-sw94ucp-shard-00-02.0swgc0s.mongodb.net:27017>, <http://ac-sw94ucp-shard-00-00.0swgc0s.mongodb.net:27017|ac-sw94ucp-shard-00-00.0swgc0s.mongodb.net:27017>, <http://ac-sw94ucp-shard-00-01.0swgc0s.mongodb.net:27017]|ac-sw94ucp-shard-00-01.0swgc0s.mongodb.net:27017]>, passives=[], arbiters=[], primary='<http://ac-sw94ucp-shard-00-02.0swgc0s.mongodb.net:27017|ac-sw94ucp-shard-00-02.0swgc0s.mongodb.net:27017>', tagSet=TagSet{[Tag{name='availabilityZone', value='use1-az2'}, Tag{name='nodeType', value='ELECTABLE'}, Tag{name='provider', value='AWS'}, Tag{name='region', value='US_EAST_1'}, Tag{name='workloadType', value='OPERATIONAL'}]}, electionId=null, setVersion=9, topologyVersion=TopologyVersion{processId=656cd2ca7767ab1c392d669b, counter=3}, lastWriteDate=Sun Dec 03 22:52:58 UTC 2023, lastUpdateTimeNanos=191066711788105}
2023-12-03T22:52:59,272+0000 [cluster-ClusterId{value='656d06c956e567163837aa99', description='null'}-<http://ac-sw94ucp-shard-00-00.0swgc0s.mongodb.net:27017]|ac-sw94ucp-shard-00-00.0swgc0s.mongodb.net:27017]> INFO  org.mongodb.driver.cluster - Monitor thread successfully connected to server with description ServerDescription{address=<http://ac-sw94ucp-shard-00-00.0swgc0s.mongodb.net:27017|ac-sw94ucp-shard-00-00.0swgc0s.mongodb.net:27017>, type=REPLICA_SET_SECONDARY, state=CONNECTED, ok=true, minWireVersion=0, maxWireVersion=17, maxDocumentSize=16777216, logicalSessionTimeoutMinutes=30, roundTripTimeNanos=805405224, setName='atlas-u82gfk-shard-0', canonicalAddress=<http://ac-sw94ucp-shard-00-00.0swgc0s.mongodb.net:27017|ac-sw94ucp-shard-00-00.0swgc0s.mongodb.net:27017>, hosts=[<http://ac-sw94ucp-shard-00-02.0swgc0s.mongodb.net:27017|ac-sw94ucp-shard-00-02.0swgc0s.mongodb.net:27017>, <http://ac-sw94ucp-shard-00-00.0swgc0s.mongodb.net:27017|ac-sw94ucp-shard-00-00.0swgc0s.mongodb.net:27017>, <http://ac-sw94ucp-shard-00-01.0swgc0s.mongodb.net:27017]|ac-sw94ucp-shard-00-01.0swgc0s.mongodb.net:27017]>, passives=[], arbiters=[], primary='<http://ac-sw94ucp-shard-00-02.0swgc0s.mongodb.net:27017|ac-sw94ucp-shard-00-02.0swgc0s.mongodb.net:27017>', tagSet=TagSet{[Tag{name='availabilityZone', value='use1-az1'}, Tag{name='nodeType', value='ELECTABLE'}, Tag{name='provider', value='AWS'}, Tag{name='region', value='US_EAST_1'}, Tag{name='workloadType', value='OPERATIONAL'}]}, electionId=null, setVersion=9,

So you deployed a SINK connector, and it looks like there wasn’t any data in the source topic,

public/default/msg2

, so the connector decided to shut down.

@David K hmm, so why was the running status true when I checked status of the connector ?

You should change the subscription setting to read from the earliest message in the topic. The default is to start from the latest.

Ok thank you @David K I will check and come back

I would have to look at the code to confirm, but IIRC, the connector uses the MongoDB client.

@David K yes I think it does, here is the repo https://github.com/apache/pulsar/blob/master/pulsar-io/mongo/src/main/java/org/apache/pulsar/io/mongodb/MongoSink.java I am just thinking how I could possible pass these rules