My historic Coherence Incubator association

Elastic Charging Engine on Raspberry Pi

Fault Tolerance in Elastic Charging Engine

Joke: The pilot screamed - The plane is overloaded and we need four people to jump off the plane to save it. Unfortunately we only have three parachutes so one person has to make a sacrifice. The Doctor said - I save lives and I must be saved and he jumps with one parachute. The Lawyer said - I am the smartest person on Earth and I must be saved  and he jumps with a parachute. Seeing no one else was jumping, the priest said to the kid - Son, you have your entire life ahead of you, you take the last parachute and jump and I will sacrifice my life. Kid said - Don't worry Father, the smartest person on Earth just jumped off with my backpack.

Overloaded systems require drastic measures to be taken and different patterns have to be applied to save the system. What is important to understand is, once overloaded the system operates on the edge of the complete meltdown and patterns that are applied to mitigate and recovery are different from other types of fault recovery.
Elastic Charging Engine was built with fault tolerance in mind. With Oracle Coherence providing the core execution platform there are implicit fault tolerance in design. ECE extended its fault tolerance in its own design:

• Redundancy
• By the virtue of Oracle Coherence's design each cache entry in a partitioned scheme are backed up on another node preferably on another host. If the primary cache entry is lost the back up becomes the primary and in process creates a new backup on another host. Coherence implicitly is NODE SAFE. Any change in the state of the primary cache entry is by default applied to its backup synchronously before calling the process success (Pattern (I) - Data Redundancy).
•  After the provisioning exercise of finding the total number of nodes required for a given data set one additional host is typically added called a (n+1) configuration to make sure the cluster can sustain a machine failure without losing any data.

• Processing are co-located with Data

• As when/if nodes die the partitions move to new members and processing scheduled on those lost partitions are moved to new members. (Pattern (II) - Processing Failover).

• Need to provision the system to minimize the time taken for partition move. Faster the data moves to new owners, faster the processing moves. Minimize the Mean Time To Recovery.

• Requests destined for the just lost member are put in a suspense queue and are retried when the partitions get its new owner (Pattern (III) - Retry).

• Checkpoints
• ECE by default bundles requests together so that a batch of requests for Customers that are managed in partitions owned by a single cluster member can be processed together. This strategy is key to support higher through puts.
• If a batch is partially processed that some requests have been processed but not all before the node dies the same batch is submitted to a suspense queue to be retried. Problem is we do not want the same request already processed before to be reprocessed. 
• When a request is submitted ECE checks if the same request has been processed before (Pattern (IV) - Duplicate Detection).

With its second released version (11.2.0.0.0) also nick named Amazon, Elastic Charging Engine introduced a new feature "Overload Protection" extending its commitment to provide a true fault tolerant system.

Overload Protection:
ECE provides a layer of APIs that allows any application to interact with its system. This layer has APIs to construct messages that ECE can understand, data structures to send the request and receive responses in and policies around how when these requests are processed. Internally this layer called Batch Request Service (Or BRS) is an amalgam of the following behaviors:

• Identifying the Member where the submitted request be sent to for processing.
• Maintaining a batch of requests, its ripe policies and its responses.
• Tuning parameters to provision the ECE for ultra low latencies or excessive high throughput or something in the middle.
• Maintaining a suspense queue for the requests that needs to be retried.
• Managing a high degree of parallelization of threads.
• To make sure the entry point to the charging system can not be overloaded either by a natural increase in network traffic or denial of service attacks.

How does it work?
BRS has a pool of Threads dedicated to process events. Most of the time these threads wait for new batches to be sent but as these batches are sent to processing nodes of the cluster the CPU usage on those nodes increase. ECE is primarily a CPU bound application and the most precious element in its design are CPU cores. Though ECE has been benchmarked for extremely high throughput (>1000 requests/sec/core) but sending an extremely high number of requests still risks high CPU usage on processing nodes to spike up resulting in long queue build ups in BRS and eventually resulting in a meltdown.
ECE v11.2.0.0.0 introduced a new overload protection that guarantees if provisioned right will never accept requests that can not be processed with in a given SLA irrespective of how high the incoming throughput is. This is a key feature of the product when it comes to deploying a fault tolerant system at different levels. And the pattern used comes right out of the Joke in the beginning of this blog.

Pattern (V) - Shed Load:
BRS monitors the internal queue of the Thread Pool Executor and at any given time if the pending count is more than a configured value it rejects the submitted request. The rejected request has to be resubmitted by the network mediation based on certain policies typically suggested by ECE. As the system being in an overload situation is dynamic the pending queue size check is done for each submitted request. It is quite possible that one request has been rejected but immediately after requests are accepted for processing. Shedding Load is a legitimate fault mitigation policy specially more so in latency sensitive applications.
Question is how do we configure the acceptable pending count? It depends on the infrastructure's latency sensitivities. Wait adds to Latencies. If it is required to have zero latency impact on acceptable requests then the acceptable pending count should be set to zero. As the requests are submitted it would go through the provisioning flow immediately after. As the acceptable pending count is increased the average latencies would increase as well and between none and equal to the throughput will dictate what the latency per request would end up in.
Following are two snapshots taken from one of a smaller test setups. One snapshot plots throughput and the other latencies. In the following test the infrastructure is configured for around 40K throughput with acceptable latencies of around 50ms. The snapshot is taken from one of the instances of the event simulators configured at 20K with total of two such simulators running on one test host. Look at how the latency increases as the acceptable count is increased in an "overloaded" system gives an insight how this feature should be configured. At 120K which for this test set up was way more than what it was provisioned for, at zero acceptable count had no impact on the processing latencies at all even if the system was receiving way more number of requests than what the infrastructure is provisioned for. Acceptable pending count @20K increases the latency but still prevents a complete system meltdown. So a single tuning parameter allows you to prevent the system from fault to failure.

 














Enjoy!

My JavaONE 2013 session

And the recorded presentation:

Colocating data in Coherence and NoSQL

Key/Value pair based distributed systems that partition data based on some hashing of keys pose a challenge of unnecessarily incurring additional network cost in storing and reading of strongly associated data with additional challenges of atomicity of operations across this data set. Fortunately both of Oracle's best of breed big distributed data source Coherence and NoSQL support co-locating associated entries and provide a rich set of APIs to create atomic operations. In both the systems association is dictated by the Map' keys, only differently.

Association in Coherence

Two entries will be co-located in the same partition if

1. The key of the associated entry implements a Coherence KeyAssociation interface (Or define a KeyAssociator and configure it in the cache config).
2. If these two entries are in two different NamedCaches then they must share the same cache service.
3. When the cache service calculates the partition Id for entries, it uses the associated key if the cache key implements the KeyAssociation interface. 
4. Coherence provides a set of APIs typically called TransactionLite framework to access associated entries from the BackingMap as a single unit of transaction. You can refer to the blog for an example.
5. Because Coherence keys can be complex objects it is much more flexible in terms of storing complex data structure.

Association in NoSQL

NoSQL has much simpler design but it is limited in a sense that it requires keys to be Strings.

1. NoSQL keys are structured as multiple major and minor keys.
2. If we are storing People in NoSQL and want everyone with the same last name to be in the same partition then, keys can be structured as:
1. /familylastname/-/myfirstname
2. /familylastname/-/wifesfirstname
3. /familylastname/-/kidsfirstname
3. Though these are three Map entries as they all share the same major key /familylastname will all end up in the same partition.
4. NoSQL then provides APIs like multiDelete() and multiGet() to delete or retrieve multiple keys as one atomic operation.

Working with Coherence and NoSQL together  

Oracle's Elastic Charging Engine is one such product that uses Coherence and NoSQL together with Coherence as a rating engine and NoSQL as its fast persistence store for the processed RatedEvents. That said aside, such integration is fast becoming very common and following pieces of codes demonstrate some of its use cases.

Code Samples

Lets start with an interface:

public interface NoSQLKeyAssociation extends KeyAssociation {
    String getRootKey();
}

public abstract class PartitionAwareNoSQLCacheStore extends AbstractCacheStore
 {
    private KVStore kvStore;

    public PartitionAwareNoSQLCacheStore(String kvServers) {
        KVStoreConfig kvStoreConfig = new KVStoreConfig("addressStore", kvServers);
        kvStoreConfig.setConsistency(Consistency.NONE_REQUIRED);
        kvStore = KVStoreFactory.getStore(kvStoreConfig);
    }

    @Override
    public void store(Object key, Object value) {
        storeAll(Collections.singletonMap(key, value));
    }

    @Override
    public void storeAll(Map map) {
        Map<NoSQLKeyAssociation, Object> transformedMap = transformer(map);

        Set<Map.Entry<NoSQLKeyAssociation,Object>> set;
        set = transformedMap.entrySet();
        ArrayList majorComponents;
        ArrayList minorComponents;

        for (Map.Entry entry : set) {
            majorComponents = new ArrayList();
            minorComponents = new ArrayList();
            NoSQLKeyAssociation associationKey = entry.getKey();
            majorComponents.add((String)associationKey.getAssociatedKey());
            minorComponents.add(associationKey.getRootKey());

            Key key = Key.createKey(majorComponents, minorComponents);
            byte [] bytes = ExternalizableHelper.toByteArray(entry.getValue())
            kvStore.put(key, Value.createValue(bytes));
        }

    }

    public abstract Map transformer(Map entries);
    
}

