Merge pull request #999: [io-cassandra] close session on AllNodesFailedException

Author: je-ik

direct/io-cassandra/src/main/java/cz/o2/proxima/direct/io/cassandra/CassandraDBAccessor.java

@@ -258,6 +258,11 @@ CqlSession ensureSession() {
     return getSession(getUri());
   }
 
+  void closeSession() {
+    final String clusterCachedKey = computeClusterKey(getUri().getAuthority(), username);
+    Optional.ofNullable(SESSION_MAP.remove(clusterCachedKey)).ifPresent(CqlSession::close);
+  }
+
   @Override
   public Optional<AttributeWriterBase> getWriter(Context context) {
     return Optional.of(newWriter());

direct/io-cassandra/src/main/java/cz/o2/proxima/direct/io/cassandra/CassandraRandomReader.java

@@ -15,6 +15,7 @@
  */
 package cz.o2.proxima.direct.io.cassandra;
 
+import com.datastax.oss.driver.api.core.AllNodesFailedException;
 import com.datastax.oss.driver.api.core.CqlSession;
 import com.datastax.oss.driver.api.core.cql.BoundStatement;
 import com.datastax.oss.driver.api.core.cql.ResultSet;
@@ -54,6 +55,10 @@ public <T> Optional<KeyValue<T>> get(
     final ResultSet result;
     try {
       result = accessor.execute(statement);
+    } catch (AllNodesFailedException ex) {
+      log.warn("Got {}, closing session.", AllNodesFailedException.class.getSimpleName(), ex);
+      accessor.closeSession();
+      throw ex;
     } catch (Exception ex) {
       throw new RuntimeException("Unable to execute query.", ex);
     }