Test: Use try with resources for the producer

mongodb · Feb 28, 2020 · c77614d · c77614d
1 parent 980ae67
commit c77614d
Showing 1 changed file with 17 additions and 15 deletions.
diff --git a/src/integrationTest/java/com/mongodb/kafka/connect/MongoSinkConnectorTest.java b/src/integrationTest/java/com/mongodb/kafka/connect/MongoSinkConnectorTest.java
@@ -52,16 +52,17 @@ void testASimpleProducerSmokeTest() {
         props.put(ProducerConfig.ACKS_CONFIG, "all");
         props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true");
 
-        KafkaProducer<Integer, String> producer = new KafkaProducer<>(props);
-        producer.initTransactions();
-        producer.beginTransaction();
+        try(KafkaProducer<Integer, String> producer = new KafkaProducer<>(props)) {
+            producer.initTransactions();
+            producer.beginTransaction();
 
-        IntStream.range(0, 10).forEach(i -> {
-            producer.send(new ProducerRecord<>(topicName, i, "Hello, World!"));
-        });
-        producer.commitTransaction();
+            IntStream.range(0, 10).forEach(i -> {
+                producer.send(new ProducerRecord<>(topicName, i, "Hello, World!"));
+            });
+            producer.commitTransaction();
 
-        assertProduced(10, topicName);
+            assertProduced(10, topicName);
+        }
     }
 
     @Test
@@ -84,14 +85,15 @@ void testSinkSavesAvroDataToMongoDB() {
         producerProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "io.confluent.kafka.serializers.KafkaAvroSerializer");
         producerProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "io.confluent.kafka.serializers.KafkaAvroSerializer");
         producerProps.put(KafkaAvroSerializerConfig.SCHEMA_REGISTRY_URL_CONFIG, KAFKA.schemaRegistryUrl());
-        KafkaProducer<String, TweetMsg> producer = new KafkaProducer<>(producerProps);
 
-        producer.initTransactions();
-        producer.beginTransaction();
-        tweets.forEach(tweet -> producer.send(new ProducerRecord<>(topicName, tweet)));
-        producer.commitTransaction();
+        try(KafkaProducer<String, TweetMsg> producer = new KafkaProducer<>(producerProps)) {
+            producer.initTransactions();
+            producer.beginTransaction();
+            tweets.forEach(tweet -> producer.send(new ProducerRecord<>(topicName, tweet)));
+            producer.commitTransaction();
 
-        assertProduced(100, topicName);
-        assertEquals(100, getCollection().countDocuments());
+            assertProduced(100, topicName);
+            assertEquals(100, getCollection().countDocuments());
+        }
     }
 }