@Override public void endWindow(){
  if (currentWindowId > windowDataManager.getLargestCompletedWindow()) {
synchronized (lock) {
      try {
        Message msg;
        while ((msg=holdingBuffer.poll()) != null) {
          processMessage(msg);
          emitCount++;
          lastMsg=msg;
        }
        windowDataManager.save(currentWindowRecoveryState,currentWindowId);
        currentWindowRecoveryState.clear();
        if (lastMsg != null) {
          acknowledge();
        }
        pendingAck.clear();
      }
 catch (      Throwable t) {
        Throwables.propagate(t);
      }
    }
    emitCount=0;
  }
 else   if (currentWindowId < windowDataManager.getLargestCompletedWindow()) {
    pendingAck.clear();
  }
  context.setCounters(counters);
}

@ Override public void endWindow ( ) { if ( currentWindowId > windowDataManager . getLargestCompletedWindow ( ) ) { synchronized ( lock ) { try { Message msg ; while ( ( msg = holdingBuffer . poll ( ) ) != null ) { processMessage ( msg ) ; emitCount ++ ; lastMsg = msg ; } windowDataManager . save ( currentWindowRecoveryState , currentWindowId ) ; currentWindowRecoveryState . clear ( ) ; if ( lastMsg != null ) { acknowledge ( ) ; } pendingAck . clear ( ) ; } catch ( Throwable t ) { Throwables . propagate ( t ) ; } } emitCount = 0 ; } else if ( currentWindowId < windowDataManager . getLargestCompletedWindow ( ) ) { pendingAck . clear ( ) ; } context . setCounters ( counters ) ; }
jms api has a drawback that it only allows acknowledgement / commitment of all the messages which have been consumed in a session instead of all the messages received till a particular message . < br / > this creates complications with recovery / idempotency as we need to ensure that the messages that are being acknowledged have been persisted because they wouldn ' t be redelivered . also if they are persisted then they shouldn ' t be re - delivered because that would cause duplicates . < br / > this is why when recovery data is persisted and messages are acknowledged , the thread that consumes message is blocked . < br / >
