@Test
public void testSerializerClose() throws Exception {

    Map<String, Object> configs = new HashMap<>();
    configs.put(ProducerConfig.CLIENT_ID_CONFIG, "testConstructorClose");
    configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999");
    configs.put(ProducerConfig.METRIC_REPORTER_CLASSES_CONFIG, MockMetricsReporter.class.getName());
    configs.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, CommonClientConfigs.DEFAULT_SECURITY_PROTOCOL);
    final int oldInitCount = MockSerializer.INIT_COUNT.get();
    final int oldCloseCount = MockSerializer.CLOSE_COUNT.get();
    KafkaProducer<byte[], byte[]> producer = new KafkaProducer<byte[], byte[]>(
            configs, new MockSerializer(), new MockSerializer());
    Assert.assertEquals(oldInitCount + 2, MockSerializer.INIT_COUNT.get());
    Assert.assertEquals(oldCloseCount, MockSerializer.CLOSE_COUNT.get());
    producer.close();
    Assert.assertEquals(oldInitCount + 2, MockSerializer.INIT_COUNT.get());
    Assert.assertEquals(oldCloseCount + 2, MockSerializer.CLOSE_COUNT.get());
}