We delegate node recovery detection to the
{@link AsyncRecoveryFailureDetector} class. When it determines that the
node has recovered, this callback is executed with the newly-recovered
node.
554727
@Override
    protected void nodeRecovered(Node node) {
        NodeStatus nodeStatus = getNodeStatus(node);

        synchronized(nodeStatus) {
            nodeStatus.resetCounters(getConfig().getTime().getMilliseconds());
            super.nodeRecovered(node);
        }
    }
1347759
public void onNodeFinished(final IExecutionNode node) {

        final Description description = descriptionMap.get(Long.valueOf(node.getId()));
        notifyTestFinished(description);

    }
695538
final void connectionErrorOccurred(final Exception exception, boolean callJCAListener) {

        final String methodName = "connectionErrorOccurred";
        if (TraceComponent.isAnyTracingEnabled() && TRACE.isEntryEnabled()) {
            SibTr.entry(this, TRACE, methodName, new Object[] {exception, Boolean.valueOf(callJCAListener)});
        }

        /*
         * Only send the event if the server isn't shutting down anyway
         */

        if (!SibRaEngineComponent.isServerStopping()) {

            final ConnectionEvent event = new ConnectionEvent(this,
                    ConnectionEvent.CONNECTION_ERROR_OCCURRED, exception);

            // Copy list to protect against concurrent modification by listener
            final List copy;
            synchronized (_eventListeners) {
                copy = new ArrayList(_eventListeners);
            }

            // We will probably want to look into some how figuring out that the listener
            // that we have is the JCA one and not some other one that is registered as we
            // want to call those ones regardless of the flag. Today we only expect JCA listeners
            // to be registered
            if (callJCAListener)
            {
              for (final Iterator iterator = copy.iterator(); iterator.hasNext();) {
                final Object object = iterator.next();
                if (object instanceof ConnectionEventListener) {
                    ((ConnectionEventListener) object)
                            .connectionErrorOccurred(event);
                }
              }
            }

        }

        if (TraceComponent.isAnyTracingEnabled() && TRACE.isEntryEnabled()) {
            SibTr.exit(this, TRACE, methodName);
        }

    }
1260276
@Override
	public void onTaskFailure(Execution taskExecution, Throwable cause) {

		executionGraph.getJobMasterMainThreadExecutor().assertRunningInMainThread();

		// to better handle the lack of resources (potentially by a scale-in), we
		// make failures due to missing resources global failures 
		if (cause instanceof NoResourceAvailableException) {
			LOG.info("Not enough resources to schedule {} - triggering full recovery.", taskExecution);
			executionGraph.failGlobal(cause);
			return;
		}

		LOG.info("Recovering task failure for {} (#{}) via individual restart.", 
				taskExecution.getVertex().getTaskNameWithSubtaskIndex(), taskExecution.getAttemptNumber());

		numTaskFailures.inc();

		// trigger the restart once the task has reached its terminal state
		// Note: currently all tasks passed here are already in their terminal state,
		//       so we could actually avoid the future. We use it anyways because it is cheap and
		//       it helps to support better testing
		final CompletableFuture<ExecutionState> terminationFuture = taskExecution.getTerminalStateFuture();
		terminationFuture.thenRun(
			() -> performExecutionVertexRestart(taskExecution.getVertex(), taskExecution.getGlobalModVersion()));
	}
1193079
public static <T> Callable<T> recover(Callable<T> callable, Function<Exception, T> exceptionHandler){
        return () -> {
            try{
                return callable.call();
            }catch (Exception exception){
                return exceptionHandler.apply(exception);
            }
        };
    }
