Skip to content

Software Development Blogs: Programming, Software Testing, Agile Project Management

Methods & Tools

Subscribe to Methods & Tools
if you are not afraid to read more than one page to be a smarter software developer, software tester or project manager!

Mark Needham
Syndicate content
Thoughts on Software Development
Updated: 3 hours 57 min ago

Neo4j 2.0.0: Cypher – Index Hints and Neo.ClientError.Schema.NoSuchIndex

Fri, 01/31/2014 - 08:14

One of the features added into the more recent versions of Neo4j’s cypher query language is the ability to tell Cypher which index you’d like to use in your queries.

We’ll use the football dataset, so let’s start by creating an index on the ‘name’ property of nodes labelled ‘Player’:

CREATE INDEX ON :Player(name)

Let’s say we want to write a query to find ‘Wayne Rooney’ while explicitly using this index. We might start with the following query:

MATCH p
USING INDEX p:Player(name)
WHERE p.name = "Wayne Rooney"
RETURN p

If we run that we’ll see this error:

Cannot use index hint in this context. The label and property comparison must be specified on a non-optional node
Label: `Player`
Property name: `name`
Neo.ClientError.Schema.NoSuchIndex

We need to specify the label on the node in the ‘MATCH’ part of the query for our hint to work e.g.

MATCH (p:Player)
USING INDEX p:Player(name)
WHERE p.name = "Wayne Rooney"
RETURN p

Now we might decide that we want to find ‘Wayne Rooney’ or ‘Robin Van Persie’ so we change our query slightly:

MATCH (p:Player)
USING INDEX p:Player(name)
WHERE p.name = "Wayne Rooney" OR p.name = "Robin Van Persie"
RETURN p

But this one fails too!

Cannot use index hint in this context. The label and property comparison must be specified on a non-optional node
Label: `Player`
Property name: `name`
Neo.ClientError.Schema.NoSuchIndex

The problem here is that when you use ‘OR’ it’s currently not using an index but rather a label scan so the hint to use the index doesn’t make sense to Cypher.

The final way I’ve seen people get confused using index hints is when matching node properties inline e.g.

MATCH (p:Player {name: "Wayne Rooney"})
USING INDEX p:Player(name)
RETURN p

If we run that we’ll be back in the land of exceptions:

Cannot use index hint in this context. The label and property comparison must be specified on a non-optional node
Label: `Player`
Property name: `name`
Neo.ClientError.Schema.NoSuchIndex

We can work around that by pulling out the inline matching of the ‘name’ property”:

MATCH (p:Player)
USING INDEX p:Player(name)
WHERE p.name = "Wayne Rooney"
RETURN p
Categories: Programming

Java: Work out the serialVersionUID of a class

Fri, 01/31/2014 - 07:51

Earlier in the week I wanted to work out the serialVersionUID of a serializable class so that I could override its toString method without breaking everything.

I came across Frank Kim’s blog post which suggested using the serialver tool which comes with the JDK.

I created a little Maven project to test this tool out on a very simple class:

import java.io.Serializable;
 
public class SerialiseMe implements Serializable
{
 
}

If we compile that class into a JAR and then run the serialver tool we see the following output:

$ serialver -classpath target/serialiser-0.0.1-SNAPSHOT.jar SerialiseMe
SerialiseMe:    static final long serialVersionUID = -6060222249255158490L;

I wanted to quickly confirm that I could serialise and deserialise this class using this value so I wrote the following bit of code to serialise the class (when it didn’t have a serial version UID):

public class Serialiser
{
    public static void main( String[] args ) throws IOException, ClassNotFoundException
    {
        ByteArrayOutputStream bout = new ByteArrayOutputStream(  );
        ObjectOutputStream oout = new ObjectOutputStream( bout );
 
        Object value = new SerialiseMe();
 
        oout.writeObject( value );
        oout.close();
        byte[] bytes = bout.toByteArray();
 
        FileOutputStream fileOuputStream = new FileOutputStream("/tmp/foo.txt");
        fileOuputStream.write(bytes);
        fileOuputStream.close();
    }
}

After I’d done that, I wrote the following bit of code to deserialise the file:

public class Deserialiser
{
    public static void main( String[] args ) throws IOException, ClassNotFoundException
    {
        FileInputStream fileInputStream = new FileInputStream( new File( "/tmp/foo.txt" ) );
        byte[] bytes = IOUtils.toByteArray( fileInputStream );
 
        ByteArrayInputStream in = new ByteArrayInputStream( bytes, 0, bytes.length );
        ObjectInputStream oin = new ObjectInputStream( in );
        Object object = oin.readObject();
    }
}

I plugged the serial version UID into the class and was able to deserialise it correctly. I tried changing one of the digits just to check it would blow up and indeed it did:

import java.io.Serializable;
 
public class SerialiseMe implements Serializable
{
    static final long serialVersionUID = -6060222249255158491L;
}
Exception in thread "main" java.io.InvalidClassException: SerialiseMe; local class incompatible: stream classdesc serialVersionUID = -6060222249255158490, local class serialVersionUID = -6060222249255158491
	at java.io.ObjectStreamClass.initNonProxy(ObjectStreamClass.java:604)
	at java.io.ObjectInputStream.readNonProxyDesc(ObjectInputStream.java:1620)
	at java.io.ObjectInputStream.readClassDesc(ObjectInputStream.java:1515)
	at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1769)
	at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1348)
	at java.io.ObjectInputStream.readObject(ObjectInputStream.java:370)
	at Deserialiser.main(Deserialiser.java:18)
	at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
	at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)
	at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
	at java.lang.reflect.Method.invoke(Method.java:601)
	at com.intellij.rt.execution.application.AppMain.main(AppMain.java:120)

serialver #ftw!

Categories: Programming

Neo4j: org.eclipse.jetty.io.EofException – Caused by: java.io.IOException: Broken pipe

Mon, 01/27/2014 - 12:32

From scouring the Neo4j google group and Stack Overflow I’ve noticed that a few people have been hitting the following exception when executing queries against Neo4j server:

