Coherence 101 - EntryProcessor traffic amplification

Oracle Coherence data grid has a powerful tool for inplace data manipulation - EntryProcessor. Using entry processor you can get reasonable atomicity guarantees without locks or transactions (and without drastic performance fees associated).

One good example of entry processor would be built-in ConditionalPut processor, which will verify certain condition before overriding value. This, in turn, could be used for implementing optimistic locking and other patterns.

ConditionalPut could accept only one value, but ConditionalPutAll processor is also available. ConditionalPutAll accepts a map of key/values. Using it, we can update multiple cache entries with single call to NamedCache API.

But there is one caveat.

We have placed values for all keys in single map instance inside of entry processor object. On the other side, in distributed cache keys are distributed across different processes.
How right values would be transferred to right keys?

Answer is simple - every node, owning at least one of keys to be updated, will receive a copy of whole map of values.
In other words, in mid size cluster (i.e. 20 nodes) you may actually transfer 20 times more data over network than really needed.

Modern networks are quite good and you may not notice this traffic amplification effect for some time (as long as you network bandwidth can handle it). But once traffic has reached network limit things are starting to break apart.

Coherence TCMP protocol is very aggressive at grabbing as much of network bandwidth as it can, so other communications protocols will likely perish first.
JDBC connections are likely victim of bandwidth shortage.
Coherence*Extend connection may also suffer (it is using TCP) and proxy nodes may start to fail in unusual ways (e.g. with OutOfMemoryError due transmission backlog overflow).

This problem may be hard to diagnose. TCP is much more vulnerable to bandwidth shortage and you will be kept distracted with TCP communication problems while root cause is excessive TCMP cluster traffic.

Monitoring TCMP statistics (available via MBean) could give you an insight about network bandwidth consumption by TCMP and network health and help to find root cause.

Isolating TCMP in separate switch is also a good practice, BTW

But how to fix it?

Manual data splitting

Simple solution is to split keys set by owning nodes, and then invoke entry processor for each subset individually. Coherence API allows you to find node owning particular key.
This approach is far from ideal though:

• it will not work for Extend clients,
• you either have to process all subset sequentially or use threads to do several parallel calls to Coherence API,
• splitting of key set complicates application logic.

Triggers

Another option is relocating your logic from entry processor to trigger and replacing invokeAll() by putAll() (putAll() does not suffer from traffic amplification). This solution is fairly good and fast, but has certain drawbacks too:

• it is less transparent (put() is not just put() now),
• trigger is configured once for all cache operations (not just one putAll() call),
• you can only have one trigger and it should handle all your data update needs.

Synthetic data keys

Finally you can use DataSplittingProcessor from CohKit project. This utility class is using virtual cache keys to transfer data associated with keys, then it is using backing map API to access real entries.

This solution has its PROs and CONs too:

• good drop-in replacement for ConditionalPutAll and alike,
• prone to deadlocks if running concurrently with other bulk updates (it is partially mitigated by sorting keys before locking).

Choosing right solution

In practice I was using all three technique listed above.

Sometimes triggers fit overall cache design quite good.
Sometimes manual data split has its advantages.
And sometimes DataSplittingProcessor is just right remedy for existing entry processors.

How to simulate Coherence node failure in JUnit test

Recently, I was working on renewed version of Coherence data loss listener. New version provides simple utility to attach partition loader to any cache. Such partition loader is guaranteed to be called for every partition of newly created cache or after partition has been lost. Unlike CacheStore, partition loader will be called on the node there it has been added. This way you could have dedicated set loader processes, which are not involved at storing a data. Also it is guaranties that

only one instance of partition loader are executed for particular partition in cluster,

if there is at least one live node with registered partition loader it would be invoked for empty partition.

But let’s get back to a topic of post. I’m actively using JUnit and my test-utils library to automate testing of my Coherence stuff. Test-utils is using class loader isolation to run multiple Coherence nodes in single JVM.

But, unlike many other tests, here I need to test disaster case. Coherence should think that one of its nodes has died. Normally, I’m using CacheFactory.shutdown() to kill virtualized Coherence node, but this way it would be a graceful shutdown.

For data loss listener, I really want to test disaster case.

How Coherence track node liveness?

Naïve approach using timeout is not working well with data grid prioritizing resilience and performance such as Coherence.

What is problem with timeout?

If you let it be too short, there will be too many false positives making grid unstable (JVM may do a GC, OS may start swapping, etc).

