A ControlResetRequestAck message tells the RME to start re-sending get requests with an eager timeout since
the DME has crashed and recovered.
557445
public void processResetRequestAck(long dmeVersion, SendDispatcher sendDispatcher)
  {
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.entry(tc, "processResetRequestAck", Long.valueOf(dmeVersion));

    // Only consider non-stale request acks
    if (dmeVersion >= _latestDMEVersion)
    {
      if (dmeVersion > _latestDMEVersion)
        _latestDMEVersion = dmeVersion;

      _slowedGetTOM.applyToEachEntry(new AddToEagerTOM(_slowedGetTOM, dmeVersion));
      sendDispatcher.sendResetRequestAckAck(dmeVersion);
    }

    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.exit(tc, "processResetRequestAck");
  }
201037
protected void eventPreUnlocked(SIMPMessage msg, TransactionCommon tran)
    {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.entry(tc, "eventPreUnlocked", new Object[] { msg, msg.guessRedeliveredCount() });

        /*
         * 166955.4
         * Check whether the retryCount has been exceeded. If it has then this is a poison msg
         * and needs to be sent to the exception destination rather than to a valid consumer
         * again
         */
        // But only check for messages that are not persistently locked (messages involved in RemoteGet)
        // Otherwise the remote get protocols dont work

        if ((msg.guessRedeliveredCount() + 1) >= _baseDestHandler.getMaxFailedDeliveries())
        {
            /*
             * Message has been delivered a greater number of times than allowed.
             * Route message on to the exception destination associated with this
             * destination.
             */

            // Deregister rollback callbacks to avoid recursive event callbacks
            msg.deregisterMessageEventListener(MessageEvents.POST_ROLLBACK_REMOVE, this);
            msg.deregisterMessageEventListener(MessageEvents.POST_ROLLBACK_ADD, this);

            UndeliverableReturnCode rc = null;
            try
            {
                if (_isPubSub)
                {
                    // F001333-14610
                    // Pass the subscription ID to the destination handler
                    // so that it can be set on the exceptioned message.
                    rc = _baseDestHandler.handleUndeliverableMessage(msg,
                                                                     SIRCConstants.SIRC0035_BACKOUT_THRESHOLD_ERROR,
                                                                     new String[] { _baseDestHandler.getName(),
                                                                                   _messageProcessor.getMessagingEngineName() },
                                                                     tran,
                                                                     dispatcherState.getSubscriberID());
                }
                else
                {
                    // Not pub sub so no subscription ID to pass
                    rc = _baseDestHandler.handleUndeliverableMessage(msg,
                                                                     SIRCConstants.SIRC0035_BACKOUT_THRESHOLD_ERROR,
                                                                     new String[] { _baseDestHandler.getName(),
                                                                                   _messageProcessor.getMessagingEngineName() },
                                                                     tran);
                }
            } catch (SIResourceException e)
            {
                // No FFDC code needed
                // Exception had already been handled
            }

            // re-register the callbacks
            msg.registerMessageEventListener(MessageEvents.POST_ROLLBACK_REMOVE, this);
            msg.registerMessageEventListener(MessageEvents.POST_ROLLBACK_ADD, this);

            if (rc == null || rc == UndeliverableReturnCode.ERROR)
            {
                // Spin of a new alarm
                _messageProcessor.getAlarmManager().
                                create(SIMPConstants.EXCEPTION_RETRY_TIMEOUT,
                                       new ExceptionDestinationRetryHandler(msg, this));
            }

            if (rc == UndeliverableReturnCode.BLOCK)
            {
                // PK03344 - Handle retry count when no exception destination available
                // When the retry count is reached and we have nowhere to send the message
                // we suspend the consumer for a period of time to avoid 100% cpu usage
                // due to continuous message retry.

                // MDBs will only have one consumer (PK03344 handles the MDB case).
                // For multiple consumers we suspend them all.

                // Pause the consumers for the specified time
                pauseConsumers(msg, _baseDestHandler.getBlockedRetryInterval());
            }

            if (rc != UndeliverableReturnCode.BLOCK)
                msg.setRedeliveryCountReached();

        } // End of 165955.4

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.exit(tc, "eventPreUnlocked");
    }
812913
def _handle_exception(self, sender, exception=None):
        
        honeybadger.notify(exception)
        if self.reset_context_after_request:
            self._reset_context()
566133
def _maybe_drop_message(self):
        
        if self.headers[_PROCESSING_EXCEPTIONS] >= self._error_max_retry:
            self.logger.warning('Dropping message with %i deaths due to '
                                'ERROR_MAX_RETRY',
                                self.headers[_PROCESSING_EXCEPTIONS])
            if self._drop_exchange:
                self._republish_dropped_message('max retries ({})'.format(
                    self.headers[_PROCESSING_EXCEPTIONS]))
            raise DropMessage