SEVERE: The response of the WebApplicationException cannot be utilized as the response is already committed. Re-throwing to the HTTP container
javax.ws.rs.WebApplicationException: javax.ws.rs.WebApplicationException: org.eclipse.jetty.io.EofException
	at org.neo4j.server.rest.repr.OutputFormat$1.write(OutputFormat.java:174)
	at com.sun.jersey.core.impl.provider.entity.StreamingOutputProvider.writeTo(StreamingOutputProvider.java:71)
	at com.sun.jersey.core.impl.provider.entity.StreamingOutputProvider.writeTo(StreamingOutputProvider.java:57)
	at com.sun.jersey.spi.container.ContainerResponse.write(ContainerResponse.java:306)
	at com.sun.jersey.server.impl.application.WebApplicationImpl._handleRequest(WebApplicationImpl.java:1437)
	at com.sun.jersey.server.impl.application.WebApplicationImpl.handleRequest(WebApplicationImpl.java:1349)
	at com.sun.jersey.server.impl.application.WebApplicationImpl.handleRequest(WebApplicationImpl.java:1339)
	at com.sun.jersey.spi.container.servlet.WebComponent.service(WebComponent.java:416)
	at com.sun.jersey.spi.container.servlet.ServletContainer.service(ServletContainer.java:537)
	at com.sun.jersey.spi.container.servlet.ServletContainer.service(ServletContainer.java:699)
	at javax.servlet.http.HttpServlet.service(HttpServlet.java:848)
	at org.eclipse.jetty.servlet.ServletHolder.handle(ServletHolder.java:698)
	at org.eclipse.jetty.servlet.ServletHandler$CachedChain.doFilter(ServletHandler.java:1506)
	at org.neo4j.server.rest.security.SecurityFilter.doFilter(SecurityFilter.java:112)
	at org.eclipse.jetty.servlet.ServletHandler$CachedChain.doFilter(ServletHandler.java:1477)
	at org.eclipse.jetty.servlet.ServletHandler.doHandle(ServletHandler.java:503)
	at org.eclipse.jetty.server.session.SessionHandler.doHandle(SessionHandler.java:211)
	at org.eclipse.jetty.server.handler.ContextHandler.doHandle(ContextHandler.java:1096)
	at org.eclipse.jetty.servlet.ServletHandler.doScope(ServletHandler.java:432)
	at org.eclipse.jetty.server.session.SessionHandler.doScope(SessionHandler.java:175)
	at org.eclipse.jetty.server.handler.ContextHandler.doScope(ContextHandler.java:1030)
	at org.eclipse.jetty.server.handler.ScopedHandler.handle(ScopedHandler.java:136)
	at org.eclipse.jetty.server.handler.HandlerList.handle(HandlerList.java:52)
	at org.eclipse.jetty.server.handler.HandlerWrapper.handle(HandlerWrapper.java:97)
	at org.eclipse.jetty.server.Server.handle(Server.java:445)
	at org.eclipse.jetty.server.HttpChannel.handle(HttpChannel.java:268)
	at org.eclipse.jetty.server.HttpConnection.onFillable(HttpConnection.java:229)
	at org.eclipse.jetty.io.AbstractConnection$ReadCallback.run(AbstractConnection.java:358)
	at org.eclipse.jetty.util.thread.QueuedThreadPool.runJob(QueuedThreadPool.java:601)
	at org.eclipse.jetty.util.thread.QueuedThreadPool$3.run(QueuedThreadPool.java:532)
	at java.lang.Thread.run(Thread.java:744)
Caused by: javax.ws.rs.WebApplicationException: org.eclipse.jetty.io.EofException
	at org.neo4j.server.rest.repr.formats.StreamingJsonFormat$StreamingListWriter.writeValue(StreamingJsonFormat.java:316)
	at org.neo4j.server.rest.repr.ListWriter.writeValue(ListWriter.java:75)
	at org.neo4j.server.rest.repr.ValueRepresentation.addTo(ValueRepresentation.java:49)
	at org.neo4j.server.rest.repr.ListRepresentation.serialize(ListRepresentation.java:65)
	at org.neo4j.server.rest.repr.Serializer.serialize(Serializer.java:75)
	at org.neo4j.server.rest.repr.MappingSerializer.putList(MappingSerializer.java:61)
	at org.neo4j.server.rest.repr.ListRepresentation.putTo(ListRepresentation.java:85)
	at org.neo4j.server.rest.repr.ObjectRepresentation$PropertyGetter.putTo(ObjectRepresentation.java:133)
	at org.neo4j.server.rest.repr.ObjectRepresentation.serialize(ObjectRepresentation.java:144)
	at org.neo4j.server.rest.repr.MappingRepresentation.serialize(MappingRepresentation.java:41)
	at org.neo4j.server.rest.repr.OutputFormat$1.write(OutputFormat.java:160)
	... 30 more
Caused by: org.eclipse.jetty.io.EofException
	at org.eclipse.jetty.io.ChannelEndPoint.flush(ChannelEndPoint.java:186)
	at org.eclipse.jetty.io.WriteFlusher.write(WriteFlusher.java:335)
	at org.eclipse.jetty.io.AbstractEndPoint.write(AbstractEndPoint.java:125)
	at org.eclipse.jetty.server.HttpConnection$ContentCallback.process(HttpConnection.java:784)
	at org.eclipse.jetty.util.IteratingCallback.iterate(IteratingCallback.java:79)
	at org.eclipse.jetty.server.HttpConnection.send(HttpConnection.java:356)
	at org.eclipse.jetty.server.HttpChannel.sendResponse(HttpChannel.java:631)
	at org.eclipse.jetty.server.HttpChannel.write(HttpChannel.java:661)
	at org.eclipse.jetty.server.HttpOutput.write(HttpOutput.java:198)
	at com.sun.jersey.spi.container.servlet.WebComponent$Writer.write(WebComponent.java:307)
	at com.sun.jersey.spi.container.ContainerResponse$CommittingOutputStream.write(ContainerResponse.java:134)
	at org.codehaus.jackson.impl.Utf8Generator._flushBuffer(Utf8Generator.java:1754)
	at org.codehaus.jackson.impl.Utf8Generator.writeNumber(Utf8Generator.java:886)
	at org.codehaus.jackson.map.ser.StdSerializers$LongSerializer.serialize(StdSerializers.java:170)
	at org.codehaus.jackson.map.ser.StdSerializers$LongSerializer.serialize(StdSerializers.java:158)
	at org.codehaus.jackson.map.ser.StdSerializerProvider._serializeValue(StdSerializerProvider.java:610)
	at org.codehaus.jackson.map.ser.StdSerializerProvider.serializeValue(StdSerializerProvider.java:256)
	at org.codehaus.jackson.map.ObjectMapper.writeValue(ObjectMapper.java:1613)
	at org.codehaus.jackson.impl.JsonGeneratorBase.writeObject(JsonGeneratorBase.java:314)
	at org.neo4j.server.rest.repr.formats.StreamingJsonFormat$StreamingListWriter.writeValue(StreamingJsonFormat.java:312)
	... 40 more
