public void delete(boolean drainQueue) {
        addTask(TbQueueConsumerManagerTask.delete(drainQueue));
    }
@Test
    public void testDelete_consumerPerPartition() {
        queue.setConsumerPerPartition(true);
        consumerManager.init(queue);
        Set<TopicPartitionInfo> partitions = createTpis(1, 2);
        consumerManager.update(partitions);
        TestConsumer consumer1 = getConsumer(1);
        TestConsumer consumer2 = getConsumer(2);
        verifySubscribedAndLaunched(consumer1, 1);
        verifySubscribedAndLaunched(consumer2, 2);
        verifyMsgProcessed(consumer1.testMsg);
        verifyMsgProcessed(consumer2.testMsg);

        consumerManager.delete(true);

        await().atMost(2, TimeUnit.SECONDS)
                .untilAsserted(() -> {
                    verify(ruleEngineMsgProducer).send(any(), any(), any());
                });
        clearInvocations(actorContext);
        verify(consumer1, never()).unsubscribe();
        verify(consumer2, never()).unsubscribe();
        int msgCount = totalConsumedMsgs.get();

        await().atLeast(2, TimeUnit.SECONDS) // based on topicDeletionDelayInSec(5) = 5 - ( 3 seconds the code may execute starting consumerManager.delete() call)
                .atMost(20, TimeUnit.SECONDS)
                .untilAsserted(() -> {
                    partitions.stream()
                            .map(TopicPartitionInfo::getFullTopicName)
                            .forEach(topic -> {
                                verify(queueAdmin).deleteTopic(eq(topic));
                            });
                });
        verify(consumer1).unsubscribe();
        verify(consumer2).unsubscribe();

        int totalMovedMsgs = totalConsumedMsgs.get() - msgCount;
        assertThat(totalMovedMsgs).isNotZero();
        verify(ruleEngineMsgProducer, atLeast(totalMovedMsgs)).send(any(), any(), any());
        verify(actorContext, never()).tell(any());
        generateQueueMsgs = false;
    }