[
https://issues.apache.org/jira/browse/FLINK-6098?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15929909#comment-15929909
]
Jakub Nowacki commented on FLINK-6098:
--------------------------------------
Well, it doesn't. :) As you can see from the above logs, which are both for the
same job, it kept running until I killed it the next day. It keeps refusing the
checkpoint though, which is an indicator but it does not help much. On the
source side, if Kafka fails the job gets, indeed, restarted.
> 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)
