void processRestartRequests() {
        List<RestartRequest> restartRequests;
        synchronized (this) {
            if (pendingRestartRequests.isEmpty()) {
                return;
            }
            //dequeue into a local list to minimize the work being done within the synchronized block
            restartRequests = new ArrayList<>(pendingRestartRequests.values());
            pendingRestartRequests.clear();
        }
        restartRequests.forEach(restartRequest -> {
            String stageDescription = "handling restart request for connector " + restartRequest.connectorName();
            try (TickThreadStage stage = new TickThreadStage(stageDescription)) {
                doRestartConnectorAndTasks(restartRequest);
            } catch (Exception e) {
                log.warn("Unexpected error while trying to process " + restartRequest + ", the restart request will be skipped.", e);
            }
        });
    }
@Test
    public void processRestartRequestsFailureSuppression() {
        doNothing().when(member).wakeup();

        final String connectorName = "foo";
        RestartRequest restartRequest = new RestartRequest(connectorName, false, false);
        doThrow(new RuntimeException()).when(herder).buildRestartPlan(restartRequest);

        configUpdateListener.onRestartRequest(restartRequest);
        assertEquals(1, herder.pendingRestartRequests.size());
        herder.processRestartRequests();
        assertTrue(herder.pendingRestartRequests.isEmpty());

        verifyNoMoreInteractions(worker, member, configBackingStore, statusBackingStore);
    }