@Override
    public void putTaskConfigs(final String connName, final List<Map<String, String>> configs, final Callback<Void> callback, InternalRequestSignature requestSignature) {
        log.trace("Submitting put task configuration request {}", connName);
        if (requestNotSignedProperly(requestSignature, callback)) {
            return;
        }

        addRequest(
            () -> {
                if (!isLeader())
                    callback.onCompletion(new NotLeaderException("Only the leader may write task configurations.", leaderUrl()), null);
                else if (!configState.contains(connName))
                    callback.onCompletion(new NotFoundException("Connector " + connName + " not found"), null);
                else {
                    writeTaskConfigs(connName, configs);
                    callback.onCompletion(null, null);
                }
                return null;
            },
            forwardErrorAndTickThreadStages(callback)
        );
    }
@Test
    public void testPutTaskConfigsInvalidSignature() {
        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.putTaskConfigs(CONN1, TASK_CONFIGS, 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, taskConfigCb);
    }