public void handleCommandWatchTopicUpdate(CommandWatchTopicUpdate update) {
        patternConsumerUpdateQueue.appendTopicsRemovedOp(update.getDeletedTopicsList());
        patternConsumerUpdateQueue.appendTopicsAddedOp(update.getNewTopicsList());
    }
@Test
    public void testWatcherCallsListenerOnUpdate() {
        ClientCnx clientCnx = mock(ClientCnx.class);
        CompletableFuture<CommandWatchTopicListSuccess> responseFuture = new CompletableFuture<>();
        ArgumentCaptor<BaseCommand> commandCaptor = ArgumentCaptor.forClass(BaseCommand.class);
        when(clientCnx.newWatchTopicList(any(BaseCommand.class), anyLong())).thenReturn(responseFuture);
        when(clientCnx.ctx()).thenReturn(mock(ChannelHandlerContext.class));
        clientCnxFuture.complete(clientCnx);
        verify(clientCnx).newWatchTopicList(commandCaptor.capture(), anyLong());
        CommandWatchTopicListSuccess success = new CommandWatchTopicListSuccess()
                .setWatcherId(7)
                .setRequestId(commandCaptor.getValue().getWatchTopicList().getRequestId())
                .setTopicsHash("FEED");
        success.addTopic("persistent://tenant/ns/topic11");
        responseFuture.complete(success);

        CommandWatchTopicUpdate update = new CommandWatchTopicUpdate()
                .setTopicsHash("F33D")
                .setWatcherId(7)
                .addAllNewTopics(Collections.singleton("persistent://tenant/ns/topic12"));

        watcher.handleCommandWatchTopicUpdate(update);
        verify(listener).onTopicsAdded(Collections.singletonList("persistent://tenant/ns/topic12"));
    }