@Override
    public ExecuteContext after(ExecuteContext context) {
        if (InvokeUtils.isKafkaInvokeBySermant(Thread.currentThread().getStackTrace())) {
            return context;
        }
        KafkaConsumerWrapper kafkaConsumerWrapper = KafkaConsumerController.getKafkaConsumerCache()
                .get(context.getObject().hashCode());
        if (kafkaConsumerWrapper == null) {
            return context;
        }
        updateKafkaConsumerWrapper(kafkaConsumerWrapper);
        if (handler != null) {
            handler.doAfter(context);
        } else {
            LOGGER.info("Try to check if it is need to disable consumption after assignment...");

            // The host application checks whether it is necessary to unsubscribe from the Topic every time it
            // subscribes
            KafkaConsumerController.disableConsumption(kafkaConsumerWrapper,
                    ProhibitionConfigManager.getKafkaProhibitionTopics());
        }
        return context;
    }
@Test
    public void testAfter() {
        ExecuteContext context = ExecuteContext.forMemberMethod(mockConsumer, null, null, null, null);
        interceptor.after(context);
        KafkaConsumerWrapper kafkaConsumerWrapper = KafkaConsumerController.getKafkaConsumerCache()
                .get(mockConsumer.hashCode());
        Assert.assertEquals(topicPartitions, kafkaConsumerWrapper.getOriginalPartitions());
        Assert.assertEquals(Collections.singleton("testTopic-1"), kafkaConsumerWrapper.getOriginalTopics());
        Assert.assertTrue(kafkaConsumerWrapper.isAssign());
    }