If you let it be too long, time of recovery from disaster would be too long.

How this can be improved? Let’s see that kind of disaster could possibly happen with your cluster:

JVM process could be killed, crushed or just exited without shutdown.

Sever could crush or become unreachable via network.

Death of process could be easily tracked if you keep open TCP connection open. OS will close all TCP connections for dead process, so you could make very good assumption that remote process is dead.

Coherence is using so called TCP ring for that purpose. Each cluster node keeps two open TCP connections to other cluster nodes (forming a ring). If cluster detects that both TCP connections have been closed, it has very good reason to disconnect node right now and start recovery procedure.

In case of server/network failure, TCP connection will not be closed immediately. In addition to TCP ring, Coherence is using IP monitor to track reachability of IP addresses. If IP address cannot be reached by rest of nodes, cluster will not hesitate to disconnect all nodes from that IP.

This two tricks allow Coherence to detect real failure very fast, yet to be very tolerant to long GC pauses and other non fatal slowdowns.

Steps to kill node in JUnit test

In JUnit test all nodes in cluster are sharing same JVM. I cannot really kill a process. To simulate node death, I’m calling Thread.suspend() on all threads related to victim node (a feature of test-utils). This is making node totally unresponsive.

Two mechanisms above should be turned off in Coherence operational configuration. Disconnect timeout also should be set to smaller value (otherwise each test will take too long).

That is it, now I can test disaster cases for Coherence using JUnit.

Below is snippet of actual test:

@Test public void verify_parallel_init_crash_case() throws InterruptedException { final int partitions = 2000; final int timeout = 15000; CacheTemplate.usePartitionedServicePartitionCount(cluster, partitions); CohHelper.setTCMPTimeout(server(0), timeout); CohHelper.disableTcpRing(server(0)); CohHelper.setTCMPTimeout(client(0), timeout); CohHelper.disableTcpRing(client(0)); ... server(0).getCache("a-cache1"); statics().initPartitionCounter(partitions); // init Coherence nodes client(0).getCache("a-cache1"); client(1).getCache("a-cache1"); ... // attaching test partition loader attachTouchMonitor(0, 20, "a-cache1"); attachTouchMonitor(1, 20, "a-cache1"); ... Thread.sleep(500); System.out.println("Simulating crash for 2,3,4,5"); // simulating client crash, verify lock revocation client(2).suspend(); client(3).suspend(); client(4).suspend(); client(5).suspend(); // waiting for all test listeners to finish statics().waitAllLatches(); System.out.println("Latches are open"); Thread.sleep(200); // checking cache state assertAllCanaries(0, "a-cache1"); }

Here is a link to full java file in SVN.

Tech meet up, distributeted caching and data grid, Moscow 17 May

I would like to announce tech meet up devoted to topic of caching in distributed systems and data grid technology. Event will be held at Moscow on May 17.

Slides from event:

Main talk by Max Alexejev

Bonus presentation by me

Asymmetric read/write path – a trend for scalable architecture

A few years ago I was blogging about architecture using composition of different storage technologies for queering and persistence of data. Rationale behind this was further explained in other my post.

Below is a sketch of such composition:

Unlike classical Von Neumann's memory hierarchy, this composition is asymmetric in terms of read and write paths.

I’m glad to see similar ideas implemented in commercial products marking a trend. In this post I want draw your attention to two very interesting middleware products having principle of composition of specialized storages in their core.

CloudTran

CloudTran is very ambitious product promising performance and scalability for a wide class of application without much extra effort.

CloudTran leverages Oracle Coherence and its integration with EclipseLink to build scalable applications using JPA for persistence. Coherence + EclipseLink (TopLink Grid) is already capable of executing JPQL queries in cache instead of database using rich querying capabilities of Coherence. Missing piece in this tandem was transaction support.

CloudTran is filling this gap adding specialized component for managing transactions. Durability of CloudTran’s transactions is provided by write-ahead disk log (many RDBMSes are using same technique). But unlike RDBMS, CloudTran’s log is not limited to single disk/server, it is distributed (same way as data in Coherence grid) and can benefit from throughput of dozens of disks in cluster.

Full picture of CouldTran based solution is triangle of technologies:

Coherence for fast data retrieval,

CloudTran transaction log for fast and durable transaction persistence,

backend database (relational or NoSQL) is a system of record and long term storage.

Backend database being updated asynchronously is on critical path for neither read nor write, thus database is not limiting application performance. On other side data in Coherence are updated synchronously, so application logic can enjoy strong consistency and ACID transactions (which is huge, eventual constancy is a lot of pain for typical enterprise application with sophisticated data model).

Datomic

Datomic is another young and interesting product promising combination of ACID and scalability. Datomic is also featuring triangle of technologies, but it has own implementation for both in-memory database/cache and transaction persistence (that component is called transactor in Datomic). For system of record you can use either RDBMS or NoSQL storage (Amazon Dynamo). Datomic is offering own API (and own unique approach) for working with data. Datomic is highly influenced by functional paradigm, which would probably make porting existing applications to Datomic non trivial, but for new projects idea of simple but scalable platform featuring ACID data manipulation may be attractive.

Cool toys for enterprise developers

I’m very glad to see such innovative products addressing scalability in not-so-fancy class of enterprise application (of cause both products are not limited to enterprise). IMHO there are enough clones of Google’s BigTable and Amazon’s dynamo in this world already. People working on inventories, reservation systems and other enterprisy stuff also need cool distributed toys.

I’m a little skeptical about future for these products (goals they have set for themselves are just too challenging), but I sincerely which luck to both projects.

Please prove my skepticism wrong ;)