444616
public void alarm(Object alarmObj)
   {
      if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) SibTr.entry(this, tc, "alarm", alarmObj);
      
      // Has the async reader already sent a message back to the client?
      boolean sessionAvailable = true;
      if (!asynchReader.isComplete())
      {
         if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) SibTr.debug(this, tc, "Async reader has not yet got a message");
         // If not stop the session...
         try 
         {
            try
            {
               asynchReader.stopSession();
            }
            catch (SISessionDroppedException e)
            {
               // No FFDC Code Needed
               if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) SibTr.debug(this, tc, "Caught a SISessionDroppedException", e);
               sessionAvailable = false;
               
               // If the session was unavailable this may be because the connection has been closed 
               // while we were in a receiveWithWait(). In this case, we should try and send an error
               // back to the client, but not worry too much if we cannot
               if (asynchReader.isConversationClosed())
               {
                  if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) 
                     SibTr.debug(this, tc, "The Conversation was closed - no need to panic");
               }
               else
               {
                  // The Conversation was not closed, throw the exception on so we can inform
                  // the client
                  throw e;
               }
            }
            catch (SISessionUnavailableException e)
            {
               // No FFDC Code Needed
               if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) SibTr.debug(this, tc, "Caught a SISessionUnavailableException", e);
               sessionAvailable = false;
               
               // See the comments above...
               if (asynchReader.isConversationClosed())
               {
                  if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) 
                     SibTr.debug(this, tc, "The Conversation was closed - no need to panic");
               }
               else
               {
                  throw e;
               }
            }
         }
         catch (SIException sis)
         {
            //No FFDC code needed
            //Only FFDC if we haven't received a meTerminated event.
            if(!asynchReader.hasMETerminated())
            {
               FFDCFilter.processException(sis, CLASS_NAME + ".alarm", 
                                           CommsConstants.CATTIMER_ALARM_01, this);
            }
            
            sessionAvailable = false;
            if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) SibTr.debug(tc, sis.getMessage(), sis);
            
            asynchReader.sendErrorToClient(sis,
                                           CommsConstants.CATTIMER_ALARM_01);
         }
         finally
         {
            //At this point we should deregister asyncReader as a SICoreConnectionListener as the current receive with wait is pretty much finished.
            //Note that there is a bit of a timing window where asyncReader.consumeMessages could be called and we end up deregistering 
            //the listener twice but that is allowed by the API. We minimize this window by only doing a dereg if the asyncReader isn't complete.
            if(!asynchReader.isComplete())
            {
               try
               {
                  final MPConsumerSession mpSession = (MPConsumerSession) asynchReader.getCATMainConsumer().getConsumerSession();
                  mpSession.getConnection().removeConnectionListener(asynchReader);
               }
               catch(SIException e)
               {
                  //No FFDC code needed   
                  if(TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) SibTr.debug(this, tc, e.getMessage(), e);
                  
                  //No need to send an exception back to the client in this case as we are only really performing tidy up processing.
                  //Also, if it is a really bad problem we are likely to have already send the exception back in the previous catch block.
               }
            }
         }
      }
      
      if (sessionAvailable)
      {
         // ...and check again.
         // If the async reader has still not sent a message at this point we
         // assume a timeout and inform the client
         if (asynchReader.isComplete())
         {
            if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) SibTr.debug(this, tc, "Async reader got a message");
         }
         else
         {
            if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) SibTr.debug(this, tc, "No message received");
            
            asynchReader.sendNoMessageToClient();
         }
      }
            
      if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) SibTr.exit(this, tc, "alarm");
   }
700488
public void onMainExecutionComplete(Throwable t) {
        try {
            // May be called twice, don't do anything the second time
            if (mainExecutionComplete) {
                return;
            }
            mainExecutionComplete = true;

            this.failure = t;

            onAttemptComplete(t);

            if (t instanceof CircuitBreakerOpenException) {
                // We didn't run anything, execution context needs closing
                close();
            }

            if (retry != null) {
                if (retry.canRetryFor(null, t)) {
                    // This is a retryable failure
                    metricRecorder.incrementRetryCallsFailureCount();
                } else {
                    // Not a retryable failure
                    if (retries > 0) {
                        metricRecorder.incrementRetryCallsSuccessRetriesCount();
                    } else {
                        metricRecorder.incrementRetryCallsSuccessImmediateCount();
                    }
                }
            }
        } catch (Exception e) {
            // Unchecked exceptions thrown here can be swallowed by Failsafe
            // This catch ensures we at least get an FFDC
            throw e;
        }

    }