The method transformer( ) allows for the Cache Store's concrete class to either transform the Map entries in some form or return the Map as is if the keys are already of type NoSQLKeyAssociation. If not this method gives an opportunity to convert the key to a type of NoSQLKeyAssociation, for one.

In example where we need to store a Customer and all his Addresses a key similar to the following can be used, both to store these values in Coherence' NamedCache (CUSTOMER and ADDRESS caches) and also to NoSQL and in each system co-located together.

public class CustomerAddressAssociationKey implements NoSQLKeyAssociation {
    String customerId;
    String addressType;

    public CustomerAddressAssociationKey(String customerId, String addressType) {
        this.customerId = customerId;
        this.addressType = addressType;
    }

    @Override
    public Object getAssociatedKey() {
        return customerId;
    }

    @Override
    public boolean equals(Object o) {
        if (this == o) return true;
        if (o == null || getClass() != o.getClass()) return false;

        CustomerAddressAssociationKey that = (CustomerAddressAssociationKey) o;

        if (!addressType.equals(that.addressType)) return false;
        if (!customerId.equals(that.customerId)) return false;

        return true;
    }

    @Override
    public int hashCode() {
        int result = customerId.hashCode();
        result = 31 * result + addressType.hashCode();
        return result;
    }

    @Override
    public String getRootKey() {
        return addressType;
    }
}

If two addresses HomeAddress (addressType = HOME) and WorkAddress (addressType = WORK) are created for a Customer with customerId = 1 then they will be stored as following:

In Coherence, Customer and his two addresses will be co-located as:
CUSTOMER Cache:

• <1, Customer>

ADDRESS Cache:

• <<customeraddressassociationkey<HOME>, HomeAddress>
• <<customeraddressassociationkey<WORK>, WorkAddress>

In NoSQL Customer's both address co-located as:
/1/-/HOME, byte[] for HomeAddress
/1/-/WORK, byte[] for WorkAddress

Enjoy!

Measuring a Toddlers weight and Object size in heap

...And the common problem is they both need some warm up phase. Toddlers are notorious to not stay standstill and if you have one you know typically the first reading is almost always wrong. But for these naughty Kids there is a way to measure it - Hold them in your arms measure your weight with the kid and then measure yours alone and subtract it. This modelling can also be applied to when Objects need to be measured for their size in a heap. Why? Because flakiness do exist with JVMs and based on sheer observation the first reading is almost always wrong. So not being part of a JVM team how am I suppose to measure an Object size fairly accurately? By reducing the flakiness. By measuring it multiple times. By minimising the affect of chaos.
Problem: To provide an API so that it can measure the size of an Object passed to it

Solution: Clean up the JVM, measure the heap size, create multiple instances of the Object, create strong references to these instances. clean up the JVM again and measure the heap. Take a difference and divide by the total number of Objects used.

If you are looking for measuring the size of Objects in a Coherence cache then look into Coherence's MemoryCalculator APIs. This solution uses Coherence's PoF framework for serialzing and deserializing non-Serialozable Objects.

1. We need to create multiple instances of the Object passed so that strong references can be maintained to these Objects and do not let GC collect these. So make multiple instances of the passed object. How? There are multiple options -

• If Object implements Serializable or any of Serializable types - Serialize the Object into byte array and to create an instance use this byte array to reconstruct the Object.
• If Object is Cloneable - To create a new instance then clone the Object
• If Object is neither Serializable nor Cloneable - Oracle Coherence provides a mechanism to serialize a non-serializable Object called Portable Object Format. PoF as it is commonly called, allows programmers to write external serializers for an Object that does not implement Serializer interface. These Serializers and the Object they serialize and deserialize can be defined in a pof configuration and loaded by a Coherence system property tangosol.pof.config. Once this is done the Object is ready to be serialized.

For the third option Coherence provides a utility in ExternalizableHelper to convert the Object into a Binary:

ConfigurablePofContext pofContext = new ConfigurablePofContext ("my-pof-config.xml");
then:
Binary binObj = ExternalizableHelper.toBinary (objToBeMeasured, pofContext);

This binObj can then be used to create new instances to make multiple strong references:

Object[] objects = new Object [1000];
Runtime runTime = Runtime.getRuntime();

for (int i = -1; i < 1000; ++i) {
     Object o = ExternalizableHelper.fromBinary(binObj, pofContext);
    // -- Reject the first object
     if (i >= 0) {
        objects[i] = o;
     } else {
            o = null;
            // -- Execute GC;
        beforeSize = runTime.totalMemory() - runTime.freeMemory();
     }
}

Execute GC again;
afterSize = runTime.totalMemory() - runTime.freeMemory();

Use the difference of (afterSize - beforSize)/1000I also found a very good implementation of "Execute Garbage Collection" from an article on the javaworld.com, that I am reproducing it here:

for (int i = 0; i < 4; ++i) {
    long m1 = runTime.totalMemory() - runTime.freeMemory();
    long m2 = Long.MAX_VALUE;
    for (int j = 0; (m1 < m2) && (j < 500): ++j) {
         runTime.runFinalization();
         runTime.gc();
         Thread.yield();
         m2 = m1;
         m1 = runTime.totalMemory() - runTime.freeMemory();
    }
}
Watching your baby's and Object's weight is critical and you know it why? Enjoy!

Coherence Transaction, Semaphore and a Joke

A DBA walks in a NOSQL bar hangs around for a few minutes, turns back and leaves. He couldn't find a table. [Quoted]

When the Data Grids came to existence it tried to solve a different set of problems that a typical relational database was not geared to address. Some failed trying imitating a Database, some remained distributed cache managers and some just got lost in space. The ones who succeeded were the ones who remained focused in solving the performance, scalability, manageability and predictability of the application state in the middle tier - actually all of them. But as they continued to sink in the Enterprise application strategies one question remained to completely solve - Transactions. Oracle Coherence couldn't defer a solution for too long either and announced a new cache scheme in its v3.6 to support true transactions extending and then deprecating its TransactionMap. Before I show how this new scheme could be used in one way lets talk about how some of these transactional problems were addressed earlier.

If an analogy has to be made between a Database and Coherence, we can think of cache schemes as a database schema and the caches as database tables.

Different solutions popped up for different level of sensitivities of desired guarantees.

Problem: Update Person and its Address together

Solving it in domain model
Solving it in the scheme
Solving it in process and, 
Solving it correctly

Domain Model
Coherence EntryProcessors provide an unique guarantee - Without an use of explicit lock only one process can modify an Entry at a given time no matter how many backups of that Entry is maintained in the grid. And as node fails the in-flight executions will move to the new owners of the partitions as Entries are repartitioned. For the clients as if nothing has happened other than some delays.

public class Person {
   private Address address;
}
NamedCache nCache = CacheFactory.getCache("Person");
nCache.put ("person1", new Person());

nCache.invoke("person1", new PersonAddressUpdaterEntryProcessor());

made sure that Person and his address were updated together while the Person was already locked. The key to this solution lies in the domain model by hiding any direct access to the Address by encapsulating it inside the Person who owns it.
But what if everytime a Person is updated its not always that his Address is touched? This requires that Person and Address can be split into multiple caches and for cases when only Person is updated the deserialization cost can be reduced.

Key Association and Transaction Lite
Everyone who has bought or rented a place knows the following three words - Location, Location, Location. When it comes to Transactions its colocation, colocation, colocation. Move related data together so that challenges of failures like network, nodes and partial failures can be contained with in a single node. In pre v3.7, not completely safe in all scenarios but as long as Addresses are not directly updated a clever solution could be used to update them together.

public class Person {
 private String id;
 public String getKey() {
    return id;
 }
}
public class Address implements KeyAssociation {
 public Address (String personId) { .. }
 public Object getAssociatedKey () {
    return personId;
 }
}
NamedCache pCache = CacheFactory.getCache("Person");
NamedCache aCache = CacheFactory.getCache("Address");

pCache.put ("person1", new Person());
aCache.put ("address1", new Address("person1"));
nCache.invoke ("person1", new MyEntryProcessor());

