@Override
    public void fenceZombieSourceTasks(final String connName, final Callback<Void> callback, InternalRequestSignature requestSignature) {
        log.trace("Submitting zombie fencing request {}", connName);
        if (requestNotSignedProperly(requestSignature, callback)) {
            return;
        }

        fenceZombieSourceTasks(connName, callback);
    }
@Test
    public void testFenceZombiesInvalidSignature() {
        // Don't have to run the whole gamut of scenarios (invalid signature, missing signature, earlier protocol that doesn't require signatures)
        // since the task config tests cover that pretty well. One sanity check to ensure that this method is guarded should be sufficient.

        when(member.currentProtocolVersion()).thenReturn(CONNECT_PROTOCOL_V2);

        InternalRequestSignature signature = mock(InternalRequestSignature.class);
        when(signature.keyAlgorithm()).thenReturn("HmacSHA256");
        when(signature.isValid(any())).thenReturn(false);

        SessionKey sessionKey = mock(SessionKey.class);
        SecretKey secretKey = mock(SecretKey.class);
        when(sessionKey.key()).thenReturn(secretKey);
        when(sessionKey.creationTimestamp()).thenReturn(time.milliseconds());

        // Read a new session key from the config topic
        configUpdateListener.onSessionKeyUpdate(sessionKey);

        Callback<Void> taskConfigCb = mock(Callback.class);
        herder.fenceZombieSourceTasks(CONN1, taskConfigCb, signature);

        ArgumentCaptor<Throwable> errorCapture = ArgumentCaptor.forClass(Throwable.class);
        verify(taskConfigCb).onCompletion(errorCapture.capture(), isNull());
        assertInstanceOf(ConnectRestException.class, errorCapture.getValue());
        assertEquals(FORBIDDEN.getStatusCode(), ((ConnectRestException) errorCapture.getValue()).statusCode());

        verifyNoMoreInteractions(member);
    }