ignite-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From a.@apache.org
Subject [08/27] ignite git commit: IGNITE-4954 - Configurable expiration timeout for Cassandra session. This closes #1785.
Date Wed, 31 May 2017 12:27:54 GMT
IGNITE-4954 - Configurable expiration timeout for Cassandra session. This closes #1785.


Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/bf104974
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/bf104974
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/bf104974

Branch: refs/heads/ignite-5232-1.7.2
Commit: bf1049741f7a64728bd433f78262ba273f969848
Parents: 4a1415a
Author: Andrey V. Mashenkov <andrey.mashenkov@gmail.com>
Authored: Mon Apr 17 19:00:30 2017 +0300
Committer: Andrey V. Mashenkov <andrey.mashenkov@gmail.com>
Committed: Mon Apr 17 19:00:30 2017 +0300

----------------------------------------------------------------------
 .../store/cassandra/datasource/DataSource.java  | 26 +++++++++++++++++++-
 .../cassandra/session/CassandraSessionImpl.java | 23 ++++++++++-------
 .../cassandra/session/pool/SessionPool.java     |  6 ++---
 .../cassandra/session/pool/SessionWrapper.java  | 15 ++++++-----
 4 files changed, 51 insertions(+), 19 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/ignite/blob/bf104974/modules/cassandra/src/main/java/org/apache/ignite/cache/store/cassandra/datasource/DataSource.java