Caused by: java.io.IOException: Broken pipe
	at sun.nio.ch.FileDispatcherImpl.writev0(Native Method)
	at sun.nio.ch.SocketDispatcher.writev(SocketDispatcher.java:51)
	at sun.nio.ch.IOUtil.write(IOUtil.java:148)
	at sun.nio.ch.SocketChannelImpl.write(SocketChannelImpl.java:524)
	at org.eclipse.jetty.io.ChannelEndPoint.flush(ChannelEndPoint.java:167)
	... 59 more

I’d not come across it myself but the error message suggests that the client has closed the connection while the server is still trying to write out a response.

This suggests that you need to write a query which runs for a long time. I created the following data set to try and force this to happen:

public class DenseNodes
{
    private static final DynamicRelationshipType FOO = DynamicRelationshipType.withName( "FOO" );
    private static final Label BAR = DynamicLabel.label( "Bar" );
 
    public static void main( String[] args ) throws IOException
    {
        String path = "/tmp/dense";
 
        FileUtils.deleteRecursively( new File( path ) );
 
        GraphDatabaseService db = new GraphDatabaseFactory().newEmbeddedDatabase( path );
 
        for ( int i = 0; i < 10; i++ )
        {
            try ( Transaction tx = db.beginTx() )
            {
                Node node = db.createNode( BAR );
 
                for ( int j = 0; j < 100_000; j++ )
                {
                    node.createRelationshipTo( db.createNode(), FOO );
                }
 
                tx.success();
            }
        }
    }
}

This script creates 10 nodes with 100,000 relationships and as a by-product we have 1,000,000 nodes with 1 relationship.

We can execute the following query using a Jersey client to take us over the timeout and force the EOFException:

public class DenseMe
{
    public static void main( String[] args ) throws IOException
    {
        String query = "MATCH (a:Bar)-[:`FOO`]->(b) RETURN a,b";
 
        long start = System.currentTimeMillis();
        executeViaHTTP( query );
        long end = System.currentTimeMillis();
 
        System.out.println(end - start);
    }
 
    private static void executeViaHTTP( String query ) throws IOException
    {
        ObjectNode entity = JsonNodeFactory.instance.objectNode();
        entity.put("query", query );
 
        ClientResponse response = client().resource( "http://localhost:7474/db/data/cypher" )
                .accept( MediaType.APPLICATION_JSON_TYPE )
                .type(MediaType.APPLICATION_JSON_TYPE)
                .post( ClientResponse.class, entity );
 
        InputStream stream = response.getEntityInputStream();
 
        BufferedReader reader = new BufferedReader( new InputStreamReader( stream ) );
        char[] buffer = new char[1024];
        int bytesRead;
        while ( (bytesRead = reader.read( buffer )) != -1 )
        {
            for ( int i = 0; i < bytesRead; i++ )
            {
                System.out.print( buffer[i] );
            }
        }
    }
 
    private static Client client()
    {
        DefaultClientConfig defaultClientConfig = new DefaultClientConfig();
        defaultClientConfig.getClasses().add( JacksonJsonProvider.class );
        Client client = Client.create( defaultClientConfig );
        return client;
    }
}

There are two ways that we can attempt to work around this problem:

  1. Increase the timeout on the Jersey Client
  2. Stream the response to the client so we don’t build up such a huge response on the server

We’ll start with the former, which requires the following tweak to the client function:

private static Client client()
{
    DefaultClientConfig defaultClientConfig = new DefaultClientConfig();
    defaultClientConfig.getClasses().add( JacksonJsonProvider.class );
    Client client = Client.create( defaultClientConfig );
    client.setConnectTimeout(1_000_000);
    client.setReadTimeout( 1_000_000 );
    return client;
}

If we run that we’ll end up with the following response using a 4GB heap:

{
  "message" : "Java heap space",
  "exception" : "OutOfMemoryError",
  "fullname" : "java.lang.OutOfMemoryError",
  "stacktrace" : [ ]
}

I was tailing the GC logs while running the query and the majority of time was being spent in Full GC while the query was running:

