This method is called when the flow is completed.
        Return 0 if success
975986
def finalize(self):
        
        if self.finalized:
            self.history.warning("Calling finalize on an already finalized flow.")
            return 1

        self.history.info("Calling flow.finalize.")
        self.finalized = True

        if self.has_db:
            self.history.info("Saving results in database.")
            try:
                self.flow.db_insert()
                self.finalized = True
            except Exception:
                 logger.critical("MongoDb insertion failed.")
                 return 2

        # Here we remove the big output files if we have the garbage collector
        # and the policy is set to "flow."
        if self.gc is not None and self.gc.policy == "flow":
            self.history.info("gc.policy set to flow. Will clean task output files.")
            for task in self.iflat_tasks():
                task.clean_output_files()

        return 0
266351
def _finish(self):
        
        if self._process.returncode is None:
            self._process.stdin.flush()
            self._process.stdin.close()
            self._process.wait()
            self.closed = True
609402
private void checkForCompletion() {
        final String methodName = "checkForCompletion(): ";
        // mark the list complete if that was the last awaited key
        if (actualIndices.isEmpty()) {
            if (tc.isDebugEnabled())
                Tr.debug(tc, methodName + "setting state=COMPLETE");
            state = failedKeys == null
                            ? COMPLETE
                            : COMPLETE_WITH_FAILURES; // state change must be last action
        }
    }
835246
def finalize_concurrency_state(self, outcome):
        
        final_outcome = outcome
        self.write_output_data()
        self.check_output_data_type()
        self.execution_history.push_return_history_item(self, CallType.CONTAINER, self, self.output_data)
        self.state_execution_status = StateExecutionStatus.WAIT_FOR_NEXT_STATE

        singleton.state_machine_execution_engine._modify_run_to_states(self)

        if self.preempted:
            final_outcome = Outcome(-2, "preempted")
        return self.finalize(final_outcome)
44414
def is_done(self):
        
        if not self.is_ready():
            return False
        done = (self._state.content['isDone'] == '1')
        return done
1293577
public void finished()
  {
    if (tc.isEntryEnabled()) 
      SibTr.entry(tc, "finished");
      
    anycastIHIterator = null;
      
    if (tc.isEntryEnabled()) 
      SibTr.exit(tc, "finished");
  }
807284
def finish(self, data: bytes=b"") -> None:
        
        if self.finished():
            if self._exc:
                raise self._exc

            if data:
                raise WriteAfterFinishedError

            return

        try:
            self._delegate.write_data(data, finished=True)

        except BaseWriteException as e:
            if self._exc is None:
                self._exc = e

            raise

        finally:
            self._finished.set()
140540
public boolean waitForCompletion() {
        try {
            return completed.get();
        } catch (InterruptedException | ExecutionException e) {
            log.warn("Failed to wait for server completion", e);
        }

        return false;
    }
687702
def finish(self) -> None:
        """"""
        if (
            self._expected_content_remaining is not None
            and self._expected_content_remaining != 0
            and not self.stream.closed()
        ):
            self.stream.close()
            raise httputil.HTTPOutputError(
                "Tried to write %d bytes less than Content-Length"
                % self._expected_content_remaining
            )
        if self._chunking_output:
            if not self.stream.closed():
                self._pending_write = self.stream.write(b"0\r\n\r\n")
                self._pending_write.add_done_callback(self._on_write_complete)
        self._write_finished = True
        # If the app finished the request while we're still reading,
        # divert any remaining data away from the delegate and
        # close the connection when we're done sending our response.
        # Closing the connection is the only way to avoid reading the
        # whole input body.
        if not self._read_finished:
            self._disconnect_on_finish = True
        # No more data is coming, so instruct TCP to send any remaining
        # data immediately instead of waiting for a full packet or ack.
        self.stream.set_nodelay(True)
        if self._pending_write is None:
            self._finish_request(None)
        else:
            future_add_done_callback(self._pending_write, self._finish_request)
69718
public boolean isFinished()
	{
		if (finished)
			return true;

		try
		{
			final int code = exitCode();

			finished(code);

			return true;
		}
		catch (IllegalThreadStateException e)
		{
			return false;
		}
	}
396588
void complete()
    {
        _file=_stack[_top].getFileName()+":"+_stack[_top].getLineNumber();
        _method=_stack[_top].getClassName()+"."+_stack[_top].getMethodName();
        _depth=_stack.length-_top;
        _thread = Thread.currentThread().getName();
    }
1253970
def is_finished(self):
        
        self.get_info()

        if 'status' not in self.info:
            return False
        if 'isFinished' not in self.info['status']:
            return False

        return self.info['status']['isFinished']
851052
public void checkExecutorHealth() {
    final Map<Optional<Executor>, List<ExecutableFlow>> exFlowMap = getFlowToExecutorMap();
    for (final Map.Entry<Optional<Executor>, List<ExecutableFlow>> entry : exFlowMap.entrySet()) {
      final Optional<Executor> executorOption = entry.getKey();
      if (!executorOption.isPresent()) {
        final String finalizeReason = "Executor id of this execution doesn't exist.";
        for (final ExecutableFlow flow : entry.getValue()) {
          logger.warn(
              String.format("Finalizing execution %s, %s", flow.getExecutionId(), finalizeReason));
          ExecutionControllerUtils
              .finalizeFlow(this.executorLoader, this.alerterHolder, flow, finalizeReason, null);
        }
        continue;
      }

      final Executor executor = executorOption.get();
      try {
        // Todo jamiesjc: add metrics to monitor the http call return time
        final Map<String, Object> results = this.apiGateway
            .callWithExecutionId(executor.getHost(), executor.getPort(),
                ConnectorParams.PING_ACTION, null, null);
        if (results == null || results.containsKey(ConnectorParams.RESPONSE_ERROR) || !results
            .containsKey(ConnectorParams.STATUS_PARAM) || !results.get(ConnectorParams.STATUS_PARAM)
            .equals(ConnectorParams.RESPONSE_ALIVE)) {
          throw new ExecutorManagerException("Status of executor " + executor.getId() + " is "
              + "not alive.");
        } else {
          // Executor is alive. Clear the failure count.
          if (this.executorFailureCount.containsKey(executor.getId())) {
            this.executorFailureCount.put(executor.getId(), 0);
          }
        }
      } catch (final ExecutorManagerException e) {
        handleExecutorNotAliveCase(entry, executor, e);
      }
    }
  }
