geode-core/src/main/java/org/apache/geode/internal/cache/partitioned/IndexCreationMsg.java - geode - Git at Google

 /*
  * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
  * agreements. See the NOTICE file distributed with this work for additional information regarding
  * copyright ownership. The ASF licenses this file to You under the Apache License, Version 2.0 (the
  * "License"); you may not use this file except in compliance with the License. You may obtain a
  * copy of the License at
  *
  * http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software distributed under the License
  * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
  * or implied. See the License for the specific language governing permissions and limitations under
  * the License.
  */
 package org.apache.geode.internal.cache.partitioned;

 import java.io.DataInput;
 import java.io.DataOutput;
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;

 import org.apache.logging.log4j.Logger;

 import org.apache.geode.CancelException;
 import org.apache.geode.DataSerializer;
 import org.apache.geode.SystemFailure;
 import org.apache.geode.cache.CacheException;
 import org.apache.geode.cache.RegionDestroyedException;
 import org.apache.geode.cache.query.Index;
 import org.apache.geode.cache.query.IndexCreationException;
 import org.apache.geode.cache.query.MultiIndexCreationException;
 import org.apache.geode.cache.query.RegionNotFoundException;
 import org.apache.geode.cache.query.internal.index.IndexCreationData;
 import org.apache.geode.cache.query.internal.index.PartitionedIndex;
 import org.apache.geode.distributed.internal.ClusterDistributionManager;
 import org.apache.geode.distributed.internal.DistributionManager;
 import org.apache.geode.distributed.internal.InternalDistributedSystem;
 import org.apache.geode.distributed.internal.ReplyException;
 import org.apache.geode.distributed.internal.ReplyMessage;
 import org.apache.geode.distributed.internal.ReplyProcessor21;
 import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
 import org.apache.geode.internal.cache.ForceReattemptException;
 import org.apache.geode.internal.cache.PartitionedRegion;
 import org.apache.geode.internal.cache.PartitionedRegionException;
 import org.apache.geode.internal.logging.log4j.LogMarker;
 import org.apache.geode.internal.serialization.DeserializationContext;
 import org.apache.geode.internal.serialization.SerializationContext;
 import org.apache.geode.internal.serialization.Version;
 import org.apache.geode.logging.internal.log4j.api.LogService;

 public class IndexCreationMsg extends PartitionMessage {
   private static final Logger logger = LogService.getLogger();

   HashSet<IndexCreationData> indexDefinitions;

   /**
    * Constructor for the index creation message to be sent over the wire with all the relevant
    * information.
    *
    * @param recipients members to which this message has to be sent
    * @param regionId partitioned region id
    * @param processor The processor to reply to
    * @param indexDefinitions definitions for the indexes
    *
    */

   IndexCreationMsg(Set recipients, int regionId, ReplyProcessor21 processor,
       HashSet<IndexCreationData> indexDefinitions) {
     super(recipients, regionId, processor);
     this.indexDefinitions = indexDefinitions;
   }

   /**
    * Empty default constructor.
    *
    */
   public IndexCreationMsg() {

   }

   /**
    * This message may be sent to nodes before the PartitionedRegion is completely initialized due to
    * the RegionAdvisor(s) knowing about the existence of a partitioned region at a very early part
    * of the initialization
    */
   @Override
   protected boolean failIfRegionMissing() {
     return false;
   }

   /**
    * This method actually operates on the partitioned region and creates given list of indexes from
    * a index creation message.
    *
    * @param dm distribution manager.
    * @param pr partitioned region on which to create an index.
    * @throws CacheException indicating a cache level error
    * @throws ForceReattemptException if the peer is no longer available
    */
   @Override
   protected boolean operateOnPartitionedRegion(ClusterDistributionManager dm, PartitionedRegion pr,
       long startTime) throws CacheException, ForceReattemptException {
     // region exists
     ReplyException replyEx = null;
     boolean result = false;
     List<Index> indexes = null;
     List<String> failedIndexNames = new ArrayList<String>();

     if (logger.isDebugEnabled()) {
       StringBuilder sb = new StringBuilder();
       for (IndexCreationData icd : indexDefinitions) {
         sb.append(icd.getIndexName()).append(" ");
       }
       logger.debug(
           "Processing index creation message on this remote partitioned region vm for indexes: {}",
           sb);
     }

     try {
       indexes = pr.createIndexes(true, indexDefinitions);
     } catch (IndexCreationException e1) {
       replyEx = new ReplyException(
           "Remote Index Creation Failed", e1);
     } catch (MultiIndexCreationException exx) {
       failedIndexNames.addAll(exx.getExceptionsMap().keySet());

       if (logger.isDebugEnabled()) {
         StringBuffer exceptionMsgs = new StringBuffer();
         for (Exception ex : exx.getExceptionsMap().values()) {
           exceptionMsgs.append(ex.getMessage()).append("\n");
         }
         logger.debug("Got an MultiIndexCreationException with \n: {}", exceptionMsgs);
         logger.debug("{} indexes were created succesfully", failedIndexNames.size());
       }
       replyEx = new ReplyException(
           "Remote Index Creation Failed", exx);
     }

     if (null == replyEx) {
       result = true;
     }

     if (result) {
       Map<String, Integer> indexBucketsMap = new HashMap<String, Integer>();
       for (Index index : indexes) {
         PartitionedIndex prIndex = (PartitionedIndex) index;
         indexBucketsMap.put(prIndex.getName(), prIndex.getNumberOfIndexedBuckets());
       }
       sendReply(getSender(), getProcessorId(), dm, replyEx, result, indexBucketsMap,
           pr.getDataStore().getAllLocalBuckets().size());
     } else {
       // add the indexes that were successfully created to the map
       Map<String, Integer> indexBucketsMap = new HashMap<String, Integer>();
       for (IndexCreationData icd : indexDefinitions) {
         // if the index was successfully created
         if (!failedIndexNames.contains(icd.getIndexName())) {
           PartitionedIndex prIndex = (PartitionedIndex) pr.getIndex(icd.getIndexName());
           indexBucketsMap.put(icd.getIndexName(), prIndex.getNumberOfIndexedBuckets());
         }
       }
       sendReply(getSender(), getProcessorId(), dm, replyEx, result, indexBucketsMap,
           pr.getDataStore().getAllLocalBuckets().size());
     }

     if (logger.isDebugEnabled()) {
       logger.debug(
           "Multi Index creation completed on remote host and has sent the reply to the originating vm.");
     }
     return false;
   }

   /**
    * Process this index creation message on the receiver.
    */
   @Override
   public void process(final ClusterDistributionManager dm) {

     final boolean isDebugEnabled = logger.isDebugEnabled();

     Throwable thr = null;
     boolean sendReply = true;
     PartitionedRegion pr = null;

     try {
       if (isDebugEnabled) {
         logger.debug("Trying to get pr with id: {}", this.regionId);
       }
       try {
         if (isDebugEnabled) {
           logger.debug("Again trying to get pr with id : {}", this.regionId);
         }
         pr = PartitionedRegion.getPRFromId(this.regionId);
         if (isDebugEnabled) {
           logger.debug("Index creation message got the pr {}", pr);
         }
         if (null == pr) {
           boolean wait = true;
           int attempts = 0;
           while (wait && attempts < 30) { // max 30 seconds of wait.
             dm.getCancelCriterion().checkCancelInProgress(null);
             if (isDebugEnabled) {
               logger.debug(
                   "Waiting for Partitioned Region to be intialized with id {}for processing index creation messages",
                   this.regionId);
             }
             try {
               boolean interrupted = Thread.interrupted();
               try {
                 Thread.sleep(500);
               } catch (InterruptedException e) {
                 interrupted = true;
                 dm.getCancelCriterion().checkCancelInProgress(e);
               } finally {
                 if (interrupted)
                   Thread.currentThread().interrupt();
               }

               pr = PartitionedRegion.getPRFromId(this.regionId);
               if (null != pr) {
                 wait = false;
                 if (isDebugEnabled) {
                   logger.debug("Indexcreation message got the pr {}", pr);
                 }
               }
               attempts++;
             } catch (CancelException ignorAndLoopWait) {
               if (isDebugEnabled) {
                 logger.debug(
                     "IndexCreationMsg waiting for pr to be properly created with prId : {}",
                     this.regionId);
               }
             }
           }

         }
       } catch (CancelException letPRInitialized) {
         // Not sure if the CacheClosedException is still thrown in response
         // to the PR being initialized.
         if (logger.isDebugEnabled()) {
           logger.debug("Waiting for notification from pr being properly created on {}",
               this.regionId);
         }

         boolean wait = true;
         while (wait) {
           dm.getCancelCriterion().checkCancelInProgress(null);
           try {
             boolean interrupted = Thread.interrupted();
             try {
               Thread.sleep(500);
             } catch (InterruptedException e) {
               interrupted = true;
               dm.getCancelCriterion().checkCancelInProgress(e);
             } finally {
               if (interrupted)
                 Thread.currentThread().interrupt();
             }
             pr = PartitionedRegion.getPRFromId(this.regionId);
             wait = false;
             if (logger.isDebugEnabled()) {
               logger.debug("Indexcreation message got the pr {}", pr);
             }
           } catch (CancelException ignorAndLoopWait) {
             if (logger.isDebugEnabled()) {
               logger.debug("IndexCreationMsg waiting for pr to be properly created with prId : {}",
                   this.regionId);
             }
           }
         }

       }

       if (pr == null /* && failIfRegionMissing() */) {
         String msg =
             String.format(
                 "Could not get Partitioned Region from Id %s for message %s received on member= %s map= %s",
                 new Object[] {Integer.valueOf(this.regionId), this, dm.getId(),
                     PartitionedRegion.dumpPRId()});
         throw new PartitionedRegionException(msg, new RegionNotFoundException(msg));
       }
       sendReply = operateOnPartitionedRegion(dm, pr, 0);

     } catch (PRLocallyDestroyedException pre) {
       if (isDebugEnabled) {
         logger.debug("Region is locally Destroyed ");
       }
       thr = pre;
     } catch (VirtualMachineError err) {
       SystemFailure.initiateFailure(err);
       // If this ever returns, rethrow the error. We're poisoned
       // now, so don't let this thread continue.
       throw err;
     } catch (Throwable t) {
       // Whenever you catch Error or Throwable, you must also
       // catch VirtualMachineError (see above). However, there is
       // _still_ a possibility that you are dealing with a cascading
       // error condition, so you also need to check to see if the JVM
       // is still usable:
       SystemFailure.checkFailure();
       // log the exception at fine level if there is no reply to the message
       if (this.processorId == 0) {
         logger.debug("{} exception while processing message:{}", this, t.getMessage(), t);
       } else if (logger.isDebugEnabled(LogMarker.DM_VERBOSE) && (t instanceof RuntimeException)) {
         logger.debug(LogMarker.DM_VERBOSE, "Exception caught while processing message: {}",
             t.getMessage(), t);
       }
       if (t instanceof RegionDestroyedException && pr != null) {
         if (pr.isClosed) {
           logger.info("Region is locally destroyed, throwing RegionDestroyedException for {}",
               pr);
           thr = new RegionDestroyedException(
               String.format("Region is locally destroyed on %s",
                   dm.getId()),
               pr.getFullPath());
         }
       } else {
         thr = t;
       }
     } finally {
       if (sendReply && this.processorId != 0) {
         ReplyException rex = null;
         if (thr != null) {
           rex = new ReplyException(thr);
         }
         sendReply(getSender(), this.processorId, dm, rex, pr, 0);
       }
     }

   }

   /**
    * Methods that sends the actual index creation message to all the members.
    *
    * @param recipient set of members.
    * @param pr partitoned region associated with the index.
    * @param indexDefinitions set of index definitions
    * @return partitionresponse a response for the index creation
    */
   public static PartitionResponse send(InternalDistributedMember recipient, PartitionedRegion pr,
       HashSet<IndexCreationData> indexDefinitions) {

     RegionAdvisor advisor = (RegionAdvisor) (pr.getDistributionAdvisor());
     final Set<InternalDistributedMember> recipients;
     /*
      * Will only send create index to all the members storing data
      */
     if (null == recipient) {
       recipients = new HashSet(advisor.adviseDataStore());
     } else {
       recipients = new HashSet<InternalDistributedMember>();
       recipients.add(recipient);
     }

     for (InternalDistributedMember rec : recipients) {
       if (rec.getVersionObject().compareTo(Version.GFE_81) < 0) {
         throw new UnsupportedOperationException(
             "Indexes should not be created during rolling upgrade");
       }
     }

     IndexCreationResponse processor = null;
     if (logger.isDebugEnabled()) {
       logger.debug("Will be sending create index msg to : {}", recipients.toString());
     }
     if (recipients.size() > 0) {
       processor =
           (IndexCreationResponse) (new IndexCreationMsg()).createReplyProcessor(pr, recipients);
     }

     IndexCreationMsg indMsg =
         new IndexCreationMsg(recipients, pr.getPRId(), processor, indexDefinitions);
     indMsg.setTransactionDistributed(pr.getCache().getTxManager().isDistributed());
     if (logger.isDebugEnabled()) {
       logger.debug("Sending index creation message: {}, to member(s) {}.", indMsg, recipients);
     }
     /* Set failures = */pr.getDistributionManager().putOutgoing(indMsg);
     // Set failures =r.getDistributionManager().putOutgoing(m);
     // if (failures != null && failures.size() > 0) {
     // throw new ForceReattemptException("Failed sending <" + indMsg + ">");
     // }
     return processor;
   }

   // override reply processor type from PartitionMessage
   @Override
   PartitionResponse createReplyProcessor(PartitionedRegion r, Set recipients) {
     return new IndexCreationResponse(r.getSystem(), recipients);
   }

   /**
    * Send a reply for index creation message.
    *
    * @param member representing the actual index creatro in the system
    * @param procId waiting processor
    * @param dm distribution manager to send the message
    * @param ex any exceptions
    * @param result represents index created properly or not.
    * @param indexBucketsMap Map of indexes created and number of buckets indexed
    * @param numTotalBuckets Number of total buckets in this vm
    */
   void sendReply(InternalDistributedMember member, int procId, DistributionManager dm,
       ReplyException ex, boolean result, Map<String, Integer> indexBucketsMap,
       int numTotalBuckets) {
     IndexCreationReplyMsg.send(member, procId, dm, ex, result, indexBucketsMap, numTotalBuckets);
   }

   @Override
   public int getDSFID() {
     return PR_INDEX_CREATION_MSG;
   }

   @Override
   public void fromData(DataInput in,
       DeserializationContext context) throws IOException, ClassNotFoundException {
     super.fromData(in, context);
     this.indexDefinitions = DataSerializer.readHashSet(in);
   }

   @Override
   public Version[] getSerializationVersions() {
     return null;
   }

   @Override
   public void toData(DataOutput out,
       SerializationContext context) throws IOException {
     super.toData(out, context);
     DataSerializer.writeHashSet(this.indexDefinitions, out);
   }

   /**
    * String representation of this message.
    */
   @Override
   public String toString() {
     StringBuffer sb = new StringBuffer();
     for (IndexCreationData icd : indexDefinitions) {
       sb.append(icd.getIndexName()).append(" ");
     }
     return sb.toString();
   }

   /**
    * Class representing index creation response. This class has all the information for successful
    * or unsuccessful index creation on this member of the partitioned region.
    *
    *
    */
   public static class IndexCreationResponse extends PartitionResponse {

     /** Map of indexes created and number of buckets indexed. */
     private Map<String, Integer> indexBucketsMap;

     /** Number of total bukets in this vm. */
     private int numTotalBuckets;

     /**
      * Construtor for index creation response message.
      *
      * @param ds distributed system for this member.
      * @param recipients all the member associated with the index
      */
     IndexCreationResponse(InternalDistributedSystem ds, Set recipients) {
       super(ds, recipients);
     }

     /**
      * Waits for the response from the members for index creation.
      *
      * @return IndexCreationResult for creation of indexes
      * @throws CacheException indicating a cache level error
      * @throws ForceReattemptException if the peer is no longer available
      */
     public IndexCreationResult waitForResult() throws CacheException, ForceReattemptException {
       try {
         waitForCacheException();
       } catch (RuntimeException re) {
         if (re instanceof PartitionedRegionException) {
           if (re.getCause() instanceof RegionNotFoundException) {
             // Region may not be available at the receiver.
             // ignore the exception.
             // This will happen when the region on the remote end is still in
             // initialization mode and is not yet created the region ID.
           } else {
             throw re;
           }
         } else {
           throw re;
         }
       }
       return new IndexCreationResult(this.indexBucketsMap, this.numTotalBuckets);
     }

     /**
      * Sets the relevant information in the response.
      *
      * @param result true if index created properly
      * @param indexBucketsMap Map of indexes created and number of buckets indexed
      * @param numTotalBuckets Number of total buckets in this vm
      */
     public void setResponse(boolean result, Map<String, Integer> indexBucketsMap,
         int numTotalBuckets) {
       this.indexBucketsMap = indexBucketsMap;
       this.numTotalBuckets = numTotalBuckets;
     }
   }

   /**
    * Class representing index creation result.
    *
    *
    */
   public static class IndexCreationResult {
     /** Map of indexes created and number of buckets indexed. */
     private Map<String, Integer> indexBucketsMap;

     /** Number of total bukets in this vm. */
     private int numTotalBuckets;

     /**
      * Constructor for index creation result.
      *
      * @param indexBucketsMap Map of indexes created and number of buckets indexed
      * @param numTotalBuckets Number of total buckets in this vm
      */
     IndexCreationResult(Map<String, Integer> indexBucketsMap, int numTotalBuckets) {
       this.indexBucketsMap = indexBucketsMap;
       this.numTotalBuckets = numTotalBuckets;
     }

     /**
      * Returns a map of index names and number of buckets indexed
      *
      */
     public Map<String, Integer> getIndexBucketsMap() {
       return this.indexBucketsMap;

     }

   }

   /**
    * Class for index creation reply. This class has the information about successful index creation.
    *
    *
    */
   public static class IndexCreationReplyMsg extends ReplyMessage {

     /** Index created or not. */
     private boolean result;

     /** Map of indexes created and number of buckets indexed. */
     private Map<String, Integer> indexBucketsMap;

     /** Number of total buckets in this vm. */
     private int numTotalBuckets;

     /** Boolean indicating weather its a data store. */
     private boolean isDataStore;

     public IndexCreationReplyMsg() {

     }

     /**
      * Constructor for index creation reply message.
      *
      * @param processorId processor id of the waiting processor
      * @param ex any exceptions
      * @param result true if index created properly else false
      * @param indexBucketsMap Map of indexes created and number of buckets indexed
      * @param numTotalBuckets Number of total buckets in this vm
      */
     IndexCreationReplyMsg(int processorId, ReplyException ex, boolean result, boolean isDataStore,
         Map<String, Integer> indexBucketsMap, int numTotalBuckets) {
       super();
       super.setException(ex);
       this.result = result;
       this.indexBucketsMap = indexBucketsMap;
       this.numTotalBuckets = numTotalBuckets;
       this.isDataStore = isDataStore;
       setProcessorId(processorId);
     }

     @Override
     public int getDSFID() {
       return PR_INDEX_CREATION_REPLY_MSG;
     }

     @Override
     public void fromData(DataInput in,
         DeserializationContext context) throws IOException, ClassNotFoundException {
       super.fromData(in, context);
       this.result = in.readBoolean();
       this.indexBucketsMap = DataSerializer.readObject(in);
       this.numTotalBuckets = in.readInt();
       this.isDataStore = in.readBoolean();

     }

     @Override
     public void toData(DataOutput out,
         SerializationContext context) throws IOException {
       super.toData(out, context);
       out.writeBoolean(this.result);
       DataSerializer.writeObject(this.indexBucketsMap, out);
       out.writeInt(this.numTotalBuckets);
       out.writeBoolean(this.isDataStore);
     }

     /**
      * Actual method sending the index creation reply message.
      *
      * @param recipient the originator of index creation message
      * @param processorId waiting processor id
      * @param dm distribution manager
      * @param ex any exceptions
      * @param result true is index created successfully
      * @param indexBucketsMap Map of indexes created and number of buckets indexed
      * @param numTotalBuckets Number of total buckets in this vm
      */
     public static void send(InternalDistributedMember recipient, int processorId,
         DistributionManager dm, ReplyException ex, boolean result,
         Map<String, Integer> indexBucketsMap, int numTotalBuckets) {
       IndexCreationReplyMsg indMsg = new IndexCreationReplyMsg(processorId, ex, result, result,
           indexBucketsMap, numTotalBuckets);
       indMsg.setRecipient(recipient);
       dm.putOutgoing(indMsg);
     }

     /**
      * Processes the index creation result.
      *
      * @param dm distribution manager
      */
     @Override
     public void process(final DistributionManager dm, final ReplyProcessor21 p) {
       if (logger.isDebugEnabled()) {
         logger.debug("Processor id is : {}", this.processorId);
       }
       IndexCreationResponse processor = (IndexCreationResponse) p;
       if (processor != null) {
         processor.setResponse(this.result, this.indexBucketsMap, this.numTotalBuckets);
         processor.process(this);
       }
     }

   }
 }
	/*
	* Licensed to the Apache Software Foundation (ASF) under one or more contributor license
	* agreements. See the NOTICE file distributed with this work for additional information regarding
	* copyright ownership. The ASF licenses this file to You under the Apache License, Version 2.0 (the
	* "License"); you may not use this file except in compliance with the License. You may obtain a
	* copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing, software distributed under the License
	* is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
	* or implied. See the License for the specific language governing permissions and limitations under
	* the License.
	*/
	package org.apache.geode.internal.cache.partitioned;

	import java.io.DataInput;
	import java.io.DataOutput;
	import java.io.IOException;
	import java.util.ArrayList;
	import java.util.HashMap;
	import java.util.HashSet;
	import java.util.List;
	import java.util.Map;
	import java.util.Set;

	import org.apache.logging.log4j.Logger;

	import org.apache.geode.CancelException;
	import org.apache.geode.DataSerializer;
	import org.apache.geode.SystemFailure;
	import org.apache.geode.cache.CacheException;
	import org.apache.geode.cache.RegionDestroyedException;
	import org.apache.geode.cache.query.Index;
	import org.apache.geode.cache.query.IndexCreationException;
	import org.apache.geode.cache.query.MultiIndexCreationException;
	import org.apache.geode.cache.query.RegionNotFoundException;
	import org.apache.geode.cache.query.internal.index.IndexCreationData;
	import org.apache.geode.cache.query.internal.index.PartitionedIndex;
	import org.apache.geode.distributed.internal.ClusterDistributionManager;
	import org.apache.geode.distributed.internal.DistributionManager;
	import org.apache.geode.distributed.internal.InternalDistributedSystem;
	import org.apache.geode.distributed.internal.ReplyException;
	import org.apache.geode.distributed.internal.ReplyMessage;
	import org.apache.geode.distributed.internal.ReplyProcessor21;
	import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
	import org.apache.geode.internal.cache.ForceReattemptException;
	import org.apache.geode.internal.cache.PartitionedRegion;
	import org.apache.geode.internal.cache.PartitionedRegionException;
	import org.apache.geode.internal.logging.log4j.LogMarker;
	import org.apache.geode.internal.serialization.DeserializationContext;
	import org.apache.geode.internal.serialization.SerializationContext;
	import org.apache.geode.internal.serialization.Version;
	import org.apache.geode.logging.internal.log4j.api.LogService;

	public class IndexCreationMsg extends PartitionMessage {
	private static final Logger logger = LogService.getLogger();

	HashSet<IndexCreationData> indexDefinitions;

	/**
	* Constructor for the index creation message to be sent over the wire with all the relevant
	* information.
	*
	* @param recipients members to which this message has to be sent
	* @param regionId partitioned region id
	* @param processor The processor to reply to
	* @param indexDefinitions definitions for the indexes
	*
	*/

	IndexCreationMsg(Set recipients, int regionId, ReplyProcessor21 processor,
	HashSet<IndexCreationData> indexDefinitions) {
	super(recipients, regionId, processor);
	this.indexDefinitions = indexDefinitions;
	}

	/**
	* Empty default constructor.
	*
	*/
	public IndexCreationMsg() {

	}

	/**
	* This message may be sent to nodes before the PartitionedRegion is completely initialized due to
	* the RegionAdvisor(s) knowing about the existence of a partitioned region at a very early part
	* of the initialization
	*/
	@Override
	protected boolean failIfRegionMissing() {
	return false;
	}

	/**
	* This method actually operates on the partitioned region and creates given list of indexes from
	* a index creation message.
	*
	* @param dm distribution manager.
	* @param pr partitioned region on which to create an index.
	* @throws CacheException indicating a cache level error
	* @throws ForceReattemptException if the peer is no longer available
	*/
	@Override
	protected boolean operateOnPartitionedRegion(ClusterDistributionManager dm, PartitionedRegion pr,
	long startTime) throws CacheException, ForceReattemptException {
	// region exists
	ReplyException replyEx = null;
	boolean result = false;
	List<Index> indexes = null;
	List<String> failedIndexNames = new ArrayList<String>();

	if (logger.isDebugEnabled()) {
	StringBuilder sb = new StringBuilder();
	for (IndexCreationData icd : indexDefinitions) {
	sb.append(icd.getIndexName()).append(" ");
	}
	logger.debug(
	"Processing index creation message on this remote partitioned region vm for indexes: {}",
	sb);
	}

	try {
	indexes = pr.createIndexes(true, indexDefinitions);
	} catch (IndexCreationException e1) {
	replyEx = new ReplyException(
	"Remote Index Creation Failed", e1);
	} catch (MultiIndexCreationException exx) {
	failedIndexNames.addAll(exx.getExceptionsMap().keySet());

	if (logger.isDebugEnabled()) {
	StringBuffer exceptionMsgs = new StringBuffer();
	for (Exception ex : exx.getExceptionsMap().values()) {
	exceptionMsgs.append(ex.getMessage()).append("\n");
	}
	logger.debug("Got an MultiIndexCreationException with \n: {}", exceptionMsgs);
	logger.debug("{} indexes were created succesfully", failedIndexNames.size());
	}
	replyEx = new ReplyException(
	"Remote Index Creation Failed", exx);
	}

	if (null == replyEx) {
	result = true;
	}

	if (result) {
	Map<String, Integer> indexBucketsMap = new HashMap<String, Integer>();
	for (Index index : indexes) {
	PartitionedIndex prIndex = (PartitionedIndex) index;
	indexBucketsMap.put(prIndex.getName(), prIndex.getNumberOfIndexedBuckets());
	}
	sendReply(getSender(), getProcessorId(), dm, replyEx, result, indexBucketsMap,
	pr.getDataStore().getAllLocalBuckets().size());
	} else {
	// add the indexes that were successfully created to the map
	Map<String, Integer> indexBucketsMap = new HashMap<String, Integer>();
	for (IndexCreationData icd : indexDefinitions) {
	// if the index was successfully created
	if (!failedIndexNames.contains(icd.getIndexName())) {
	PartitionedIndex prIndex = (PartitionedIndex) pr.getIndex(icd.getIndexName());
	indexBucketsMap.put(icd.getIndexName(), prIndex.getNumberOfIndexedBuckets());
	}
	}
	sendReply(getSender(), getProcessorId(), dm, replyEx, result, indexBucketsMap,
	pr.getDataStore().getAllLocalBuckets().size());
	}

	if (logger.isDebugEnabled()) {
	logger.debug(
	"Multi Index creation completed on remote host and has sent the reply to the originating vm.");
	}
	return false;
	}

	/**
	* Process this index creation message on the receiver.
	*/
	@Override
	public void process(final ClusterDistributionManager dm) {

	final boolean isDebugEnabled = logger.isDebugEnabled();

	Throwable thr = null;
	boolean sendReply = true;
	PartitionedRegion pr = null;

	try {
	if (isDebugEnabled) {
	logger.debug("Trying to get pr with id: {}", this.regionId);
	}
	try {
	if (isDebugEnabled) {
	logger.debug("Again trying to get pr with id : {}", this.regionId);
	}
	pr = PartitionedRegion.getPRFromId(this.regionId);
	if (isDebugEnabled) {
	logger.debug("Index creation message got the pr {}", pr);
	}
	if (null == pr) {
	boolean wait = true;
	int attempts = 0;
	while (wait && attempts < 30) { // max 30 seconds of wait.
	dm.getCancelCriterion().checkCancelInProgress(null);
	if (isDebugEnabled) {
	logger.debug(
	"Waiting for Partitioned Region to be intialized with id {}for processing index creation messages",
	this.regionId);
	}
	try {
	boolean interrupted = Thread.interrupted();
	try {
	Thread.sleep(500);
	} catch (InterruptedException e) {
	interrupted = true;
	dm.getCancelCriterion().checkCancelInProgress(e);
	} finally {
	if (interrupted)
	Thread.currentThread().interrupt();
	}

	pr = PartitionedRegion.getPRFromId(this.regionId);
	if (null != pr) {
	wait = false;
	if (isDebugEnabled) {
	logger.debug("Indexcreation message got the pr {}", pr);
	}
	}
	attempts++;
	} catch (CancelException ignorAndLoopWait) {
	if (isDebugEnabled) {
	logger.debug(
	"IndexCreationMsg waiting for pr to be properly created with prId : {}",
	this.regionId);
	}
	}
	}

	}
	} catch (CancelException letPRInitialized) {
	// Not sure if the CacheClosedException is still thrown in response
	// to the PR being initialized.
	if (logger.isDebugEnabled()) {
	logger.debug("Waiting for notification from pr being properly created on {}",
	this.regionId);
	}

	boolean wait = true;
	while (wait) {
	dm.getCancelCriterion().checkCancelInProgress(null);
	try {
	boolean interrupted = Thread.interrupted();
	try {
	Thread.sleep(500);
	} catch (InterruptedException e) {
	interrupted = true;
	dm.getCancelCriterion().checkCancelInProgress(e);
	} finally {
	if (interrupted)
	Thread.currentThread().interrupt();
	}
	pr = PartitionedRegion.getPRFromId(this.regionId);
	wait = false;
	if (logger.isDebugEnabled()) {
	logger.debug("Indexcreation message got the pr {}", pr);
	}
	} catch (CancelException ignorAndLoopWait) {
	if (logger.isDebugEnabled()) {
	logger.debug("IndexCreationMsg waiting for pr to be properly created with prId : {}",
	this.regionId);
	}
	}
	}

	}

	if (pr == null /* && failIfRegionMissing() */) {
	String msg =
	String.format(
	"Could not get Partitioned Region from Id %s for message %s received on member= %s map= %s",
	new Object[] {Integer.valueOf(this.regionId), this, dm.getId(),
	PartitionedRegion.dumpPRId()});
	throw new PartitionedRegionException(msg, new RegionNotFoundException(msg));
	}
	sendReply = operateOnPartitionedRegion(dm, pr, 0);

	} catch (PRLocallyDestroyedException pre) {
	if (isDebugEnabled) {
	logger.debug("Region is locally Destroyed ");
	}
	thr = pre;
	} catch (VirtualMachineError err) {
	SystemFailure.initiateFailure(err);
	// If this ever returns, rethrow the error. We're poisoned
	// now, so don't let this thread continue.
	throw err;
	} catch (Throwable t) {
	// Whenever you catch Error or Throwable, you must also
	// catch VirtualMachineError (see above). However, there is
	// _still_ a possibility that you are dealing with a cascading
	// error condition, so you also need to check to see if the JVM
	// is still usable:
	SystemFailure.checkFailure();
	// log the exception at fine level if there is no reply to the message
	if (this.processorId == 0) {
	logger.debug("{} exception while processing message:{}", this, t.getMessage(), t);
	} else if (logger.isDebugEnabled(LogMarker.DM_VERBOSE) && (t instanceof RuntimeException)) {
	logger.debug(LogMarker.DM_VERBOSE, "Exception caught while processing message: {}",
	t.getMessage(), t);
	}
	if (t instanceof RegionDestroyedException && pr != null) {
	if (pr.isClosed) {
	logger.info("Region is locally destroyed, throwing RegionDestroyedException for {}",
	pr);
	thr = new RegionDestroyedException(
	String.format("Region is locally destroyed on %s",
	dm.getId()),
	pr.getFullPath());
	}
	} else {
	thr = t;
	}
	} finally {
	if (sendReply && this.processorId != 0) {
	ReplyException rex = null;
	if (thr != null) {
	rex = new ReplyException(thr);
	}
	sendReply(getSender(), this.processorId, dm, rex, pr, 0);
	}
	}

	}

	/**
	* Methods that sends the actual index creation message to all the members.
	*
	* @param recipient set of members.
	* @param pr partitoned region associated with the index.
	* @param indexDefinitions set of index definitions
	* @return partitionresponse a response for the index creation
	*/
	public static PartitionResponse send(InternalDistributedMember recipient, PartitionedRegion pr,
	HashSet<IndexCreationData> indexDefinitions) {

	RegionAdvisor advisor = (RegionAdvisor) (pr.getDistributionAdvisor());
	final Set<InternalDistributedMember> recipients;
	/*
	* Will only send create index to all the members storing data
	*/
	if (null == recipient) {
	recipients = new HashSet(advisor.adviseDataStore());
	} else {
	recipients = new HashSet<InternalDistributedMember>();
	recipients.add(recipient);
	}

	for (InternalDistributedMember rec : recipients) {
	if (rec.getVersionObject().compareTo(Version.GFE_81) < 0) {
	throw new UnsupportedOperationException(
	"Indexes should not be created during rolling upgrade");
	}
	}

	IndexCreationResponse processor = null;
	if (logger.isDebugEnabled()) {
	logger.debug("Will be sending create index msg to : {}", recipients.toString());
	}
	if (recipients.size() > 0) {
	processor =
	(IndexCreationResponse) (new IndexCreationMsg()).createReplyProcessor(pr, recipients);
	}

	IndexCreationMsg indMsg =
	new IndexCreationMsg(recipients, pr.getPRId(), processor, indexDefinitions);
	indMsg.setTransactionDistributed(pr.getCache().getTxManager().isDistributed());
	if (logger.isDebugEnabled()) {
	logger.debug("Sending index creation message: {}, to member(s) {}.", indMsg, recipients);
	}
	/* Set failures = */pr.getDistributionManager().putOutgoing(indMsg);
	// Set failures =r.getDistributionManager().putOutgoing(m);
	// if (failures != null && failures.size() > 0) {
	// throw new ForceReattemptException("Failed sending <" + indMsg + ">");
	// }
	return processor;
	}

	// override reply processor type from PartitionMessage
	@Override
	PartitionResponse createReplyProcessor(PartitionedRegion r, Set recipients) {
	return new IndexCreationResponse(r.getSystem(), recipients);
	}

	/**
	* Send a reply for index creation message.
	*
	* @param member representing the actual index creatro in the system
	* @param procId waiting processor
	* @param dm distribution manager to send the message
	* @param ex any exceptions
	* @param result represents index created properly or not.
	* @param indexBucketsMap Map of indexes created and number of buckets indexed
	* @param numTotalBuckets Number of total buckets in this vm
	*/
	void sendReply(InternalDistributedMember member, int procId, DistributionManager dm,
	ReplyException ex, boolean result, Map<String, Integer> indexBucketsMap,
	int numTotalBuckets) {
	IndexCreationReplyMsg.send(member, procId, dm, ex, result, indexBucketsMap, numTotalBuckets);
	}

	@Override
	public int getDSFID() {
	return PR_INDEX_CREATION_MSG;
	}

	@Override
	public void fromData(DataInput in,
	DeserializationContext context) throws IOException, ClassNotFoundException {
	super.fromData(in, context);
	this.indexDefinitions = DataSerializer.readHashSet(in);
	}

	@Override
	public Version[] getSerializationVersions() {
	return null;
	}

	@Override
	public void toData(DataOutput out,
	SerializationContext context) throws IOException {
	super.toData(out, context);
	DataSerializer.writeHashSet(this.indexDefinitions, out);
	}

	/**
	* String representation of this message.
	*/
	@Override
	public String toString() {
	StringBuffer sb = new StringBuffer();
	for (IndexCreationData icd : indexDefinitions) {
	sb.append(icd.getIndexName()).append(" ");
	}
	return sb.toString();
	}

	/**
	* Class representing index creation response. This class has all the information for successful
	* or unsuccessful index creation on this member of the partitioned region.
	*
	*
	*/
	public static class IndexCreationResponse extends PartitionResponse {

	/** Map of indexes created and number of buckets indexed. */
	private Map<String, Integer> indexBucketsMap;

	/** Number of total bukets in this vm. */
	private int numTotalBuckets;

	/**
	* Construtor for index creation response message.
	*
	* @param ds distributed system for this member.
	* @param recipients all the member associated with the index
	*/
	IndexCreationResponse(InternalDistributedSystem ds, Set recipients) {
	super(ds, recipients);
	}

	/**
	* Waits for the response from the members for index creation.
	*
	* @return IndexCreationResult for creation of indexes
	* @throws CacheException indicating a cache level error
	* @throws ForceReattemptException if the peer is no longer available
	*/
	public IndexCreationResult waitForResult() throws CacheException, ForceReattemptException {
	try {
	waitForCacheException();
	} catch (RuntimeException re) {
	if (re instanceof PartitionedRegionException) {
	if (re.getCause() instanceof RegionNotFoundException) {
	// Region may not be available at the receiver.
	// ignore the exception.
	// This will happen when the region on the remote end is still in
	// initialization mode and is not yet created the region ID.
	} else {
	throw re;
	}
	} else {
	throw re;
	}
	}
	return new IndexCreationResult(this.indexBucketsMap, this.numTotalBuckets);
	}

	/**
	* Sets the relevant information in the response.
	*
	* @param result true if index created properly
	* @param indexBucketsMap Map of indexes created and number of buckets indexed
	* @param numTotalBuckets Number of total buckets in this vm
	*/
	public void setResponse(boolean result, Map<String, Integer> indexBucketsMap,
	int numTotalBuckets) {
	this.indexBucketsMap = indexBucketsMap;
	this.numTotalBuckets = numTotalBuckets;
	}
	}

	/**
	* Class representing index creation result.
	*
	*
	*/
	public static class IndexCreationResult {
	/** Map of indexes created and number of buckets indexed. */
	private Map<String, Integer> indexBucketsMap;

	/** Number of total bukets in this vm. */
	private int numTotalBuckets;

	/**
	* Constructor for index creation result.
	*
	* @param indexBucketsMap Map of indexes created and number of buckets indexed
	* @param numTotalBuckets Number of total buckets in this vm
	*/
	IndexCreationResult(Map<String, Integer> indexBucketsMap, int numTotalBuckets) {
	this.indexBucketsMap = indexBucketsMap;
	this.numTotalBuckets = numTotalBuckets;
	}

	/**
	* Returns a map of index names and number of buckets indexed
	*
	*/
	public Map<String, Integer> getIndexBucketsMap() {
	return this.indexBucketsMap;

	}

	}

	/**
	* Class for index creation reply. This class has the information about successful index creation.
	*
	*
	*/
	public static class IndexCreationReplyMsg extends ReplyMessage {

	/** Index created or not. */
	private boolean result;

	/** Map of indexes created and number of buckets indexed. */
	private Map<String, Integer> indexBucketsMap;

	/** Number of total buckets in this vm. */
	private int numTotalBuckets;

	/** Boolean indicating weather its a data store. */
	private boolean isDataStore;

	public IndexCreationReplyMsg() {

	}

	/**
	* Constructor for index creation reply message.
	*
	* @param processorId processor id of the waiting processor
	* @param ex any exceptions
	* @param result true if index created properly else false
	* @param indexBucketsMap Map of indexes created and number of buckets indexed
	* @param numTotalBuckets Number of total buckets in this vm
	*/
	IndexCreationReplyMsg(int processorId, ReplyException ex, boolean result, boolean isDataStore,
	Map<String, Integer> indexBucketsMap, int numTotalBuckets) {
	super();
	super.setException(ex);
	this.result = result;
	this.indexBucketsMap = indexBucketsMap;
	this.numTotalBuckets = numTotalBuckets;
	this.isDataStore = isDataStore;
	setProcessorId(processorId);
	}

	@Override
	public int getDSFID() {
	return PR_INDEX_CREATION_REPLY_MSG;
	}

	@Override
	public void fromData(DataInput in,
	DeserializationContext context) throws IOException, ClassNotFoundException {
	super.fromData(in, context);
	this.result = in.readBoolean();
	this.indexBucketsMap = DataSerializer.readObject(in);
	this.numTotalBuckets = in.readInt();
	this.isDataStore = in.readBoolean();

	}

	@Override
	public void toData(DataOutput out,
	SerializationContext context) throws IOException {
	super.toData(out, context);
	out.writeBoolean(this.result);
	DataSerializer.writeObject(this.indexBucketsMap, out);
	out.writeInt(this.numTotalBuckets);
	out.writeBoolean(this.isDataStore);
	}

	/**
	* Actual method sending the index creation reply message.
	*
	* @param recipient the originator of index creation message
	* @param processorId waiting processor id
	* @param dm distribution manager
	* @param ex any exceptions
	* @param result true is index created successfully
	* @param indexBucketsMap Map of indexes created and number of buckets indexed
	* @param numTotalBuckets Number of total buckets in this vm
	*/
	public static void send(InternalDistributedMember recipient, int processorId,
	DistributionManager dm, ReplyException ex, boolean result,
	Map<String, Integer> indexBucketsMap, int numTotalBuckets) {
	IndexCreationReplyMsg indMsg = new IndexCreationReplyMsg(processorId, ex, result, result,
	indexBucketsMap, numTotalBuckets);
	indMsg.setRecipient(recipient);
	dm.putOutgoing(indMsg);
	}

	/**
	* Processes the index creation result.
	*
	* @param dm distribution manager
	*/
	@Override
	public void process(final DistributionManager dm, final ReplyProcessor21 p) {
	if (logger.isDebugEnabled()) {
	logger.debug("Processor id is : {}", this.processorId);
	}
	IndexCreationResponse processor = (IndexCreationResponse) p;
	if (processor != null) {
	processor.setResponse(this.result, this.indexBucketsMap, this.numTotalBuckets);
	processor.process(this);
	}
	}

	}
	}