rebalanceMqList must run after commit

@throws MQClientException
7147
public void rebalanceMqList() throws Exception {
		LOG.info("Begin to do rebalance operation");
		Set<MessageQueue> newMqs = getMQ();
		
		Set<MessageQueue> oldMqs = currentOffsets.keySet();
		
		if (oldMqs.equals(newMqs) == true) {
			LOG.info("No change of meta queues " + newMqs);
			return ;
		}
		
		Set<MessageQueue> removeMqs = new HashSet<MessageQueue>();
		removeMqs.addAll(oldMqs);
		removeMqs.removeAll(newMqs);
		
		Set<MessageQueue> addMqs = new HashSet<MessageQueue>();
		addMqs.addAll(newMqs);
		addMqs.removeAll(oldMqs);
		
		LOG.info("Remove " + removeMqs);
		for (MessageQueue mq : removeMqs) {
			Long offset = frontOffsets.remove(mq);
			updateOffsetToZk(mq, offset);
			
			backendOffset.remove(mq);
			
		}
		
		LOG.info("Add " + addMqs);
		for (MessageQueue mq : addMqs) {
			long offset = getOffsetFromZk(mq);
			frontOffsets.put(mq, offset);
			backendOffset.put(mq, offset);
		}
	}
1304746
private void validateRebalancePlanState(RebalancePlan rebalancePlan) {
        logger.info("Validating rebalance plan state.");

        Cluster currentCluster = rebalancePlan.getCurrentCluster();
        List<StoreDefinition> currentStores = rebalancePlan.getCurrentStores();
        Cluster finalCluster = rebalancePlan.getFinalCluster();
        List<StoreDefinition> finalStores = rebalancePlan.getFinalStores();

        RebalanceUtils.validateClusterStores(currentCluster, currentStores);
        RebalanceUtils.validateClusterStores(finalCluster, finalStores);
        RebalanceUtils.validateCurrentFinalCluster(currentCluster, finalCluster);
        RebalanceUtils.validateRebalanceStore(currentStores);
        RebalanceUtils.validateRebalanceStore(finalStores);
    }
201405
public void flushUpdate()
    {
        for (int i = 0; i < numberOfQueues; i++)
        {
            if (!dbUpdateLists[i].isEmpty())
            {
                this.submitTaskToQueue(this.createUpdateTask(dbUpdateLists[i], fileUpdateLists[i]), i, processedUpdates);
                this.dbUpdateLists[i] = new ArrayList(batchSize);
                this.fileUpdateLists[i] = new ArrayList(batchSize);
                this.updatesQueued++;
            }
        }
    }
3795
@Override
    protected void onPersist(EntityMetadata entityMetadata, Object entity, Object id, List<RelationHolder> rlHolders)
    {
        Object connection = getConnection();
        // Create a hashset and populate data into it
        //

        Pipeline pipeLine = null;
        try
        {
            if (isBoundTransaction())
            {
                pipeLine = ((Jedis) connection).pipelined();
                onPersist(entityMetadata, entity, id, rlHolders, pipeLine);
            }
            else
            {
                onPersist(entityMetadata, entity, id, rlHolders, connection);
            }
        }
        finally
        {
            //
            if (pipeLine != null)
            {
                pipeLine.sync(); // send I/O.. as persist call. so no need to
                                 // read
            } // response?

            onCleanup(connection);
        }

    }
