[jira] [Updated] (FLINK-6098) Cassandra sink freezes after write error

["Flink Jira Bot (Jira)" <ji...@apache.org>]

     [ https://issues.apache.org/jira/browse/FLINK-6098?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]

Flink Jira Bot updated FLINK-6098:
----------------------------------
      Labels: auto-deprioritized-major auto-deprioritized-minor  (was: auto-deprioritized-major stale-minor)
    Priority: Not a Priority  (was: Minor)

This issue was labeled "stale-minor" 7 days ago and has not received any updates so it is being deprioritized. If this ticket is actually Minor, please raise the priority and ask a committer to assign you the issue or revive the public discussion.


> Cassandra sink freezes after write error
> ----------------------------------------
>
>                 Key: FLINK-6098
>                 URL: https://issues.apache.org/jira/browse/FLINK-6098
>             Project: Flink
>          Issue Type: Bug
>          Components: Connectors / Cassandra
>    Affects Versions: 1.2.0
>         Environment: Flink 1.2.0, standalone cluster, Debian GNU/Linux 8.7 (jessie)
>            Reporter: Jakub Nowacki
>            Priority: Not a Priority
>              Labels: auto-deprioritized-major, auto-deprioritized-minor
>
> 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
(v8.20.1#820001)