----------------------------------------------------------------------
diff --git a/modules/cassandra/src/main/java/org/apache/ignite/cache/store/cassandra/datasource/DataSource.java
b/modules/cassandra/src/main/java/org/apache/ignite/cache/store/cassandra/datasource/DataSource.java
index 915eebd..1fa2a1d 100644
--- a/modules/cassandra/src/main/java/org/apache/ignite/cache/store/cassandra/datasource/DataSource.java
+++ b/modules/cassandra/src/main/java/org/apache/ignite/cache/store/cassandra/datasource/DataSource.java
@@ -46,6 +46,9 @@ import org.apache.ignite.internal.util.typedef.internal.S;
  * Data source abstraction to specify configuration of the Cassandra session to be used.
  */
 public class DataSource {
+    /** Default expiration timeout for Cassandra driver session. */
+    public static final long DFLT_SESSION_EXPIRATION_TIMEOUT = 300000; // 5 minutes.
+
     /** Number of rows to immediately fetch in CQL statement execution. */
     private Integer fetchSize;
 
@@ -123,6 +126,9 @@ public class DataSource {
     /** Netty options to use for connection. */
     private NettyOptions nettyOptions;
 
+    /** Expiration timeout for Cassandra driver session. */
+    private long sessionExpirationTimeout = DFLT_SESSION_EXPIRATION_TIMEOUT;
+
     /** Cassandra session wrapper instance. */
     private volatile CassandraSession ses;
 
@@ -442,6 +448,23 @@ public class DataSource {
     }
 
     /**
+     * Sets expiration timeout for Cassandra driver session. Idle sessions that are not
+     * used during this timeout value will be automatically closed and recreated later
+     * on demand.
+     * <p>
+     * If set to {@code 0}, timeout is disabled.
+     * <p>
+     * Default value is {@link #DFLT_SESSION_EXPIRATION_TIMEOUT}.
+     *
+     * @param sessionExpirationTimeout Expiration timeout for Cassandra driver session.
+     */
+    public void setSessionExpirationTimeout(long sessionExpirationTimeout) {
+        this.sessionExpirationTimeout = sessionExpirationTimeout;
+
+        invalidate();
+    }
+
+    /**
      * Creates Cassandra session wrapper if it wasn't created yet and returns it
      *
      * @param log logger
@@ -523,7 +546,8 @@ public class DataSource {
         if (nettyOptions != null)
             builder = builder.withNettyOptions(nettyOptions);
 
-        return ses = new CassandraSessionImpl(builder, fetchSize, readConsistency, writeConsistency,
log);
+        return ses = new CassandraSessionImpl(
+            builder, fetchSize, readConsistency, writeConsistency, sessionExpirationTimeout,
log);
     }
 
     /**

http://git-wip-us.apache.org/repos/asf/ignite/blob/bf104974/modules/cassandra/src/main/java/org/apache/ignite/cache/store/cassandra/session/CassandraSessionImpl.java
----------------------------------------------------------------------
diff --git a/modules/cassandra/src/main/java/org/apache/ignite/cache/store/cassandra/session/CassandraSessionImpl.java
b/modules/cassandra/src/main/java/org/apache/ignite/cache/store/cassandra/session/CassandraSessionImpl.java
index 95b8581..cee776b 100644
--- a/modules/cassandra/src/main/java/org/apache/ignite/cache/store/cassandra/session/CassandraSessionImpl.java
+++ b/modules/cassandra/src/main/java/org/apache/ignite/cache/store/cassandra/session/CassandraSessionImpl.java
@@ -17,6 +17,13 @@
 
 package org.apache.ignite.cache.store.cassandra.session;
 
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
+import javax.cache.Cache;
 import com.datastax.driver.core.BatchStatement;
 import com.datastax.driver.core.BoundStatement;
 import com.datastax.driver.core.Cluster;
@@ -30,13 +37,6 @@ import com.datastax.driver.core.Statement;
 import com.datastax.driver.core.exceptions.AlreadyExistsException;
 import com.datastax.driver.core.exceptions.InvalidQueryException;
 import com.datastax.driver.core.querybuilder.Batch;
-import java.io.IOException;
-import java.util.HashMap;
-import java.util.LinkedList;
-import java.util.List;
-import java.util.Map;
-import java.util.concurrent.atomic.AtomicInteger;
-import javax.cache.Cache;
 import org.apache.ignite.IgniteException;
 import org.apache.ignite.IgniteLogger;
 import org.apache.ignite.cache.store.cassandra.common.CassandraHelper;
@@ -82,6 +82,9 @@ public class CassandraSessionImpl implements CassandraSession {
     /** Consistency level for Cassandra WRITE operations (insert/update/delete). */
     private ConsistencyLevel writeConsistency;
 
+    /** Expiration timeout. */
+    private long expirationTimeout;
+
     /** Logger. */
     private IgniteLogger log;
 
@@ -101,11 +104,12 @@ public class CassandraSessionImpl implements CassandraSession {
      * @param log Logger.
      */
     public CassandraSessionImpl(Cluster.Builder builder, Integer fetchSize, ConsistencyLevel
readConsistency,
-        ConsistencyLevel writeConsistency, IgniteLogger log) {
+        ConsistencyLevel writeConsistency, long expirationTimeout, IgniteLogger log) {
         this.builder = builder;
         this.fetchSize = fetchSize;
         this.readConsistency = readConsistency;
         this.writeConsistency = writeConsistency;
+        this.expirationTimeout = expirationTimeout;
         this.log = log;
     }
 
@@ -404,7 +408,8 @@ public class CassandraSessionImpl implements CassandraSession {
     /** {@inheritDoc} */
     @Override public synchronized void close() throws IOException {
         if (decrementSessionRefs() == 0 && ses != null) {
-            SessionPool.put(this, ses);
+            SessionPool.put(this, ses, expirationTimeout);
+
             ses = null;
         }
     }

http://git-wip-us.apache.org/repos/asf/ignite/blob/bf104974/modules/cassandra/src/main/java/org/apache/ignite/cache/store/cassandra/session/pool/SessionPool.java
----------------------------------------------------------------------
diff --git a/modules/cassandra/src/main/java/org/apache/ignite/cache/store/cassandra/session/pool/SessionPool.java
b/modules/cassandra/src/main/java/org/apache/ignite/cache/store/cassandra/session/pool/SessionPool.java
index fc4a907..86db713 100644
--- a/modules/cassandra/src/main/java/org/apache/ignite/cache/store/cassandra/session/pool/SessionPool.java
+++ b/modules/cassandra/src/main/java/org/apache/ignite/cache/store/cassandra/session/pool/SessionPool.java
@@ -17,13 +17,13 @@
 
 package org.apache.ignite.cache.store.cassandra.session.pool;
 
-import com.datastax.driver.core.Session;
 import java.lang.Thread.State;
 import java.util.Collection;
 import java.util.HashMap;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
+import com.datastax.driver.core.Session;
 import org.apache.ignite.cache.store.cassandra.session.CassandraSessionImpl;
 
 /**
@@ -98,14 +98,14 @@ public class SessionPool {
      * @param cassandraSes Session wrapper.
      * @param driverSes Driver session.
      */
-    public static void put(CassandraSessionImpl cassandraSes, Session driverSes) {
+    public static void put(CassandraSessionImpl cassandraSes, Session driverSes, long expirationTimeout)
{
         if (cassandraSes == null || driverSes == null)
             return;
 
         SessionWrapper old;
 
         synchronized (sessions) {
-            old = sessions.put(cassandraSes, new SessionWrapper(driverSes));
+            old = sessions.put(cassandraSes, new SessionWrapper(driverSes, expirationTimeout));
 
             if (monitorSingleton == null || State.TERMINATED.equals(monitorSingleton.getState()))
{
                 monitorSingleton = new SessionMonitor();

http://git-wip-us.apache.org/repos/asf/ignite/blob/bf104974/modules/cassandra/src/main/java/org/apache/ignite/cache/store/cassandra/session/pool/SessionWrapper.java
----------------------------------------------------------------------
diff --git a/modules/cassandra/src/main/java/org/apache/ignite/cache/store/cassandra/session/pool/SessionWrapper.java
b/modules/cassandra/src/main/java/org/apache/ignite/cache/store/cassandra/session/pool/SessionWrapper.java
index 7c5722b..68b9dd4 100644
--- a/modules/cassandra/src/main/java/org/apache/ignite/cache/store/cassandra/session/pool/SessionWrapper.java
+++ b/modules/cassandra/src/main/java/org/apache/ignite/cache/store/cassandra/session/pool/SessionWrapper.java
@@ -24,12 +24,12 @@ import org.apache.ignite.cache.store.cassandra.common.CassandraHelper;
  * Wrapper for Cassandra driver session, responsible for monitoring session expiration and
its closing.
  */
 public class SessionWrapper {
-    /** Expiration timeout for Cassandra driver session. */
-    public static final long DFLT_EXPIRATION_TIMEOUT = 300000;  // 5 minutes.
-
     /** Cassandra driver session. */
     private Session ses;
 
+    /** Expiration timeout. */
+    private long expirationTimeout;
+
     /** Wrapper creation time.  */
     private long time;
 
@@ -38,9 +38,11 @@ public class SessionWrapper {
      *
      * @param ses Cassandra driver session.
      */
-    public SessionWrapper(Session ses) {
+    public SessionWrapper(Session ses, long expirationTimeout) {
         this.ses = ses;
-        this.time = System.currentTimeMillis();
+        this.expirationTimeout = expirationTimeout;
+
+        time = System.currentTimeMillis();
     }
 
     /**
@@ -49,7 +51,7 @@ public class SessionWrapper {
      * @return true if session expired.
      */
     public boolean expired() {
-        return System.currentTimeMillis() - time > DFLT_EXPIRATION_TIMEOUT;
+        return expirationTimeout > 0 && System.currentTimeMillis() - time >
expirationTimeout;
     }
 
     /**
@@ -66,6 +68,7 @@ public class SessionWrapper {
      */
     public void release() {
         CassandraHelper.closeSession(ses);
+
         ses = null;
     }
 }


Mime
View raw message