KafkaProducer implements Producer<K, V> { @Override public void close() { close(Long.MAX_VALUE, TimeUnit.MILLISECONDS); } KafkaProducer(Map<String, Object> configs);  KafkaProducer(Map<String, Object> configs, Serializer<K> keySerializer, Serializer<V> valueSerializer);  KafkaProducer(Properties properties);  KafkaProducer(Properties properties, Serializer<K> keySerializer, Serializer<V> valueSerializer); @SuppressWarnings({ "unchecked", "deprecation" }) private  KafkaProducer(ProducerConfig config, Serializer<K> keySerializer, Serializer<V> valueSerializer); @Override Future<RecordMetadata> send(ProducerRecord<K, V> record); @Override Future<RecordMetadata> send(ProducerRecord<K, V> record, Callback callback); @Override void flush(); @Override List<PartitionInfo> partitionsFor(String topic); @Override Map<MetricName, ? extends Metric> metrics(); @Override void close(); @Override void close(long timeout, TimeUnit timeUnit);  }
@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()); }
@Test public void testInterceptorConstructClose() throws Exception { try { Properties props = new Properties(); 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 { MockProducerInterceptor.resetCounters(); } }