Clearing Hazelcast data grid cache with Oracle Database change notification

UP1: if you are interested in in-memory computing, we recommended the book "High performance in-memory computing with Apache Ignite".

A few days ago we decided to use 2nd level cache for better java scalability in our legacy system. Everything goes fine with hazelcast as a 2nd level cache, whenever our a few 3rd party applications starts uploading data directly to the Oracle schema. Generally, a middle-tier data cache duplicates some data from the back-end database server. Its goal is to avoid redundant queries to the database. However, this is efficient only when the data rarely changes in the database. The data cache has to be updated or invalidated when the data changes in the database. If application operates DML operations through cache it's simply your life, but in our case some of our 3rd party can't use hazelcast data grid and we decided to get the proper way to update our caches or clear it whenever some entity on tables updates. In this post i will provide a simple way to clear Hazelcast cache (Hibernate region) whenever data base event occurs on the oracle database. 
It's very easy to plug Hazelcast as a 2nd level cache on Hibernate Project. 
To enable 2nd level cache on hibernate, do the following:

      AnnotationConfiguration aconf=new AnnotationConfiguration();
      .... 
      aconf.setProperty("hibernate.cache.use_query_cache","true");
      aconf.setProperty("hibernate.cache.usesecondlevel_cache","true");
      aconf.setProperty("hibernate.cache.useminimalputs","true");
      aconf.setProperty("hibernate.cache.provider_class","com.hazelcast.hibernate.provider.HazelcastCacheProvider"); 

as usual you can set these above properties in the hibernate.cfg.xml.
Now put hibernate cache annotations @Cache annotation on your entities and collections.

@Cache(usage = CacheConcurrencyStrategy.READ)@SuppressWarnings("serial")
public class FdcVt extends FdcDocBase implements java.io.Serializable, IReportable, ITrXmlGenerationable {
}

Application is ready to use of 2nd level caches (assume you also hazelcast-hibernate-.jar in your calsspath), if you run the application you should see the following logs on your console or file:

22.01.2011 15:04:50 com.hazelcast.hibernate.provider.HazelcastCacheProvider
INFO: Starting up HazelcastCacheProvider...
22.01.2011 15:04:50 com.hazelcast.config.XmlConfigBuilder
INFO: Looking for hazelcast.xml config file in classpath.
22.01.2011 15:04:50 com.hazelcast.config.XmlConfigBuilder
WARNING: Could not find hazelcast.xml in classpath.
Hazelcast will use hazelcast-default.xml config file in jar.
22.01.2011 15:04:50 com.hazelcast.config.XmlConfigBuilder
INFO: Using configuration file /hazelcast-default.xml in the classpath.
22.01.2011 15:04:51 com.hazelcast.system
INFO: [dev] Hazelcast 1.9.1 (20110103) starting at Address[192.168.157.1:5701]
22.01.2011 15:04:51 com.hazelcast.system
INFO: [dev] Copyright (C) 2008-2010 Hazelcast.com
22.01.2011 15:04:51 com.hazelcast.impl.LifecycleServiceImpl
INFO: [dev] Address[192.168.157.1:5701] is STARTING
22.01.2011 15:04:53 com.hazelcast.impl.Node
INFO: [dev] 


Members [1] {
    Member [192.168.157.1:5701] this
}

22.01.2011 15:04:53 com.hazelcast.impl.LifecycleServiceImpl
INFO: [dev] Address[192.168.157.1:5701] is STARTED
22.01.2011 15:04:54 com.hazelcast.hibernate.provider.HazelcastCache
INFO: Creating new HazelcastCache with region name: ru.fors.lsadb.datamodel.FdcVt
22.01.2011 15:04:54 com.hazelcast.hibernate.provider.HazelcastCache
INFO: Creating new HazelcastCache with region name: org.hibernate.cache.UpdateTimestampsCache
22.01.2011 15:04:54 com.hazelcast.hibernate.provider.HazelcastCache
INFO: Creating new HazelcastCache with region name: org.hibernate.cache.StandardQueryCache 

Hazelcast create hibernate region for every entity, in our case it's ru.fors.lsadb.datamodel.FdcVt. You can change your region name on @Cache annotations also in hazelcast configuration file.
If you will run any query against fdcVt entity, first time query will runs on database table and the second time should use hazelcast cache. If you enable hibernate statistics through JMX, you would also see the query result hit as follows:

Now it's time to meet with Oracle database change notification, from the database version Oracle 10g (10.2) it's possible to get notification whenever any Database object's change. Oracle provide these unique features on JDBC driver for 11g. You should download the drive from the oracle site and you are ready for coding. Even more oracle given a fine grained example to use of the features.
The following paragraph i have copy and paste :-)

To use Oracle JDBC driver support for Database Change Notification, perform the following:

1. Registration: You first need to create a registration.
   
2. Query association: After you have created a registration, you can associate SQL queries with it. These queries are part of the registration.
   