Data Grid Pattern - Canary key set

Most of modern in-memory-data-grid products have grown from distributed caching. For traditional cache, loss of data is not a big deal, missing data could always be recovered from backend storage via read through. Main focus of distributed cache is data coherence between nodes (preventing stale reads, etc).

Advanced patterns like all-in-memory and proactive caching may provide considerable benefits over traditional read through cache. But simple fall back to read-through as a strategy for data recovery may not be an option for these advanced cache usages. Read-through have two prerequisite to be implemented:

• data should be accessed by primary key,
• for given primary key, you should known its master data source.

Both these prerequisite may be broken in advanced solution. Having all-data in memory will allow you to execute queries in cache layer instead of backend database. Products like Oracle Coherence, GemStone GemFire and GigaSpaces have advanced queering capabilities (including non-primary indexes support). Offloading queries from database is a huge win, but the price is that you cannot rely on read-through any more. If some data is missing in cache, queries will produce incomplete results without warning.

Second read-through prerequisite also may be sacrificed, i.e. by using multiple backends (cache acting as aggregator for data set scattered across multiple databases). You can more details in my previous article.

Data loss imminent

Please mention, that loss of data in modern in-memory-data-grid is an exceptional event. Data is usually protected by multiple replicas and grid can tolerate server failure. But it still possible and you cannot ignore this aspect as you cannot ignore e.g. backing up your database.

Through all reliability provided by data grid technology data may be lost and it means they will be lost eventually. Next question, what is your desired strategy to cope with incomplete data set?

It depends on type of application.

•  For some applications, it is ok to have incomplete results from application during recovery window.
• For some application, incomplete response is worst than no response. Application should guaranty that every response is complete and if it cannot provide complete response (e.g. part of data set is missing in cache) it should raise an error.

First strategy is rather simple, you should monitor your grid and automatically trigger recovery procedure if disastrous event is detected.

Second type of strategy is more tricky in implementation. Just monitoring is not an option.  There would be a gap between data loss event and reaction of monitoring system (which e.g. can switch service to offline mode for duration recovery process). Some application are totally intolerant to inconsistent data. This data, for example, could be used in complex batch of financial risk calculations (running for few hours in large HPC grid) and single inconsistent piece of input data could invalidate whole batch of work.

We need a solution better than monitoring for such kind of applications.

Canary keys to detect missing data

We must guaranty that result of each query is consistent (i.e. all data that has to be processed has been processed, often this means - whole dataset). Here we have a paradox at hands: we must check presence of  certain key/value pairs (data grid is a key/value storage) in cache, but we cannot know keys of these pairs or even their total number.

Solution to this paradox lies in base approach of data distribution used by data grids. Technique described in this article has been used by me with Oracle Coherence grid, but idea can also be applied to other products using similar type of DHT. Oracle Coherence is using partitioned distributed hash table (DHT). In practice this means that key/value pairs are not distributed individually, but whole partition is assigned to particular node. It also means, that you cannot lose individual key/value pair but only whole partition at once (if you lose all replicas of that partition at the same time). Number of partitions is fixed for live time of grid (changing that number requires rebuild of DHT).