The key association between the Person and the Address will make sure that no matter if two different caches are used to put these two Entries they will end up together as long as these two caches use the same cache scheme. And then the following:

public class MyEntryProcessor implements InvocableMap.EntryProcessor {
public Object process (InvocableMap.Entry entry) {
  final BackingMapManagerContext addressCtx = ((BinaryEntry) entry).getContext();
  Map map = addressCtx.getBackingMap ("Address");
  Address address = map.get ("address1");
  ...
  entry.setValue(...);
}
}

This works as long as no other process updates the Address while the Person is locked by the EntryProcessor.
If you are already using v3.7 the getBackingMap() API has been deprecated and replaced with thread safe getBackingMapEntry() API and this new framework is called TransactionLite.  If Entries could be co-located then Transaction Lite is worth looking at.
Following code replaces the one above with TransactionLite framework:

// -- Assuming EntryProcessor is executed on "Person" cache and the address in the "Address" cache with its key AddressKey key associated to the Person's key.

final BackingMapManagerContext ctx = ((BinaryEntry) entry).getContext();
BackingMapContext backingMapContext = ctx.getBackingMapContext("Address");
Map.Entry backingMapEntry = backingMapContext.getBackingMapEntry(
                ctx.getKeyToInternalConverter().convert(new AddressKey(personId)));
((BinaryEntry) backingMapEntry).setValue(.., true);

A dirty little way if you know your finger crossing has yielded positive results
Problem of transactions become severe when associated objects are meant to be updated in parallel. What if the same Address is being shared by multiple People? Using KeyAssociation is not recommended. As more and more People share the same Address (like a Dorm) KeyAssociation will make the partition bloated and unevenly balanced something that preferably be avoided. If provisioning is done right (or even over provisioned) and cluster nodes appear to not fail and atomicity is not absolutely required but good to have then an EntryProcessor can be invoked with in another EntryProcessor as long as the Person and Address caches use different cache schemes (Different service names). But all bets are off once say Address's EP succeeds but the Person's fail. Setting the backup count to '0' would minimize the error window but still this is just not a solution. Coherence did have a now deprecated API TransactionMap that allowed multiple operations to be committed in a single transaction albeit nothing goes wrong as described here. This API has now been deprecated and replaced with the Transaction scheme.

Transactional Scheme
A new cache scheme has been added post v3.6 that allows updates to multiple cache entries in a single transaction as long as the caches belong to the same transactional scheme. Make sure following cache scheme is defined:

<cache-config> ...
<caching-schemes>
 <transactional-scheme>
     <scheme-name>transactional-scheme</scheme-name>
      <service-name>TransactionalCache</service-name>
      <autostart>></autostart>
  </transactional-scheme>
</caching-schemes>
</cache-config>

If Person and Address use the same TransactionalCache scheme

This is a simpler problem to solve with the current scheme.

DefaultConnectionFactory factory = new DefaultConnectionFactory();
Connection connection = factory.createConnection("TransactionalCache");
connection.setAutoCommit (false);
connection.setIsolationLevel(Isolation.READ_COMMITTED);

OptimisiticNamedCache personCache = connection.getNamedCache("Person");
OptimisticNamedCache addressCache = connection.getNamedCache("Address");

Person p = personCache("p1");
Address a = addressCache("a1");

update(p, a);

try {
 connection.commit();
} catch (..) { }
finally {
  connection.close();
}

Twist - Person and Address in two different cache schemes

Modified Problem: Update two Persons in a single Transaction and then update the Address if the previous transaction succeeds

This is tricky and this is what I call a dependency transaction

Lets take a twist and introduce a new type of InvocableMap.Entry:

public interface TransactionalEntry extends InvocableMap.Entry { 
  Status status getTxStatus();
  Object getAnotherEntry (Object key); 
  void update (Object key, Object value, Filter predicate);
}

Lets use this new interface in line with running an EntryProcessor:

public class SomeUtilClass {
   public Object invoke (final K key, final InvocableMap.EntryProcessor processor) {
   final DefaultConnectionFactory factory = new DefaultConnectionFactory();
   final Connection connection = factory.createConnection ("TransactionalCache");
   connection.setAutoCommit(false);
   connection.setIsolationLevel (Isolation.READ_COMMITTED);

   final TransactionState state = connection.getTransactionState();

   TransactionEntry entry = new TransactionalEntry () {
     private final OptimisticNamedCache oCache = connection.getNamedCache(cacheName);

     @Override
     public Object getKey() {
       return key;
     }

     @Override
     public Object getValue() {
       return oCache.get(key);
     }

     @Override
     public Object getAnotherEntry(Object relatedKey) {
       return oCache.get(relatedKey);
     }

     @Override
     public void update(Object key, Object value, Filter predicate) {
       oCache.update(key, value, predicate);
     }

     @Override
     public Status getTransactionStatus() {
       return state.getStatus();
     }

     // -- Other methods
     ...
  };
    try {
        processor.process (entry);
        connection.commit();
    } catch (Exception exp) {
         connection.rollback();
    } finally {
          connection.close();
          synchronized (entry.getTransactionStatus()) {
                entry.getTxStatus().notifyAll();
           }
    }

}

}

Now which processor is passed in the previously declared invoke method? It is an EntryProcessor that is not necessarily executed as an EntryProcessor but this model could gain some solid points in the design consistency.

public class MyProcessor implements InvocableMap.EntryProcessor {
   private final ValueExtractor extractor = new Reflectionxtractor ("currentVersion", ...);

 public Object process (final InvocableMap.Entry entry) {
   final TransactionalEntry txEntry = (TransactionalEntry) entry;
   final Person firstPerson = (Person) txEntry.getValue();
   Filter predicate = new WhateverFilter (extractor, firstPerson.currentVersion());
   firstPerson.incrementVersion();

   Person anotherPerson = (Person) txEntry.getAnotherEntry ("anotherPersonKey");
   anotherPerson.incrementVersion();

   // -- This needs to be transactional
   doSomething (firstPerson, anotherPerson);

   txEntry.update (entry.getKey(), firstPerson, predicate);

     // -- Use the Status as a Semaphore
   executerService.execute (new Runnable () {
         @Override
         public void run () {
            Status status = ((TransactionalEntry) entry).getTransactionStatus();
            synchronized (status) {
              try {
                status.wait();
              } catch (InterruptedException i) { ... }
                     
              switch (status) {
                 case COMMITTED:
                   // -- Update the Address.
                   // -- Now the two Person objects are guaranteed to be single transactionally
                   // -- updated and address could have been done in the same way, had address
                   // -- used the same Transactional scheme. 
                   // -- If Address is in a different cache scheme and the system is provisioned
                   // -- right that put() succeeds then the Address will only get updated after
                   // -- Person(s) have been successfully updated. 
                   // -- The updateAddress() could use another EntryProcessor 
                   updateAddress (address);
                   break;
                 case ROLLEDBACK:
                   break;
                 default:
              }
            }
         }
      });
    return firstPerson;
 }
}
Enjoy!

How to name your cache

When my daughter was born one of the first things we had to deal with was to come up with her name. As we had not discussed it earlier, the arguments ensued and for a long time me and my wife did not have a complete agreement. Now besides her official name we call her by ten other names and 8 out of these 10 does not even have a literary meaning. But we love to call her by those names as they express hers and ours different moods. Naming your cache is one such process.

Common Cache Conventions

• All upper case
• All lower case
• Upper or lower case with delimiters
• Application Name dot cache name
• Camel case
• Cryptic internal looking name
• No exposed names
  

All upper case
Though by Java conventions all upper case words are considered constants you would need to be cautious with cache names. Cache configurations are considered to be "updatable" at deployment time. As the updates are manual, cache names should also be humane. All upper case though are okay if short but long words or a combination of words all upper case does have a screaming feel and should be avoided.

All lower case
Though much easier on eyes all lower case cache names suffer from the problem of illegibility if the name is either long or a combination of words. Use of "-" or "_" can also be considered as long as this convention is followed throughout the configuration.

Schema qualification
Some coherence grids are configured to provide cache services for more that one application. Usually the cache access to this type of infrastructure are also backed up by a policy file that could define access permissions like application(A) can read but not write into a cache belonging to application (B) etc. These kinds of caches are better configured with names qualified by the application name. The application name should be short max 4 letters and could be all upper case. Like, a cache "Positions" of a Finance application can be named FIN.Positions while an audit cache "Statistics" could be AUD.Statistics and the Java policy (if any exists) can use regular expressions based on application names to set the Permissions.

Camel case
This is likely the most human readable format. As it also follows the Java class naming convention there exist an uniformity in how the classes and the caches are named.