1237741
private void reconstituteRemoteQueuePoints(int startMode,
                                               boolean isSystem)
                    throws SIDiscriminatorSyntaxException, MessageStoreException, SIResourceException
    {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.entry(tc,
                        "reconstituteRemoteQueuePoints",
                        new Object[] {
                                      Integer.valueOf(startMode),
                                      Boolean.valueOf(isSystem),
                                      this });

        // Next look for any remote transmission itemstreams to queue points
        int transmissionQCount = 0;
        PtoPXmitMsgsItemStream ptoPXmitMsgsItemStream = null;

        // There can be one or more transmission itemstreams in the BaseDestinationHandler.
        NonLockingCursor cursor =
                        _baseDestinationHandler.newNonLockingItemStreamCursor(
                                        new ClassEqualsFilter(PtoPXmitMsgsItemStream.class));

        do
        {
            ptoPXmitMsgsItemStream = (PtoPXmitMsgsItemStream) cursor.next();

            if (ptoPXmitMsgsItemStream != null)
            {
                ptoPXmitMsgsItemStream.reconstitute(_baseDestinationHandler);

                if (ptoPXmitMsgsItemStream.isToBeDeleted())
                {
                    ptoPXmitMsgsItemStream.setDeleteRequiredAtReconstitute(true);
                    SIBUuid8 remoteMEUuid = ptoPXmitMsgsItemStream.getLocalizingMEUuid();
                    _postMediatedItemStreamsRequiringCleanup.put(
                                                                 remoteMEUuid, ptoPXmitMsgsItemStream);
                    _baseDestinationHandler.setHasReconciledStreamsToBeDeleted(true);
                    _localisationManager.
                                    attachRemotePtoPLocalisation(ptoPXmitMsgsItemStream,
                                                                 _remoteSupport);
                    // Reconstitute source streams for PtoPOutputHandler
                    _remoteSupport.reconstituteSourceStreams(startMode,
                                                             (PtoPOutputHandler) ptoPXmitMsgsItemStream.getOutputHandler());
                    assignQueuePointOutputHandler(
                                                  ptoPXmitMsgsItemStream.getOutputHandler(),
                                                  ptoPXmitMsgsItemStream.getLocalizingMEUuid());
                }
                else
                {
                    transmissionQCount++;
                    _localisationManager.
                                    attachRemotePtoPLocalisation(ptoPXmitMsgsItemStream,
                                                                 _remoteSupport);
                    // Reconstitute source streams for PtoPOutputHandler
                    _remoteSupport.reconstituteSourceStreams(startMode,
                                                             (PtoPOutputHandler) ptoPXmitMsgsItemStream.getOutputHandler());

                    /* Feature 176658.3.2 */
                    assignQueuePointOutputHandler(
                                                  ptoPXmitMsgsItemStream.getOutputHandler(),
                                                  ptoPXmitMsgsItemStream.getLocalizingMEUuid());

                    //Restore the guess sets for system destinations
                    if (isSystem)
                    {
                        _localisationManager.addMEToQueuePointGuessSet(ptoPXmitMsgsItemStream.getLocalizingMEUuid());
                        _localisationManager.addMEToRemoteQueuePointGuessSet(ptoPXmitMsgsItemStream.getLocalizingMEUuid());
                    }
                }
            }
        } while (ptoPXmitMsgsItemStream != null);

        cursor.finished();

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.exit(tc, "reconstituteRemoteQueuePoints");

    }
1136660
def reload(self, reload_timeout=300, save_config=True, no_reload_cmd=False):
        
        begin = time.time()
        self._chain.target_device.clear_info()
        result = False
        try:
            result = self._chain.target_device.reload(reload_timeout, save_config, no_reload_cmd)

        except ConnectionStandbyConsole as exc:
            message = exc.message
            self.log("Active RP became standby: {}".format(message))
            self.disconnect()
            # self._conn_status.clear()
            result = self.reconnect()

        # connection error caused by device booting up
        except ConnectionError as exc:
            message = exc.message
            self.log("Connection error: {}".format(message))
            self.disconnect()
            result = self.reconnect()

        if result:
            elapsed = time.time() - begin
            self.emit_message("Target device reload last {:.0f}s.".format(elapsed), log_level=logging.INFO)
        else:
            self.emit_message("Target device not reloaded.", log_level=logging.ERROR)
        return result
1238033
def _no_answer_do_retry(self, pk, pattern):
        """"""
        logger.info('Resending for pattern %s', pattern)
        # Set the timer to None before trying to send again
        self.send_packet(pk, expected_reply=pattern, resend=True)
