public void commit(int taskId) throws InterruptedException, ReplicatorException {
  ConcurrentLinkedQueue<ReplDBMSEvent> queue=taskQueues.get(taskId);
  if (queue.size() == 0) {
    if (logger.isDebugEnabled()) {
      logger.debug("Empty commit on serial queue: taskId=" + taskId);
    }
    return;
  }
  if (commitAction != null)   commitAction.execute(taskId);
synchronized (serialQueue) {
    while (queue.peek() != null) {
      ReplDBMSEvent event=queue.poll();
      boolean ok=serialQueue.offer(event,commitTimeout,TimeUnit.SECONDS);
      if (!ok) {
        throw new ReplicatorException("Commit timed out: taskId=" + taskId + " timeout="+ commitTimeout);
      }
      if (logger.isDebugEnabled()) {
        logger.debug("Committing on serial queue: taskId=" + taskId + " seqno="+ event.getSeqno()+ " shardId="+ event.getShardId());
      }
    }
    queue.clear();
    commitCount++;
  }
}

public void commit ( int taskId ) throws InterruptedException , ReplicatorException { ConcurrentLinkedQueue < ReplDBMSEvent > queue = taskQueues . get ( taskId ) ; if ( queue . size ( ) == 0 ) { if ( logger . isDebugEnabled ( ) ) { logger . debug ( "Empty commit on serial queue: taskId=" + taskId ) ; } return ; } if ( commitAction != null ) commitAction . execute ( taskId ) ; synchronized ( serialQueue ) { while ( queue . peek ( ) != null ) { ReplDBMSEvent event = queue . poll ( ) ; boolean ok = serialQueue . offer ( event , commitTimeout , TimeUnit . SECONDS ) ; if ( ! ok ) { throw new ReplicatorException ( "Commit timed out: taskId=" + taskId + " timeout=" + commitTimeout ) ; } if ( logger . isDebugEnabled ( ) ) { logger . debug ( "Committing on serial queue: taskId=" + taskId + " seqno=" + event . getSeqno ( ) + " shardId=" + event . getShardId ( ) ) ; } } queue . clear ( ) ; commitCount ++ ; } }
commits all pending events for a particular task .
