@Test
public void testInterceptorConstructClose() throws Exception {

    try {
        Properties props = new Properties();
        // test with client ID assigned by KafkaProducer
        props.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999");
        props.setProperty(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, MockProducerInterceptor.class.getName());
        props.setProperty(MockProducerInterceptor.APPEND_STRING_PROP, "something");
        KafkaProducer<String, String> producer = new KafkaProducer<String, String>(
                props, new StringSerializer(), new StringSerializer());
        Assert.assertEquals(1, MockProducerInterceptor.INIT_COUNT.get());
        Assert.assertEquals(0, MockProducerInterceptor.CLOSE_COUNT.get());
        producer.close();
        Assert.assertEquals(1, MockProducerInterceptor.INIT_COUNT.get());
        Assert.assertEquals(1, MockProducerInterceptor.CLOSE_COUNT.get());
    } finally {
        // cleanup since we are using mutable static variables in MockProducerInterceptor
        MockProducerInterceptor.resetCounters();
    }
}