public static AbstractFileOutputOperator checkpoint(AbstractFileOutputOperator writer,long windowId){
  if (windowId >= Stateless.WINDOW_ID) {
    writer.beforeCheckpoint(windowId);
  }
  Kryo kryo=new Kryo();
  ByteArrayOutputStream bos=new ByteArrayOutputStream();
  Output loutput=new Output(bos);
  kryo.writeObject(loutput,writer);
  loutput.close();
  Input lInput=new Input(bos.toByteArray());
  @SuppressWarnings("unchecked") AbstractFileOutputOperator checkPointedWriter=kryo.readObject(lInput,writer.getClass());
  lInput.close();
  return checkPointedWriter;
}

public static AbstractFileOutputOperator checkpoint ( AbstractFileOutputOperator writer , long windowId ) { if ( windowId >= Stateless . WINDOW_ID ) { writer . beforeCheckpoint ( windowId ) ; } Kryo kryo = new Kryo ( ) ; ByteArrayOutputStream bos = new ByteArrayOutputStream ( ) ; Output loutput = new Output ( bos ) ; kryo . writeObject ( loutput , writer ) ; loutput . close ( ) ; Input lInput = new Input ( bos . toByteArray ( ) ) ; @ SuppressWarnings ( "unchecked" ) AbstractFileOutputOperator checkPointedWriter = kryo . readObject ( lInput , writer . getClass ( ) ) ; lInput . close ( ) ; return checkPointedWriter ; }
this method checkpoints the given writer .
