Hello!
One of my bolt in Storm is a module that adds edge on Neo4j. If I use
parallelism I have duplicates. Here's the code:
try {
Connection connection = getNeo4jConnection();
try {
Trip trip = (Trip) tuple.getValueByField("trip");
try (Statement stmt = connection.createStatement()) {
String query = "MATCH (u:Airport {name:'" +
trip.getOutgoingAirport() + "'}), (r:Airport {name:'" +
trip.getIngoingAirport() + "'})" +
" CREATE UNIQUE (u)-[:FLIGHT_TO { amount: '"+ 1 +"'
}]->(r)";
System.out.println("QUERY " + query);
stmt.execute(query);
}
} finally {
connection.close();
}
} catch (SQLException e) {
System.out.println(e.getMessage());
}
collector.ack(tuple);
I'm experimenting so don't mind the ugly try catch, it's not production
code. The connection is created with:
I'm experimenting so don't mind the ugly try catch, it's not production
code. The connection is created with:
private static Connection getNeo4jConnection() throws SQLException {
Driver driver = new org.neo4j.jdbc.Driver();
Properties properties = new Properties();
properties.put("user", "neo4j");
properties.put("password", "neo4j");
String url = "jdbc:neo4j://localhost:7474/";
return driver.connect(url, properties);}
As you can see a query is something like: MATCH (u:Airport {name:'CIA'}),
(r:Airport {name:'STN'}) CREATE UNIQUE (u)-[:FLIGHT_TO { amount: '1' }]->(r).
Now I would have expected, as you can read on the docs, that neo4j has acid
transaction so I would assume that I had no duplicates but I have them when
using concurrent bolts. I would assume it's a racing condition because if I
use 1 thread, it doesn't happen.
Could you help me to figure out what I'm doing wrong?
The version is 2.3.1
--
You received this message because you are subscribed to the Google Groups
"Neo4j" group.
To unsubscribe from this group and stop receiving emails from it, send an email
to [email protected].
For more options, visit https://groups.google.com/d/optout.