How this could be helpful to ensure data completeness?

We may not be caring about presence of individual key/value pairs, instead we may check presence of all partitions (and we know their exact IDs and total number). But how we can check presence of partition (technically partition cannot be missing, it will be just empty)? Also we should join data completeness check with queering of data in the same operation otherwise we will always have a gap of uncertainty.

Canary keys is a trick to solve this problem. Canary keys are synthetic keys, you put just on key to each partition. Every partition should have a canary key. So if your grid is configured to have N partitions, it should contain exactly N canary keys. If number of canary keys is less than N, that means portion of data has been lost (poor canary has perished) and is not recovered yet. Of cause your data loading/recovery procedure should put canary keys back in cache once data is restored.

It is also possible (though quite awkward to be honest) to integrate canary keys check in single request with data queering. In each query you should select canary keys along with actual data you need to retrieve. Once query is executed you should check presence of all canaries and then strip them from result set. If all of them are there, you can be sure that your result set in complete and no SLA would be broken. If some canaries are missing, you should trigger data recovery and/or raise an error indicating that request cannot be completed until recovery is done.

Conclusion

Technique described in this article is very advanced. Most applications do not require such rigorous consistency checks for every operations. But few do, and for them this approach may be useful (still implementation may have its quirks).

On high level canary keys technique demonstrate how understanding of  core principles of DHT can help solving challenging task. Understanding of low level grid operations, their guaranties and limitation is a cornerstone in engineering of complex data processing solution using distributed ACID-less data grid as a storage.

See also

Open source implementation of canary keys framework - http://code.google.com/p/gridkit/wiki/DataLossListener.

Data Grid Pattern - Time series index for managing versioned data

Many critical application are using append only approach for dealing with transactional data. In other words, they never update data records, but instead insert new records with greater timestamp (or sequence number, or any other kind of version, they are using to find latest record). Common challenge for such data model, is how to retrieve an appropriate version of record (e.g. latest version, or version at certain moment in time). A simple query for latest version for key 'A' would translate into query as complex as

select * from versions where series='A' and version = (select max(version) from versions where key='A')

This query is already too complex for data grid (and even RDBMS would not be too happy).

Accidently Ben Stopford has recently published a great article about this problem, outlining a lot of important aspects. I do not want to repeat him here, I suggest you read his article now, then continue with mine, which complements his two approaches this third one using custom index in Coherence.

Using custom index for accessing versioned data

In this  approach, each version is stored as a separate entry in cache.

Entry key is composite key, including logical key (series key) and some additional field to make version key unique (e.g. transaction ID, sequence number etc). Value contains actual business data (payload) and timestamp, we are using in our queries (technically timestamp could be part of composite key).

Series key should be an affinity key also - all versions related to same series should be physically on one Coherence node (or affinity key can be a part of series key, this will also satisfy this requirement).

Normally, if you want to find certain version by series key and timestamp you have to do aggregation of all versions for this series. In two approaches mentioned by Ben Stopford, latest version is separated from all other versions (using separate cache - approach 1, using marker - approach 2). It solves problem of finding latest version, but doesn't help if we need to find version for certain moment at time.

Time series index structure

Normal Coherence indexes cannot help us much, due to complexity of query, but it is possible to create custom index, tailored specifically for this task.

Time series index is similar to traditional inverted index, but instead of storing set of entry references, it is storing a nested index, indexing  only versions belonging to certain series by timestamp. Using this index structure you could find latest version or version for certain moment in time, without any aggregation, just by index lookup.

This index goes beyond standard index Coherence API, so it requires a complementary implementation of custom filter.

PROs and CONs

Below and PROs and CONs of this approach, compared to approaches from Ben's article.

PRO

• Inserting new version doesn't require modifications of any other versions. In particular, you do not need to use hack, directly accessing to backing map, and you do not create extra replication traffic.
• Time series index works efficiently for any point in time, not only latest versions.
• It can be used with any kind of caches (even with continuous queries).

CON

• Through custom index usage is straightforward, troubleshooting could be very tricky unless you understand index mechanics very well.

Source code

Time series index implementation is available at GridKit project.

Data Grid Pattern - Proactive caching

Classic and most widely used approach for caching is read through pattern. Look up in cache, then try to load from primary data source if entry is missing in cache - that is how it works. This pattern is easy to implement but it has few unpleasant limitations:

