public static void disableConsumption(KafkaConsumerWrapper kafkaConsumerWrapper, Set<String> prohibitionTopics) {
        Set<String> originalTopics = kafkaConsumerWrapper.getOriginalTopics();

        // Not subscribed to any Topic, so no action is required
        if (originalTopics.size() == 0) {
            return;
        }
        Collection<TopicPartition> originalPartitions = kafkaConsumerWrapper.getOriginalPartitions();
        KafkaConsumer<?, ?> kafkaConsumer = kafkaConsumerWrapper.getKafkaConsumer();
        Collection<String> subtractTopics = CollectionUtils.subtract(originalTopics, prohibitionTopics);
        if (kafkaConsumerWrapper.isAssign()) {
            kafkaConsumer.assign(originalPartitions.stream().filter(obj -> subtractTopics.contains(obj.topic()))
                    .collect(Collectors.toSet()));
            return;
        }
        kafkaConsumer.subscribe(subtractTopics);
    }
@Test
    public void testDisableConsumptionByAssign() {
        KafkaConsumer<?, ?> mockConsumer = Mockito.mock(KafkaConsumer.class);
        KafkaConsumerWrapper kafkaConsumerWrapper = new KafkaConsumerWrapper(mockConsumer);
        HashSet<String> originalTopics = new HashSet<>();
        originalTopics.add("testTopic-1");
        originalTopics.add("testTopic-2");
        HashSet<TopicPartition> originalPartitions = new HashSet<>();
        TopicPartition topicPartition1 = new TopicPartition("testTopic-1", 0);
        TopicPartition topicPartition2 = new TopicPartition("testTopic-1", 1);
        TopicPartition topicPartition3 = new TopicPartition("testTopic-2", 0);
        originalPartitions.add(topicPartition1);
        originalPartitions.add(topicPartition2);
        originalPartitions.add(topicPartition3);
        kafkaConsumerWrapper.setOriginalTopics(originalTopics);
        kafkaConsumerWrapper.setOriginalPartitions(originalPartitions);
        kafkaConsumerWrapper.setAssign(true);
        Set<String> prohibitionTopics = new HashSet<>();
        prohibitionTopics.add("testTopic-2");
        prohibitionTopics.add("testTopic-3");
        KafkaConsumerController.disableConsumption(kafkaConsumerWrapper, prohibitionTopics);
        originalPartitions.remove(topicPartition3);
        Mockito.verify(mockConsumer, Mockito.times(1)).assign(originalPartitions);
    }