[jboss-user] [JBoss Cache: Core Edition] - Re: invalidation messages lost/ignored?

[dukehoops]

The possible solution I described above works for me. The code's below, I would still appreciate peer review and feedback, particularly on:
-whether option.setCacheModeLocal(true) is enough to ensure non-propagation of the remove call (to other cluster nodes)
-why was cache getting into this (indefinite) state and whether that's to be expected or a sign of a bug/greater problem


  | package com.doppelganger.framework.hibernate;
  | 
  | import com.doppelganger.framework.cache.DGEntityTransactionalAccess;
  | import javax.transaction.Transaction;
  | import org.hibernate.HibernateException;
  | import org.hibernate.StaleObjectStateException;
  | import org.hibernate.cache.CacheKey;
  | import org.hibernate.event.EventSource;
  | import org.hibernate.event.FlushEvent;
  | import org.hibernate.event.def.DefaultFlushEventListener;
  | import org.hibernate.impl.SessionFactoryImpl;
  | import org.hibernate.persister.entity.EntityPersister;
  | import org.slf4j.Logger;
  | import org.slf4j.LoggerFactory;
  | import org.springframework.beans.factory.annotation.Autowired;
  | import org.springframework.stereotype.Component;
  | import org.springframework.transaction.jta.JtaTransactionManager;
  | 
  | /**
  |  *
  |  * @author nikita
  |  */
  | @Component("cacheUpdatingFlushEventListener")
  | public class CacheUpdatingFlushEventListener extends DefaultFlushEventListener {
  | 
  |     final static private Logger LOG = LoggerFactory.getLogger(CacheUpdatingFlushEventListener.class);
  |     @Autowired
  |     private JtaTransactionManager jtaTransactionManager;
  | 
  |     @Override
  |     public void onFlush(FlushEvent event) throws HibernateException {
  |         try {
  |             super.onFlush(event);
  |         } catch (StaleObjectStateException ex) {
  | 
  |             //if possible, remove offending entity from 2nd level cache
  |             //remove offending ENTITY cache node (only locally - no cluster-wide propagation)
  | 
  |             Transaction tx = null;
  |             try {
  |                 SessionFactoryImpl sfImpl = (SessionFactoryImpl) event.getSession().getSessionFactory();
  |                 EntityPersister persister = sfImpl.getEntityPersister(ex.getEntityName());
  |                 if (persister.hasCache()) {
  |                     EventSource session = event.getSession();
  |                     CacheKey ck = new CacheKey(
  |                             ex.getIdentifier(),
  |                             persister.getIdentifierType(),
  |                             persister.getRootEntityName(),
  |                             session.getEntityMode(),
  |                             session.getFactory());
  | 
  | 
  |                     if (persister.getCacheAccessStrategy() instanceof DGEntityTransactionalAccess) {
  | 
  |                         LOG.debug("removing Entity cache node with key=" + ck + " after concurrency exception: " + ex);
  | 
  |                         tx = jtaTransactionManager.getTransactionManager().suspend();
  | 
  |                         DGEntityTransactionalAccess entityCacheAccessStrategy = (DGEntityTransactionalAccess) persister.getCacheAccessStrategy();
  |                         entityCacheAccessStrategy.remove(ck, true);
  |                     }
  |                 }
  |             } catch (Throwable t) {
  |                 LOG.error("error removing Entity cache node after detecting exception: " + ex + ". Proceeding to rethrow detected exception", t);
  |             } finally {
  |                 if (tx != null) {
  |                     try {
  |                         jtaTransactionManager.getTransactionManager().resume(tx);
  |                     } catch (Throwable t) {
  |                         LOG.error("failed to resume a tx - INVESTIGATE!", t);
  |                     }
  |                 }
  |                 //rethrow original exception
  |                 throw ex;
  |             }
  |         }
  |     }
  | }
  | 


  | package com.doppelganger.framework.cache;
  | 
  | import org.hibernate.cache.CacheException;
  | import org.hibernate.cache.CacheKey;
  | import org.hibernate.cache.jbc2.entity.EntityRegionImpl;
  | import org.hibernate.cache.jbc2.entity.TransactionalAccess;
  | import org.jboss.cache.config.Option;
  | 
  | /**
  |  *
  |  * @author nikita
  |  */
  | public class DGEntityTransactionalAccess extends TransactionalAccess {
  | 
  |     final private DGTransactionalAccessDelegate accessDelegate;
  | 
  |     public DGEntityTransactionalAccess(EntityRegionImpl region, DGTransactionalAccessDelegate delegate) {
  |         super(region, delegate);
  |         if (delegate == null) {
  |             throw new IllegalStateException("accessDelegate cannot be null and must wrap the same 'region' as passed to this constructor");
  |         }
  |         this.accessDelegate = delegate;
  |     }
  | 
  |     /**
  |      * same as TransactionalAccess.remove(key) except adds switch to NOT send replication/invalidation messages
  |      * @param key
  |      * @param localOnly
  |      * @throws org.hibernate.cache.CacheException
  |      */
  |     public void remove(CacheKey key, boolean localOnly) throws CacheException {
  |         if (localOnly) {
  |             Option option = new Option();
  |             option.setCacheModeLocal(true);
  |             getAccessDelegate().remove(key, option);
  |         } else {
  |             remove(key);
  |         }
  |     }
  | 
  |     /**
  |      * @return the accessDelegate
  |      */
  |     public DGTransactionalAccessDelegate getAccessDelegate() {
  |         return accessDelegate;
  |     }
  | }
  | 


  | /*
  |  * To change this template, choose Tools | Templates
  |  * and open the template in the editor.
  |  */
  | package com.doppelganger.framework.cache;
  | 
  | import org.hibernate.cache.CacheException;
  | import org.hibernate.cache.CacheKey;
  | import org.hibernate.cache.jbc2.BasicRegionAdapter;
  | import org.hibernate.cache.jbc2.access.TransactionalAccessDelegate;
  | import org.hibernate.cache.jbc2.util.CacheHelper;
  | import org.jboss.cache.config.Option;
  | 
  | /**
  |  *
  |  * @author nikita
  |  */
  | public class DGTransactionalAccessDelegate extends TransactionalAccessDelegate {
  | 
  |     DGTransactionalAccessDelegate(BasicRegionAdapter adapter) {
  |         super(adapter);
  |     }
  | 
  |     /**
  |      * same as remove(key), but with an Option
  |      * @param key
  |      * @param option
  |      */
  |     void remove(CacheKey key, Option option) throws CacheException {
  |         region.ensureRegionRootExists();
  |         CacheHelper.remove(cache, regionFqn, key, option);
  |     //cache.removeNode(new Fqn(region, key));
  |     //cache.evict(new Fqn(region, key));
  |     }
  | }
  | 


  | /*
  |  * To change this template, choose Tools | Templates
  |  * and open the template in the editor.
  |  */
  | package com.doppelganger.framework.cache;
  | 
  | import org.hibernate.cache.CacheDataDescription;
  | import org.hibernate.cache.CacheException;
  | import org.hibernate.cache.access.AccessType;
  | import org.hibernate.cache.access.EntityRegionAccessStrategy;
  | import org.hibernate.cache.jbc2.entity.EntityRegionImpl;
  | import org.jboss.cache.Cache;
  | 
  | /**
  |  *
  |  * @author nikita
  |  */
  | public class DGEntityRegionImpl extends EntityRegionImpl {
  | 
  |     public DGEntityRegionImpl(Cache jbcCache, String regionName, String regionPrefix, CacheDataDescription metadata) {
  |         super(jbcCache, regionName, regionPrefix, metadata);
  |     }
  | 
  |     /**
  |      * supply our own Entity TransactionalAccess impl; assume we don't need OptimisticTransactionalAccess (since we're using MVCC)
  |      * @param accessType
  |      * @return
  |      * @throws org.hibernate.cache.CacheException
  |      */
  |     @Override
  |     public EntityRegionAccessStrategy buildAccessStrategy(AccessType accessType) throws CacheException {
  |         if (AccessType.READ_ONLY.equals(accessType)) {
  |             return new DGEntityReadOnlyAccess(this, new DGTransactionalAccessDelegate(this));
  |         }
  |         if (AccessType.TRANSACTIONAL.equals(accessType)) {
  |             return new DGEntityTransactionalAccess(this, new DGTransactionalAccessDelegate(this));
  |         }
  | 
  |         throw new CacheException("unsupported access type [" + accessType.getName() + "]");
  |     }
  | }
  | 
  | 

View the original post : http://www.jboss.org/index.html?module=bb&op=viewtopic&p=4221430#4221430

Reply to the post : http://www.jboss.org/index.html?module=bb&op=posting&mode=reply&p=4221430