payment: fix potential DB leak in CompletionTaskBase

This fixes #757.

Signed-off-by: Pierre-Alexandre Meyer <pierre@mouraf.org>

payment/src/main/java/org/killbill/billing/payment/core/janitor/CompletionTaskBase.java

@@ -18,6 +18,7 @@
 package org.killbill.billing.payment.core.janitor;
 
 import java.io.IOException;
+import java.util.Iterator;
 
 import org.killbill.billing.account.api.AccountApiException;
 import org.killbill.billing.account.api.AccountInternalApi;
@@ -86,16 +87,25 @@ public void run() {
             log.info("Janitor was requested to stop");
             return;
         }
-        final Iterable<T> items = getItemsForIteration();
-        for (final T item : items) {
-            if (isStopped) {
-                log.info("Janitor was requested to stop");
-                return;
+
+        final Iterator<T> iterator = getItemsForIteration().iterator();
+        try {
+            while (iterator.hasNext()) {
+                final T item = iterator.next();
+                if (isStopped) {
+                    log.info("Janitor was requested to stop");
+                    return;
+                }
+                try {
+                    doIteration(item);
+                } catch (final Exception e) {
+                    log.warn(e.getMessage());
+                }
             }
-            try {
-                doIteration(item);
-            } catch (final IllegalStateException e) {
-                log.warn(e.getMessage());
+        } finally {
+            // In case the loop stops early, make sure to close the underlying DB connection
+            while (iterator.hasNext()) {
+                iterator.next();
             }
         }
     }

payment/src/test/java/org/killbill/billing/payment/PaymentTestSuiteWithEmbeddedDB.java

@@ -33,7 +33,9 @@
 import org.killbill.billing.payment.core.PaymentProcessor;
 import org.killbill.billing.payment.core.janitor.IncompletePaymentTransactionTask;
 import org.killbill.billing.payment.core.janitor.Janitor;
+import org.killbill.billing.payment.core.sm.PaymentControlStateMachineHelper;
 import org.killbill.billing.payment.core.sm.PaymentStateMachineHelper;
+import org.killbill.billing.payment.core.sm.PluginControlPaymentAutomatonRunner;
 import org.killbill.billing.payment.dao.PaymentDao;
 import org.killbill.billing.payment.glue.PaymentModule;
 import org.killbill.billing.payment.glue.TestPaymentModuleWithEmbeddedDB;
@@ -103,6 +105,10 @@ public abstract class PaymentTestSuiteWithEmbeddedDB extends GuicyKillbillTestSu
     protected IncompletePaymentTransactionTask incompletePaymentTransactionTask;
     @Inject
     protected GlobalLocker locker;
+    @Inject
+    protected PluginControlPaymentAutomatonRunner pluginControlPaymentAutomatonRunner;
+    @Inject
+    protected PaymentControlStateMachineHelper paymentControlStateMachineHelper;
 
     @Override
     protected KillbillConfigSource getConfigSource() {

payment/src/test/java/org/killbill/billing/payment/core/janitor/TestCompletionTaskBase.java

@@ -0,0 +1,91 @@
+/*
+ * Copyright 2014-2017 Groupon, Inc
+ * Copyright 2014-2017 The Billing Project, LLC
+ *
+ * The Billing Project licenses this file to you under the Apache License, version 2.0
+ * (the "License"); you may not use this file except in compliance with the
+ * License.  You may obtain a copy of the License at:
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.  See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.killbill.billing.payment.core.janitor;
+
+import java.util.Iterator;
+import java.util.List;
+
+import org.killbill.billing.account.api.AccountInternalApi;
+import org.killbill.billing.payment.PaymentTestSuiteWithEmbeddedDB;
+import org.killbill.billing.payment.api.PaymentApiException;
+import org.killbill.billing.payment.core.sm.PaymentControlStateMachineHelper;
+import org.killbill.billing.payment.core.sm.PaymentStateMachineHelper;
+import org.killbill.billing.payment.core.sm.PluginControlPaymentAutomatonRunner;
+import org.killbill.billing.payment.dao.PaymentAttemptModelDao;
+import org.killbill.billing.payment.dao.PaymentDao;
+import org.killbill.billing.util.callcontext.InternalCallContextFactory;
+import org.killbill.billing.util.config.definition.PaymentConfig;
+import org.killbill.clock.Clock;
+import org.killbill.commons.locker.GlobalLocker;
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+import com.google.common.collect.ImmutableList;
+
+public class TestCompletionTaskBase extends PaymentTestSuiteWithEmbeddedDB {
+
+    @Test(groups = "slow", description = "https://github.com/killbill/killbill/issues/757")
+    public void testHandleRuntimeExceptions() throws PaymentApiException {
+        final List<PaymentAttemptModelDao> paymentAttemptModelDaos = ImmutableList.<PaymentAttemptModelDao>of(new PaymentAttemptModelDao(),
+                                                                                                              new PaymentAttemptModelDao());
+        final Iterator<PaymentAttemptModelDao> paymentAttemptModelDaoIterator = paymentAttemptModelDaos.iterator();
+        final Iterable<PaymentAttemptModelDao> itemsForIteration = new Iterable<PaymentAttemptModelDao>() {
+            @Override
+            public Iterator<PaymentAttemptModelDao> iterator() {
+                return paymentAttemptModelDaoIterator;
+            }
+        };
+        Assert.assertTrue(paymentAttemptModelDaoIterator.hasNext());
+
+        final Runnable incompletePaymentAttemptTaskWithException = new IncompletePaymentAttemptTaskWithException(itemsForIteration,
+                                                                                                                 internalCallContextFactory,
+                                                                                                                 paymentConfig,
+                                                                                                                 paymentDao,
+                                                                                                                 clock,
+                                                                                                                 paymentSMHelper,
+                                                                                                                 paymentControlStateMachineHelper,
+                                                                                                                 accountApi,
+                                                                                                                 pluginControlPaymentAutomatonRunner,
+                                                                                                                 locker);
+
+        incompletePaymentAttemptTaskWithException.run();
+
+        // Make sure we cycled through all entries
+        Assert.assertFalse(paymentAttemptModelDaoIterator.hasNext());
+    }
+
+    private final class IncompletePaymentAttemptTaskWithException extends IncompletePaymentAttemptTask {
+
+        private final Iterable<PaymentAttemptModelDao> itemsForIteration;
+
+        public IncompletePaymentAttemptTaskWithException(final Iterable<PaymentAttemptModelDao> itemsForIteration, final InternalCallContextFactory internalCallContextFactory, final PaymentConfig paymentConfig, final PaymentDao paymentDao, final Clock clock, final PaymentStateMachineHelper paymentStateMachineHelper, final PaymentControlStateMachineHelper retrySMHelper, final AccountInternalApi accountInternalApi, final PluginControlPaymentAutomatonRunner pluginControlledPaymentAutomatonRunner, final GlobalLocker locker) {
+            super(internalCallContextFactory, paymentConfig, paymentDao, clock, paymentStateMachineHelper, retrySMHelper, accountInternalApi, pluginControlledPaymentAutomatonRunner, locker);
+            this.itemsForIteration = itemsForIteration;
+        }
+
+        @Override
+        public Iterable<PaymentAttemptModelDao> getItemsForIteration() {
+            return itemsForIteration;
+        }
+
+        @Override
+        public void doIteration(final PaymentAttemptModelDao attempt) {
+            throw new NullPointerException("NPE for tests");
+        }
+    }
+}