Going nuts with write timeouts... please help if you can! =D

[Tony Chong]

Hi There,

My issues are similar to most but not sure if they are the same root cause. The issue is:

write timeouts using LOCAL_QUORUM.

Running spark 1.3.0, with Cassandra 2.1.2. We have included the latest maven release (1.3.0-m1) into my jar and spark-submit the job.


Spark and Cassandra are co-located on each machine, and I have a total of 8 machines in Ec2. We are talking about 8 machines with 30GB of ram, although my Cassandra JVMs max heap is only 8GB.

bin/spark-submit --conf spark.cassandra.output.throughput_mb_per_sec=5 --conf spark.cassandra.connection.timeout_ms=20000 --conf spark.cassandra.read.timeout_ms=20000 --class MyClass --deploy-mode cluster --executor-memory 4G --driver-memory 4G --master spark://spark-datacrunch-ingest-master01.prod.vungle.com:7077 s3n://key:secret@s3_bucket/MY.jar

The Driver and Application start fine, and I can see in the spark logs that things are connecting and everything looks good.

That's when spark lies to me! Here's the stack trace.


15/06/10 00:07:01 ERROR writer.QueryExecutor: Failed to execute: com.datastax.spark.connector.writer.RichBoundStatement@78d67619
com.datastax.driver.core.exceptions.WriteTimeoutException: Cassandra timeout during write query at consistency LOCAL_QUORUM (2 replica were required but only 1 acknowledged the write)
at com.datastax.driver.core.exceptions.WriteTimeoutException.copy(WriteTimeoutException.java:54)
at com.datastax.driver.core.Responses$Error.asException(Responses.java:99)
at com.datastax.driver.core.DefaultResultSetFuture.onSet(DefaultResultSetFuture.java:140)
at com.datastax.driver.core.RequestHandler.setFinalResult(RequestHandler.java:293)
at com.datastax.driver.core.RequestHandler.onSet(RequestHandler.java:467)
at com.datastax.driver.core.Connection$Dispatcher.messageReceived(Connection.java:734)
at org.jboss.netty.channel.SimpleChannelUpstreamHandler.handleUpstream(SimpleChannelUpstreamHandler.java:70)
at org.jboss.netty.handler.timeout.IdleStateAwareChannelUpstreamHandler.handleUpstream(IdleStateAwareChannelUpstreamHandler.java:36)
at org.jboss.netty.channel.DefaultChannelPipeline.sendUpstream(DefaultChannelPipeline.java:564)
at org.jboss.netty.channel.DefaultChannelPipeline$DefaultChannelHandlerContext.sendUpstream(DefaultChannelPipeline.java:791)
at org.jboss.netty.handler.timeout.IdleStateHandler.messageReceived(IdleStateHandler.java:294)
at org.jboss.netty.channel.SimpleChannelUpstreamHandler.handleUpstream(SimpleChannelUpstreamHandler.java:70)
at org.jboss.netty.channel.DefaultChannelPipeline.sendUpstream(DefaultChannelPipeline.java:564)
at org.jboss.netty.channel.DefaultChannelPipeline$DefaultChannelHandlerContext.sendUpstream(DefaultChannelPipeline.java:791)
at org.jboss.netty.channel.Channels.fireMessageReceived(Channels.java:296)
at org.jboss.netty.handler.codec.oneone.OneToOneDecoder.handleUpstream(OneToOneDecoder.java:70)
at org.jboss.netty.channel.DefaultChannelPipeline.sendUpstream(DefaultChannelPipeline.java:564)
at org.jboss.netty.channel.DefaultChannelPipeline$DefaultChannelHandlerContext.sendUpstream(DefaultChannelPipeline.java:791)
at org.jboss.netty.channel.Channels.fireMessageReceived(Channels.java:296)
at org.jboss.netty.handler.codec.frame.FrameDecoder.unfoldAndFireMessageReceived(FrameDecoder.java:462)
at org.jboss.netty.handler.codec.frame.FrameDecoder.callDecode(FrameDecoder.java:443)
at org.jboss.netty.handler.codec.frame.FrameDecoder.messageReceived(FrameDecoder.java:303)
at org.jboss.netty.channel.SimpleChannelUpstreamHandler.handleUpstream(SimpleChannelUpstreamHandler.java:70)
at org.jboss.netty.channel.DefaultChannelPipeline.sendUpstream(DefaultChannelPipeline.java:564)
at org.jboss.netty.channel.DefaultChannelPipeline.sendUpstream(DefaultChannelPipeline.java:559)
at org.jboss.netty.channel.Channels.fireMessageReceived(Channels.java:268)
at org.jboss.netty.channel.Channels.fireMessageReceived(Channels.java:255)
at org.jboss.netty.channel.socket.nio.NioWorker.read(NioWorker.java:88)
at org.jboss.netty.channel.socket.nio.AbstractNioWorker.process(AbstractNioWorker.java:109)
at org.jboss.netty.channel.socket.nio.AbstractNioSelector.run(AbstractNioSelector.java:312)
at org.jboss.netty.channel.socket.nio.AbstractNioWorker.run(AbstractNioWorker.java:90)
at org.jboss.netty.channel.socket.nio.NioWorker.run(NioWorker.java:178)
at org.jboss.netty.util.ThreadRenamingRunnable.run(ThreadRenamingRunnable.java:108)
at org.jboss.netty.util.internal.DeadLockProofWorker$1.run(DeadLockProofWorker.java:42)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
at java.lang.Thread.run(Thread.java:745)
Caused by: com.datastax.driver.core.exceptions.WriteTimeoutException: Cassandra timeout during write query at consistency LOCAL_QUORUM (2 replica were required but only 1 acknowledged the write)
at com.datastax.driver.core.Responses$Error$1.decode(Responses.java:58)
at com.datastax.driver.core.Responses$Error$1.decode(Responses.java:38)
at com.datastax.driver.core.Message$ProtocolDecoder.decode(Message.java:168)
at org.jboss.netty.handler.codec.oneone.OneToOneDecoder.handleUpstream(OneToOneDecoder.java:66)
... 21 more