• caching may reduce average response time, but maximum response time is still bound to back end data source response time,

• cache may have stale data, expiry policy may relive this problem to some extent, but aggressive expiry is drastically reducing performance gain from caching.

All in memory pattern

Caching concept is close relative to memory hierarchy principle. Memory hierarchy is one of cornerstones of Von Neumann architecture relies on fact that we have different kinds (in terms of capacity and performance) of memory in system.  With modern hardware dynamic memory capacity is often large enough to keep whole dataset. All-in-memory is term used to describe caching architecture there you have 100% of your data in cache at all times. Having all data in cache allow you to guaranty that no request will have to hit slow backend data source and thus provide more aggressive SLA for max response time. It also may be required for workloads with highly random access to data, there traditional assumptions like 80/20 are not working.

While all-in-memory approach is definitely win in terms of performance, its implementation has few serious challenges:

•  cache should have enough capacity to hold 100% of our data,
• cache should be fault tolerant (losing a portion of data in cache will render it defunct until missing data would be reloaded),
• preloading procedure is often non-trivial due to scale of data set,
• cache should be kept in sync with backend data source.

Capacity and fault tolerance are provided by modern distributed caches out of box, but preloading procedures and cache update strategy are very application specific and fairly challenging to implement.

Below are few practical approaches for keeping cache in synch with primary data source.

Refresh ahead

This is approach similar to expiry policy, but instead of invalidating data, cache proactively refreshing them from master source. Refresh ahead pattern is quite simple, but not very practical though. If cached data set is large (and we are talking about all-in-memory pattern) automatic refreshing is like to overwhelm backend with requests. And even if data set is reasonably small we still have to use fairly long expiry time to make it practical.

So if you looking for all-in-memory cache, refresh ahead are unlikely to help you.

Proactive caching

In contrast with traditional (I would say reactive) caching, with proactive caching pattern you insert/update value in cache as soon as it is updated in backend data source, not at the moment data was requested from cache.

Proactive caching is not necessary should be used with all-in-memory pattern, but combination of these two is very powerful.

Polling updates from DB

An evolutionary step from refresh ahead to proactive caching, would be polling changes for database. Some daemon component should periodically (and frequently) poll database and fetch changes since last cache update. Sounds simple but you have to come out with a way how to "fetch changes since last cache update".  Usually some kind of timestamp is used - each record in table to be cached has a kind of last modified field. Another few challenges:

·         what if cache is representing a query result, not just a single table,

·         'last modified' field should be indexed, otherwise frequent polls will bring database to its knees,

·         polling daemon should be fault tolerant,

·         cache timestamp should be stored somewhere on cache side.

So, implementing this approach will require some amount of work (on both sides, database and cache), but in the end you will have very robust solution.

Only sever limitation of this approach is lag between changes in database and cache, which is no less than poll period.

Long poll

If database has some means for wait/notification in its query language, you can use long poll pattern. 

Using long poll will reduce load on your database server and probably reduce lag between cache and database. Disadvantages of this approach: more code on database side and need to use dedicated thread(s) for polling (because thread will be blocked waiting notification of database side for most time).

In Oracle database long poll could be implemented using DBMS_ALERT package. Though use of DBMS_ALERT may cause serialization of update transaction and harms database performance.

Database notifications

Some databases can push data change notifications directly to clients, without need for polling. E.g. Oracle database has DCN (data change notification) mechanism. Using DCN you can register callbacks which would be notified that certain data have been changed in database. Notification mechanism has few advantages over polling

• less load of database while no data is actually changing,
• smaller lag between changing data in database are reaction in cache.

Notifications approach have disadvantages also

• API usually more complicated, you have to learn more quirks to make it work,
• connection hang problems - application is listening to event on connection which is defunct for some reason,
• notifications may be lost in transition for some reason.

One particular problem with using Oracle DCN in java, was leaking of subscriptions. DCN subscription is remaining active on database side after termination java process (unless it was deregistered explicitly) and eventually you are going to hit limit for active subscriptions. Of cause you should deregister subscription before terminating of client process, but you cannot always guaranty graceful shutdown in practice.

Hooking into database replication

All mature databases have replication feature and thus replication wire protocol. Sometimes, they also provide API to hook into replication channel and programmatically receive all updates (essentially change notifications).

Replication is implemented differently in various databases (or even with different replication solution for same database). But general idea is to make cache act as replication slave for its source database.