2014-01-27T10:27:26.101+0000: 239848.812: Total time for which application threads were stopped: 5.5309550 seconds
2014-01-27T10:27:26.101+0000: 239848.812: [Full GC2014-01-27T10:27:26.101+0000: 239848.812: [CMS: 3512768K->3512768K(3512768K), 5.4359920 secs] 4126207K->4126207K(4126208K), [CMS Perm : 41512K->41512K(69300K)], 5.4360820 secs] [Times: user=5.43 sys=0.00, real=5.43 secs]
2014-01-27T10:27:31.537+0000: 239854.249: [Full GC2014-01-27T10:27:31.537+0000: 239854.249: [CMS: 3512768K->3512768K(3512768K), 5.4878690 secs] 4126207K->4126207K(4126208K), [CMS Perm : 41512K->41512K(69300K)], 5.4879470 secs] [Times: user=5.49 sys=0.01, real=5.49 secs]
2014-01-27T10:27:37.025+0000: 239859.737: Total time for which application threads were stopped: 10.9243140 seconds
2014-01-27T10:27:37.025+0000: 239859.737: [Full GC2014-01-27T10:27:37.025+0000: 239859.737: [CMS: 3512768K->3512768K(3512768K), 5.4437040 secs] 4126207K->4126207K(4126208K), [CMS Perm : 41512K->41512K(69300K)], 5.4437790 secs] [Times: user=5.44 sys=0.01, real=5.44 secs]
2014-01-27T10:27:42.469+0000: 239865.181: [Full GC2014-01-27T10:27:42.469+0000: 239865.181: [CMS: 3512768K->3512768K(3512768K), 5.4283480 secs] 4126207K->4126207K(4126208K), [CMS Perm : 41512K->41512K(69300K)], 5.4284400 secs] [Times: user=5.43 sys=0.00, real=5.43 secs]
2014-01-27T10:27:47.898+0000: 239870.609: Total time for which application threads were stopped: 10.8724950 seconds
2014-01-27T10:27:47.898+0000: 239870.609: [Full GC2014-01-27T10:27:47.898+0000: 239870.609: [CMS: 3512768K->3512768K(3512768K), 5.4385630 secs] 4126208K->4126207K(4126208K), [CMS Perm : 41512K->41512K(69300K)], 5.4386540 secs] [Times: user=5.43 sys=0.01, real=5.44 secs]
2014-01-27T10:27:53.337+0000: 239876.048: Total time for which application threads were stopped: 5.4389110 seconds

The second option is to stream back the response by adding the following header to our request:

ClientResponse response = client().resource( "http://localhost:7474/db/data/cypher" )
    .accept( MediaType.APPLICATION_JSON_TYPE )
    .type(MediaType.APPLICATION_JSON_TYPE)
    .header( "X-Stream", "true" )
    .post( ClientResponse.class, entity );

If we do that then we’ll get back the first rows of the query immediately although although we’ll have to be careful with what we do with the response or we could see an OutOfMemory exception on the client instead.

We might also want to think whether we actually need to return that many rows in the first place. A lot of the time a subset is more than enough.

Categories: Programming

Neo4j HA: org.neo4j.graphdb.TransactionFailureException: Timeout waiting for database to allow new transactions. Blocking components (1): []

Mon, 01/27/2014 - 10:42

As I mentioned in my previous post, I’ve been spending quite a bit of time working with Neo4j HA and recently came across the following exception in data/graph.db/messages.log:

org.neo4j.graphdb.TransactionFailureException: Timeout waiting for database to allow new transactions. Blocking components (1): []
	at org.neo4j.kernel.ha.HighlyAvailableGraphDatabase.beginTx(HighlyAvailableGraphDatabase.java:199)
	at org.neo4j.kernel.TransactionBuilderImpl.begin(TransactionBuilderImpl.java:43)
	at org.neo4j.kernel.InternalAbstractGraphDatabase.beginTx(InternalAbstractGraphDatabase.java:949)
	at org.neo4j.server.rest.transactional.TransactionalRequestDispatcher.dispatch(TransactionalRequestDispatcher.java:52)
	at com.sun.jersey.server.impl.uri.rules.HttpMethodRule.accept(HttpMethodRule.java:288)
	at com.sun.jersey.server.impl.uri.rules.ResourceClassRule.accept(ResourceClassRule.java:108)
	at com.sun.jersey.server.impl.uri.rules.RightHandPathRule.accept(RightHandPathRule.java:147)
	at com.sun.jersey.server.impl.uri.rules.RootResourceClassesRule.accept(RootResourceClassesRule.java:84)
	at com.sun.jersey.server.impl.application.WebApplicationImpl._handleRequest(WebApplicationImpl.java:1469)
	at com.sun.jersey.server.impl.application.WebApplicationImpl._handleRequest(WebApplicationImpl.java:1400)
	at com.sun.jersey.server.impl.application.WebApplicationImpl.handleRequest(WebApplicationImpl.java:1349)
	at com.sun.jersey.server.impl.application.WebApplicationImpl.handleRequest(WebApplicationImpl.java:1339)
	at com.sun.jersey.spi.container.servlet.WebComponent.service(WebComponent.java:416)
	at com.sun.jersey.spi.container.servlet.ServletContainer.service(ServletContainer.java:537)
	at com.sun.jersey.spi.container.servlet.ServletContainer.service(ServletContainer.java:699)
	at javax.servlet.http.HttpServlet.service(HttpServlet.java:848)
	at org.eclipse.jetty.servlet.ServletHolder.handle(ServletHolder.java:698)
	at org.eclipse.jetty.servlet.ServletHandler$CachedChain.doFilter(ServletHandler.java:1506)
	at org.neo4j.server.rest.security.SecurityFilter.doFilter(SecurityFilter.java:112)
	at org.eclipse.jetty.servlet.ServletHandler$CachedChain.doFilter(ServletHandler.java:1477)
	at org.eclipse.jetty.servlet.ServletHandler.doHandle(ServletHandler.java:503)
	at org.eclipse.jetty.server.session.SessionHandler.doHandle(SessionHandler.java:211)
	at org.eclipse.jetty.server.handler.ContextHandler.doHandle(ContextHandler.java:1096)
	at org.eclipse.jetty.servlet.ServletHandler.doScope(ServletHandler.java:432)
	at org.eclipse.jetty.server.session.SessionHandler.doScope(SessionHandler.java:175)
	at org.eclipse.jetty.server.handler.ContextHandler.doScope(ContextHandler.java:1030)
	at org.eclipse.jetty.server.handler.ScopedHandler.handle(ScopedHandler.java:136)
	at org.eclipse.jetty.server.handler.HandlerList.handle(HandlerList.java:52)
	at org.eclipse.jetty.server.handler.HandlerWrapper.handle(HandlerWrapper.java:97)
	at org.eclipse.jetty.server.Server.handle(Server.java:445)
	at org.eclipse.jetty.server.HttpChannel.handle(HttpChannel.java:268)
	at org.eclipse.jetty.server.HttpConnection.onFillable(HttpConnection.java:229)
	at org.eclipse.jetty.io.AbstractConnection$ReadCallback.run(AbstractConnection.java:358)
	at org.eclipse.jetty.util.thread.QueuedThreadPool.runJob(QueuedThreadPool.java:601)
	at org.eclipse.jetty.util.thread.QueuedThreadPool$3.run(QueuedThreadPool.java:532)
	at java.lang.Thread.run(Thread.java:744)