903484
private void updateFailure(DownloadRequest request, int statusCode, String errMsg) {
    updateState(request, DownloadState.FAILURE);

    /* if the status code is 0, may be cause by the net error */
    int leftRetryTime = request.retryTime();
    if (leftRetryTime >= 0) {
      try {
        /* sleep a while before retrying */
        sleep(request.retryInterval());
      } catch (InterruptedException e) {
        /* we may have been interrupted because it was time to quit */
        if (quit) {
          request.finish();
          return;
        }
      }

      /* retry downloading */
      if (!request.isCanceled()) {
        logger.log("Retry DownloadRequest: "
            + request.downloadId()
            + " left retry time: "
            + leftRetryTime);
        updateRetry(request);
        executeDownload(request);
      }

      return;
    }

    /* notify the request that downloading has finished */
    request.finish();

    /* deliver failure message */
    delivery.postFailure(request, statusCode, errMsg);
  }
419180
private void doFail(long id) {
    if (log.isDebugEnabled()) {
      log.debug(String.format("%s - Received resend request for messages starting at %d", this, id));
    }

    // Ack all the entries before the given ID.
    doAck(id);

    // Now that all the entries before the given ID have been removed,
    // just iterate over the messages map and resend all the messages.
    Iterator<Map.Entry<Long, JsonObject>> iter = messages.entrySet().iterator();
    while (iter.hasNext()) {
      eventBus.send(inAddress, iter.next().getValue());
    }
  }
322067
private static void resumeRLS(RLSSuspendToken token) throws  RLSInvalidSuspendTokenException
    {
        if (tc.isEntryEnabled()) Tr.entry(tc, "resumeRLS", token);
        
        synchronized(SUSPEND_LOCK)
        {
            try
            {
            	_tokenManager.registerResume(token);
            }
            catch (RLSInvalidSuspendTokenException exc)
            {
                FFDCFilter.processException(exc, "com.ibm.ws.recoverylog.spi.RecoveryLogService.resume", "120");
                if (tc.isEventEnabled()) Tr.event(tc, "Rethrow RLSInvalidSuspendTokenException - attempt to resume an invalid token", token);
                if (tc.isEntryEnabled()) Tr.exit(tc, "resumeRLS", exc);
                throw exc;
            }
                
            // Check if there are any outstanding suspends
            if (_tokenManager.isResumable())
            {
                if (tc.isEventEnabled()) Tr.event(tc, "There are no outstanding suspends - resuming recovery log service");
                
                Tr.info(tc, "CWRLS0002_RESUME_RLS", token);
                
                _isSuspended = false;
                
                // Notify all waiting threads that
                // normal service is resumed
                SUSPEND_LOCK.notifyAll();
            }
            else
            {
            	Tr.info(tc, "CWRLS0021_RESUME_ATTEMPT_RLS", token);
                if (tc.isEventEnabled()) Tr.event(tc, "There are still outstanding suspensions - recovery log service remains suspended");   
            }
        }
        
        if (tc.isEntryEnabled()) Tr.exit(tc, "resumeRLS");
    }
900974
def __send_retry_requests(self, last_send_failure_time):
        """"""
        # make sure multiple failures having set multiple times do not run concurrently
        with self.__send_retry_requests_lock:
            with self.__requests:
                # produce list instead of generator as requests mapping can change during subsequent loop
                retry_reqs = [req for req in self.__requests.values()
                              if req._sent_without_response(last_send_failure_time)]

            retry_req_count = 0
            # don't continue if another network failure has occured (which will trigger this function again)
            while retry_reqs and self.__amqplink.last_send_exc_time <= last_send_failure_time:
                req = retry_reqs.pop()
                # lock individuallly so incoming request handling does not 'pause' for too long
                with self.__requests:
                    # might have received a response (or finished since)
                    if not (req.id_ in self.__requests and req._sent_without_response(last_send_failure_time)):
                        logger.debug('Not resending request %s (finished or has received response)', req.id_)
                        continue
                logger.debug('Resending request %s', req.id_)
                if not self.__retry_enqueue(PreparedMessage(req._inner_msg_out, req.id_)):
                    # client shutdown
                    break
                retry_req_count += 1

        if retry_req_count:
            logger.debug('Resending of %d request(s) complete (before %s)', retry_req_count, last_send_failure_time)