841650
public void reconcileMQLinks()
    {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.entry(tc, "reconcileMQLinks");

        // Set reconciling flag to false
        reconciling = false;

        LinkTypeFilter filter = new LinkTypeFilter();
        filter.MQLINK = Boolean.TRUE;
        filter.UNRECONCILED = Boolean.TRUE;
        SIMPIterator itr = linkIndex.iterator(filter);

        //  Get the MQLinkManager
        MQLinkManager mqlinkManager = getMQLinkManager();

        while (itr.hasNext())
        {
            MQLinkHandler dh = (MQLinkHandler) itr.next();

            // Retrieve the uuid for the Handler.
            String mqLinkUuid = dh.getMqLinkUuid().toString();

            // Call MQLink Component to alert it to create appropriate resources for
            // a link that previously existed
            MQLinkObject mqlinkObj = null;
            try
            {
                mqlinkObj = mqlinkManager.create(messageProcessor.createMQLinkDefinition(mqLinkUuid),
                                                 (MQLinkLocalization) dh,
                                                 (ControllableRegistrationService) messageProcessor.getMEInstance(SIMPConstants.JS_MBEAN_FACTORY),
                                                 true); // We've reconciled and this destination is to be deleted
            } catch (SIResourceException e)
            {
                // No FFDC code needed

                SibTr.exception(tc, e);
            } catch (SIException e)
            {
                // No FFDC code needed

                SibTr.exception(tc, e);
            }

            // Store mqlinkObj in the handler
            dh.setMQLinkObject(mqlinkObj);
            // remove from the list - we won't attempt to delete it until requested
            // by the MQLink component
            linkIndex.cleanup(dh);
            linkIndex.defer(dh);
        }
        itr.finished();

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.exit(tc, "reconcileMQLinks");
    }
425202
public void commit() throws HandlerException {
        if (_isInitiator) {
            try {
                _userTx.commit();
            } catch (SystemException se) {
                throw CommonKnowledgeMessages.MESSAGES.userTransactionCommitFailedSystem(se);
            } catch (HeuristicRollbackException hre) {
                throw CommonKnowledgeMessages.MESSAGES.userTransactionCommitFailedRollback(hre);
            } catch (HeuristicMixedException hme) {
                throw CommonKnowledgeMessages.MESSAGES.userTransactionCommitFailedMixed(hme);
            } catch (RollbackException re) {
                throw CommonKnowledgeMessages.MESSAGES.userTransactionCommitFailed(re);
            }
        }
    }
545145
private void cardinalityChangeConsumersDetached() throws SIResourceException
  {
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) SibTr.entry(tc, "cardinalityChangeConsumersDetached");

    // Tell the AIH to start the flush.
    // this call will acquire a lock on AIH such that it is guaranteed that all calls to
    // RDC.put() for messages on this stream will have returned before rejectAll() returns
    // Therefore we can safely go through itemStream and remove the messages that have
    // already been put and reject them.
    _aih.rejectAll(); // reject everything and start flushing the stream

    // then delete all the messages, since know that put() must have completed
    removeAllMessagesItemStream();

    synchronized (consumerPoints)
    {
      // set cardinalityChange to false. Note that any consumers that try to attach right after this will not
      // be able to use the old stream since already initiated the flush
      _cardinalityChange = false;
      // tell the AIH to change its cardinality
      _aih.changeReceiveExclusive(_cardinalityOne);

    }
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) SibTr.exit(tc, "cardinalityChangeConsumersDetached");

  }
718260
private void reconstituteMQLink(int startMode) throws MessageStoreException, SIResourceException
    {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.entry(tc, "reconstituteMQLink", new Object[] { Integer.valueOf(startMode), this });

        int localisationCount = 0;
        MQLinkMessageItemStream mqlinkMessageItemStream = null;

        // There can be one or more localisations in the BaseDestinationHandler.
        NonLockingCursor cursor =
                        _baseDestinationHandler.newNonLockingItemStreamCursor(
                                        new ClassEqualsFilter(MQLinkMessageItemStream.class));

        do
        {
            mqlinkMessageItemStream = (MQLinkMessageItemStream) cursor.next();

            if (mqlinkMessageItemStream != null)
            {
                localisationCount++;
                mqlinkMessageItemStream.reconstitute(_baseDestinationHandler);

                attachLocalPtoPLocalisation(mqlinkMessageItemStream);

                /* Feature 176658.3.2 */
                //TODO: Check need all this for MQLinks
                assignQueuePointOutputHandler(
                                              mqlinkMessageItemStream.getOutputHandler(),
                                              mqlinkMessageItemStream.getLocalizingMEUuid());
                ConsumerDispatcher consumerDispatcher =
                                (ConsumerDispatcher) mqlinkMessageItemStream.getOutputHandler();
                consumerDispatcher.setReadyForUse();

                //If the local queue point is awaiting deletion ensure it is correctly configured
                if (mqlinkMessageItemStream.isToBeDeleted())
                {
                    dereferenceLocalisation(mqlinkMessageItemStream);
                }
            }
        } while (mqlinkMessageItemStream != null);

        cursor.finished();

        // Sanity - There should never be more than one mediation itemstream
        if (localisationCount > 1)
        {
            SIErrorException e =
                            new SIErrorException(
                                            nls.getFormattedMessage(
                                                                    "INTERNAL_MESSAGING_ERROR_CWSIP0005",
                                                                    new Object[] {
                                                                                  "com.ibm.ws.sib.processor.impl.destination.JSPtoPRealization",
                                                                                  "1:458:1.24.1.7",
                                                                                  _baseDestinationHandler.getName() },
                                                                    null));

            FFDCFilter.processException(
                                        e,
                                        "com.ibm.ws.sib.processor.impl.destination.JSPtoPRealization.reconstituteMQLink",
                                        "1:454:1.24.1.25",
                                        this);

            SibTr.exception(tc, e);
            SibTr.error(tc, "INTERNAL_MESSAGING_ERROR_CWSIP0005",
                        new Object[] {
                                      "com.ibm.ws.sib.processor.impl.destination.JSPtoPRealization",
                                      "1:461:1.24.1.25",
                                      _baseDestinationHandler.getName() });

            if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
                SibTr.exit(tc, "reconstituteMQLink", e);
            throw e;
        }

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.exit(tc, "reconstituteMQLink");
    }
