ERRO[0015] timeout reached waiting for Out channels to clear [MongoDB->ElasticSearch]

[ReMuSoMeGa]

I am trying to Sync a MongoDB to ElasticSearch.

First I execute ./transporter init mongodb elasticsearch
Then I modified my pipe file to reflect my db.
I execute ./transporter run

I get this error & everything fails:

INFO[0000] boot map[source:mongodb sink:elasticsearch]   ts=1500340480703118452
INFO[0000] starting with metadata map[]                  name=source path=source type=mongodb
INFO[0000] adaptor Listening...                          name=sink path="source/sink" type=elasticsearch
INFO[0000] adaptor Starting...                           name=source path=source type=mongodb
INFO[0000] starting Read func                            db=ontology
INFO[0000] collection count                              db=ontology num_collections=4
INFO[0000] adding for iteration...                       collection=chebinames db=ontology
INFO[0000] adding for iteration...                       collection=entries db=ontology
INFO[0000] adding for iteration...                       collection=ontology db=ontology
INFO[0000] adding for iteration...                       collection=suggestions db=ontology
INFO[0000] done iterating collections                    db=ontology
INFO[0000] iterating...                                  collection=chebinames
INFO[0000] Establishing new connection to localhost:27017 (timeout=1h0m0s)... 
INFO[0000] Connection to localhost:27017 established.   
INFO[0005] metrics source records: 2006                  path=source ts=1500340485702848350
INFO[0005] metrics source/sink records: 1994             path="source/sink" ts=1500340485702862441
INFO[0005] adaptor Stopping...                           name=source path=source type=mongodb
INFO[0005] adaptor Listen closed...                      name=sink path="source/sink" type=elasticsearch
INFO[0010] metrics source records: 2006                  path=source ts=1500340490704648315
INFO[0010] metrics source/sink records: 1994             path="source/sink" ts=1500340490704660715
INFO[0015] metrics source records: 2006                  path=source ts=1500340495704578839
INFO[0015] metrics source/sink records: 1994             path="source/sink" ts=1500340495704590150
INFO[0015] Establishing new connection to localhost:27017 (timeout=15s)... 
INFO[0015] Connection to localhost:27017 established.   
INFO[0015] Ping for localhost:27017 is 0 ms             
ERRO[0015] timeout reached waiting for Out channels to clear 
INFO[0015] adaptor Stopped                               name=source path=source type=mongodb
INFO[0015] adaptor Stopping...                           name=sink path="source/sink" type=elasticsearch
INFO[0015] adaptor Stopped                               name=sink path="source/sink" type=elasticsearch
INFO[0015] closing BulkProcessor                         version=5 writer=elasticsearch
ERRO[0019] elastic: Error 400 (Bad Request): Validation Failed: 1: index is missing;2: index is missing;3: index is missing;4: index is missing;5: index is missing;6: index is missing;7: index is missing;8: index is missing;9: index is missing;10: index is missing;11: index is missing;12: index is missing;13: index is missing;14: index is missing;15: index is missing;16: index is missing;17: index is missing;18: index is missing;19: index is missing;20: index is missing;21: index is missing;22: index is missing;23: index is missing;24: index is missing;25: index is missing;26: index is missing;27: index is missing;28: index is missing;29: index is missing;30: index is missing;31: index is missing;32: index is missing;33: index is missing;34: index is missing;35: index is missing;36: index is missing;37: index is missing;38: index is missing;39: index is missing;40: index is missing;41: index is missing;42: index is missing;43: index is missing;44: index is missing;45: index is missing;46: index is missing;47: index is missing;48: index is missing;49: index is missing;50:

Here is my Schema:

const ChebiName = new mySchema({
  name: {
    type: String,
    required: true,
    es_indexed: true
  },

  id: {
    type: Number,
    required: true
  }
});

My pipeline.js file:

var source = mongodb({
  "uri": "mongodb://localhost/ontology"
   "timeout": "3000s",
  // "tail": false,
  // "ssl": false,
  // "cacerts": ["/path/to/cert.pem"],
  // "wc": 1,
  // "fsync": false,
  // "bulk": false,
  // "collection_filters": "{}"
})

var sink = elasticsearch({
  "uri": "http://localhost:9200/"
   "timeout": "3000s", // defaults to 30s
  // "aws_access_key": "ABCDEF", // used for signing requests to AWS Elasticsearch service
  // "aws_access_secret": "ABCDEF" // used for signing requests to AWS Elasticsearch service
})

t.Source("source", source, "/.*/").Save("sink", sink, "/.*/")

I don't know what version this is, but I downloaded it today. It says "dev" as the version, if that helps! I'm running OS X 10.12.5

[jipperinbham]

based on the logs you provided, it appears as though the elasticsearch cluster went into a bad state causing the bulk insert to either fail or take longer than the default write timeout of 5 seconds

[pmjhonwang]

same question
via 0.3.1 is ok 0.4.0 got crash