I traced the error message back to AvailabilityGuard#describeWhoIsBlocking and discovered that this error message is thrown when we don’t want the cluster to accept transactions.

Usually the blocking component (the thing that’s stopping transactions from being accepted) is described but in this case the error message indicates that the cluster hasn’t been formed.

I was able to reproduce the error message by trying to commit a transaction from a slave in a cluster which was misconfigured so that the slaves were unable to interact with the master.

To find the root cause for this error message we need to scroll up the log a bit and look for the exception that happened immediately before this one. In this case we see the following:

2014-01-24 11:48:53.889+0000 WARN  [o.n.k.h.c.HighAvailabilityModeSwitcher]: Consistency checker failed
org.neo4j.com.ComException: MasterClient20 could not connect to /10.137.62.4:6001
	at org.neo4j.com.Client$1.create(Client.java:141) ~[neo4j-com-2.0.0.jar:2.0.0]
	at org.neo4j.com.Client$1.create(Client.java:123) ~[neo4j-com-2.0.0.jar:2.0.0]
	at org.neo4j.com.ResourcePool.acquire(ResourcePool.java:175) ~[neo4j-com-2.0.0.jar:2.0.0]
	at org.neo4j.com.Client.getChannel(Client.java:336) ~[neo4j-com-2.0.0.jar:2.0.0]
	at org.neo4j.com.Client.sendRequest(Client.java:216) ~[neo4j-com-2.0.0.jar:2.0.0]
	at org.neo4j.kernel.ha.MasterClient20.getMasterIdForCommittedTx(MasterClient20.java:359) ~[neo4j-ha-2.0.0.jar:2.0.0]
	at org.neo4j.kernel.ha.cluster.HighAvailabilityModeSwitcher.checkDataConsistencyWithMaster(HighAvailabilityModeSwitcher.java:679) [neo4j-ha-2.0.0.jar:2.0.0]
	at org.neo4j.kernel.ha.cluster.HighAvailabilityModeSwitcher.checkDataConsistency(HighAvailabilityModeSwitcher.java:498) [neo4j-ha-2.0.0.jar:2.0.0]
	at org.neo4j.kernel.ha.cluster.HighAvailabilityModeSwitcher.access$900(HighAvailabilityModeSwitcher.java:110) [neo4j-ha-2.0.0.jar:2.0.0]
	at org.neo4j.kernel.ha.cluster.HighAvailabilityModeSwitcher$2.run(HighAvailabilityModeSwitcher.java:393) [neo4j-ha-2.0.0.jar:2.0.0]
	at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:471) [na:1.7.0_51]
	at java.util.concurrent.FutureTask.run(FutureTask.java:262) [na:1.7.0_51]
	at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:178) [na:1.7.0_51]
	at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:292) [na:1.7.0_51]
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145) [na:1.7.0_51]
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615) [na:1.7.0_51]
	at java.lang.Thread.run(Thread.java:744) [na:1.7.0_51]

A quick configuration change was able to get me going again but you could also see this error message if firewall rules were misconfigured so that’s something else to look out for.

Categories: Programming

Neo4j HA: Election could not pick a winner

Fri, 01/24/2014 - 11:30

Recently I’ve been spending a reasonable chunk of my time helping people get up and running with their Neo4j High Availability cluster and there’s sometimes confusion around how it should be configured.

A Neo4j cluster typically consists of a master and two slaves and you’d usually have it configured so that any machine can be the master.

However, there is a configuration parameter ‘ha.slave_only’ which can be set to ‘true’ to ensure that a machine will never be elected as master when an election takes place.

We might configure our machine with that setting if it’s acting as a reporting instance but we need to make sure that 2 members don’t have that setting otherwise we won’t have any failover in the cluster.

For example, if we set two of the machines in the cluster to be slave only and then stop the master we’ll see output similar to the following in data/graph.db/messages.log:

2014-01-23 11:17:24.510+0000 INFO  [o.n.c.p.a.m.MultiPaxosContext$ElectionContextImpl]: Doing elections for role coordinator
2014-01-23 11:17:24.510+0000 DEBUG [o.n.c.p.e.ElectionState$2]: ElectionState: election-[performRoleElections]->election from:cluster://10.239.8.251:5001 conversation-id:3/13#
2014-01-23 11:17:24.513+0000 DEBUG [o.n.c.p.e.ElectionState$2]: ElectionState: election-[vote:coordinator]->election from:cluster://10.151.24.237:5001 conversation-id:3/13#
2014-01-23 11:17:24.515+0000 DEBUG [o.n.c.p.e.ElectionState$2]: ElectionState: election-[voted]->election from:cluster://10.138.29.197:5001 conversation-id:3/13#
2014-01-23 11:17:24.516+0000 DEBUG [o.n.c.p.e.ElectionState$2]: ElectionState: election-[voted]->election from:cluster://10.151.24.237:5001 conversation-id:3/13#
2014-01-23 11:17:24.519+0000 DEBUG [o.n.c.p.a.m.MultiPaxosContext$ElectionContextImpl$2]: Elections ended up with list []
2014-01-23 11:17:24.519+0000 WARN  [o.n.c.p.e.ElectionState]: Election could not pick a winner

This message initially looks confusing but what it’s telling us is that the cluster was unable to elect a new master, in this case because there were no machines that could be elected as master.

So if you see that message in your logs, check your config to make sure that there are actually machines to choose from.

Categories: Programming

Neo4j Backup: Store copy and consistency check

Wed, 01/22/2014 - 18:36

One of the lesser known things about the Neo4j online backup tool, which I wrote about last week, is that conceptually there are two parts to it:

  1. Copying the store files to a location of your choice
  2. Verifying that those store files are consistent.

By default both of these run when you run the ‘neo4j-backup’ script but sometimes it’s useful to be able to run them separately.

If we want to just run the copying the store files part of the process we can tell the backup tool to skip the consistency check by using the ‘verify‘ flag:

$ pwd
/Users/markneedham/Downloads/neo4j-enterprise-2.0.0
$ ./bin/neo4j-backup -from single://127.0.0.1 -to /tmp/foo -verify false
Performing full backup from 'single://127.0.0.1'
Files copied
................        done
Done

If we ran that without the ‘verify’ flag we’d see the output of the consistency checker as well:

$ ./bin/neo4j-backup -from single://127.0.0.1 -to /tmp/foo
Performing full backup from 'single://127.0.0.1'
Files copied
................        done
Full consistency check
....................  10%
....................  20%
....................  30%
....................  40%
....................  50%
....................  60%
....................  70%
....................  80%
....................  90%
.................... 100%
Done

If we already have a backup and only want to run the consistency checker we can run the following command:

$ java -cp 'lib/*:system/lib/*' org.neo4j.consistency.ConsistencyCheckTool /tmp/foo
Full consistency check
....................  10%
....................  20%
....................  30%
....................  40%
....................  50%
....................  60%
....................  70%
....................  80%
....................  90%
.................... 100%

The consistency tool itself takes a ‘config‘ flag which gives you some control over what things you want to consistency check.

The various options are defined in org.neo4j.consistency.ConsistencyCheckSettings.

For example, if we want to change the file that the consistency check report is written to we could add the following property to our config file:

$ tail -n 1 conf/neo4j.properties
consistency_check_report_file=/tmp/foo.txt

And then run the consistency tool like so:

$ java -cp 'lib/*:system/lib/*' org.neo4j.consistency.ConsistencyCheckTool -config conf/neo4j.properties /tmp/foo

If there are any inconsistencies they’ll now be written to that file rather than to a file in the store directory.

You can also pass that ‘config’ flag to the backup tool and it will make use of it when it runs the consistency check. e.g.

$ ./bin/neo4j-backup -from single://127.0.0.1 -to /tmp/foo -verify false -config conf/neo4j.properties

Most of the time you don’t need to worry too much about either of these commands but I always forget what the various options are so I thought I’d better write it up while it’s fresh in my mind.

Categories: Programming

Neo4j Backup: java.lang.ClassCastException: org.jboss.netty.buffer.BigEndianHeapChannelBuffer cannot be cast to org.neo4j.cluster.com.message.Message

Sun, 01/19/2014 - 20:29

When using Neo4j’s online backup facility there are two ways of triggering it, either by using the ‘single://‘ or ‘ha://‘ syntax and these behave slightly differently.

If you’re using the ‘single://’ syntax and don’t specify a port then it will connect to ’6362′ by default:

./neo4j-backup -from single://192.168.1.34 -to /mnt/backup/neo4j-backup

If you’ve changed the backup port via the ‘online_backup_server’ property in conf/neo4j.properties you’ll need to set the port explicitly:

online_backup_server=192.168.1.34:6363
./neo4j-backup -from single://192.168.1.34:6363 -to /mnt/backup/neo4j-backup

If you’re using the ‘ha://’ syntax then the backup client joins the HA cluster, works out which machine is the master and then creates a backup from that machine.

In order for the backup client to join the cluster it connects to port ’5001′ by default:

./neo4j-backup -from ha://192.168.1.34 -to /mnt/backup/neo4j-backup

If you’ve changed the ‘ha.cluster_server’ property then you’ll need to set the port explicitly:

ha.cluster_server=192.168.1.34:5002
./neo4j-backup -from ha://192.168.1.34:5002 -to /mnt/backup/neo4j-backup

A mistake that I made when first using this utility was to use the ‘ha://’ syntax with the backup port. e.g.

./neo4j-backup -from ha://192.168.1.34:6362 -to /mnt/backup/neo4j-backup

If you do this you’ll end up with the following exception:

2014-01-19 19:24:30.842+0000 ERROR [o.n.c.c.NetworkSender]: Receive exception:
java.lang.ClassCastException: org.jboss.netty.buffer.BigEndianHeapChannelBuffer cannot be cast to org.neo4j.cluster.com.message.Message
	at org.neo4j.cluster.com.NetworkSender$NetworkMessageSender.messageReceived(NetworkSender.java:409) ~[neo4j-cluster-2.0.0.jar:2.0.0]
	at org.jboss.netty.channel.Channels.fireMessageReceived(Channels.java:268) ~[netty-3.6.3.Final.jar:na]
	at org.jboss.netty.channel.Channels.fireMessageReceived(Channels.java:255) ~[netty-3.6.3.Final.jar:na]
	at org.jboss.netty.channel.socket.nio.NioWorker.read(NioWorker.java:88) ~[netty-3.6.3.Final.jar:na]
	at org.jboss.netty.channel.socket.nio.AbstractNioWorker.process(AbstractNioWorker.java:107) ~[netty-3.6.3.Final.jar:na]
	at org.jboss.netty.channel.socket.nio.AbstractNioSelector.run(AbstractNioSelector.java:312) ~[netty-3.6.3.Final.jar:na]
	at org.jboss.netty.channel.socket.nio.AbstractNioWorker.run(AbstractNioWorker.java:88) ~[netty-3.6.3.Final.jar:na]
	at org.jboss.netty.channel.socket.nio.NioWorker.run(NioWorker.java:178) ~[netty-3.6.3.Final.jar:na]
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145) [na:1.7.0_45]
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615) [na:1.7.0_45]
	at java.lang.Thread.run(Thread.java:744) [na:1.7.0_45]

Let me know in the comments if any of this doesn’t make sense. There are lots of other examples to follow on neo4j-backup’s man page in the manual as well.

Categories: Programming

Learning about bitmaps

Sun, 01/12/2014 - 18:44

A few weeks ago Alistair and I were working on the code used to model the labels that a node has attached to it in a Neo4j database.

The way this works is that chunks of 32 nodes ids are represented as a 32 bit bitmap for each label where a 1 for a bit means that a node has the label and a 0 means that it doesn’t.

For example, let’s say we have node ids 0-31 where 0 is the highest bit and 31 is the lowest bit. If only node 0 has the label then that’d be represented as the following value:

java> int bitmap = 1 << 31;
int bitmap = -2147483648

If we imagine the 32 bits positioned next to each other it would look like this:

2014 01 12 15 45 16
java> 0X80000000;
Integer res16 = -2147483648

The next thing we want to do is work out whether a node has a label applied or not. We can do this by using a bitwise AND.

For example to check whether the highest bit is set we would write the following code:

java> bitmap & (1 << 31);
Integer res10 = -2147483648

That is set as we would imagine. Now let’s check a a few bits that we know aren’t set:

java> bitmap & (1 << 0);
Integer res11 = 0
 
java> bitmap & (1 << 1);
Integer res12 = 0
 
java> bitmap & (1 << 30);
Integer res13 = 0

Another operation we might want to do is set another bit on our existing bitmap for which we can use a bitwise inclusive OR.

A bitwise inclusive OR means that a bit will be set if either value has the bit set or if both have it set.

Let’s set the second highest bit. and visualise that calculation:

2014 01 12 15 45 16

If we evaluate that we’d expect the two highest bits to be set:

java> bitmap |= (1 << 30);
Integer res14 = -1073741824

Now if we visualise the bitmap we’ll see that is indeed the case:

2014 01 12 17 16 21
java> 0XC0000000;
Integer res15 = -1073741824

The next operation we want to do is to unset a bit that we’re already set for which we can use a bitwise exclusive OR.

An exclusive OR means that a bit will only remain set if there’s a combination of (0 and 1) or (1 and 0) in the calculation. If there are two 1′s or 2 0′s then it’ll be unset.

Let’s unset the 2nd highest bit so that we’re left with just the top bit being set.

If we visualise that we have the following calculation:

2014 01 12 17 33 21

And if we evaluate that we’re back to our original bitmap:

java> bitmap ^= (1 << 30);
Integer res2 = -2147483648

I used the Java REPL to evaluate the code samples in this post and this article explains bitshift operators very clearly.

The Neo4j version of the bitmap described in this post is in the BitmapFormat class on github.

Categories: Programming

RxJava: From Future to Observable

Sat, 12/28/2013 - 22:46

I first came across Reactive Extensions about 4 years ago on Matthew Podwysocki’s blog but then haven’t heard much about it until I saw Matthew give a talk at Code Mesh a few weeks ago.

It seems to have grown in popularity recently and I noticed that’s there’s now a Java version called RxJava written by Netflix.

I thought I’d give it a try by changing some code I wrote while exploring cypher’s MERGE function to expose an Observable instead of Futures.

To recap, we have 50 threads and we do 100 iterations where we create random (user, event) pairs. We create a maximum of 10 users and 50 events and the goal is to concurrently send requests for the same pairs.

In the example of my other post I was throwing away the result of each query whereas here I returned the result back so I had something to subscribe to.

The outline of the code looks like this:

public class MergeTimeRx
{
    public static void main( final String[] args ) throws InterruptedException, IOException
    {
        String pathToDb = "/tmp/foo";
        FileUtils.deleteRecursively( new File( pathToDb ) );
 
        GraphDatabaseService db = new GraphDatabaseFactory().newEmbeddedDatabase( pathToDb );
        final ExecutionEngine engine = new ExecutionEngine( db );
 
        int numberOfThreads = 50;
        int numberOfUsers = 10;
        int numberOfEvents = 50;
        int iterations = 100;
 
        Observable<ExecutionResult> events = processEvents( engine, numberOfUsers, numberOfEvents, numberOfThreads, iterations );
 
        events.subscribe( new Action1<ExecutionResult>()
        {
            @Override
            public void call( ExecutionResult result )
            {
                for ( Map<String, Object> row : result )
                {
                }
            }
        } );
 
        ....
    }
 
}

The nice thing about using RxJava is that there’s no mention of how we got our collection of ExecutionResults, it’s not important. We just have a stream of them and by calling the subscribe function on the Observable we’ll be informed whenever another one is made available.

Most of the examples I found show how to generate events from a single thread but I wanted to use a thread pool so that I could fire off lots of requests at the same time. The processEvents method ended up looking like this:

    private static Observable<ExecutionResult> processEvents( final ExecutionEngine engine, final int numberOfUsers, final int numberOfEvents, final int numberOfThreads, final int iterations )
    {
        final Random random = new Random();
        final List<Integer> userIds = generateIds( numberOfUsers );
        final List<Integer> eventIds = generateIds( numberOfEvents );
 
        return Observable.create( new Observable.OnSubscribeFunc<ExecutionResult>()
        {
            @Override
            public Subscription onSubscribe( final Observer<? super ExecutionResult> observer )
            {
                final ExecutorService executor = Executors.newFixedThreadPool( numberOfThreads );
 
                List<Future<ExecutionResult>> jobs = new ArrayList<>();
                for ( int i = 0; i < iterations; i++ )
                {
                    Future<ExecutionResult> job = executor.submit( new Callable<ExecutionResult>()
                    {
                        @Override
                        public ExecutionResult call()
                        {
                            Integer userId = userIds.get( random.nextInt( numberOfUsers ) );
                            Integer eventId = eventIds.get( random.nextInt( numberOfEvents ) );
 
                            return engine.execute(
                                    "MERGE (u:User {id: {userId}})\n" +
                                    "MERGE (e:Event {id: {eventId}})\n" +
                                    "MERGE (u)-[:HAS_EVENT]->(e)\n" +
                                    "RETURN u, e",
                                    MapUtil.map( "userId", userId, "eventId", eventId ) );
                        }
                    } );
                    jobs.add( job );
                }
 
                for ( Future<ExecutionResult> future : jobs )
                {
                    try
                    {
                        observer.onNext( future.get() );
                    }
                    catch ( InterruptedException | ExecutionException ignored )
                    {
                    }
                }
 
                observer.onCompleted();
                executor.shutdown();
 
                return Subscriptions.empty();
            }
        } );
    }

I’m not sure if that’s the correct way of using Observables so please let me know in the comments if I’ve got it wrong.

I wasn’t sure what the proper way of handling errors was. I initially had a call to observer#onError in the catch block but that means that no further events are produced which wasn’t what I wanted.

