Issues/#5 Add fallback-mode and health-check (#6)

* Modify HibernateArcusRegionFactory.shutdown to call clientPool.shutdown()

* Remove HibernateArcusRegionFactory.nextTimestamp() not to override

* Add fallbackMode and fallBackEnabled configs

* Delete main/resources/logback.xml
log config conflicts the project that uses the hibernate-arcus

* Modify HibernateArcusRegionFactory constructor param

* Modify ArcusClientConfig params by removing connectionObserver and adding reconnectIntervalInSec

* Add HibernateArcusClientFactory.healthCheckArcusCluster setting fallbackMode on and off

* Modify ArcusClientConfig.createArcusClientPool not to set TimeoutExceptionThreshold

* Extract scheduledExecutorService in HibernateArcusClientFactory.healthCheckArcusCluster into class field

* Update version 1.1.0-SNAPSHOT

* Add "hibernate.cache.arcus.healthCheckIntervalInSec" properties default 10

* Modify HibernateArcusClientFactory.healthCheckArcusCluster() not to execute in case of fallbackDisabled

* Add HibernateArcusClientFactoryTest for fallback mode

Co-authored-by: rica.rdo <rica.rdo@kakaopaycorp.com>

Author: rica-v3

build.gradle

@@ -6,7 +6,7 @@ plugins {
 
 group='com.github.kims-developergroup'
 archivesBaseName='hibernate-arcus'
-version='1.0.0-RELEASE'
+version='1.1.0-SNAPSHOT'
 
 repositories {
     mavenCentral()

src/main/java/com/devookim/hibernatearcus/client/HibernateArcusClientFactory.java

@@ -3,42 +3,73 @@
 import com.devookim.hibernatearcus.config.ArcusClientConfig;
 import lombok.extern.slf4j.Slf4j;
 import net.spy.memcached.ArcusClientPool;
-import net.spy.memcached.ConnectionObserver;
 
 import java.net.SocketAddress;
+import java.util.Collection;
+import java.util.Map;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 @Slf4j
-public class HibernateArcusClientFactory implements ConnectionObserver {
+public class HibernateArcusClientFactory {
+    public final boolean fallbackEnabled;
     private final ArcusClientPool clientPool;
+    private final AtomicBoolean fallbackMode;
+    private final ScheduledExecutorService scheduledExecutorService;
+    private final int healthCheckIntervalInSec;
 
     public HibernateArcusClientFactory(ArcusClientConfig clientConfig) {
-        this.clientPool = clientConfig.getArcusClientPool();
+        this.clientPool = clientConfig.createArcusClientPool();
+        fallbackEnabled = clientConfig.fallbackEnabled;
+        fallbackMode = new AtomicBoolean(clientConfig.initFallbackMode);
+        healthCheckIntervalInSec = clientConfig.healthCheckIntervalInSec;
+        scheduledExecutorService = Executors.newScheduledThreadPool(1,
+                runnable -> new Thread(runnable,"HibernateArcusClientFactoryHealthChecker"));
+        healthCheckArcusCluster(scheduledExecutorService);
     }
 
     public void shutdown() {
+        scheduledExecutorService.shutdown();
+        clientPool.shutdown();
         log.info("ArcusClient shutdown");
     }
 
-    public Long nextTimeStamp(long currentTimeStamp, String TIMESTAMP_KEY) {
-        long timeStamp = (long) clientPool.get(TIMESTAMP_KEY);
-        if (timeStamp < currentTimeStamp) {
-            clientPool.set(TIMESTAMP_KEY, -1, currentTimeStamp);
-        }
-        return Math.max(timeStamp, currentTimeStamp);
+    public ArcusClientPool getClientPool() {
+        return clientPool;
     }
 
-    public ArcusClientPool getClient() {
-        return clientPool;
+    public void setFallbackMode(boolean onAndOff) {
+        if(fallbackMode.compareAndSet(!onAndOff, onAndOff)) {
+            log.info("Fallback mode changed: {}", onAndOff);
+        }
     }
 
-    @Override
-    public void connectionEstablished(SocketAddress sa, int reconnectCount) {
-        log.info("Arcus client connection established. addr: {} reconnect: {}", sa, reconnectCount);
+    public boolean isFallbackModeOn() {
+        return fallbackMode.get();
     }
 
-    @Override
-    public void connectionLost(SocketAddress sa) {
-        log.info("Arcus client connection is lost. addr: {}", sa);
+    public void healthCheckArcusCluster(ScheduledExecutorService scheduledExecutorService) {
+        if (!fallbackEnabled) {
+            return;
+        }
+        scheduledExecutorService.scheduleWithFixedDelay(() -> {
+            log.trace("ping...");
+            Map<SocketAddress, Map<String, String>> stats = clientPool.getStats();
+            if(stats.size() == 0) {
+                setFallbackMode(true);
+                return;
+            }
 
+            Collection<Map<String, String>> nodes = stats.values();
+            for (Map<String, String> node: nodes) {
+                if (node.containsKey("zk_connected")
+                        && Boolean.parseBoolean(node.get("zk_connected"))) {
+                    setFallbackMode(false);
+                    return;
+                }
+            }
+        }, healthCheckIntervalInSec, healthCheckIntervalInSec, TimeUnit.SECONDS);
     }
 }

src/main/java/com/devookim/hibernatearcus/config/ArcusClientConfig.java

@@ -1,31 +1,39 @@
 package com.devookim.hibernatearcus.config;
 
-import net.spy.memcached.ArcusClient;
-import net.spy.memcached.ArcusClientPool;
-import net.spy.memcached.ConnectionFactoryBuilder;
+import lombok.extern.slf4j.Slf4j;
+import net.spy.memcached.*;
 
+import java.util.Map;
+
+@Slf4j
 public class ArcusClientConfig {
+    public final boolean fallbackEnabled;
+    public final boolean initFallbackMode;
+    public final int healthCheckIntervalInSec;
     private final int poolSize;
+    private final Map<String, String> properties;
     private final String host;
     private final String serviceCode;
 
-    public ArcusClientConfig(String host,
-                             String serviceCode,
-                             int poolSize) {
-        this.host = host;
-        this.serviceCode = serviceCode;
-        this.poolSize = poolSize;
-    }
-
-    public ArcusClientPool getArcusClientPool() {
-        return createArcusClientPool();
+    public ArcusClientConfig(Map<String, String> properties) {
+        this.host = properties.getOrDefault("hibernate.cache.arcus.host", "localhost:2181");
+        this.serviceCode = properties.getOrDefault("hibernate.cache.arcus.serviceCode", "");
+        this.poolSize = Integer.parseInt(properties.getOrDefault("hibernate.cache.arcus.poolSize", "1"));
+        this.fallbackEnabled = Boolean.parseBoolean(properties.getOrDefault("hibernate.cache.arcus.fallbackEnabled", "true"));
+        this.initFallbackMode = Boolean.parseBoolean(properties.getOrDefault("hibernate.cache.arcus.initFallbackMode", "false"));
+        this.healthCheckIntervalInSec = Integer.parseInt(properties.getOrDefault("hibernate.cache.arcus.healthCheckIntervalInSec", "10"));
+        this.properties = properties;
     }
 
-    private ArcusClientPool createArcusClientPool() {
+    public ArcusClientPool createArcusClientPool() {
+        log.info("Creating arcus client pool");
+        ConnectionFactoryBuilder cfb = new ConnectionFactoryBuilder();
+        cfb.setMaxReconnectDelay(Long.parseLong(properties.getOrDefault("hibernate.cache.arcus.reconnectIntervalInSec", "10000")));
+        cfb.setOpTimeout(Long.parseLong(properties.getOrDefault("hibernate.cache.arcus.opTimeout", "10000")));
         return ArcusClient.createArcusClientPool(
                 host,
                 serviceCode,
-                new ConnectionFactoryBuilder(),
+                cfb,
                 poolSize
         );
     }

src/main/java/com/devookim/hibernatearcus/factory/HibernateArcusRegionFactory.java

@@ -30,11 +30,10 @@ public class HibernateArcusRegionFactory extends RegionFactoryTemplate {
 
     private HibernateArcusClientFactory hibernateArcusClientFactory;
     private CacheKeysFactory cacheKeysFactory;
-    protected boolean fallback;
 
     @PostConstruct
     public void postConstruct() {
-        log.debug("HibernateArcusRegionFactory is initialized");
+        log.info("HibernateArcusRegionFactory is initialized");
     }
 
     @Override
@@ -45,13 +44,8 @@ protected CacheKeysFactory getImplicitCacheKeysFactory() {
     @Override
     protected void prepareForUse(SessionFactoryOptions settings, Map properties) throws CacheException {
 
-        ArcusClientConfig arcusClientConfig = new ArcusClientConfig(
-                properties.getOrDefault("hibernate.cache.arcus.host", "localhost:2181").toString(),
-                properties.getOrDefault("hibernate.cache.arcus.serviceCode", "").toString(),
-                Integer.parseInt(properties.getOrDefault("hibernate.cache.arcus.poolSize", 1).toString())
-        );
+        ArcusClientConfig arcusClientConfig = new ArcusClientConfig(properties);
         this.hibernateArcusClientFactory = new HibernateArcusClientFactory(arcusClientConfig);
-        fallback = true;
         StrategySelector selector = settings.getServiceRegistry().getService(StrategySelector.class);
         cacheKeysFactory = selector.resolveDefaultableStrategy(CacheKeysFactory.class,
                 properties.get(Environment.CACHE_KEYS_FACTORY), new HibernateArcusCacheKeysFactory());
@@ -72,19 +66,6 @@ public AccessType getDefaultAccessType() {
         return AccessType.TRANSACTIONAL;
     }
 
-    @Override
-    public long nextTimestamp() {
-        long currentTimeStamp = System.currentTimeMillis() << 12;
-        try {
-            return hibernateArcusClientFactory.nextTimeStamp(currentTimeStamp, "TIME_STAMP");
-        } catch (Exception e) {
-            if (fallback) {
-                return super.nextTimestamp();
-            }
-            throw e;
-        }
-    }
-
     @Override
     public DomainDataRegion buildDomainDataRegion(
             DomainDataRegionConfig regionConfig,
@@ -101,8 +82,7 @@ public DomainDataRegion buildDomainDataRegion(
 
     @Override
     protected DomainDataStorageAccess createDomainDataStorageAccess(DomainDataRegionConfig regionConfig, DomainDataRegionBuildingContext buildingContext) {
-        ArcusClientPool cacheClientPool = getCache(qualify(regionConfig.getRegionName()));
-        return new DomainDataHibernateArcusStorageAccess(cacheClientPool,  qualify(regionConfig.getRegionName()));
+        return new DomainDataHibernateArcusStorageAccess(getClientFactory(qualify(regionConfig.getRegionName())),  qualify(regionConfig.getRegionName()));
     }
 
     @Override
@@ -112,18 +92,16 @@ public String qualify(String regionName) {
 
     @Override
     protected StorageAccess createQueryResultsRegionStorageAccess(String regionName, SessionFactoryImplementor sessionFactory) {
-        ArcusClientPool cacheClientPool = getCache(qualify(regionName));
-        return new QueryCacheHibernateArcusStorageAccess(cacheClientPool, qualify(regionName));
+        return new QueryCacheHibernateArcusStorageAccess(getClientFactory(qualify(regionName)), qualify(regionName));
     }
 
     @Override
     protected StorageAccess createTimestampsRegionStorageAccess(String regionName, SessionFactoryImplementor sessionFactory) {
-        ArcusClientPool cacheClientPool = getCache(qualify(regionName));
-        return new HibernateArcusStorageAccess(cacheClientPool, qualify(regionName));
+        return new HibernateArcusStorageAccess(getClientFactory(qualify(regionName)), qualify(regionName));
     }
 
-    protected ArcusClientPool getCache(String regionName) {
+    protected HibernateArcusClientFactory getClientFactory(String regionName) {
         log.debug("getCache region: {}", regionName);
-        return hibernateArcusClientFactory.getClient();
+        return hibernateArcusClientFactory;
     }
 }

src/main/java/com/devookim/hibernatearcus/storage/DomainDataHibernateArcusStorageAccess.java

@@ -1,14 +1,14 @@
 package com.devookim.hibernatearcus.storage;
 
+import com.devookim.hibernatearcus.client.HibernateArcusClientFactory;
 import lombok.extern.slf4j.Slf4j;
-import net.spy.memcached.ArcusClientPool;
 import org.hibernate.cache.spi.support.AbstractReadWriteAccess;
 import org.hibernate.engine.spi.SharedSessionContractImplementor;
 
 @Slf4j
 public class DomainDataHibernateArcusStorageAccess extends HibernateArcusStorageAccess {
-    public DomainDataHibernateArcusStorageAccess(ArcusClientPool arcusClientPool, String regionName) {
-        super(arcusClientPool, regionName);
+    public DomainDataHibernateArcusStorageAccess(HibernateArcusClientFactory arcusClientFactory, String regionName) {
+        super(arcusClientFactory, regionName);
     }
 
     @Override