code: @SuppressWarnings("unchecked")
    private void processRecord(Object convertedRecord, ForkOperator forkOperator, RowLevelPolicyChecker rowChecker,
        RowLevelPolicyCheckResults rowResults, int branches, AcknowledgableWatermark watermark)
        throws Exception {
      // Skip the record if quality checking fails
      if (!rowChecker.executePolicies(convertedRecord, rowResults)) {
        if (watermark != null) {
          watermark.ack();
        }
        return;
      }
  
      List<Boolean> forkedRecords = forkOperator.forkDataRecord(this.taskState, convertedRecord);
      if (forkedRecords.size() != branches) {
        throw new ForkBranchMismatchException(String
            .format("Number of forked data records [%d] is not equal to number of branches [%d]", forkedRecords.size(),
                branches));
      }
  
      boolean needToCopy = inMultipleBranches(forkedRecords);
      // we only have to copy a record if it needs to go into multiple forks
      if (needToCopy && !(CopyHelper.isCopyable(convertedRecord))) {
        throw new CopyNotSupportedException(convertedRecord.getClass().getName() + " is not copyable");
      }
  
      int branch = 0;
      int copyInstance = 0;
      for (Optional<Fork> fork : this.forks.keySet()) {
        if (fork.isPresent() && forkedRecords.get(branch)) {
          Object recordForFork = needToCopy ? CopyHelper.copy(convertedRecord) : convertedRecord;
          copyInstance++;
          if (isStreamingTask()) {
            // Send the record, watermark pair down the fork
            ((RecordEnvelope) recordForFork).addCallBack(watermark.incrementAck());
          }
          // Put the record into the record queue of each fork. A put may timeout and return a false, in which
          // case the put is retried until it is successful.
          boolean succeeded = false;
          while (!succeeded) {
            succeeded = fork.get().putRecord(recordForFork);
          }
        }
        branch++;
      }
      if (watermark != null) {
        watermark.ack();
      }
    }
docs: Process a (possibly converted) record.