INFO[0000] Establishing new connection to 10.200.20.23:27118 (timeout=1h0m0s)...
INFO[0000] Connection to 10.200.20.23:27118 established.
INFO[0005] metrics source records: 2010                  path=source ts=1502939310710042479
INFO[0005] metrics source/sink records: 1998             path="source/sink" ts=1502939310710052588
INFO[0005] adaptor Listen closed...                      name=sink path="source/sink" type=elasticsearch
INFO[0005] adaptor Stopping...                           name=source path=source type=mongodb
INFO[0010] metrics source records: 2010                  path=source ts=1502939315710152792
INFO[0010] metrics source/sink records: 1998             path="source/sink" ts=1502939315710167902
DEBU[0010] _bulk flush completed                         executionID=1 failed=0 succeeeded=1000 took=10606ms version=5 writer=elasticsearch
DEBU[0012] _bulk flush completed                         executionID=2 failed=0 succeeeded=998 took=1508ms version=5 writer=elasticsearch
DEBU[0014] _bulk flush completed                         executionID=3 failed=0 succeeeded=1 took=1513ms version=5 writer=elasticsearch
INFO[0015] metrics source records: 2010                  path=source ts=1502939320710123309
INFO[0015] metrics source/sink records: 1998             path="source/sink" ts=1502939320710138139
INFO[0015] Establishing new connection to 10.200.20.23:27118 (timeout=15s)...
INFO[0015] Ping for 10.200.20.23:27120 is 0 ms
INFO[0015] Ping for 10.200.20.23:27119 is 0 ms
INFO[0015] Connection to 10.200.20.23:27118 established.
INFO[0015] Ping for 10.200.20.23:27118 is 0 ms
ERRO[0015] timeout reached waiting for Out channels to clear  path=source
INFO[0015] adaptor Stopped                               name=source path=source type=mongodb
INFO[0015] adaptor Stopping...                           name=sink path="source/sink" type=elasticsearch
INFO[0015] adaptor Stopped                               name=sink path="source/sink" type=elasticsearch
INFO[0015] closing BulkProcessor                         version=5 writer=elasticsearch
INFO[0015] metrics source records: 2010                  path=source ts=1502939321011812192
INFO[0015] metrics source/sink records: 1998             path="source/sink" ts=1502939321011816379
INFO[0015] exit map[source:mongodb sink:elasticsearch]   ts=1502939321011817342
context deadline exceeded

[marczuk]

and the same problem on my side:
I'm using version 0.4.0, on 0.3.1 the same configuration was working fine

time="2017-08-23T09:57:22+02:00" level=info msg="Ping for localhost:27017 is 0 ms" 
time="2017-08-23T09:57:27+02:00" level=info msg="metrics source records: 115328" path=source ts=1503475047583570354 
time="2017-08-23T09:57:27+02:00" level=info msg="metrics source/sink records: 115316" path="source/sink" ts=1503475047583578824 
time="2017-08-23T09:57:29+02:00" level=info msg="adaptor Listen closed..." name=sink path="source/sink" type=elasticsearch 
time="2017-08-23T09:57:29+02:00" level=info msg="adaptor Stopping..." name=source path=source type=mongodb 
time="2017-08-23T09:57:32+02:00" level=info msg="metrics source records: 115328" path=source ts=1503475052583532310 
time="2017-08-23T09:57:32+02:00" level=info msg="metrics source/sink records: 115316" path="source/sink" ts=1503475052583553740 
time="2017-08-23T09:57:33+02:00" level=info msg="SYNC Starting full topology synchronization..." 
time="2017-08-23T09:57:33+02:00" level=info msg="SYNC Processing localhost:27017..." 
time="2017-08-23T09:57:33+02:00" level=info msg="SYNC Synchronization was complete (got data from primary)." 
time="2017-08-23T09:57:33+02:00" level=info msg="SYNC Synchronization completed: 1 master(s) and 0 slave(s) alive." 
time="2017-08-23T09:57:37+02:00" level=info msg="metrics source records: 115328" path=source ts=1503475057583585302 
time="2017-08-23T09:57:37+02:00" level=info msg="metrics source/sink records: 115316" path="source/sink" ts=1503475057583596115 
time="2017-08-23T09:57:37+02:00" level=info msg="Ping for localhost:27017 is 0 ms" 
time="2017-08-23T09:57:39+02:00" level=error msg="timeout reached waiting for Out channels to clear" path=source 
time="2017-08-23T09:57:39+02:00" level=info msg="adaptor Stopped" name=source path=source type=mongodb 
time="2017-08-23T09:57:39+02:00" level=info msg="adaptor Stopping..." name=sink path="source/sink" type=elasticsearch 
time="2017-08-23T09:57:39+02:00" level=info msg="adaptor Stopped" name=sink path="source/sink" type=elasticsearch 
time="2017-08-23T09:57:39+02:00" level=info msg="closing BulkProcessor" version=5 writer=elasticsearch 
time="2017-08-23T09:57:39+02:00" level=info msg="stopping compaction routine" name=source path=source type=mongodb 
time="2017-08-23T09:57:39+02:00" level=info msg="metrics source records: 115328" path=source ts=1503475059384668464 
time="2017-08-23T09:57:39+02:00" level=info msg="metrics source/sink records: 115316" path="source/sink" ts=1503475059384673174 
time="2017-08-23T09:57:39+02:00" level=info msg="waitForConfirms stopped" name=sink path="source/sink" type=elasticsearch 
time="2017-08-23T09:57:39+02:00" level=info msg="exit map[source:mongodb sink:elasticsearch]" ts=1503475059384674138 
context deadline exceeded

[johnjjung]

Maybe we should make the write timeout default to 5 seconds, but add it to elastic configs to increase it? @jipperinbham let me know, it's still fresh in my head to add that param

[johnjjung]

@jipperinbham I think you can close this issue, it was addressed in the documentation

https://github.com/compose/transporter/tree/master/adaptor/elasticsearch