Compared to polling or data change notifications, using of replication usually requires more effort from DBA side (they should setup replication slave of master database). It may also cost you some in licensing fees dependent on your replication solution.

MySQL slave protocol does not require any setup on master, so replication links can be created ad hoc. But MySQL has another catch, you should setup row based replication on master database unless you want to parse and execute SQL statements in your cache.

Single cache in front of multiple sources

In large scale system, primary data source may be distrusted itself  (e.g. using sharded database). Having single read through cache may be a problem in this case (doing read through, you have to know which shard to consult about data missing in cache), but with proactive caching such setup it much straightforward. While in read through caching, cache responsibilities of serving requests and acquiring data are coupled. With proactive caching, responsibilities of storing data/serving read requests and feeding cache with data updates may be separated. This way, you can have single cache instance and multiple other components pushing data into it (e.g. each sharing could push its data in single cache).

In this role cache can be though as a kind "materialized view" based up on data in primary (potentially distributed) data source.

Few more links

·         Coherence SIG: Using Oracle Coherence to Enable Database Partitioning and DC Level Fault Tolerance

Using Database Change Notification (DCN) with a Coherence Cache

Coherence write behind, finding not-yet-stored entries

Write behind strategy may be very useful in certain cases. Using Coherence you can use this pattern in very smart way - synchronously replicate your data in memory over several nodes, then asynchronously write to slow external storage. This way you will not lose any data in case of single server outage. One unpleasant thing is that Coherence does not retain sequence of updates in certain cases, but wait,  this is a distributed system after all :)

But there is a catch. Let's assume you are receiving message via JMS, put data to Coherence grid, start processing etc. But you do not want to acknowledge message until it is written to persistent storage.

While Coherence let your data survive single node failure, there could be network failure, logical bug in your system or outage of persistent storage your are writing to. You want to be consistent, you can process message, but you do not want to confirm its delivery (delete it from one persistent storage) until it is not written into another persistent storage.

So, is it possible to find which entries in cache are not persisted yet?

Yes, it is. Internally Coherence marks not-yet-stored entries with special flag. This flag is usually invisible for application, but we can access it using some low level Coherence API.

Below is StoreFlagExtractor returning FALSE for vulnerable entries.

public class StoreFlagExtractor extends AbstractExtractor implements PortableObject {

    private static final long serialVersionUID = 20010915L;

    public StoreFlagExtractor() {

    }

   

    @Override

    public int compare(Object object1, Object object2) {

        // make no sense

        throw new UnsupportedOperationException();

    }

    @Override

    public int compareEntries(com.tangosol.util.QueryMap.Entry entry1, com.tangosol.util.QueryMap.Entry entry2) {

        // make no sense

        throw new UnsupportedOperationException();

    }

    @Override

    public Object extract(Object object) {

        // decorator can be extracted only from binary entry

        throw new UnsupportedOperationException();

    }

    @Override

    @SuppressWarnings("rawtypes")

    public Object extractFromEntry(java.util.Map.Entry entry) {

        BinaryEntry binEntry = (BinaryEntry) entry;

        Binary binValue = binEntry.getBinaryValue();

        return extractInternal(binValue, binEntry);

    }

    @Override

    public Object extractOriginalFromEntry(com.tangosol.util.MapTrigger.Entry entry) {

        BinaryEntry binEntry = (BinaryEntry) entry;

        Binary binValue = binEntry.getOriginalBinaryValue();

        return extractInternal(binValue, binEntry);

    }

   

    private Object extractInternal(Binary binValue, BinaryEntry entry) {

        if (ExternalizableHelper.isDecorated(binValue)) {

            Binary store = ExternalizableHelper.getDecoration(binValue, ExternalizableHelper.DECO_STORE);

            if (store != null) {

                Object st = ExternalizableHelper.fromBinary(store, entry.getSerializer());

                return st;

            }

        }

        return Boolean.TRUE;

    }

    @Override

    public void readExternal(PofReader paramPofReader) throws IOException {

        // do nothing

    }

    @Override

    public void writeExternal(PofWriter paramPofWriter) throws IOException {

        // do nothing

    }

}

Using this extractor you can easily query unstored entries, set listeners and even build indexes and CQ for such entries.
UPDATE: Changes of STORE decoration are not triggering cache events (only backing map events) and index updates. Nor CQ nor indexes nor listener would not work.