@Test
public void testInterceptorConstructorClose() throws Exception {

    try {
        Properties props = new Properties();
        // test with client ID assigned by KafkaConsumer
        props.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999");
        props.setProperty(ConsumerConfig.INTERCEPTOR_CLASSES_CONFIG, MockConsumerInterceptor.class.getName());
        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(
                props, new StringDeserializer(), new StringDeserializer());
        assertEquals(1, MockConsumerInterceptor.INIT_COUNT.get());
        assertEquals(0, MockConsumerInterceptor.CLOSE_COUNT.get());
        consumer.close();
        assertEquals(1, MockConsumerInterceptor.INIT_COUNT.get());
        assertEquals(1, MockConsumerInterceptor.CLOSE_COUNT.get());
    } finally {
        // cleanup since we are using mutable static variables in MockConsumerInterceptor
        MockConsumerInterceptor.resetCounters();
    }
}