Cryptic Names
A new convention has been introduced since Coherence transaction schemes after version 3.6. Some caches that are supposed to be used internally i.e, not meant for business usage can be defined with cryptic names. For example the Transaction scheme uses cache names starting with prefix "V$TXN-". If you desire to follow this convention then it would be good to create an internal cache configuration and use the DefaultConfigurableCacheFactory to initialize the caches through its ensureCache APIs. It will also be the responsibility of the application developer to tune these named caches - typically. Do not expose such a cache configuration for deployment time tuning.

No Names?
Some application may not want to expose the names at all. This may fall under two categories:

1. Coherence bundles a default cache configuration that can be used to create caches with names starting with dist- or repl- etc. Following this convention makes the configuration very short and clean. But it also risks a loose usage pattern where caches either can be wrongly created or not known at all unless the code flow initializes it. A typical use of this pattern is when we do not know the cache names up front. Like if an application provides employee Id to email address mapping for companies. As companies use this service the application can create dist-CustomerA or dist-CustomerB as they join this service. No infrastructure changes would be needed and it is easier to expand the application.
   
2. Cache mappings does not need to exist in the configuration and the application can use the name of the cache service to create its own caches. Though this either should be discouraged or be used in some advanced use cases. Either way, the naming convention should be consistent across all the components of the application.

Though there is no right or a wrong convention when it comes to cache names care should be taken to make it consistent, self explanatory and in some cases to show their usage. At the end of the day it is your baby, name it what you are comfortable with.

How to do Grid Computing right?

Nah.. There are plenty of forums and consultants who would talk and guide you on complex challenges with Grid Computing. Following are seven simple premise to do Grid computing right. No this is not off of a book or a result of any theoretical analysis but my field experience with numerous clients over the last three years of Coherence consulting on high volume systems.

1. Process where data are - Its a proven fact that processing data where they are is much more efficient than bringing it over the wire and then processing it - A concept popularized by PL/SQL.
   
2. Storage Disabled - This challenge I have seen often with products that have embedded Coherence. Application objects competing for the same heap that Coherence uses to manage data. Remember Garbage Collection in Java still remains an unresolved necessity. GC can not be avoided but its affect can be minimized. Make a rule of thumb - Make your application storage disabled member of the Grid if you really want a stable Grid with predictable performance. Application layer can be scaled out independently of storage nodes. Application can go under long GC and the load-balancer can choose another application instance while the Data Grid remains responsive and stable. An effective way of building a high responsive system.
   
3. Key Association - In any serious application we always talk about associated data. Person associated with an Address. If computing demands Person and Address to be processed together then keep them together and use Coherence Key Association or partition affinity. Partition affinity can have some major performance impact.
4. Domain Model - Typically not talked about much. You don't have to put every object of your application in Coherence just because you have them. Put only those objects in Cache that are needed for computation. A stock is associated with an Account which is associated with an address. Why to keep Address in the Cache if it has no bearing on stock's buy or sell? Selective association also trims the composite object size and helps in effective partition affinity.
   
5. Thread Count - Make sure you use the Machine to its fullest. Don't just go by some rule of threads and cores ratio. Run the application in real-life-like environment and see which number makes sense? Appropriate thread count with Invocation Service can have real impact.
6. Backing Map Listeners - If done right BMLs can make the application layer very light weight. BML using Executor service to do asynchronous processing is a proven architecture but take it with salt and see if that makes sense for your application. But when it comes to raw end-to-end computing performance take a stab at it.
7. Custom Backing Map - By default in distributed scheme, Objects are stored in Binary format. This is the most efficient and effective format to manage partitioned data but it doesn't have to be. If your application requires to minimize deserialization cost during an expensive computation then you can change the Object format to what ever you like it to be or keep them unchanged by writing a custom Backing Map. Work with someone who has done this before because Backing Map should be the most efficient piece of the solution.

I haven't seen a single application that has violated any of these seven rules and have not have had any problems, yet. So Enjoy!

Bay Area SIG - 29-Apr-2010

So this was my second Coherence SIG (the first was in New York almost three years ago). This time it was all about monitoring monitoring monitoring. It was nice to meet familiar faces. Few who recognized me right away, including Jon, Peter, and with a slight hick up Desmond and Raanan. Rao from Oracle EM remembered me from my blog photo (I think I still look the same). Ones who did not, included Craig. A few I knew I have met before or worked with was good to see in the Coherence crowd. It was a good presentation from Randy Stafford and his tips were:

• Monitor everything - CPU, Network, Heap and Application.
  
• Different tools for different things - Jamon, sar/ksar, JMX
• Re-architect if you have to.
• No comprehensive patterns available for Grid Computing so keep yourself up-to-date
• Mission Control rocks - May be single biggest reason to use JRockit

Tom from SL.com's presentation was informative and stressed the importance of Monitoring and why it is not an optional piece of the solution. I had introduced them to one of the clients I worked with and was to good to see that they were in their customers list as shown in his slide. It was good to see Tom recognizing features of Oracle EM not offered by RTView. Rao must feel happy. At the end Taylor from Grid Dynamics talked about the proven design patterns also implemented by Oracle Consulting at numerous engagements and was good to see common solutions to common problems by two professional services. Overall was a good night out.

Coherence with Spring on OSGi - Oh My!

OSGi has gotten some real traction as an Enterprise services platform. For one of the projects I was involved in a viability study for Coherence with Spring on OSGi recently. Even though not all can be divulged here but we overcame some interesting challenges that are worth to blog about.

Why OSGi? Why Spring? and Why Coherence (Is that a question)?

• OSGi - It has championed the modularity.
  
• Spring - Dependency Injection probably is still its biggest strengths.
• Coherence - Solves World Peace with simplicity, elegance and Powerful APIs.

Challenges:

1. OSGi - It needs patience. Lots of it specially if you are just starting with it. You do not need to change the application code to be OSGi compliant. It requires a specific MANIFEST.MF that describes the bundle's attributes and packages it exports and imports with other bundle dependencies. Make sure you have BND handy. I would not recommend you writing the Manifest file your own. Let BND do it for you. Equinox and Knopflerfish are two very popular platforms. I used Equinox.
   
2. Spring - Do not work with vanilla Spring APIs. You would need Spring Dynamic Modules that are Spring for OSGi. Also download Springsource Tool Suit. You would need some jars from it's STS to be installed. Also make sure you deploy the 3.0.0 version of the libraries. It will solve a lot of pain you would rather end up going through.
   
3. Coherence - Coherence does not have an OSGi bundle off the shelf so you would need to create one. Creating an OSGi bundle for Coherence is fairly easy by using Eclipse IDE. File->New->Project->Plug-in from Existing jar Archives->Select coherence.jar and follow the rest of the instructions. Export the artifact. That's it. Watch out for serialization issues if multiple modules are involved.
   

Spring DMs:
Install the following modules in the same order (use appropriate start levels):

1. com.springsource.org.aopalliance
2. com.springsource.org.apache.commons.logging
3. org.springframework.aop (3.0.0.RELEASE)
4. org.springframework.asm ( 3.0.0.RELEASE)
5. org.springframework.beans (3.0.0.RELEASE)
6. org.springframework.context (3.0.0.RELEASE)
   
7. org.springframework.core (3.0.0.RELEASE)
8. org.springframework.expression (3.0.0.RELEASE)
9. org.springframework.osgi.core (1.2.1)
10. org.springframework.osgi.extender (1.2.1)
11. org.springframework.osgi.io (1.2.1)
12. org.springframework.transaction (3.0.0.RELEASE)
13. com.tangosol.coherence (Or whatever name is given for the Coherence bundle created)
    

Don't shy away by restarting the Equinox a few times - you may need it.

Lets build a service: Say we want to build a service that puts a String key and a String value in a given data source.

public interface IPut {
   public void put (String ds, String key, String value);  
}

For an ability to replace Coherence as a Datasource with another lets use a Template that can be injected by Spring later with Application remaining agnostic to the data source provider.

public interface ITemplate {
   public void put(String ds, String key, String val);  
}

Even though IPut and ITemplate are similar it doesn't have to be. IPut represents the exposed service and ITemplate is for persistence.

Implementations:

public class Putter implements IPut {
   private ITemplate template;

   public Putter(ITemplate template) {
       this.template = template;
   }

   public void put (String ds, String key, String value) {
       if (template != null) {
           template.put(ds, key, value);
       }
   }
}

And,

public class CoherenceTemplate implements ITemplate {

   public CoherenceTemplate() {
   }

   public void put (String cacheName, String key, String value) {
         ClassLoader oldLoader = Thread.currentThread().getContextClassLoader();
         ClassLoader newLoader = com.tangosol.net.CacheFactory.class.getClassLoader();
         Thread.currentThread().setContextClassLoader(newLoader);
         NamedCache nCache = CacheFactory.getCache (cacheName);
         nCache.put (key, value);
         Thread.currentThread().setContextClassLoader(oldLoader);"

   }

}

