Jakub Nowacki created FLINK-6098:
------------------------------------
Summary: Cassandra sink freezes after write error
Key: FLINK-6098
URL:
https://issues.apache.org/jira/browse/FLINK-6098 Project: Flink
Issue Type: Bug
Components: Cassandra Connector
Affects Versions: 1.2.0
Environment: Flink 1.2.0, standalone cluster, Debian GNU/Linux 8.7 (jessie)
Reporter: Jakub Nowacki
I am having problem with a very simple pipeline taking messages form Kafka and writing them into Cassandra. The pipeline runs fine for a number of days and at some point I am getting the below error in the taskmanager logs:
{code}
2017-03-13 16:01:50,699 ERROR org.apache.flink.streaming.connectors.cassandra.CassandraSinkBase - Error while sending value.
com.datastax.driver.core.exceptions.WriteTimeoutException: Cassandra timeout during write query at consistency LOCAL_ONE (1 replica were required but only 0 acknowledged the write)
Caused by: com.datastax.driver.core.exceptions.WriteTimeoutException: Cassandra timeout during write query at consistency LOCAL_ONE (1 replica were required but only 0 acknowledged the write)
{code}
The job seems to be running fine, but it does not process any messages, which is only visible in the metrics and in the JobManager log:
{code}
2017-03-14 14:00:44,611 INFO org.apache.flink.runtime.checkpoint.CheckpointCoordinator - Triggering checkpoint 42288 @ 1489496444610
2017-03-14 14:00:44,612 INFO org.apache.flink.runtime.checkpoint.CheckpointCoordinator - Discarding checkpoint 42288 because of checkpoint decline from task 35926157acfb1b68d1f6c9abcd90c8b4 : Task Source: Custom Source -> Map -> Map -> Sink: Cassandra Sink (1/2) was not running
{code}
I know this is some Cassandra hiccup, but in theory pipeline should recover after a failure or fail and stop.
Everything seems fine and I didn't find any information specific to the reconnect after failure for the Cassandra Connector. The only thing I'm not sure if it's done correctly is the ClusterBuilder; i.e I use the below code in the job definition main method (in Scala):
{code:java}
val cassandraServers = parameters.get("cassandra.servers", "localhost").split(",")
val cassandraUser = parameters.get("cassandra.user")
val cassandraPassword = parameters.get("cassandra.password")
val clusterBuilder = new ClusterBuilder() {
override def buildCluster(builder: Cluster.Builder): Cluster = {
cassandraServers.foreach(builder.addContactPoint)
if (cassandraUser != null && cassandraPassword != null)
builder.withCredentials(cassandraUser, cassandraPassword)
builder.build()
}
}
{code}
The job starts correctly but I'm not sure if the configuration from the properties is pulled correctly on the taskmanages, as I understand the {{buildCluster}} call is done on its side.
--
This message was sent by Atlassian JIRA
(v6.3.15#6346)
Locked
