[fix][test] Close KafkaProducer in integration tests to prevent thread leak
@@ -118,14 +118,14 @@ public class KafkaSourceTester extends SourceTester<KafkaContainer> {
@Override
public Map<String, String> produceSourceMessages(int numMessages) throws Exception{
- KafkaProducer<String, String> producer = new KafkaProducer<>(
+ try (KafkaProducer<String, String> producer = new KafkaProducer<>(
ImmutableMap.of(
ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaContainer.getBootstrapServers(),
ProducerConfig.CLIENT_ID_CONFIG, UUID.randomUUID().toString()
),
new StringSerializer(),
new StringSerializer()
- );
+ )) {
LinkedHashMap<String, String> kvs = new LinkedHashMap<>();
for (int i = 0; i < numMessages; i++) {
String key = "key-" + i;
@@ -142,6 +142,7 @@ public class KafkaSourceTester extends SourceTester<KafkaContainer> {
log.info("Successfully produced {} messages to kafka topic {}", numMessages, kafkaTopicName);
return kvs;
}
+ }
@Override
public void close() throws Exception {
