protected long releaseMemory(){
  long memoryFreed=0;
  while (!windowsForFreeMemory.isEmpty()) {
    long windowId=windowsForFreeMemory.poll();
    long originSize=keyStream.size() + valueStream.size();
    keyStream.completeWindow(windowId);
    valueStream.completeWindow(windowId);
    memoryFreed+=originSize - (keyStream.size() + valueStream.size());
  }
  if (memoryFreed > 0) {
    LOG.debug("Total freed memory size: {}",memoryFreed);
    sizeInBytes.getAndAdd(-memoryFreed);
  }
  return memoryFreed;
}

protected long releaseMemory ( ) { long memoryFreed = 0 ; while ( ! windowsForFreeMemory . isEmpty ( ) ) { long windowId = windowsForFreeMemory . poll ( ) ; long originSize = keyStream . size ( ) + valueStream . size ( ) ; keyStream . completeWindow ( windowId ) ; valueStream . completeWindow ( windowId ) ; memoryFreed += originSize - ( keyStream . size ( ) + valueStream . size ( ) ) ; } if ( memoryFreed > 0 ) { LOG . debug ( "Total freed memory size: {}" , memoryFreed ) ; sizeInBytes . getAndAdd ( - memoryFreed ) ; } return memoryFreed ; }
this operation must be called from operator thread . it won ' t do anything if no memory to be freed