703765
private void handleRpcException(RpcException e, int attemptNumber) throws IOException {
        String messageStart;

        if (!(e.getStatus().equals(RpcStatus.NETWORK_ERROR))) {
            messageStart = "rpc";
        } else {
            // check whether to retry
            if (attemptNumber + 1 < _maximumRetries) {
                try {
                    int waitTime = _retryWait * (attemptNumber + 1);
                    Thread.sleep(waitTime);
                } catch (InterruptedException ie) {
                    // restore the interrupt status
                    Thread.currentThread().interrupt();
                }
                LOG.warn("network error happens, server {}, attemptNumber {}", new Object[] { _server, attemptNumber });
                return;
            }

            messageStart = "network";
        }

        throw new NfsException(NfsStatus.NFS3ERR_IO,
                String.format("%s error, server: %s, RPC error: %s", messageStart, _server, e.getMessage()), e);
    }
658012
def _handle_retryable_error(self, e, retry_count):
        """"""
        if isinstance(e, error.RateLimitEnforcedError):
            time.sleep(e.retry_after)
        else:
            time.sleep(self.RETRY_DELAY * (self.RETRY_BACKOFF ** retry_count))
915186
def retry_with_recovery(
        transport: 'UDPTransport',
        messagedata: bytes,
        message_id: UDPMessageID,
        recipient: Address,
        stop_event: Event,
        event_healthy: Event,
        event_unhealthy: Event,
        backoff: Iterable[int],
) -> bool:
    

    # The underlying unhealthy will be cleared, care must be taken to properly
    # clear stop_or_unhealthy too.
    stop_or_unhealthy = event_first_of(
        stop_event,
        event_unhealthy,
    )

    acknowledged = False
    while not stop_event.is_set() and not acknowledged:

        # Packets must not be sent to an unhealthy node, nor should the task
        # wait for it to become available if the message has been acknowledged.
        if event_unhealthy.is_set():
            log.debug(
                'waiting for recipient to become available',
                node=pex(transport.raiden.address),
                recipient=pex(recipient),
            )
            wait_recovery(
                stop_event,
                event_healthy,
            )

            # Assume wait_recovery returned because unhealthy was cleared and
            # continue execution, this is safe to do because stop_event is
            # checked below.
            stop_or_unhealthy.clear()

            if stop_event.is_set():
                return acknowledged

        acknowledged = retry(
            transport,
            messagedata,
            message_id,
            recipient,

            # retry will stop when this event is set, allowing this task to
            # wait for recovery when the node becomes unhealthy or to quit if
            # the stop event is set.
            stop_or_unhealthy,

            # Intentionally reusing backoff to restart from the last
            # timeout/number of iterations.
            backoff,
        )

    return acknowledged
394446
@Override
  public RetryDecision onErrorResponse(
      @NonNull Request request, @NonNull CoordinatorException error, int retryCount) {

    RetryDecision decision =
        (error instanceof ReadFailureException || error instanceof WriteFailureException)
            ? RetryDecision.RETHROW
            : RetryDecision.RETRY_NEXT;

    if (decision == RetryDecision.RETRY_NEXT && LOG.isTraceEnabled()) {
      LOG.trace(RETRYING_ON_ERROR, logPrefix, retryCount, error);
    }

    return decision;
  }
955518
private void maybeSendNewPermitRequest() {
    if (!this.requestSemaphore.tryAcquire()) {
      return;
    }
    if (!this.retryStatus.canRetryNow()) {
      this.requestSemaphore.release();
      return;
    }
    try {
      long permits = computeNextPermitRequest();
      if (permits <= 0) {
        this.requestSemaphore.release();
        return;
      }

      PermitRequest permitRequest = this.basePermitRequest.copy();
      permitRequest.setPermits(permits);
      permitRequest.setMinPermits((long) this.permitsOutstanding.getAverageWeightOrZero());
      permitRequest.setVersion(ThrottlingProtocolVersion.WAIT_ON_CLIENT.ordinal());
      if (BatchedPermitsRequester.this.restRequestHistogram != null) {
        BatchedPermitsRequester.this.restRequestHistogram.update(permits);
      }

      log.debug("Sending permit request " + permitRequest);

      this.requestSender.sendRequest(permitRequest, new AllocationCallback(
          BatchedPermitsRequester.this.restRequestTimer == null ? NoopCloseable.INSTANCE :
              BatchedPermitsRequester.this.restRequestTimer.time(), new Sleeper()));
    } catch (CloneNotSupportedException cnse) {
      // This should never happen.
      this.requestSemaphore.release();
      throw new RuntimeException(cnse);
    }
  }
1363311
protected void reconnect(Throwable cause) {
		if (cause != null) {
			String msg = cause.getMessage();
			if (msg == null || msg.isEmpty()) {
				msg = "Unable to connect to JMS server!";
			} else if (!msg.endsWith("!") && !msg.endsWith(".")) {
				msg += "!";
			}
			logger.warn(msg);
		}
		disconnect();
		logger.info("Trying to reconnect...");
		scheduler.schedule(this::connect, 5, TimeUnit.SECONDS);
	}