Merge pull request #23 from leancloud/fix/close-producer

close producer in integration test

Author: ylgrgyq

src/test/java/cn/leancloud/kafka/consumer/integration/Bootstrap.java

@@ -48,13 +48,14 @@ public static void main(String[] args) throws Exception {
         final List<TestRunner> runners = prepareRunners();
 
         try {
-            for (TestRunner runner : runners) {
-                runner.runTest(new SingleConsumerTest());
-                runner.runTest(new TwoConsumerTest());
-                runner.runTest(new JoinLeaveGroupTest());
+            for (int i = 0; i < runners.size(); i++) {
+                try (TestRunner runner = runners.get(i)){
+                    runner.runTest(new SingleConsumerTest());
+                    runner.runTest(new TwoConsumerTest());
+                    runner.runTest(new JoinLeaveGroupTest());
+                }
             }
         } finally {
-            shutdown(runners);
             workerPool.shutdown();
         }
     }
@@ -108,12 +109,6 @@ private static List<TestRunner> prepareRunners() {
         return runners;
     }
 
-    private static void shutdown(List<TestRunner> runners) throws Exception {
-        for (TestRunner runner : runners) {
-            runner.close();
-        }
-    }
-
     private static class AutoCommitWithWorkerPoolConsumerFactory implements ConsumerFactory {
         @Override
         public String type() {

src/test/java/cn/leancloud/kafka/consumer/integration/TestContext.java

@@ -22,8 +22,4 @@ public TestingProducer producer() {
     public String topic() {
         return topic;
     }
-
-    public TestStatistics newStatistics() {
-        return new TestStatistics();
-    }
 }

src/test/java/cn/leancloud/kafka/consumer/integration/TestRunner.java

@@ -9,22 +9,24 @@
 
 public class TestRunner implements Closeable {
     private static final Logger logger = LoggerFactory.getLogger(TestRunner.class);
+    private static final Duration defaultProducerSendMsgInterval = Duration.ofMillis(100);
+    private static final int defaultConcurrentProducerCount = 2;
 
     private final TestingProducer producer;
     private final ConsumerFactory factory;
     private final TestContext context;
 
     TestRunner(String topic, ConsumerFactory factory) {
         this.factory = factory;
-        this.producer = new TestingProducer(Duration.ofMillis(100), 2);
+        this.producer = new TestingProducer(defaultProducerSendMsgInterval, defaultConcurrentProducerCount);
         this.context = new TestContext(topic, producer, factory);
     }
 
     void runTest(IntegrationTest test) throws Exception {
         logger.info("\n\n\n------------------------------------ Start {} with consumer: {} ------------------------------------\n",
                 test.name(), factory.type());
 
-        final TestStatistics statistics = context.newStatistics();
+        final TestStatistics statistics = new TestStatistics();
         try {
             test.runTest(context, statistics);