The code is available as a gist if you want to play around with it. I added the following dependency to get the RxJava library:

    <dependency>
      <groupId>com.netflix.rxjava</groupId>
      <artifactId>rxjava-core</artifactId>
      <version>0.15.1</version>
    </dependency>
Categories: Programming

Neo4j: Cypher – Using MERGE with schema indexes/constraints

Mon, 12/23/2013 - 14:30

A couple of weeks about I wrote about cypher’s MERGE function and over the last few days I’ve been exploring how it works when used with schema indexes and unique constraints.

A common use case with Neo4j is to model users and events where an event could be a tweet, Facebook post or Pinterest pin. The model might look like this:

2013 12 22 20 14 04

We’d have a stream of (user, event) pairs and a cypher statement like the following to get the data into Neo4j:

MERGE (u:User {id: {userId}})
MERGE (e:Event {id: {eventId}})
MERGE (u)-[:CREATED_EVENT]->(m)
RETURN u, e

We’d like to ensure that we don’t get duplicate users or events and MERGE provides the semantics to do this:

MERGE ensures that a pattern exists in the graph. Either the pattern already exists, or it needs to be created.

I wanted to see what would happen if I wrote a script that tried to create the same (user, event) pairs concurrently and ended up with the following:

import org.neo4j.cypher.javacompat.ExecutionEngine;
import org.neo4j.cypher.javacompat.ExecutionResult;
import org.neo4j.graphdb.GraphDatabaseService;
import org.neo4j.graphdb.factory.GraphDatabaseFactory;
import org.neo4j.helpers.collection.MapUtil;
import org.neo4j.kernel.impl.util.FileUtils;
 
...
 
public class MergeTime
{
    public static void main(String[] args) throws Exception
    {
        String pathToDb = "/tmp/foo";
        FileUtils.deleteRecursively(new File(pathToDb));
 
        GraphDatabaseService db = new GraphDatabaseFactory().newEmbeddedDatabase( pathToDb );
        final ExecutionEngine engine = new ExecutionEngine( db );
 
        ExecutorService executor = Executors.newFixedThreadPool( 50 );
        final Random random = new Random();
 
        final int numberOfUsers = 10;
        final int numberOfEvents = 50;
        int iterations = 100;
        final List<Integer> userIds = generateIds( numberOfUsers );
        final List<Integer> eventIds = generateIds( numberOfEvents );
        List<Future> merges = new ArrayList<>(  );
        for ( int i = 0; i < iterations; i++ )
        {
            Integer userId = userIds.get(random.nextInt(numberOfUsers));
            Integer eventId = eventIds.get(random.nextInt(numberOfEvents));
            merges.add(executor.submit(mergeAway( engine, userId, eventId) ));
        }
 
        for ( Future merge : merges )
        {
            merge.get();
        }
 
        executor.shutdown();
 
        ExecutionResult userResult = engine.execute("MATCH (u:User) RETURN u.id as userId, COUNT(u) AS count ORDER BY userId");
 
        System.out.println(userResult.dumpToString());
 
    }
 
    private static Runnable mergeAway(final ExecutionEngine engine,
                                      final Integer userId, final Integer eventId)
    {
        return new Runnable()
        {
            @Override
            public void run()
            {
                try
                {
                    ExecutionResult result = engine.execute(
                            "MERGE (u:User {id: {userId}})\n" +
                            "MERGE (e:Event {id: {eventId}})\n" +
                            "MERGE (u)-[:CREATED_EVENT]->(m)\n" +
                            "RETURN u, e",
                            MapUtil.map( "userId", userId, "eventId", eventId) );
 
                    // throw away
                    for ( Map<String, Object> row : result ) { }
                }
                catch ( Exception e )
                {
                    e.printStackTrace();
                }
            }
        };
    }
 
    private static List<Integer> generateIds( int amount )
    {
        List<Integer> ids = new ArrayList<>();
        for ( int i = 1; i <= amount; i++ )
        {
            ids.add( i );
        }
        return ids;
    }
}

We create a maximum of 10 users and 50 events and then do 100 iterations of random (user, event) pairs with 50 concurrent threads. Afterwards we execute a query which checks how many users of each id have been created and get the following output:

+----------------+
| userId | count |
+----------------+
| 1      | 6     |
| 2      | 3     |
| 3      | 4     |
| 4      | 8     |
| 5      | 9     |
| 6      | 7     |
| 7      | 5     |
| 8      | 3     |
| 9      | 3     |
| 10     | 2     |
+----------------+
10 rows

Next I added in a schema index on users and events to see if that would make any difference, something Javad Karabi recently asked on the user group.

CREATE INDEX ON :User(id)
CREATE INDEX ON :Event(id)

We wouldn’t expect this to make a difference as schema indexes don’t ensure uniqueness but I ran it anyway t and got the following output:

+----------------+
| userId | count |
+----------------+
| 1      | 2     |
| 2      | 9     |
| 3      | 7     |
| 4      | 2     |
| 5      | 3     |
| 6      | 7     |
| 7      | 7     |
| 8      | 6     |
| 9      | 5     |
| 10     | 3     |
+----------------+
10 rows

If we want to ensure uniqueness of users and events we need to add a unique constraint on the id of both of these labels:

CREATE CONSTRAINT ON (user:User) ASSERT user.id IS UNIQUE
CREATE CONSTRAINT ON (event:Event) ASSERT event.id IS UNIQUE

Now if we run the test we’ll only end up with one of each user:

+----------------+
| userId | count |
+----------------+
| 1      | 1     |
| 2      | 1     |
| 3      | 1     |
| 4      | 1     |
| 5      | 1     |
| 6      | 1     |
| 7      | 1     |
| 8      | 1     |
| 9      | 1     |
| 10     | 1     |
+----------------+
10 rows

We’d see the same type of result if we ran a similar query checking for the uniqueness of events.

As far as I can tell this duplication of nodes that we merge on only happens if you try and create the same node twice concurrently. Once the node has been created we can use MERGE with a non-unique index and a duplicate node won’t get created.

All the code from this post is available as a gist if you want to play around with it.

Categories: Programming