killbill-aplcache

diff --git a/beatrix/src/test/java/org/killbill/billing/beatrix/integration/overdue/TestOverdueIntegration.java b/beatrix/src/test/java/org/killbill/billing/beatrix/integration/overdue/TestOverdueIntegration.java
index 41261c8..8beca38 100644
--- a/beatrix/src/test/java/org/killbill/billing/beatrix/integration/overdue/TestOverdueIntegration.java
+++ b/beatrix/src/test/java/org/killbill/billing/beatrix/integration/overdue/TestOverdueIntegration.java
@@ -280,6 +280,10 @@ public class TestOverdueIntegration extends TestOverdueBase {
         addDaysAndCheckForCompletion(5);
         invoiceChecker.checkChargedThroughDate(baseEntitlement.getId(), new LocalDate(2012, 7, 31), callContext);
 
+        // Make sure the 'invoice-service:next-billing-date-queue' gets processed before we continue and since we are in AUTO_INVOICING_OFF
+        // no event (NULL_INVOICE) will be generated and so we can't synchronize on any event, and we need to add a small amount of sleep
+        Thread.sleep(1000);
+
         allowPaymentsAndResetOverdueToClearByPayingAllUnpaidInvoices(true);
 
         invoiceChecker.checkInvoice(account.getId(), 3, callContext,
@@ -946,6 +950,7 @@ public class TestOverdueIntegration extends TestOverdueBase {
     }
 
     private void allowPaymentsAndResetOverdueToClearByPayingAllUnpaidInvoices(final boolean extraPayment) {
+
         // Reset plugin so payments should now succeed
         paymentPlugin.makeAllInvoicesFailWithError(false);
 

diff --git a/beatrix/src/test/java/org/killbill/billing/beatrix/integration/TestIntegrationBase.java b/beatrix/src/test/java/org/killbill/billing/beatrix/integration/TestIntegrationBase.java
index f55940d..a9605bd 100644
--- a/beatrix/src/test/java/org/killbill/billing/beatrix/integration/TestIntegrationBase.java
+++ b/beatrix/src/test/java/org/killbill/billing/beatrix/integration/TestIntegrationBase.java
@@ -245,9 +245,6 @@ public class TestIntegrationBase extends BeatrixTestSuiteWithEmbeddedDB {
     protected RecordIdApi recordIdApi;
 
     @Inject
-    protected IDBI idbi;
-
-    @Inject
     protected NonEntityDao nonEntityDao;
 
     @Inject

diff --git a/pom.xml b/pom.xml
index 217a483..3cb9325 100644
--- a/pom.xml
+++ b/pom.xml
@@ -21,7 +21,7 @@
     <parent>
         <artifactId>killbill-oss-parent</artifactId>
         <groupId>org.kill-bill.billing</groupId>
-        <version>0.91.seventeen-SNAPSHOT</version>
+        <version>0.93.seventeen-SNAPSHOT</version>
     </parent>
     <artifactId>killbill</artifactId>
     <version>0.17.0.seventeen-SNAPSHOT</version>

diff --git a/util/pom.xml b/util/pom.xml
index 66b639c..0789c81 100644
--- a/util/pom.xml
+++ b/util/pom.xml
@@ -70,6 +70,11 @@
             <artifactId>jmustache</artifactId>
         </dependency>
         <dependency>
+            <groupId>com.zaxxer</groupId>
+            <artifactId>HikariCP-java6</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
             <groupId>io.dropwizard.metrics</groupId>
             <artifactId>metrics-core</artifactId>
         </dependency>

diff --git a/util/src/test/java/org/killbill/billing/api/TestApiListener.java b/util/src/test/java/org/killbill/billing/api/TestApiListener.java
index f9c28db..61c1252 100644
--- a/util/src/test/java/org/killbill/billing/api/TestApiListener.java
+++ b/util/src/test/java/org/killbill/billing/api/TestApiListener.java
@@ -42,6 +42,7 @@ import org.killbill.billing.events.PaymentInfoInternalEvent;
 import org.killbill.billing.events.PaymentPluginErrorInternalEvent;
 import org.killbill.billing.events.TagDefinitionInternalEvent;
 import org.killbill.billing.events.TagInternalEvent;
+import org.killbill.clock.Clock;
 import org.skife.jdbi.v2.Handle;
 import org.skife.jdbi.v2.IDBI;
 import org.skife.jdbi.v2.tweak.HandleCallback;
@@ -66,6 +67,7 @@ public class TestApiListener {
 
     private final List<NextEvent> nextExpectedEvent;
     private final IDBI idbi;
+    private final Clock clock;
 
     private boolean isListenerFailed = false;
     private String listenerFailedMsg;
@@ -73,10 +75,11 @@ public class TestApiListener {
     private volatile boolean completed;
 
     @Inject
-    public TestApiListener(final IDBI idbi) {
+    public TestApiListener(final IDBI idbi, final Clock clock) {
         nextExpectedEvent = new Stack<NextEvent>();
         this.completed = false;
         this.idbi = idbi;
+        this.clock = clock;
     }
 
     public void assertListenerStatus() {
@@ -296,16 +299,9 @@ public class TestApiListener {
                         await().atMost(timeout, TimeUnit.MILLISECONDS).until(new Callable<Boolean>() {
                             @Override
                             public Boolean call() throws Exception {
-                                final long inProcessing = idbi.withHandle(new HandleCallback<Long>() {
-                                    @Override
-                                    public Long withHandle(final Handle handle) throws Exception {
-                                        return (Long) handle.select("select count(distinct record_id) count from bus_events").get(0).get("count") +
-                                               // We assume all ready notifications have been picked up
-                                               (Long) handle.select("select count(distinct record_id) count from notifications where processing_state = 'IN_PROCESSING'").get(0).get("count");
-                                    }
-                                });
-                                log.debug("Events still in processing: {}", inProcessing);
-                                return inProcessing == 0;
+                                final long pending = idbi.withHandle(new PendingBusOrNotificationCallback(clock));
+                                log.debug("Events still in processing: {}", pending);
+                                return pending == 0;
                             }
                         });
                         return completed;
@@ -313,19 +309,9 @@ public class TestApiListener {
                     final long after = System.currentTimeMillis();
                     waitTimeMs -= (after - before);
                 } catch (final Exception ignore) {
-                    final StringBuilder errorBuilder = new StringBuilder("isCompleted got interrupted. Exception: ").append(ignore)
-                                                                                                                    .append("\nRemaining bus events:\n");
-                    idbi.withHandle(new HandleCallback<Void>() {
-                        @Override
-                        public Void withHandle(final Handle handle) throws Exception {
-                            final List<Map<String, Object>> busEvents = handle.select("select * from bus_events");
-                            for (final Map<String, Object> busEvent : busEvents) {
-                                errorBuilder.append(busEvent).append("\n");
-                            }
-                            return null;
-                        }
-                    });
-                    log.error(errorBuilder.toString());
+                    // Rerun one more time to provide details
+                    final long pending = idbi.withHandle(new PendingBusOrNotificationCallback(clock));
+                    log.error("isCompleted : Received all events but found remaining unprocessed bus events/notifications =  {}", pending);
                     return false;
                 }
             } while (waitTimeMs > 0 && !completed);
@@ -335,10 +321,24 @@ public class TestApiListener {
             final Joiner joiner = Joiner.on(" ");
             log.error("TestApiListener did not complete in " + timeout + " ms, remaining events are " + joiner.join(nextExpectedEvent));
         }
-
         return completed;
     }
 
+    private static class PendingBusOrNotificationCallback implements HandleCallback<Long> {
+
+        private final Clock clock;
+
+        public PendingBusOrNotificationCallback(final Clock clock) {
+            this.clock = clock;
+        }
+        @Override
+        public Long withHandle(final Handle handle) throws Exception {
+            return (Long) handle.select("select count(distinct record_id) count from bus_events").get(0).get("count") +
+                   (Long) handle.select("select count(distinct record_id) count from notifications where effective_date < ?", clock.getUTCNow().toDate()).get(0).get("count") +
+                   (Long) handle.select("select count(distinct record_id) count from notifications where processing_state = 'IN_PROCESSING'").get(0).get("count");
+        }
+    }
+
     private void notifyIfStackEmpty() {
         log.debug("TestApiListener notifyIfStackEmpty ENTER");
         synchronized (this) {

diff --git a/util/src/test/java/org/killbill/billing/DBTestingHelper.java b/util/src/test/java/org/killbill/billing/DBTestingHelper.java
index e6bb516..7023039 100644
--- a/util/src/test/java/org/killbill/billing/DBTestingHelper.java
+++ b/util/src/test/java/org/killbill/billing/DBTestingHelper.java
@@ -21,12 +21,14 @@ package org.killbill.billing;
 import java.io.IOException;
 import java.net.URL;
 import java.util.Enumeration;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 import org.killbill.billing.platform.test.PlatformDBTestingHelper;
 import org.killbill.billing.util.dao.AuditLogModelDaoMapper;
 import org.killbill.billing.util.dao.RecordIdIdMappingsMapper;
 import org.killbill.billing.util.io.IOUtils;
 import org.killbill.billing.util.security.shiro.dao.SessionModelDao;
+import org.killbill.commons.embeddeddb.EmbeddedDB;
 import org.killbill.commons.jdbi.mapper.LowerToCamelBeanMapperFactory;
 import org.skife.jdbi.v2.DBI;
 import org.skife.jdbi.v2.IDBI;
@@ -37,6 +39,8 @@ public class DBTestingHelper extends PlatformDBTestingHelper {
 
     private static DBTestingHelper dbTestingHelper = null;
 
+    private AtomicBoolean initialized;
+
     public static synchronized DBTestingHelper get() {
         if (dbTestingHelper == null) {
             dbTestingHelper = new DBTestingHelper();
@@ -44,20 +48,26 @@ public class DBTestingHelper extends PlatformDBTestingHelper {
         return dbTestingHelper;
     }
 
-    protected DBTestingHelper() {
+    private DBTestingHelper() {
         super();
+        initialized = new AtomicBoolean(false);
     }
 
     @Override
-    public synchronized IDBI getDBI() throws IOException {
+    public IDBI getDBI() {
         final DBI dbi = (DBI) super.getDBI();
-        dbi.registerMapper(new AuditLogModelDaoMapper());
-        dbi.registerMapper(new RecordIdIdMappingsMapper());
-        dbi.registerMapper(new LowerToCamelBeanMapperFactory(SessionModelDao.class));
+        // Register KB specific mappers
+        if (initialized.compareAndSet(false, true)) {
+            dbi.registerMapper(new AuditLogModelDaoMapper());
+            dbi.registerMapper(new RecordIdIdMappingsMapper());
+            dbi.registerMapper(new LowerToCamelBeanMapperFactory(SessionModelDao.class));
+        }
         return dbi;
     }
 
     protected synchronized void executePostStartupScripts() throws IOException {
+
+        final EmbeddedDB instance = getInstance();
         final String databaseSpecificDDL = "org/killbill/billing/util/" + "ddl-" + instance.getDBEngine().name().toLowerCase() + ".sql";
         installDDLSilently(databaseSpecificDDL);
 
@@ -181,7 +191,7 @@ public class DBTestingHelper extends PlatformDBTestingHelper {
             final String ddl;
             try {
                 ddl = IOUtils.toString(inputStream.openStream());
-                instance.executeScript(ddl);
+                getInstance().executeScript(ddl);
             } catch (final Exception ignored) {
                 // The test doesn't have this module ddl in the classpath - that's fine
             }