Make sure you do have appropriate coherence-cache-config.xml deployed.

Configurations:
Spring beans.xml:

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
      xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
      xmlns:osgi="http://www.springframework.org/schema/osgi"
       xmlns:aop="http://www.springframework.org/schema/aop"
       xmlns:tx="http://www.springframework.org/schema/tx"
       xsi:schemaLocation="http://www.springframework.org/schema/beans
       http://www.springframework.org/schema/beans/spring-beans-2.0.xsd>
           
 <bean name="template" class="CoherenceTemplate" scope="prototype">
 </bean>

 <bean name="daoService" class="Putter">
   <constructor-arg ref="template"/>
 </bean>

</beans>

That's not it. We need another configuration to expose this service for OSGi:

<?xml version="1.0" encoding="UTF-8" ?>
<beans xmlns="http://www.springframework.org/schema/beans"
      xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
      xmlns:osgi="http://www.springframework.org/schema/osgi"
      xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd                       http://www.springframework.org/schema/osgi http://www.springframework.org/schema/osgi/spring-osgi.xsd">
   <osgi:service id="daoOSGiService" ref="daoService"
                 interface="IPut"></osgi:service>
</beans>

It's also important to place these two configurations under META-INF/spring directory.

What about the MANIFEST.MF?
Use bundlor to generate the MANIFEST.MF for you. You will have to copy the manifest back to the project and then re-bundle the final artifact jar.

Once you have it all together thats all you need to deploy this application on OSGi.

What about the Client now?
Client would be another OSGi bundle and the process of developing it is similar. As it is still all Spring driven make sure the services are appropriately injected.

First the OSGi configuration:

<?xml version="1.0" encoding="UTF-8" ?>
<beans xmlns="http://www.springframework.org/schema/beans"
      xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
      xmlns:osgi="http://www.springframework.org/schema/osgi"
      xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd                       http://www.springframework.org/schema/osgi http://www.springframework.org/schema/osgi/spring-osgi.xsd">
   <osgi:reference id="daoService" interface="IPut"/>
</beans>

Then for other beans:

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
      xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
      xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">
 <bean name="persister" class="MyClient"
       init-method="start" destroy-method="stop">
   <property name="putter" ref="daoService"/>
 </bean>
</beans>

And the client is like any typical Spring client:

public class MyClient {
   IPut putter;

   public MyClient () {
   }

   public IPut getPutter() {
       return putter;
   }

   public void setPutter(IPut putter) {
       this.putter = putter;
   }

   public void start() throws Exception {
       putter.put ("MyDataSource", "key", "Value");
   }

   public void stop() throws Exception {
       System.out.println("Goodbye");
   }
}

Don't forget to use the BND again to generate the manifest and re-bundling it with the client artifact jar.

Now deploy the client jar and start it.

Enjoy!

How to SIP a Coherence soup - Part(I)

A few times a requirement has passed our way to find if Oracle Coherence can be integrated with hand-held devices. Either to get a notification of an Entry or as a task submission unit. The short answer is No and the long answer is Yes. No - because Coherence requires at least a J2SE to be used natively. J2ME platform is not supported. Yes - because there is an architectural pattern called edging. Edging is a pattern of platform integration when a system does not support the protocol of another and we have to create an intermediary system that integrates these two protocols by understanding both. This intermediary system is "edged" or proxied in between. So lets cook this recipe.

Problem Statement: Integrating a cellphone with Coherence.

Ingredients:

1. A clean servlet container
2. Coherence jar
   
3. A SIP emulator and,
4. An IDE as per your taste

For a spicy recipe we have used an Oracle WebLogic Application Server, JDeveloper and Sun's now called Oracle's Java ME Platform SDK.

How to Cook:

• Download Weblogic Application Server from OTN and install it.
• Install Java ME Platform SDK from http://java.sun.com, install it and start Java_Platform_ME_SDK under toolbar/bin.
  
• While j2me SDK is warming up start JDeveloper and a simple Coherence cluster.
• If this is the first time you are cooking keep coherence-cache-config pretty simple with a distributed cache "A". If you like a little hotter you can spice the cache configuration up as needed.
  
• By now JDeveloper should be boiling. Create a new J2EE project and put the following servlet:

 public class SipRequestInitiator extends HttpServlet  {

   private NamedCache nCache = CacheFactory.getCache("A");
   public SipRequestInitiator() {
   }
   public void init(ServletConfig sc) throws ServletException {
     super.init(sc);
   }
   protected void doGet(HttpServletRequest req, HttpServletResponse res)
   throws ServletException, IOException {
     int num = 10;
     try {
         num = Integer.parseInt(req.getParameter("num"));
     } catch (NumberFormatException nExp) {
         System.out.println("Wrong number, defaulting to 10");
     }
     Filter filter =
         new LimitFilter(new LikeFilter(new KeyExtractor("toString"),
                                        "news-%", '*', true), num);
     Set news = nCache.keySet(filter);
     StringBuilder sB = new StringBuilder();
     Iterator iter = news.iterator();
     while (iter.hasNext()) {
         String k = (String)iter.next();
         sB.append((nCache.get(k)).toString());
         sB.append("\n");
     }
     res.setContentType("text/plain");
     res.setContentLength(sB.toString().intern().length());
     res.getWriter().println(sB.toString());
   }
}

• Deploy the servlet with a web.xml. WebLogic plate should be ready now.
• Once J2ME is hot enough start a MIDP project and cook it with the following Midlet:

 public class CohMIDlet extends MIDlet
 implements CommandListener, SipServerConnectionListener {

   private Display display;
   private Form form;
   private Command myMessages;
   private Command exitCommand; // The exit command
   private TextField number;

   public CohMIDlet() {
      display = Display.getDisplay(this);
      form = new Form("Receive Message");
      exitCommand = new Command("Exit", Command.EXIT, 0);
      myMessages = new Command("Messages", Command.ITEM, 1);
      number = new TextField ("No. Of Messages", "10", 3, TextField.LAYOUT_LEFT);
      form.append (number);
      form.addCommand(exitCommand);
      form.addCommand(myMessages);
      form.setCommandListener(this);
   }

   public void startApp() {
      display.setCurrent(form);
   }

   public void pauseApp() {
   }

   public void destroyApp(boolean unconditional) {
      notifyDestroyed();
   }

   public void commandAction(Command c, Displayable s) {
      if (c == myMessages) {
          Thread t = new Thread() {
              public void run() {
                  getMessages();
              }
          };
          t.start();
      }

      if (c == exitCommand) {
          destroyApp(true);
      }
   }

   private void getMessages() {
      int num = Integer.parseInt(number.getString());
      HttpConnection hc = null;
      InputStream in = null;     
      String url = "http://localhost:7001/sipinitiator?&num=" + num;
      try {
          hc = (HttpConnection) Connector.open(url);
          in = hc.openInputStream();

          int contentLength = (int) hc.getLength();
          byte[] raw = new byte[contentLength];
          int length = in.read(raw);

          in.close();
          hc.close();

          // Show the response to the user.
          String s = new String(raw, 0, length);
          form.append(s);
      } catch (IOException ioe) {
          form.append(ioe.getMessage());
      }
   }

   public void notifyRequest(SipConnectionNotifier arg0) {
   }
 }

• put two entries in cache A with keys as "news-$i", where $i is the article number.

• Choose a Cellphone type as per your choice. I like the good old style flip-phones. Run the emulator. It will deploy the application. Press the Menu button and select Messages. Do you see the following?

In this part we have prepared the soup and tasted it using a Midlet. In part(II) we will try to find how we can SIP it. Enjoy!

Customizing JMX Reporter

JMX Reporter is a Coherence utility that generates statistical logs about nodes, networks, memory, proxy and cache usage. If enabled (-Dtangosol.coherence.management.report.autostart=true) by default every minute it prints these statistical logs and every hour creates a new set of files. What if we need to customize it?

1. Easiest is to unzip coherence.jar and extract reports directory. This is what we would customize by putting in the classpath before coherence.jar
   
2. Edit report-group.xml to change the frequency and the location where these reports should be created.
3. Configure the <report-list> to have a list of <report-config> for the reports we want to have. The report-config takes a location of where the configuration of a specific type is defined.
   
4. Copy the report configurations in the directory as defined in the report-group.xml

Following is a sample of report-group.xml under reports/ directory that prints memory status and network health reports at a frequency of 10mins:

<report-group>
  <frequency>10m<</frequency>
  <output-directory>../StatisticalReports</output-directory>
  <report-list>
       <report-config>
           <location>reports/report-memory-status.xml</location>
       </report-config>
       <report-config>
           <location>reports/report-network-health.xml</location>
       </report-config>
  </report-list>
</report-group>

reports/report-memory-status.xml and reports/report-network-health.xml has a list of respective attributes to be reported. Each report-config sets the file-name and the delimiter typically set as {date}-<stat-name>.txt and {tab} respectively. The report generated are CSV files that can be opened in MS Excel or Openoffice and historically analyzed. What if at the end of the day you want to open in Excel and analyze it? Following is a simple shell script that concatenates the reports appropriately (by removing repeated headers) and names it aggregate-<report-type>.xls:

#!/bin/bash
p=0;
k=`ls *$1.txt 2>/dev/null | wc -l`
if [[ $k == 0 ]]
then echo "Valid types are: ";
        echo -e "\tmemory-status";
        echo -e "\tnetwork-health";
        exit;
else
        for i in `ls --sort time -r *$1.txt`
          do
            if [[ $p == 0 ]]
            then
                 cat $i >> aggregate-$1.xls;
                 p=1;
             else
                 sed '1d' $i | cat >> aggregate-$1.xls
             fi
          done
         dos2unix aggregate-$1.xls
fi

Enjoy!

I don't know which cache this Object goes to

..This was something that came up during a sluggish talk at a customer's site. A developer had just jokingly remarked.

Problem Statement: How to make Coherence objects destination aware?

One way is to annotate the objects that are to be cached. So lets create some simple annotations:

@Documented
@Retention(RetentionPolicy.RUNTIME)
@Target ({ElementType.TYPE, ElementType.PACKAGE})
public @interface Cache {
    String name();
    int ttl() default -1;
    String pk() default "getId";
}

All this annotation is telling is the name of the cache, what should be the TTL (default being -1) and how to find the primary key for this object? The annotation only complements the coherence-cache-config.xml and does not replace it (Unlike JPA annotations). You still need to define all cache names in the configuration file. Next step is to create a sample domain object annotated with Cache:

@Cache(name = "CacheDestination", ttl=-1, pk="getId")
public class Domain implements Serializable {
    private Key id;
    private int age;
    private String name;
    
    public Domain() {
    }

    public void setId(Key id) {
        this.id = id;
    }

    public Key getId() {
        return Key;
    }

    public void setAge(int age) {
        this.age = age;
    }

    public int getAge() {
        return age;
    }

    public void setName(String name) {
        this.name = name;
    }

    public String getName() {
        return name;
    }
    
    @Override
    public String toString () {
        return id + ":" + age + ":" + name;
    }
}

Unfortunately this destination aware class will not do anything unless the class that puts this Object in the cache extracts the destination. Cache annotation also tells which method to call to find it's primary key, in this sample it is getId () (Remember JPA?). For developers who do not want to "know" which cache to put this object to, need an Abstraction layer. Quite a few customers that I have worked with try to create an Abstraction layer on top of Coherence to make "all caching" logic to be centralized in one layer. Coherence APIs are already highly abstracted. NamedCache is probably the most important interface in all of Coherence APIs. NamedCache already extends java.util.Map. So having an abstraction layer that returns an instance of NamedCache is good enough for Map centric applications. Problem is that NamedCache is not just a Map. It also provides quereablility feature on top of this data structure. It also provides support for Events, transactional support and process invocation features that are missing from a vanilla Map contract. Most likely this abstraction layer will miss some of these native features of NamedCache that even JCACHE (JSR107) dictates only a subset of features of. The challenge here is that NamedCache has no knowledge of objects being destination aware or in simple English does not understand the Cache annotation, unless I find Product team drunk enough to put this idea forward to. So we need to build one. Lets create a ReducedMap:

public interface ReducedMap {
    public Object put (Object value) throws NoSuchMethodException,
           IllegalAccessException, InvocationTargetException,
                                   OperationNotSupportedException; 
}

and implement it in a custom Map using Java delegation pattern to pass cache invocations to Coherence.

public class AnnoMap extends AbstractMap implements ReducedMap {
    private volatile NamedCache nCache;

    public AnnoMap() {
    }

    public int size() {
        return (nCache == null) ? 0 : nCache.size();
    }

    public boolean isEmpty() {
        return (nCache == null) ? false : nCache.isEmpty();
    }

    public Set entrySet() {
        return (nCache == null) ? Collections.EMPTY_SET : nCache.entrySet();
    }

    public boolean containsKey(Object key) {
        if (nCache == null) {
            Class clz = key.getClass();
            Cache c = (Cache)clz.getAnnotation(Cache.class);
            if (c != null) {
                String name = (c.name() == null) ? clz.getName() : c.name();
                nCache = CacheFactory.getCache(name);
            }
        }
        return (nCache == null) ? false : nCache.containsKey(key);
    }

    public Object put(Object value) throws NoSuchMethodException,
          IllegalAccessException, InvocationTargetException,
                                  OperationNotSupportedException {
        Class clz = value.getClass();
        Cache c = (Cache)clz.getAnnotation(Cache.class);
        if (c != null) {
            String name = (c.name() == null) ? clz.getName() : c.name();
            nCache = CacheFactory.getCache(name);
            Method m = value.getClass().getMethod(c.pk(), new Class[0]);
            Object key = m.invoke(value, null);
            return nCache.put(key, value, c.ttl());
        }
        throw new OperationNotSupportedException("Class not annotated");
    }

    public Object get(Object key) {
        if (nCache == null) {
            Class clz = key.getClass();
            Cache c = (Cache)clz.getAnnotation(Cache.class);
            if (c != null) {
                String name = (c.name() == null) ? clz.getName() : c.name();
                nCache = CacheFactory.getCache(name);
            }
        }
        return (nCache == null) ? null : nCache.get(key);
    }

    public Object put(Object key, Object value) {
        Class clz = value.getClass();
        Cache c = (Cache)clz.getAnnotation(Cache.class);
        if (c == null)
            throw new UnsupportedOperationException("Value not annotated");
        String name = (c.name() == null) ? clz.getName() : c.name();
        nCache = CacheFactory.getCache(name);
        return nCache.put(key, value);
    }

}

Now as you can see AnnoMap is very limited in its capability but if you are taking this route probably this Map is the place to enhance. Lets run a quick test:

public static void main(String[] args) throws NoSuchMethodException,
                          IllegalAccessException, InvocationTargetException,
                                                 OperationNotSupportedException {
   Domain d = new Domain ();
   // -- KeyObject will be similar to Domain object with the same Cache 
   // -- destination defined in the Cache annotation.
   Key key = new KeyObject (...);
   d.setId(k);
   d.setXXX(..);
   AnnoMap map = new AnnoMap ();
   map.put(d);
        
   System.out.println(map.get (key));
}

Enjoy!

Functional Programming with Coherence

If you come across the following, don't get surprised. Its called Functional programming.

object FP {
  def doSomething (callback: () => Unit) {
    callback();
  }
  def letsPrint() {
    println ("Ha ha!");
  }
  def main (args: Array[String]) {
    doSomething (letsPrint)
  }
}

Functions in Scala are treated as Objects that can be passed around. The concept of Object has still not changed - Its a place holder of application state. So in Functional Programming Model, functions are states and if it is legal then why not manage them in a state repository like Oracle Coherence?

Problem Statement: Executing new Entry Processors without having to deploy them in the cluster and continue to achieve 100% up time?

In a simpler English it means how can I push new EntryProcessor(s) and execute them without having to bounce the cluster to deploy the new class. The solution lies in the concept of Functional Programming - Passing functions to the Executor but this time slightly in a different way. Lets again, begin with a cache configuration:

<!DOCTYPE cache-config SYSTEM "cache-config.dtd">
<cache-config>
    <caching-scheme-mapping>
        <cache-mapping>
            <cache-name>EPFeeder</cache-name>
            <scheme-name>feeder-scheme</scheme-name>
        </cache-mapping>        
        <cache-mapping>
            <cache-name>EPCache</cache-name>
            <scheme-name>ep-scheme</scheme-name>
        </cache-mapping>        
    </caching-scheme-mapping>
    
    <caching-schemes>
        <class-scheme>
            <scheme-name>ep-scheme</scheme-name>
            <class-name>DynaEPCache</class-name>
            <init-params>
                <init-param>
                    <param-type>string</param-type>
                    <param-value>{cache-name}</param-value>
                </init-param>
                <init-param>
                    <param-type>string</param-type>
                    <param-value>coherence-cache-config.xml</param-value>
                </init-param>
            </init-params>
        </class-scheme>        
        <distributed-scheme>
            <scheme-name>feeder-scheme</scheme-name>
            <service-name>EPFeederScheme</service-name>
            <backing-map-scheme>
                <read-write-backing-map-scheme>
                    <internal-cache-scheme>
                        <local-scheme>
                            <high-units>100MB</high-units>
                            <listener>
                                <class-scheme>
                                    <class-name>EPListener</class-name>
                                </class-scheme>
                            </listener>
                        </local-scheme>
                    </internal-cache-scheme>
                </read-write-backing-map-scheme>
            </backing-map-scheme>
            <autostart>true</autostart>
        </distributed-scheme>        

        <proxy-scheme>
          ...
        </proxy-scheme>

        <invocation-scheme>
            <scheme-name>InvocationService</scheme-name>
            <service-name>InvocationService</service-name>
            <thread-count>2</thread-count>
            <autostart>true</autostart>
        </invocation-scheme>        
    </caching-schemes>