235662
private void handleDiscoveryEvent(NodeDiscoveryEvent event) {
    switch (event.type()) {
      case JOIN:
        handleJoinEvent(event.subject());
        break;
      case LEAVE:
        handleLeaveEvent(event.subject());
        break;
      default:
        throw new AssertionError();
    }
  }
1123533
public void onFailure(Throwable t)
    {
        MessagingService.instance().sendOneWay(new SyncComplete(desc, request.src, request.dst, false).createMessage(), request.initiator);
    }
398703
public void directTermination(FailureScope failureScope) throws TerminationFailedException {
        if (tc.isEntryEnabled())
            Tr.entry(tc, "directTermination", new Object[] { failureScope, this });

        Tr.info(tc, "CWRLS0014_HALT_PEER_RECOVERY", failureScope.serverName());

        // If callbacks are registered, drive then now.
        if (_registeredCallbacks != null) {
            driveCallBacks(CALLBACK_TERMINATIONSTARTED, failureScope);
        }

        if (Configuration.HAEnabled()) {
            Configuration.getRecoveryLogComponent().leaveCluster(failureScope);
        }

        // Extract the 'values' collection from the _registeredRecoveryAgents map and create an iterator
        // from it. This iterator will return ArrayList objects each containing a set of RecoveryAgent
        // objects. Each ArrayList corrisponds to a different sequence priority value.
        final Collection registeredRecoveryAgentsValues = _registeredRecoveryAgents.values();
        final Iterator registeredRecoveryAgentsValuesIterator = registeredRecoveryAgentsValues.iterator();

        while (registeredRecoveryAgentsValuesIterator.hasNext()) {
            // Extract the next ArrayList and create an iterator from it. This iterator will return RecoveryAgent
            // objects that are registered at the same sequence priority value.
            final ArrayList registeredRecoveryAgentsArray = (java.util.ArrayList) registeredRecoveryAgentsValuesIterator.next();
            final Iterator registeredRecoveryAgentsArrayIterator = registeredRecoveryAgentsArray.iterator();

            while (registeredRecoveryAgentsArrayIterator.hasNext()) {
                // Extract the next RecoveryAgent object
                final RecoveryAgent recoveryAgent = (RecoveryAgent) registeredRecoveryAgentsArrayIterator.next();

                // Record the fact that we have an outstanding termination request for the RecoveryAgent.
                addTerminationRecord(recoveryAgent, failureScope);

                // Direct the RecoveryAgent instance to terminate processing of this failure scope
                try {
                    recoveryAgent.terminateRecovery(failureScope);
                } catch (TerminationFailedException exc) {
                    FFDCFilter.processException(exc, "com.ibm.ws.recoverylog.spi.RecoveryDirectorImpl.directTermination", "540", this);
                    if (tc.isEntryEnabled())
                        Tr.exit(tc, "directTermination", exc);
                    throw exc;
                } catch (Exception exc) {
                    FFDCFilter.processException(exc, "com.ibm.ws.recoverylog.spi.RecoveryDirectorImpl.directTermination", "576", this);
                    if (tc.isEntryEnabled())
                        Tr.exit(tc, "directTermination", exc);
                    throw new TerminationFailedException(exc);
                }

                // Wait for 'terminationComplete' to be called. This callback may be issued from another thread.
                synchronized (_outstandingTerminationRecords) {
                    while (terminationOutstanding(recoveryAgent, failureScope)) {
                        try {
                            _outstandingTerminationRecords.wait();
                        } catch (InterruptedException exc) {
                            // This exception is recieved if another thread interrupts this thread by calling this threads
                            // Thread.interrupt method. The RecoveryDirectorImpl class does not use this mechanism for
                            // breaking out of the wait call - it uses notifyAll to wake up all waiting threads. This
                            // exception should never be generated. If for some reason it is called then ignore it and
                            //start to wait again.
                            FFDCFilter.processException(exc, "com.ibm.ws.recoverylog.spi.RecoveryDirectorImpl.directTermination", "549", this);
                        }
                    }
                }
            }
        }

        // If callbacks are registered, drive then now.
        if (_registeredCallbacks != null) {
            driveCallBacks(CALLBACK_TERMINATIONCOMPLETE, failureScope);
        }

        if (tc.isEntryEnabled())
            Tr.exit(tc, "directTermination");
    }