In OpsCenter, I can see that my write requests are about 7/sec, and my Write request latency is about 1700ms/op. I have even increased my

write_request_timeout_in_ms

in the cassandra.yaml. I've tried ridiculous timeouts and all it does is delay the inevitable error message.

Memory and CPU are relatively low. Please..anyone..have a heart and throw me a bone!


Thanks all,

Tony

[Russell Spitzer]

--conf spark.cassandra.output.throughput_mb_per_sec=5 ## Is per executor core so this is most likely to high (we also should probably let you set it in KB as well). Try setting this to 1 to limit the write throughput.

To unsubscribe from this group and stop receiving emails from it, send an email to spark-connector-...@lists.datastax.com.

[Tony Chong]

Thanks Russell for the super quick reply. I have tried setting
--conf spark.cassandra.output.throughput_mb_per_sec=1

and still have the same problem. I think I've tried every spark config I could think of. Is there any other thing you think I could try?

[Russell Spitzer]

Yeah you can try shrinking the write batch sizes

batch.size.rows or batch.size.bytes

or lowering concurrent.writes

If all else fails (or as a good sanity check)

Try running with only a single executor. If it fails with a single executor we can explore some other options :)

https://github.com/datastax/spark-cassandra-connector/blob/master/doc/5_saving.md

[Tony Chong]

Thanks Russell.

--conf spark.cassandra.output.concurrent.writes=1 seems to have done the trick for now. I will keep everyone posted, but this has gotten me passed my write timeout issues for now. Looks like there are still some bugs in the code, but at this was a god send.

BTW, your video is top notch.

https://academy.datastax.com/demos/how-spark-cassandra-connector-reads-data

[Tony Chong]

Well, looks like all isn't well.

After about 2.5 hours of running, I get the strangest error about the the inability to write statements and then all the machines are "unavailable".

Taking your advice, I shut down all spark workers except for one.

Here is the job:

bin/spark-submit --conf spark.cassandra.output.concurrent.writes=1 --conf batch.size.rows=10 --conf spark.cassandra.output.throughput_mb_per_sec=1 --conf spark.cassandra.connection.timeout_ms=20000 --conf spark.cassandra.read.timeout_ms=20000 --class "$CLASS_NAME" --deploy-mode cluster --executor-memory 4G --driver-memory 4G --master $SPARK_NAME s3n://$S3_CREDS@bucket/$JAR_NAME


