Mads Chr. Olesen created FLINK-13059:
----------------------------------------
Summary: Cassandra Connector leaks Semaphore on Exception; hangs
on close
Key: FLINK-13059
URL: https://issues.apache.org/jira/browse/FLINK-13059
Project: Flink
Issue Type: Bug
Components: Connectors / Cassandra
Affects Versions: 1.8.0
Reporter: Mads Chr. Olesen
In CassandraSinkBase the following code is present (comments are mine):
{code:java}
public void invoke(IN value) throws Exception {
checkAsyncErrors();
tryAcquire();
//Semaphore held here
final ListenableFuture<V> result = send(value);
Futures.addCallback(result, callback); //Callback releases semaphore
}{code}
Any Exception happening inside send(value) will result in the semaphore not
being released. Such exceptions are possible, e.g.
{code:java}
com.datastax.driver.core.exceptions.InvalidQueryException: Some partition key
parts are missing: hest
at
com.datastax.driver.core.exceptions.InvalidQueryException.copy(InvalidQueryException.java:50)
at
com.datastax.driver.core.DriverThrowables.propagateCause(DriverThrowables.java:37)
at com.datastax.driver.core.AbstractSession.prepare(AbstractSession.java:98)
at com.datastax.driver.mapping.Mapper.getPreparedQuery(Mapper.java:118)
at com.datastax.driver.mapping.Mapper.saveQuery(Mapper.java:201)
at com.datastax.driver.mapping.Mapper.saveQuery(Mapper.java:163)
at
org.apache.flink.streaming.connectors.cassandra.CassandraPojoSink.send(CassandraPojoSink.java:128)
at
org.apache.flink.streaming.connectors.cassandra.CassandraSinkBase.invoke(CassandraSinkBase.java:131)
at
org.apache.flink.streaming.api.functions.sink.SinkFunction.invoke(SinkFunction.java:52)
{code}
The result of the semaphore not being released will be that when the exception
bubbles out and causes the job to close, CassandraSinkBase.flush() will
eventually be called. Flush will be deadlocked trying to acquire
config.getMaxConcurrentRequests() from the semaphore, which has 1 less than
that available.
The Flink job will thus be half-way closed, but marked as "RUNNING".
Checkpointing will however fail with
{noformat}
INFO org.apache.flink.runtime.checkpoint.CheckpointCoordinator - Discarding
checkpoint 201325 of job XXX.
org.apache.flink.runtime.checkpoint.decline.CheckpointDeclineTaskNotReadyException:
Task Source: XXX (3/4) was not running {noformat}
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