293295
@Override
    public void commit()
        throws HeuristicMixedException,
               HeuristicRollbackException,
               RollbackException,
               SystemException
    {
        try {
            if (!Context.isTMNoTransaction()) {
                if (Context.isTMActive()) {
                    Context.commit();
                } else {
                    Context.rollback();
                }
            }
        } catch (final EFapsException e) {
            throw new SystemException(e.getMessage());
        }
    }
371161
private synchronized void rebalanceJobNode(List<String> nodes) {
		while(updatingStatus);
		Collection<JobConfig> jobs = schedulerConfgs.values();
		int nodeIndex = 0;
		for (JobConfig job : jobs) {
			String nodeId = nodes.get(nodeIndex++);
			if(!StringUtils.equals(job.getCurrentNodeId(), nodeId)){				
				job.setCurrentNodeId(nodeId);
				logger.info("rebalance Job[{}-{}] To Node[{}] ",job.getGroupName(),job.getJobName(),nodeId);
			}
			if(nodeIndex >= nodes.size()){
				nodeIndex = 0;
			}
			//
			updateJobConfig(job);
		}
		
	}
785290
@Override
    public void commitChanges(SynchronizationBarrier barrier) throws JMSException
    {
    	checkNotClosed();
    	checkTransactionLock();
    	
    	if (!committables.isEmpty())
        {
    		long start = System.currentTimeMillis();
    		
    		Iterator<Committable> allCommitables = committables.iterator();
        	while (allCommitables.hasNext())
        	{
        		Committable committable = allCommitables.next();
        		committable.commitChanges(barrier);
        	}
        	
        	long end = System.currentTimeMillis();
        	notifyCommitTime(end-start);
        	
        	pendingChanges = false;
        }
    }
332001
public function commit()
    {
        try {
            $this->connection->commit();
        } catch (DBALException $e) {
            throw new QueryException($e->getMessage(), $e->getCode(), $e);
        }
    }
1373820
private void startExecutingUpdates() throws ClosedException
  {
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.entry(tc, "startExecutingUpdates");

    // swap the enqueuedUnits and executingUnits.
    ArrayList temp = executingUnits;
    executingUnits = enqueuedUnits;
    enqueuedUnits = temp;
    enqueuedUnits.clear();
    // enqueuedUnits is now ready to accept AsyncUpdates in enqueueWork()

    executing = true;
    try
    {
      LocalTransaction tran = tranManager.createLocalTransaction(false);
      ExecutionThread thread = new ExecutionThread(executingUnits, tran);
      mp.startNewSystemThread(thread);
    }
    catch (InterruptedException e)
    {
      // this object cannot recover from this exception since we don't know how much work the ExecutionThread
      // has done. should not occur!
      FFDCFilter.processException(
        e,
        "com.ibm.ws.sib.processor.impl.store.AsyncUpdateThread.startExecutingUpdates",
        "1:222:1.28",
        this);
      SibTr.exception(tc, e);

      if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
        SibTr.exit(tc, "startExecutingUpdates", e);

      closed = true;
      throw new ClosedException(e.getMessage());
    }

    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.exit(tc, "startExecutingUpdates");

  }