Here is the error:

15/06/11 17:19:25 ERROR writer.QueryExecutor: Failed to execute: com.datastax.spark.connector.writer.RichBoundStatement@274a916f
com.datastax.driver.core.exceptions.NoHostAvailableException: All host(s) tried for query failed (tried: /10.102.191.85:9042 (com.datastax.driver.core.TransportException: [/10.102.191.85:9042] Connection has been closed), /10.45.70.167:9042 (com.datastax.driver.core.TransportException: [/10.45.70.167:9042] Connection has been closed), spark-datacrunch-ingest-worker02.prod.vungle.com/10.63.137.147:9042 (com.datastax.driver.core.TransportException: [spark-datacrunch-ingest-worker02.prod.vungle.com/10.63.137.147:9042] Connection has been closed), /10.81.193.169:9042 [only showing errors of first 3 hosts, use getErrors() for more details])
at com.datastax.driver.core.RequestHandler.sendRequest(RequestHandler.java:107)
at com.datastax.driver.core.RequestHandler$1.run(RequestHandler.java:210)

at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
at java.lang.Thread.run(Thread.java:745)

AND

15/06/11 17:27:43 ERROR writer.QueryExecutor: Failed to execute: com.datastax.spark.connector.writer.RichBatchStatement@4afa4953
com.datastax.driver.core.exceptions.WriteTimeoutException: Cassandra timeout during write query at consistency ONE (1 replica were required but only 0 acknowledged the write)



I'm at a loss now because all nodes are up. Logs in cassandra say nothing. I'm starting to think this isn't a valid solution.

[Russell Spitzer]

The Cassandra logs should be showing some kind of GC event when the failure occurs. These would not be visible in passing (the nodes are down because the stop the world event stops gossip from responding.) But if you are hitting this with only a single executor core  something is much more wrong. The errors are a symptom of ring instability and are usually traced to a build up of hints or compaction backing up and the system not being able to keep up. 

Unless you C* system is extremely fragile (on ebs? something else taking a lot of ram? jvm being paged out? ) this kind of stability issue shouldn't be seen. Basically what you are seeing is C* not being able to keep up with what should be a very very modest rate of injest (with one core). 

I would set batch.size.rows = 1 and cores to 1 and concurrent executors to 1. This will effectively just be inserting rows in serial ... if that causes system instability then it's time to start debugging the C* side. 

[Tony Chong]

Thanks again for the quick reply. I have also been taking a look at the C* side, and not spark connector side.

I've been looking at cstar_perf and cassandra-stress right now because like you said, this should be working with such a small load of inserts.

Immediately I'm running into cassandra issues so looks like I've been going down the wrong rabbit hole. Going to investigate and report back.

[chen qiang]

I came across of a problem like this one, except that I'm facing OOM. I changed batch size to 1, lowered executor count, and lowered total data when using saveToCassandra. My problem is very clear, cassandra is sufferring from compacting sstables which eventually lead to OOM. I'm using cassandra connector 1.1.1 though.

[Tony Chong]

Just to give everyone a follow up, the cluster turned out to be okay.

I eventually ran the job with these settings:

bin/spark-submit --conf spark.cassandra.output.batch.size.rows=1 --conf spark.cassandra.output.concurrent.writes=1 --conf spark.cassandra.output.throughput_mb_per_sec=1 --conf spark.cassandra.connection.timeout_ms=20000 --conf spark.cassandra.read.timeout_ms=20000 --class "$CLASS_NAME" --deploy-mode cluster --executor-memory 4G --driver-memory 4G --master $SPARK_NAME s3n://$S3_CREDS@s3bucket/$JAR_NAME

It's slowly chugging along. The process eventually dies on a read (I believe) due to out of memory heap space errors, but I think at this point it's about fine tuning the job. I've rerun the job with a smaller memory footprint since running it with a larger executor size at 8G, instantly throws a fit. I have 30GB total on the machine, with 8gb of the RAM allocated to Cassandra, since I'm running spark on the same nodes for locality.