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)