149477
def on_channel_flow(self, method):
        
        if method.active:
            LOGGER.info('Channel flow is active (READY)')
            self.state = self.STATE_READY
            if self.on_ready:
                self.on_ready(self)
        else:
            LOGGER.warning('Channel flow is inactive (BLOCKED)')
            self.state = self.STATE_BLOCKED
            if self.on_unavailable:
                self.on_unavailable(self)
100315
def finalize(self, outcome=None):
        

        # Set the final outcome of the state
        if outcome is not None:
            self.final_outcome = outcome

        # If we are within a concurrency state, we have to notify it about our finalization
        if self.concurrency_queue:
            self.concurrency_queue.put(self.state_id)

        logger.debug("Finished execution of {0}: {1}".format(self, self.final_outcome))

        return None
283504
@Override
    public void afterCompletion(int status) {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) { 
            Tr.entry(this, tc, "afterCompletion", status);
        }

        if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) { 
            Tr.debug(this, tc, "Using transaction wrapper@" + Integer.toHexString(this.hashCode()));
        }


        // When afterCompletionCode is called we need to reset the registeredForSync flag.
        registeredForSync = false;

        /*
         * Mark the transaction complete in the wrapper
         */

        // Do NOT catch runtime exceptions here.  Let them flow out
        // of the component.
        mcWrapper.transactionComplete();

        // Check if any connection handles are still in use
        if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) { 
            if (mcWrapper.getHandleCount() != 0) {
                // Issue warning that Connections have not been closed by
                // the end of the current UOW scope and will be closed
                // by the CM.
                //            Tr.warning(this, tc,"HANDLE_NOT_CLOSED_J2CA0055");
                // removed null check on mcWrapper since it is redundant from the above code (mcWrapper.getUOWCoordinator())
                String pmiName = mcWrapper.gConfigProps.cfName;
                Tr.debug(this, tc, "Information:  handle not closed at end of UOW for resource " + pmiName);
            }
        }

        // Do NOT catch the runtime exception which getConnectionManager
        // might throw. If this is thrown it is an internal bug and needs
        // to be fixed. Allow the rte to flow up to the container.
        boolean shareable = mcWrapper.getConnectionManager().shareable();

        //
        // If the connection is shareable, or it is non-shareable and the
        // handleCount is zero, then release it back to the poolManger,
        // otherwise simply reset the transaction related variables.
        //
        // Note: shareable connections are released at the end of
        // transactions regardless of outstanding handles because we don't
        // support handled being associated to an active
        //  managedConnection outside of a sharing boundary.
        if ((shareable) || (!shareable && mcWrapper.getHandleCount() == 0)) {
            if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) { 
                Tr.debug(this, tc, "afterCompletionCode: Releasing the connection to the pool. Shareable = " + shareable);
            }

            try {
                mcWrapper.releaseToPoolManager();
            } catch (Exception e) {
                // Do not rethrow the exception here since the client has
                // successfully completed the transaction.
                // Log the error and go on.

                // need to change the "291" number below
                com.ibm.ws.ffdc.FFDCFilter.processException(e, "com.ibm.ejs.j2c.RRSLocalTransactionWrapper.afterCompletion", "291", this);
                if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) { 
                    String pmiName = mcWrapper.gConfigProps.cfName;
                    Tr.debug(this, tc, "afterCompletionCode for resource " + pmiName + ":  caught Exception", e);
                }
            }

        } // end ((shareable) || (!shareable && mcWrapper.getHandleCount() == 0))
        else {
            /*
             * The cleanup of the coordinator and enlisted flag should
             * only be done here for non sharable connections. In other cases,
             * the coord is set to null in mcWrapper.releaseToPoolManager().
             * We only need to worry about this here if the handleCount is
             * greater than 0 (in which case the cleanup will be done
             * by the poolManager). Same thing for the tranFailed flag in
             * the MCWrapper.
             */
            enlisted = false;
            //     mcWrapper.setTranFailed(false);
            mcWrapper.setUOWCoordinator(null);
        }

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            Tr.exit(this, tc, "afterCompletion");

    }
637697
@Override
    public void complete(VirtualConnection vc, TCPWriteRequestContext wsc) {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) {
            Tr.entry(tc, "complete, vc=" + getVCHash());
        }

        synchronized (closeSync) {
            if (closeCalled) {
                if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) {
                    Tr.exit(tc, "complete");
                }
                return;
            }
        }

        VirtualConnection rc = vc;
        if (0 < this.asyncBytesToWrite) {
            // previously went async partly through the app data, continue now
            rc = encryptAndWriteAsync(this.asyncBytesToWrite, false, this.asyncTimeout);
        }

        if (null != rc) {
            // Nothing else needs to be done here. Report success up the chain.
            this.callback.complete(rc, this);
        }
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) {
            Tr.exit(tc, "complete");
        }
    }