@RunWith(CustomRunner.class)
public class QueryStalenessTest {
   SessionFactory sf1, sf2;

   @BeforeClassOnce
   public void init() {
      TestResourceTracker.testStarted(getClass().getSimpleName());
      sf1 = createSessionFactory();
      sf2 = createSessionFactory();
   }

   @AfterClassOnce
   public void destroy() {
      sf1.close();
      sf2.close();
      TestResourceTracker.testFinished(getClass().getSimpleName());
   }

   public SessionFactory createSessionFactory() {
      Configuration configuration = new Configuration()
            .setProperty(Environment.USE_SECOND_LEVEL_CACHE, "true")
            .setProperty(Environment.USE_QUERY_CACHE, "true")
            .setProperty(Environment.CACHE_REGION_FACTORY, TestInfinispanRegionFactory.class.getName())
            .setProperty(Environment.DEFAULT_CACHE_CONCURRENCY_STRATEGY, "transactional")
            .setProperty(AvailableSettings.SHARED_CACHE_MODE, "ALL")
            .setProperty(Environment.HBM2DDL_AUTO, "create-drop");
      configuration.addAnnotatedClass(Person.class);
      return configuration.buildSessionFactory();
   }

   @Test
   public void test() {
      Session s1 = sf1.openSession();
      Person person = new Person("John", "Smith", 29);
      s1.persist(person);
      s1.flush();
      s1.close();

      TimestampsRegion timestampsRegion = ((CacheImplementor) sf1.getCache()).getUpdateTimestampsCache().getRegion();
      DistributionInfo distribution = ((ClusteredTimestampsRegionImpl) timestampsRegion).getCache().getDistributionManager().getCacheTopology().getDistribution(Person.class.getSimpleName());
      SessionFactory qsf = distribution.isPrimary() ? sf2 : sf1;

      Session s2 = qsf.openSession();
      List<Person> list1 = s2.createCriteria(Person.class).setCacheable(true).add(Restrictions.le("age", 29)).list();
      assertEquals(1, list1.size());
      s2.close();

      Session s3 = qsf.openSession();
      Person p2 = s3.load(Person.class, person.getName());
      p2.setAge(30);
      s3.persist(p2);
      List<Person> list2 = s3.createCriteria(Person.class).setCacheable(true).add(Restrictions.le("age", 29)).list();
      assertEquals(0, list2.size());
      s3.close();
   }
}