3. Notification: Notifications are created in response to changes in tables or result set. Oracle database communicates these notifications to the JDBC drivers through a dedicated network connection and JDBC drivers convert these notifications to Java events.
   

Also, you need to grant the CHANGE NOTIFICATION privilege to the user. For example, if you connect to the database using the SCOTT user name, then you need to run the following command in the database:

grant change notification to scott;

For detail information you should visit the Database change notification page.
Here is the complete quick start example which will clear the hazelcast cache whenever the entity will updated:

package com.blu.misc;

import oracle.jdbc.driver.OracleConnection;
import oracle.jdbc.driver.OracleDriver;
import oracle.jdbc.dcn.DatabaseChangeRegistration;
import oracle.jdbc.dcn.DatabaseChangeListener;
import oracle.jdbc.dcn.DatabaseChangeEvent;
import oracle.jdbc.dcn.TableChangeDescription;
import oracle.jdbc.OracleStatement;

import java.util.Properties;
import java.util.Map;
import java.sql.SQLException;
import java.sql.Statement;
import java.sql.ResultSet;

import com.hazelcast.core.HazelcastInstance;
import com.hazelcast.core.Transaction;
import com.hazelcast.client.HazelcastClient;

public class GetNotify {
    private static final String USERNAME="xyz";
    private static final String PASSWORD = "w";
    private static final String URL="jdbc:oracle:thin:@xyz:1521:orcl";

    public static void main(String[] args) {
        System.out.println("Notify start");
        GetNotify notif = new GetNotify();
        OracleConnection con= null;
        DatabaseChangeRegistration dcr = null;
        try{
            con =  notif.getConnection();
            Properties prop = new Properties();
            // set the registration propetries
            prop.setProperty(OracleConnection.DCN_NOTIFY_ROWIDS,"true");
            //prop.setProperty(OracleConnection.DCN_QUERY_CHANGE_NOTIFICATION,"true");

            dcr = con.registerDatabaseChangeNotification(prop);
            // add the listener
            dcr.addListener(new DatabaseChangeListener(){
                public void onDatabaseChangeNotification(DatabaseChangeEvent e) {
                    Thread t = Thread.currentThread();
                    System.out.println("QCNDemoListener: got an event ("+this+" running on thread "+t+")");
                    System.out.println("====================================");
                    System.out.println(e.toString());
                    System.out.println("====================================");
                    TableChangeDescription[] tchanges =  e.getTableChangeDescription();
                    for(TableChangeDescription tdesc : tchanges){
                        System.out.println("Changed Object:"+ tdesc.getTableName());
                    }
                    // clear cache
                    HazelcastInstance instance = HazelcastClient.newHazelcastClient("dev", "dev-pass", "192.168.157.1", "192.168.157.1:5702");
                    Map<string, org.hibernate.cache.readwritecache.item=""> vtMaps =  instance.getMap("ru.fors.lsadb.datamodel.FdcVt");
                    System.out.println("Cache size by region:"+ vtMaps.size());
                    
                    Transaction transaction = instance.getTransaction();
                    transaction.begin();
                    vtMaps.clear();
                    transaction.commit();
                    System.out.println("Object allocated after clear cache.."+ vtMaps.size());
                }
            });

            String query = "select * from dbf_kbk";
            Statement stm = con.createStatement();
            ((OracleStatement) stm).setDatabaseChangeRegistration(dcr);

            ResultSet rs = stm.executeQuery(query);
            stm.executeQuery("select 1 from fdc_vt where 1!=2");
            stm.executeQuery("select 1 from fdc_pt where 1!=2");
            while(rs.next()){
            }
            // get tables from dcr
            String[] tables = dcr.getTables();
            for(String str : tables){
                System.out.println("Tables:"+ str);
            }
            rs.close();
            stm.close();
        }catch(SQLException e){
            System.out.println("SQLException:"+ e.getMessage());
            try{
            if(con != null && !con.isClosed()){
                con.unregisterDatabaseChangeNotification(dcr);
                con.close();

            }
            }catch(SQLException e1){
                System.out.println("e1"+ e1.getMessage());
            }
        }finally{
            try{
            if(con != null && !con.isClosed()){
                con.close();
            }
            }catch(SQLException e){
                System.out.println("e2:"+e.getMessage());
            }

        }
    }
    private OracleConnection getConnection() throws SQLException {
        OracleDriver driver = new OracleDriver();
        Properties prop = new Properties();
        prop.setProperty("user",this.USERNAME);
        prop.setProperty("password",this.PASSWORD);
        return (OracleConnection)driver.connect(this.URL,prop);
    }

}

Above snippet is very self explainable, first we creates registrations and register Entity through the sql query. Add a DatabaseChangeListner and implement the hazelcast client operation on it. Through the hazelcast client we gets cache of the hibernate region, turn on the transaction and clear the cache. Whenever you commit the changes, all the member of the Hazelcast gets the notification. Now hibernate has to run the query on the database and will fill the 2nd level cache, which will help to always getting synchronous query result on the client side.
Thank'x for reading.