@Subscribe
    public void inputUpdated(InputUpdated inputUpdatedEvent) {
        final String inputId = inputUpdatedEvent.id();
        LOG.debug("Input updated: {}", inputId);
        final Input input;
        try {
            input = inputService.find(inputId);
        } catch (NotFoundException e) {
            LOG.warn("Received InputUpdated event but could not find input {}", inputId, e);
            return;
        }

        final boolean startInput;
        final IOState<MessageInput> inputState = inputRegistry.getInputState(inputId);
        if (inputState != null) {
            startInput = inputState.getState() == IOState.Type.RUNNING;
            inputRegistry.remove(inputState);
        } else {
            startInput = false;
        }

        if (startInput && (input.isGlobal() || this.nodeId.getNodeId().equals(input.getNodeId()))) {
            startInput(input);
        }
    }
@Test
    public void inputUpdatedRestartsLocalInputOnLocalNode() throws Exception {
        final String inputId = "input-id";
        final Input input = mock(Input.class);
        @SuppressWarnings("unchecked")
        final IOState<MessageInput> inputState = mock(IOState.class);
        when(inputState.getState()).thenReturn(IOState.Type.RUNNING);
        when(inputService.find(inputId)).thenReturn(input);
        when(inputRegistry.getInputState(inputId)).thenReturn(inputState);
        when(input.getNodeId()).thenReturn(THIS_NODE_ID);
        when(input.isGlobal()).thenReturn(false);

        final MessageInput messageInput = mock(MessageInput.class);
        when(inputService.getMessageInput(input)).thenReturn(messageInput);

        listener.inputUpdated(InputUpdated.create(inputId));

        verify(inputLauncher, times(1)).launch(messageInput);
    }