Bulk Insert In Janusgraph

[sharma-kunal]

I have to insert around 10 million vertices and edges in my Graph. I am using cassandra for backend db and elastic for indexing. I am doing some data pre-processing and then add the vertices like this to the transaction

g.tx.open();
Vertex seeker = g.addV("seeker").property("name", "seeker").next();
Vertex institute = g.addV("institute").property("institute_name", "Oxford").next();
g.V(seeker).addE("studied").to(institute).iterate();

I add around 500 entries at once and then I commit the data

g.tx().commit();
g.tx().close()

and this goes on, and it works fine for some time, but after some time, it fails with NullPointerException: Graph has been closed.

Now my question is, if graph closes and I am not able to commit, than what happens with the data that I have added in the transaction ??. Do I need to reinitialize the graph and then again do the same processing everytime I encounter this error ? or is there any other way to handle such scenario ?

[li-boxuan]

Could you provide the full stacktrace for the exception please?

If you see an exception, it won't affect the data you have already successfully committed.

[sharma-kunal]

I made some small changes in my code, but it still fails with NullPointerException.

Now the code looks like,

// do this when application starts
JanusGraph graph = JanusGraphFactory.build().set(<graph config>).open()
GraphTraversalSource g = graph.traversal()
// Start a RabbitMQ Consumer

// Consume msg from RabbitMQ (a range of profiles)
g.tx().open()
while(<some condition>) {
     // process these profiles
    Vertex seeker = g.addV("seeker").property("name", "seeker").next();
    Vertex institute = g.addV("institute").property("institute_name", "Oxford").next();
    g.V(seeker).addE("studied").to(institute).iterate();
}
g.tx().commit();
g.tx().close();

Now the issue with the above code I am facing is, sometimes the graph closes while inside the while loop and sometime the g.tx() closes. (I have checked it in the try catch block with 'graph.isOpen()' and 'g.tx().isOpen()')

So, what I do now is, I catch the Exception and if it is due to GRaph Closure, break the loop and push the current msg to a fail queue.

if (graph == null || !graph.isOpen()) {
    JanusGraph graph = JanusGraphFactory.build().set(<graph config>).open()
    GraphTraversalSource g = graph.traversal()
}
if(!g.tx().isOpen()) g.tx().open()

// Same old code with try catch block
// Consume msg from RabbitMQ (a range of profiles)
g.tx().open()
try {
    while(<some condition>) {
         // process these profiles
        Vertex seeker = g.addV("seeker").property("name", "seeker").next();
        Vertex institute = g.addV("institute").property("institute_name", "Oxford").next();
        g.V(seeker).addE("studied").to(institute).iterate();
    } 
    g.tx().commit();
    g.tx().close();
} catch (Exception e) {
    if (!graph.isOpen() || !g.tx().isOpen()) // push to fail queue
    // log the msg
}

But it still fails with same error msg: java.lang.NullPointerException

StackTrace:

[org.janusgraph.graphdb.tinkerpop.JanusGraphBlueprintsGraph.startNewTx(JanusGraphBlueprintsGraph.java:86), org.janusgraph.graphdb.tinkerpop.JanusGraphBlueprintsGraph.access$000(JanusGraphBlueprintsGraph.java:59), org.janusgraph.graphdb.tinkerpop.JanusGraphBlueprintsGraph$GraphTransaction.doOpen(JanusGraphBlueprintsGraph.java:292), org.apache.tinkerpop.gremlin.structure.util.AbstractTransaction.open(AbstractTransaction.java:95), com.connectioninsight.janusgraph.GraphConnections.index_core_profiles(GraphConnections.java:461), com.connectioninsight.janusgraph.Consumer.callback(Consumer.java:77), com.connectioninsight.janusgraph.Consumer.access$000(Consumer.java:16), com.connectioninsight.janusgraph.Consumer$1.handleDelivery(Consumer.java:51), com.rabbitmq.client.impl.ConsumerDispatcher$5.run(ConsumerDispatcher.java:149), com.rabbitmq.client.impl.ConsumerWorkService$WorkPoolRunnable.run(ConsumerWorkService.java:111), java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149), java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624), java.lang.Thread.run

I am unable to figure out the issue here ? and what is the best way to handle this issue ?

[sharma-kunal]

Hi, I found the issue, it was not related to JanusGraph. It was due to some internal health check that my application was doing. Sorry for the misunderstanding.