844243
public void resetQuotaAndRecoverEnforcement() {
        for(Integer nodeId: nodeIds) {
            boolean quotaEnforcement = mapNodeToQuotaEnforcingEnabled.get(nodeId);
            adminClient.metadataMgmtOps.updateRemoteMetadata(Arrays.asList(nodeId),
                                                             MetadataStore.QUOTA_ENFORCEMENT_ENABLED_KEY,
                                                             Boolean.toString(quotaEnforcement));
        }
        for(String storeName: storeNames) {
            adminClient.quotaMgmtOps.rebalanceQuota(storeName);
        }
    }
110807
@Override
	public void commit() throws TransactionManagementException {
		logger.entering(CLASSNAME, "commit");
		try {
			userTran.commit();
			logger.log(Level.FINE, "javax.transaction.Status: {0}", userTran.getStatus());
		} catch (SecurityException e) {
			throw new TransactionManagementException(e);
		} catch (IllegalStateException e) {
			throw new TransactionManagementException(e);
		} catch (RollbackException e) {
			throw new TransactionManagementException(e);
		} catch (HeuristicMixedException e) {
			throw new TransactionManagementException(e);
		} catch (HeuristicRollbackException e) {
			throw new TransactionManagementException(e);
		} catch (SystemException e) {
			throw new TransactionManagementException(e);
		}
		logger.exiting(CLASSNAME, "commit");
	}
736105
@Override
    protected void onPersist(EntityMetadata entityMetadata, Object entity, Object id, List<RelationHolder> rlHolders) {
        // Insert, update is fine
        try {
            cqlClient.persist(entityMetadata, entity, null, rlHolders,
                getTtlValues().get(entityMetadata.getTableName()));
        } catch (InvalidRequestException e) {
            log.error("Error while persisting record, Caused by: .", e);
            throw new KunderaException(e);
        } catch (TException e) {
            log.error("Error while persisting record, Caused by: .", e);
            throw new KunderaException(e);
        } catch (UnsupportedEncodingException e) {
            log.error("Error while persisting record, Caused by: .", e);
            throw new KunderaException(e);
        }

    }
188727
public void refreshIndexList() throws IOException
   {
      synchronized (updateMonitor)
      {
         // release reader if any
         releaseMultiReader();
         // prepare added/removed sets
         Set<String> newList = new HashSet<String>(indexNames.getNames());

         // remove removed indexes
         Iterator<PersistentIndex> iterator = indexes.iterator();
         while (iterator.hasNext())
         {
            PersistentIndex index = iterator.next();
            String name = index.getName();
            // if current index not in new list, close it, cause it is deleted.
            if (!newList.contains(name))
            {
               index.close();
               iterator.remove();
            }
            else
            {
               // remove from list, cause this segment of index still present and indexes list contains
               // PersistentIndex instance related to this index..
               newList.remove(name);
               // Release everything to make sure that we see the
               // latest changes
               index.releaseWriterAndReaders();
            }
         }
         // now newList contains ONLY new, added indexes, deleted indexes, are removed from list.
         for (String name : newList)
         {
            // only open if it still exists
            // it is possible that indexNames still contains a name for
            // an index that has been deleted, but indexNames has not been
            // written to disk.
            if (!directoryManager.hasDirectory(name))
            {
               LOG.debug("index does not exist anymore: " + name);
               // move on to next index
               continue;
            }
            PersistentIndex index =
               new PersistentIndex(name, handler.getTextAnalyzer(), handler.getSimilarity(), cache, directoryManager,
                  modeHandler);
            index.setMaxFieldLength(handler.getMaxFieldLength());
            index.setUseCompoundFile(handler.getUseCompoundFile());
            index.setTermInfosIndexDivisor(handler.getTermInfosIndexDivisor());
            indexes.add(index);
         }
          // Reset the volatile index to be exactly like the master
          resetVolatileIndex();
      }
   }