</cache-config>

EPFeeder Cache is where we store the Functions (Implementations of process() method of the EntryProcessor) against the EP's class name. EPCache is a demo cache where your data resides and you would be executing new EPs against. Next Step is to see how the client code will look like:

public class DynaCUtilTest extends TestCase {
    ....
    private String getEPImpl() {

        StringBuffer sBuffer = new StringBuffer();
        sBuffer.append("public Object process (Entry entry) {");
        sBuffer.append ("System.out.println(\"In process\");");
        sBuffer.append ("System.out.println(\"Key:\" + entry.getKey());");
        sBuffer.append ("System.out.println(\"Value:\" + entry.getValue());");
        sBuffer.append("return null;");
        sBuffer.append("}");

        return sBuffer.toString();
    }

    public void testCreateEP()
        String impl = getEPImpl();
        String clzName = "EPClass_v1";
        NamedCache eCache = CacheFactory.getCache("EPFeeder");
        eCache.put(clzName, impl);

        NamedCache nCache = CacheFactory.getCache("EPCache");
        nCache.invoke("A", new EPClass_v1());
    }    
}

getEPImpl () is the implementation that we would replace the process() method with as we feed new EP classes, in this case the first version of it named EPClass_v1. So what happens next?
When a new implementation is put in the EPFeeder Cache, A Backing Map Listener picks up this event and creates a new class (EntryProcessor) on all the cluster members dynamically using an Invocation Service. This step achieves 100% up time for Coherence. The Backing Map Listener (EPListener) looks something like this:

public class EPListener extends MultiplexingMapListener {

    public EPListener() {
    }

    protected void onMapEvent(MapEvent mapEvent) {
        String key = ...;
        String impl = ...;
        InvocationService iS = 
            (InvocationService)CacheFactory.getService("InvocationService");
        Invocable inv = new EPCreator (key, impl);
        // -- Create a new Class on all nodes
        Set set = CacheFactory.getCluster ().getMemberSet();
        iS.query(inv, set);        
    }

}

The core of this listener is the magic EPCreator Invocable but before we look at the EPCreator let see an EPInterface:

import com.tangosol.util.InvocableMap;

public interface EPInterface extends InvocableMap.EntryProcessor {
}

The most critical piece of this puzzle is the Invocable and how it does its magic. The EPCreator executes the following in it's run() method. Lets put it in its own Util class (DynaCUtil):

    public static Class createEP(String clzName, String impl) {
         ClassPool pool = ClassPool.getDefault();
        pool.importPackage("com.tangosol.util.InvocableMap.EntryProcessor");
        pool.importPackage("com.tangosol.util.InvocableMap.Entry");
        Class clz = null;
        CtClass eClass = null;
        boolean shouldCreate = false;
        try {
            eClass = pool.get(clzName);
        } catch (NotFoundException e) {
            shouldCreate = true;
        }
        if (shouldCreate) {
            eClass = pool.makeClass(clzName);
                eClass.setInterfaces(new CtClass[] { 
                pool.makeClass("EPInterface") });
            try {
                eClass.addConstructor(CtNewConstructor.defaultConstructor(eClass));
            } catch (CannotCompileException e) {
                e.printStackTrace();
            }
            try {
                eClass.addMethod(CtNewMethod.make(impl, eClass));
                StringBuffer sBuffer = new StringBuffer ();
                sBuffer.append ("public java.util.Map processAll(java.util.Set set) {");
                sBuffer.append ("System.out.println(\"In processAll\");");
                sBuffer.append ("return java.util.Collections.EMPTY_MAP;");
                sBuffer.append ("}");
                eClass.addMethod (CtNewMethod.make (sBuffer.toString(), eClass));
                
            } catch (CannotCompileException e) {
                e.printStackTrace();
            }
            try {
                clz = eClass.toClass();
            } catch (CannotCompileException e) {
                e.printStackTrace();
            }
        }

What the heck was it? The Invocable uses JavaAssist to create a new EntryProcessor on the fly. Now the last question is as Coherence is a self-healing system where new nodes can join and leave anytime, how to make sure new EP Classes are available to the new nodes? And the answer is a Custom NamedCache which also is the last piece in the puzzle. The class would look something like the following:

public class DynaEPCache extends WrapperNamedCache {

    public Object invoke(Object oKey, InvocableMap.EntryProcessor agent) {
        String name = agent.getClass().getName();
        createEP (name, (String) CacheFactory.getCache("EPFeeder").get (name));
        return super.invoke(oKey, agent);
    }

    public Map invokeAll(Collection collKeys, EntryProcessor agent) {
        ....
    }

    private void createEP (String name, String impl) {
        if (impl == null) {
            throw new RuntimeException ("EntryProcessor not created yet!");
        }
        DynaCUtil.createEP(name, impl);
    }
}

A much more advanced implementation is sitting on my laptop that a pieces I will soon upload to http://sites.google.com/site/miscellaneouscomponents/Home. In the meantime just Enjoy!

Why measuring exact size in memory could be a futile exercise?

Coherence being an In-memory data grid, it is important to provision the hardware right. Many factors play different roles - Total RAM on the box, avoiding paging, providing linear scalability without stepping on Out of Memory errors and so on. Now the problem is how to measure how many additional nodes (Cache Servers) one would need and in result how many new boxes when we have to scale out? Also, if Indexes are created how to measure the additional space required and how to do it right?
There are two ways to measure things - First, like measuring Gold and Second like measuring Onions. Onions are always approximate. Coherence data sizing is like measuring Onions and its not like you can not measure it like Gold - accurate and precise but in most cases it is not needed. Why? Because the dynamic/auto provisioning nature of the cluster and cheaper memory by the day. It is much easier to approximate the size and add new nodes or boxes in the cluster than to be a Mathematician and calculate the size in bytes. If you are an Operations person you need quick and almost correct formulas. If you are a Coherence enthusiast you might already know it - On a 32-bit machine 1.2GB of RAM is needed to run a JVM with 1GB heap. Off of 1GB heap having only 375MB of space for primary data storage in distributed data scheme with one backup count. Keeping 30% of scratch space left per JVM to keep the GC profile in check and so on. What about Indexes? That's easy too.. Account for 30% overhead for each Index added. So watch for how many Indexes are added as it is easy to cross the data size itself. Are these numbers accurate? Nope. They are not meant to be either. Are they simple? Yes and close to correct. After all when it comes to provisioning a system like Coherence its okay to just measure it like Onions.

Coherence - Back to Basics

One question that keeps cropping up is when to configure an application as a cluster member and when over *Extend? The answer has always been pretty simple - If application stability cannot be guaranteed then it cannot be considered as a cluster member. There is more to it, btw. Following questions you must ask before opting for any configuration strategy:

• If an application is written in either C++ or .NET then there is no other choice but to configure it as an *Extend client.
• If the interface to Coherence is Java (including JNI) then choices are a little deeper.

Lets talk about it in more details (The content is not a dictation to its readers but an assistance to put them on a right track).

When to configure as a cluster member?

1. Applications deployed inside a container (Like an Application server) are typically configured as storage disabled cluster members. The reason is application servers themselves being "server-side" components are considered fairly stable.
   
2. Applications demanding "extreme" performance are configured as cluster members.
   
3. *Extend introduces an additional network hop and if this additional couple of milliseconds are not acceptable then configure it as a cluster member.
4. If proxy-reconnect is an issue.

When to *Extend?
I am a big proponent of *Extend configuration because it provides us a framework that we are already so used to. With new enhancements in Coherence v3.5 and later in Proxy services and serialization mechanism, overhead on proxy has reduced considerably and this opens up a some nice architectural options.

1. If the application is a desktop application where reliability cannot be guaranteed.
2. If network over which the application is accessing Coherence resources is slow or unreliable.
3. If application is written in .NET or C++

Besides these there are some advanced considerations:

1. If application is already very TCP/IP centric. New services can be developed on top of Proxy as is demonstrated here, here and here.
   
2. If an application arbiter is needed or something like a Gatekeeper.
   
3. With *Extend proxy layer can be scaled independent of other storage nodes. *Extend configuration allows three layers of scalability - First application clients, second the proxy layer itself and the third the core storage members or non-proxy cluster members. Storage member layer is about data scalability and proxy layer is about request scalability.
4. In line with the popular Adapter pattern typically found in BPEL designs. Application specific adapters can be plugged into Coherence using Proxy's TCP/IP service making application clients even more agnostic of Coherence infrastructure beyond the Map implementation.
5. If a new component is added that introduces new features that are not available out of the box from Coherence, it is much safer to deploy it on proxy nodes. Always remember we do not have any access to either the underline protocol (TCMP) or the core coherence component layer. Any feature that simulate either of the two must be performed by proxy.
6. Anyone still remembers SOA? Even though SOA does not dictate the transport layer but it is much cleaner to have well defined services running and accessing coherence resources over *Extend. If services are deployed inside a container then revisit (1), (2) and (3).
   

Enjoy!

Integrating LDAP with Coherence

Please read Securing a Coherence Cache as a precursor to this blog. The link talks about how to externalize the configuration of cache security provider that can be configured in coherence cache configuration. The security provider class that implements SecurityProvider interface has to implement a simple method checkAccess (Subject). Subject needs to be passed by the "cache client" and has to be authenticated/authorized in the security provider. Since I started Oracle coherence consulting it has come up time and again to integrate Coherence with an LDAP provider so that application/user accounts can be controlled on what access they can have to which cache with user accounts being managed in a Directory server. So lets think about it again and see if we can streamline this solution and something generic can be built.

Problem Statement: To setup Coherence cache in such a way that discrete cache access can be set up driven by Enterprise directory.

Lets, think about architectural decision points:

• LDAP Server is an external data source. Use CacheLoader.
• Avoid accessing LDAP for each Cache request. Cache user authentications (An admin cache).
• Protect the admin cache that manages the user-auth to not allow any access. Protect the protector.
• Protect the proxy from *Extend access. Cluster member has inherent trust. Use authorized-hosts for cluster members.
• Use JAAS
• Manage authorization locally but authentication centrally.

How about a quick activity diagram?

Authentication
User authentication has to happen only once (typically once in 24 hrs). This is not such a bad cost that can be incurred once in a day as user accounts do not change and if changes it changes very infrequently. Authentication information can be cached once an account is verified against a Directory server. We also need to make sure that the cache that manages account authentication information is inaccessible to any unauthorized user or applications. Now how to do it?

• Create an Admin cache.
• Plug a custom CacheLoader that interacts with an external Directory server.
• Build the cache key to include cache name and user credentials, the cache value be Boolean.TRUE or Boolean.FALSE.
• Using the <entitled> XmlElement configure a DisAllowSecurityProvider as it's security-provider.
• DisAllowSecurityProvider denies all requests to this cache other than made by a very "few chosen". Scroll down for its implementation.

So how would such an Admin cache configuration look like?
<distributed-scheme>
<scheme-name>admin-distributed-scheme</scheme-name>
<service-name>AdminDistributedService</service-name>
<backing-map-scheme>
<read-write-backing-map-scheme>
<internal-cache-scheme>
<local-scheme>
<high-units>200KB</high-units>
<unit-calculator>BINARY</unit-calculator>
<expiry-delay>86400000</expiry-delay>
</local-scheme>
</internal-cache-scheme>
<cachestore-scheme>
<class-scheme>
<class-name>LDAPCacheLoader</class-name>
<init-params>
<init-param>
<param-type>string</param-type>
<param-value>ldap.server.com</param-value>
</init-param>
<init-param>
<param-type>int</param-type>
<param-value>389</param-value>
</init-param>
</init-params>
</class-scheme>
</cachestore-scheme>
</read-write-backing-map-scheme>
</backing-map-scheme>
<autostart>true</autostart>
<entitled>
<security-provider>DisAllowSecurityProvider</security-provider>
</entitled></distributed-scheme>

So Admin cache is size limited and expires every 24 hrs of the first authentication and does not allow any access. How could that be? LDAPCacheLoader's load () method can be very simple. The cache key passed could be a "username$password" that can be parsed and authenticated against a Directory server using LDAP APIs. If authentication succeeds return a Boolean.TRUE otherwise false. So how is this load () invoked and from where?

Default Security Provider
Caching user authentication is a luxury that can be centralized. Applications deal with two aspects of cache security - Authentication and Authorization and these can be split in two classes. Combined with cached-authentication lets write an abstract Default security provider. Any security provider that extends it gets the "performance" for free.

public abstract class DefaultSecurityProvider implements SecurityProvider {
private NamedCache nCache = CacheFactory.getCache ("USER_CRED");
public boolean checkAccess (Subject subject) {
String user_pw = ((Principal) principals.iterator().next()).getName();
String userName = getUserName (user_pw);
Boolean isPresent = (Boolean) nCache.get (user_pw + "$$" + cacheName);
boolean isAuth = false;
if (isPresent.booleanValue()) {
isAuth = authorize (userName);
}
return isAuth;
}
public abstract boolean authorize(String userName);
}
Authorization
Like authentication authorization should be relatively inexpensive too. There could be two approaches. One, using Directory server to store authorization attributes too. Even though it is perfectly doable but authorization is owned by Coherence or application and should be "owned" by it. Central governance should only be applied to authentication and not to authorization. So lets find an inexpensive way... how about Java Permission object driven by a policy file? Lets write a Policy file:
grant Principal CustomPrincipal "Principal1" {
permission java.util.PropertyPermission "Cache1", "read, write";
.. More can be added here...
};
grant Principal CustomPrincipal "Principal2" {
permission java.util.PropertyPermission "Cache2", "read, write";
... More can be added here...
};

What about the custom security provider?

public class MyCustomSP extends DefaultSecurityProvider {

public MyCustomSP (String cacheName) {
super(cacheName);
this.cacheName = cacheName;
}

public boolean authorize(final String user) {
if (user == null) {
System.out.println("Auth not in USER_CRED cache");
return false;
}
try {
PropertyPermission fp =
new PropertyPermission(cacheName, "write");
new SecurityManager().checkPermission(fp);
return true;
} catch (SecurityException exp) {
...
}
return false;
}
}
Now in this implementation if a User Principal has "write" permission then it gets the access. But, out of NamedCache behaviors if each method can be classified into two - Either read or write then the method that was invoked can also be passed along with it's classification to checkAccess () method. Instead of hard-coded "write" for every access, NamedCache's each method can have a fine grained user authorization. Of course you reserve the right to create your own Permission object and a set of Actions and use that.

I am not done yet!
In the activity diagram there is a logical concept of Gatekeeper. Who is it? And how does it do it? This gatekeeper is a combination of a custom NamedCache (EntitledNamedCache) and a SecurityProvider called DisAllowSecurityProvider. EntitledNamedCache is auto-magically configured for caches that has &entitled> Element defined (Read Securing a Coherence Cache for more information). While, DisAllowSecurityProvider is configured on the Admin Cache (USER_CRED) that stores the authentication info.

What does DisAllowSecurityProvider do?
public class DisAllowSecurityProvider implements SecurityProvider {
public DisAllowSecurityProvider() {
}

public DisAllowSecurityProvider(String cacheName) {
}

public boolean checkAccess(Subject subject) {
StackTraceElement[] elements = new Throwable().getStackTrace();
StackTraceElement e3 = (StackTraceElement) elements[3];
StackTraceElement e0 = (StackTraceElement) elements[0];

try {
if (SecurityProvider.class.isAssignableFrom(Class.forName(e0.getClassName())) ||
SecurityProvider.class.isAssignableFrom(Class.forName(e3.getClassName()))) {
return true;
} else {
return false;
}
} catch (ClassNotFoundException f) {
Base.log (f);
return false;
}
}
}
So here you go, you get a decently flexible Coherence Cache security implementation. Enjoy!

**One of my colleagues Steve Brockman asked if it was possible to extend the security to other cluster nodes too besides the proxy nodes. The solution is a little different but easy to make. Following are the steps how to do it:

1. Copy coherence-cache-config.xml to say alt-cache-config.xml
2. Open alt-cache-config.xml in an editor and remove all the <entitled> section from the configuration.
3. Edit ExtendedCacheFactory and look for FILE_CFG_CACHE in the file. The next line is where the class sets the cache configuration name. Hardcode the param-value to alt-cache-config.xml (Or, be more creative but set it to alt-cache-config.xml).
4. Deploy the alt-cache-config.xml on all the cluster nodes.
5. Set -Dtangosol.coherence.override=proxy-override.xml on all cluster nodes.