475012
public void driveLocalRecovery() throws RecoveryFailedException {
        if (tc.isEntryEnabled())
            Tr.entry(tc, "driveLocalRecovery", this);

        if (tc.isDebugEnabled())
            Tr.debug(tc, "RLSHA: configuring for local only recovery");

        synchronized (this) {
            if (_currentFailureScope == null) {
                // Create a FailureScope that defines the current server.
                _currentFailureScope = new FileFailureScope();
            }
        }

        try {
            directInitialization(_currentFailureScope);
        } catch (RecoveryFailedException exc) {
            FFDCFilter.processException(exc, "com.ibm.ws.recoverylog.spi.RecoveryDirectorImpl.driveLocalRecovery", "620", this);
            if (tc.isEntryEnabled())
                Tr.exit(tc, "driveLocalRecovery", exc);
            throw exc;
        }

        if (tc.isEntryEnabled())
            Tr.exit(tc, "driveLocalRecovery");
    }
115460
@Override
    public void replayCompletion(String globalId) {
        if (rc == null) {
            // Find a factory to create the RecoveryCoordinator
            try {
                rc = getRecoveryCoordinator();
            } catch (RecoveryCoordinatorNotAvailableException e) {
                if (tc.isDebugEnabled())
                    Tr.debug(tc, "replayCompletion", e);
            }
        }

        if (rc != null) {
            rc.replayCompletion(globalId);
        }
    }
269641
private synchronized boolean datanodeReportsReceived(boolean checkDatanodes) {
    try {
      boolean received = this.getDatanodeReportRatio() >=
        this.outStandingReportThreshold;
      if (!received && checkDatanodes) {
        checkDatanodes();
        return this.getDatanodeReportRatio() >= this.outStandingReportThreshold;
      }
      return received;
    } catch (Exception e) {
      LOG.warn("Failover - caught exception when checking reports", e);
      return false;
    }
  }
65328
public void onNodeFailed(final IExecutionNode node, final Throwable cause) {

        final Description description = descriptionMap.get(Long.valueOf(node.getId()));
        notifyTestFailed(description, cause);

    }
592785
@Override
	public void afterCompletion(int status) {

		CompletionDelegate delegate = this.delegates.get(status);
		if (delegate != null) {
			delegate.execute(this.solrOperations);
		}
	}
1289503
@Override
	public void onTaskFailure(Execution taskExecution, Throwable cause) {
		final ExecutionVertex ev = taskExecution.getVertex();
		final FailoverRegion failoverRegion = vertexToRegion.get(ev);

		if (failoverRegion == null) {
			executionGraph.failGlobal(new FlinkException(
					"Can not find a failover region for the execution " + ev.getTaskNameWithSubtaskIndex(), cause));
		}
		else {
			LOG.info("Recovering task failure for {} #{} ({}) via restart of failover region",
					taskExecution.getVertex().getTaskNameWithSubtaskIndex(),
					taskExecution.getAttemptNumber(),
					taskExecution.getAttemptId());

			failoverRegion.onExecutionFail(taskExecution, cause);
		}
	}
55053
public static <V> CompletionHandler<V, Void> createAsyncHandler(final Consumer<V> success, final Consumer<Throwable> fail) {
        return new CompletionHandler<V, Void>() {
            @Override
            public void completed(V result, Void attachment) {
                if (success != null) success.accept(result);
            }

            @Override
            public void failed(Throwable exc, Void attachment) {
                if (fail != null) fail.accept(exc);
            }
        };
    }