Re: [infinispan-dev] [ISPN-116] Async cache store: aggregation of multiple changes on a single key

Hi Galder, I checked out latest version (from subversion trunk) and used it for my test. Here is config in xml: --- <?xml version="1.0" encoding="UTF-8"?> <infinispan xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns="urn:infinispan:config:4.0"> <global> <transport clusterName="SmppGatewayCluster"/> </global> <namedCache name="accounts"> <eviction wakeUpInterval="5000" maxEntries="100" strategy="FIFO"/> <expiration lifespan="600" maxIdle="1000"/> <loaders passivation="false" shared="false" preload="true"> <loader class="com.magfa.gateway.dao.infinispan.jdbcstore.AccountJdbcCacheStore" fetchPersistentState="true" ignoreModifications="false" purgeOnStartup="true"> <properties/> <singletonStore enabled="true" pushStateWhenCoordinator="true" pushStateTimeout="20000"/> <async enabled="true" threadPoolSize="5"/> </loader> </loaders> </namedCache> </infinispan> --- And here is my similar config code: ---        Configuration config = new Configuration();        GlobalConfiguration globalConfiguration = GlobalConfiguration.getNonClusteredDefault();        AsyncStoreConfig asyncStoreConfig = new AsyncStoreConfig();        asyncStoreConfig.setEnabled(true);        CacheStoreConfig cacheStoreConfig = new AccountJdbcCacheStoreConfig();        cacheStoreConfig.setAsyncStoreConfig(asyncStoreConfig);        cacheStoreConfig.setPurgeSynchronously(false);        CacheLoaderManagerConfig clmc = new CacheLoaderManagerConfig();        List<CacheLoaderConfig> cacheStoreConfigs = new ArrayList<CacheLoaderConfig>(1);        cacheStoreConfigs.add(cacheStoreConfig);        clmc.setCacheLoaderConfigs(cacheStoreConfigs);        config.setCacheLoaderManagerConfig(clmc);        config.setEvictionStrategy(EvictionStrategy.FIFO);        config.setExpirationLifespan(10000);        config.setExpirationMaxIdle(10000);        config.setEvictionMaxEntries(100);        config.setEvictionWakeUpInterval(60 * 1000);        config.setCacheMode(Configuration.CacheMode.INVALIDATION_ASYNC); // todo: important        //CacheStoreConfig        CacheManager manager = new DefaultCacheManager(globalConfiguration, config); --- and here is my custom store: --- public class AccountJdbcCacheStore extends LockSupportCacheStore {    private static final Logger log = Logger.getLogger(AccountJdbcCacheStore.class);    private final ConnectionPool pool;    // todo; why constructor is called twice?    public AccountJdbcCacheStore() {        Properties properties = new Properties();        properties.setProperty("driver", "com.mysql.jdbc.Driver");        properties.setProperty("url", "jdbc:mysql://dbsrv:3306/accounts?autoReconnect=true");        properties.setProperty("user", "*****");        properties.setProperty("password", "*****");        pool = new ProxoolConnectionPool(properties);    }    protected void clearLockSafe() throws CacheLoaderException {        throw new CacheLoaderException("Will not clear AccountBillingTable");    }    private static final String LOAD_ACCOUNTS_QUERY = "SELECT * FROM accountsBilling";    protected Set<InternalCacheEntry> loadAllLockSafe() throws CacheLoaderException {        log.trace("loadAllLockSafe()");        Connection connection = null;        Statement st = null;        ResultSet rs = null;        try {            connection = pool.getConnection();            st = connection.createStatement();            rs = st.executeQuery(LOAD_ACCOUNTS_QUERY); // todo: where enable=1            final Set<InternalCacheEntry> results = new HashSet<InternalCacheEntry>();            while (rs.next()) {                results.add(InternalEntryFactory.create(rs.getLong("id"), EngineCompatibleBillingBackend.fetchFromResultSet(rs)));            }            return results;        } catch (SQLException e) {            throw new RuntimeException("Exception in loading accounts from good old accountsBilling ", e);        } finally {            pool.attemptClose(rs);            pool.attemptClose(st);            pool.attemptClose(connection);        }    }    protected void toStreamLockSafe(ObjectOutput oos) throws CacheLoaderException {        throw new CacheLoaderException("toStreamLockSafe(); What should i do with ObjectOutput: " + oos);    }    protected void fromStreamLockSafe(ObjectInput ois) throws CacheLoaderException {        throw new CacheLoaderException("fromStreamLockSafe(); What should i do with ObjectInput: " + ois);    }    protected boolean removeLockSafe(Object key, String lockingKey) throws CacheLoaderException {        log.trace("removeLockSafe(); will not remove anything from AccountsTable with key: " + key + ", lockingKey: " + lockingKey);        return true;    }    private final static double OUTGOING_FARSI_COST_BASE = 71.2d;    private final static double OUTGOING_ENGLISH_COST_BASE = 177.6d;    private final static String UPDATE_ACCOUNT_QUERY;    static {        final String TOTAL_MESSAGE_COST_FORMULA = "(?*(farsiCost+" + OUTGOING_FARSI_COST_BASE + ") + (?*(englishCost+"                + OUTGOING_ENGLISH_COST_BASE + "))+(?*incomingCost))";        UPDATE_ACCOUNT_QUERY = "UPDATE accountsBilling SET totalSentMessages = totalSentMessages + ?, " +                "totalReceivedMessages = totalReceivedMessages + ?, farsiSentMessages = farsiSentMessages + ? , " +                "englishSentMessages = englishSentMessages + ?, " +                "usedCredit2 = usedCredit2 + " + TOTAL_MESSAGE_COST_FORMULA + ',' +                "credit2 = credit2 - " + TOTAL_MESSAGE_COST_FORMULA + ' ' +                "WHERE id=?";    }    private static final AtomicInteger storeCounter = new AtomicInteger();    protected void storeLockSafe(InternalCacheEntry ed, String lockingKey) throws CacheLoaderException {        final Account account = (Account) ed.getValue();        log.info("-------> storeLockSafe(" + storeCounter.incrementAndGet() + "); storing account: " + account + ", lockingKey: " + lockingKey);        final long sent = account.getSentAndReset();        final long received = account.getReceivedAndReset();        final long dlr = account.getDlrAndReset();        final long farsiSent = account.getFarsiSentAndReset();        final long englishSent = account.getEnglishSentAndReset();        Connection connection = null;        PreparedStatement st = null;        try {            connection = pool.getConnection();            st = connection.prepareStatement(UPDATE_ACCOUNT_QUERY);            st.setLong(1, sent);            st.setLong(2, received);            st.setLong(3, farsiSent);            st.setLong(4, englishSent);            st.setLong(5, farsiSent);            st.setLong(6, englishSent);            st.setLong(7, received);            st.setLong(8, farsiSent);            st.setLong(9, englishSent);            st.setLong(10, received);            st.setLong(11, account.getId());            st.executeUpdate();        } catch (SQLException e) {            throw new RuntimeException("Exception in updating accountsBilling from engine DB", e);        } finally {            pool.attemptClose(st);            pool.attemptClose(connection);        }    }    protected InternalCacheEntry loadLockSafe(Object key, String lockingKey) throws CacheLoaderException {        log.info("loadLockSafe(), key: " + key + ", lockingKey: " + lockingKey);        Connection conn = null;        Statement st = null;        ResultSet rs = null;        try {            conn = pool.getConnection();            st = conn.createStatement();            rs = st.executeQuery("SELECT * FROM accountsBilling WHERE id=" + lockingKey);            if (rs.next()) {                return InternalEntryFactory.create(key, EngineCompatibleBillingBackend.fetchFromResultSet(rs));            }            return null;        } catch (SQLException e) {            final String message = "SQL error while fetching strored entry with key:" + key + " lockingKey: " + lockingKey;            log.error(message, e);            throw new CacheLoaderException(message, e);        } finally {            pool.attemptClose(rs);            pool.attemptClose(st);            pool.attemptClose(conn);        }    }    protected String getLockFromKey(Object key) throws CacheLoaderException {        return key.toString();    }    public Class<? extends CacheLoaderConfig> getConfigurationClass() {        return AccountJdbcCacheStoreConfig.class;    } } ---- and custom config: --- public class AccountJdbcCacheStoreConfig extends LockSupportCacheStoreConfig {    private static final Logger log = Logger.getLogger(AccountJdbcCacheStoreConfig.class);    public LockSupportCacheStoreConfig clone() {        return new AccountJdbcCacheStoreConfig();    }    public String getCacheLoaderClassName() {        return AccountJdbcCacheStore.class.getName();    }    public void setCacheLoaderClassName(String s) {    } } --- but I still have two issues: 1. Using config via code, I get the flush to db working but still no aggregation. For 100 even I have 100 stores. 2. Using xml config, no storeLockSafe is called at all? Thanks for your help, Amin Abbaspour On Mon, Jul 20, 2009 at 7:09 PM, Galder Zamarreno&lt;galder.zamarreno(a)redhat.com&gt; wrote: