How to do some action on all the cache enties?

[szhem]

Hello guys,

What is the best way to do some kind of action on all the cache entries?

Let's say I'd like to trigger all cache entry listeners for the all the entries.

Regards,

Sergey

[matlach]

Hi Sergey,

I think you could use the Map Reduce framework which in your Mapper will get called for each Key Value entries of a given Cache.

see : https://docs.jboss.org/author/display/ISPN/Map+Reduce+framework

Hope this helps,

Mat

[szhem]

Hello Mat,

Thanks a lot for the link. I've already tried it.

I understood that I'd like to use the same code for all my environments (unit testing, local, uat, prod, etc.)

As map-reduce works only with distributed caching it's hardly possible to use the same code for different cache configurations in different environments.

As a rule distributed executor service works wito replicated and distributed caches, but not with a local ones.

Regards,

Sergey

[matlach]

Yep sadly MapReduceTask only works with distributed cachev(just tried it with replicated cache and it doesn't work).

I also guess using cache.values() is not an acceptable approach (I dont even know if in either mode the cache store value are returned and if so, if it could turn into an OOME if too much data is returned).

Though, by looking at the source code of MapReduceTask, the MapCombineCommand, ReduceCommand are managed differently when executed locally.

like :

  protected Set<KOut> executeMapPhase(boolean useCompositeKeys) throws InterruptedException,
            ExecutionException {
      RpcManager rpc = cache.getRpcManager();
      MapCombineCommand<KIn, VIn, KOut, VOut> cmd = null;
      Set<KOut> mapPhasesResult = new HashSet<KOut>();
      List<MapTaskPart<Set<KOut>>> futures = new ArrayList<MapTaskPart<Set<KOut>>>();
      if (inputTaskKeysEmpty()) {
         for (Address target : rpc.getMembers()) {
            if (target.equals(rpc.getAddress())) { // <<<<<<<< if myself build command in a specific way
               cmd = buildMapCombineCommand(taskId.toString(), clone(mapper), clone(combiner),
                        null, true, useCompositeKeys);
            } else {
               cmd = buildMapCombineCommand(taskId.toString(), mapper, combiner, null, true,
                        useCompositeKeys);
            }
            MapTaskPart<Set<KOut>> part = createTaskMapPart(cmd, target, true);
            part.execute();
            futures.add(part);
         }

What would be needed is to enhance the

private void ensureProperCacheState(AdvancedCache<KIn, VIn> cache) throws NullPointerException,
            IllegalStateException {

     if (cache.getRpcManager() == null)
      //   throw new IllegalStateException("Can not use non-clustered cache for MapReduceTask"); // do not fail
         performLocalMapReduce();


      if (cache.getStatus() != ComponentStatus.RUNNING)
         throw new IllegalStateException("Invalid cache state " + cache.getStatus());

      if (cache.getDistributionManager() == null) {
      //   throw new IllegalStateException("Cache mode should be DIST, rather than "
      //                     + cache.getCacheConfiguration().clustering().cacheModeString()); // do not fail
         performLocalMapReduce(); // as data is the same accross the cluster
      }
   }

That's just my idea, what do you think about it ?

Maybe we should fill a JIRA feature request issue with this ?

[szhem]

Hello Mat,

Thanks a lot for analysis and clarification.

I really like your idea and I suppose the same code must be runnable event if the cache topologies are different (local, replicated, distributed, etc.).

The configuration should not force the developer to write the code in any special way.

I've raise a JIRA - https://issues.jboss.org/browse/ISPN-2812.

[matlach]

just voted for it,

thanks you sergey !

[amanukyan]

Hi Sergey,

For your information, the Distributed Executors support LOCAL caches starting from version 5.2.0 (see the following enhancement - https://issues.jboss.org/browse/ISPN-2192 ).

Regards,

Anna.

[szhem]

Hello Anna,

Many thanks for the notice.

So, Distributed Executors support local, replicated, distributed envs.

Will be great to have the same thing for the map-reduce.

Regards,

Sergey