docs: public void sendOutState(State<Serializable, Serializable> state,
                             String checkpointId,
                             boolean spillState,
                             String location) {
      lock.lock();
      try {
        // flush all the current data before sending the state
        flushRemaining();
  
        // serialize the state
        byte[] serializedState = serializer.serialize(state);
  
        CheckpointManager.InstanceStateCheckpoint.Builder instanceStateBuilder =
            CheckpointManager.InstanceStateCheckpoint.newBuilder();
        instanceStateBuilder.setCheckpointId(checkpointId);
  
        if (spillState) {
          FileUtils.cleanDir(location);
  
          String stateLocation = location + checkpointId + "-" + UUID.randomUUID();
          if (!FileUtils.writeToFile(stateLocation, serializedState, true)) {
            throw new RuntimeException("failed to spill state. Bailing out...");
          }
          instanceStateBuilder.setStateLocation(stateLocation);
        } else {
          instanceStateBuilder.setState(ByteString.copyFrom(serializedState));
        }
  
        CheckpointManager.StoreInstanceStateCheckpoint storeRequest =
            CheckpointManager.StoreInstanceStateCheckpoint.newBuilder()
                .setState(instanceStateBuilder.build())
                .build();
  
        // Put the checkpoint to out stream queue
        outQueue.offer(storeRequest);
      } finally {
        lock.unlock();
      }
    }