killbill-aplcache

diff --git a/account/src/main/java/com/ning/billing/account/api/user/DefaultAccountChangeEvent.java b/account/src/main/java/com/ning/billing/account/api/user/DefaultAccountChangeEvent.java
index 59bcb78..d6b4351 100644
--- a/account/src/main/java/com/ning/billing/account/api/user/DefaultAccountChangeEvent.java
+++ b/account/src/main/java/com/ning/billing/account/api/user/DefaultAccountChangeEvent.java
@@ -187,17 +187,4 @@ public class DefaultAccountChangeEvent extends DefaultBusInternalEvent implement
             inputList.add(new DefaultChangedField(key, oldData, newData));
         }
     }
-
-    @Override
-    public Long getTenantRecordId() {
-        // TODO Auto-generated method stub
-        return null;
-    }
-
-    @Override
-    public Long getAccountRecordId() {
-        // TODO Auto-generated method stub
-        return null;
-    }
-
 }

diff --git a/analytics/src/main/java/com/ning/billing/analytics/AnalyticsListener.java b/analytics/src/main/java/com/ning/billing/analytics/AnalyticsListener.java
index 5c29bbf..b0e1e1d 100644
--- a/analytics/src/main/java/com/ning/billing/analytics/AnalyticsListener.java
+++ b/analytics/src/main/java/com/ning/billing/analytics/AnalyticsListener.java
@@ -188,6 +188,6 @@ public class AnalyticsListener {
     }
 
     private InternalCallContext createCallContext(final BusInternalEvent event) {
-        return internalCallContextFactory.createInternalCallContext("AnalyticsService", CallOrigin.INTERNAL, UserType.SYSTEM, event.getUserToken());
+        return internalCallContextFactory.createInternalCallContext(event.getTenantRecordId(), event.getAccountRecordId(), "AnalyticsService", CallOrigin.INTERNAL, UserType.SYSTEM, event.getUserToken());
     }
 }

diff --git a/overdue/src/main/java/com/ning/billing/overdue/applicator/OverdueStateApplicator.java b/overdue/src/main/java/com/ning/billing/overdue/applicator/OverdueStateApplicator.java
index da3ff18..e10fb5e 100644
--- a/overdue/src/main/java/com/ning/billing/overdue/applicator/OverdueStateApplicator.java
+++ b/overdue/src/main/java/com/ning/billing/overdue/applicator/OverdueStateApplicator.java
@@ -194,6 +194,7 @@ public class OverdueStateApplicator<T extends Blockable> {
             final List<Subscription> toBeCancelled = new LinkedList<Subscription>();
             computeSubscriptionsToCancel(blockable, toBeCancelled, context);
             for (final Subscription cur : toBeCancelled) {
+                // STEPH Need conversion toCallContext because we are calling a public API through the Subscription object
                 cur.cancelWithPolicy(clock.getUTCNow(), actionPolicy, context.toCallContext());
             }
         } catch (EntitlementUserApiException e) {

diff --git a/overdue/src/main/java/com/ning/billing/overdue/listener/OverdueListener.java b/overdue/src/main/java/com/ning/billing/overdue/listener/OverdueListener.java
index 3e2b8d5..795e5c7 100644
--- a/overdue/src/main/java/com/ning/billing/overdue/listener/OverdueListener.java
+++ b/overdue/src/main/java/com/ning/billing/overdue/listener/OverdueListener.java
@@ -77,7 +77,7 @@ public class OverdueListener {
     }
 
     private InternalCallContext createCallContext(@Nullable final BusInternalEvent event) {
-        return internalCallContextFactory.createInternalCallContext("OverdueService", CallOrigin.INTERNAL, UserType.SYSTEM,
+        return internalCallContextFactory.createInternalCallContext(event.getTenantRecordId(), event.getAccountRecordId(), "OverdueService", CallOrigin.INTERNAL, UserType.SYSTEM,
                                                                     event == null ? null : event.getUserToken());
     }
 }

diff --git a/payment/src/main/java/com/ning/billing/payment/core/RefundProcessor.java b/payment/src/main/java/com/ning/billing/payment/core/RefundProcessor.java
index 2054f0e..c37b6ed 100644
--- a/payment/src/main/java/com/ning/billing/payment/core/RefundProcessor.java
+++ b/payment/src/main/java/com/ning/billing/payment/core/RefundProcessor.java
@@ -16,6 +16,8 @@
 
 package com.ning.billing.payment.core;
 
+import static com.ning.billing.payment.glue.PaymentModule.PLUGIN_EXECUTOR_NAMED;
+
 import java.math.BigDecimal;
 import java.util.ArrayList;
 import java.util.Collection;
@@ -64,8 +66,6 @@ import com.google.common.collect.Collections2;
 import com.google.common.collect.ImmutableMap;
 import com.google.inject.name.Named;
 
-import static com.ning.billing.payment.glue.PaymentModule.PLUGIN_EXECUTOR_NAMED;
-
 public class RefundProcessor extends ProcessorBase {
 
     private static final Logger log = LoggerFactory.getLogger(RefundProcessor.class);
@@ -234,7 +234,7 @@ public class RefundProcessor extends ProcessorBase {
             throw new PaymentApiException(ErrorCode.PAYMENT_NO_SUCH_REFUND, refundId);
         }
 
-        if (completePluginCompletedRefund(filteredInput)) {
+        if (completePluginCompletedRefund(filteredInput, context)) {
             result = paymentDao.getRefund(refundId, context);
         }
         return new DefaultRefund(result.getId(), result.getCreatedDate(), result.getUpdatedDate(),
@@ -245,7 +245,7 @@ public class RefundProcessor extends ProcessorBase {
     public List<Refund> getAccountRefunds(final Account account, final InternalTenantContext context)
             throws PaymentApiException {
         List<RefundModelDao> result = paymentDao.getRefundsForAccount(account.getId(), context);
-        if (completePluginCompletedRefund(result)) {
+        if (completePluginCompletedRefund(result, context)) {
             result = paymentDao.getRefundsForAccount(account.getId(), context);
         }
         final List<RefundModelDao> filteredInput = filterUncompletedPluginRefund(result);
@@ -255,7 +255,7 @@ public class RefundProcessor extends ProcessorBase {
     public List<Refund> getPaymentRefunds(final UUID paymentId, final InternalTenantContext context)
             throws PaymentApiException {
         List<RefundModelDao> result = paymentDao.getRefundsForPayment(paymentId, context);
-        if (completePluginCompletedRefund(result)) {
+        if (completePluginCompletedRefund(result, context)) {
             result = paymentDao.getRefundsForPayment(paymentId, context);
         }
         final List<RefundModelDao> filteredInput = filterUncompletedPluginRefund(result);
@@ -282,7 +282,7 @@ public class RefundProcessor extends ProcessorBase {
         }));
     }
 
-    private boolean completePluginCompletedRefund(final List<RefundModelDao> refunds) throws PaymentApiException {
+    private boolean completePluginCompletedRefund(final List<RefundModelDao> refunds, final InternalTenantContext tenantContext) throws PaymentApiException {
 
         final Collection<RefundModelDao> refundsToBeFixed = Collections2.filter(refunds, new Predicate<RefundModelDao>() {
             @Override
@@ -295,7 +295,7 @@ public class RefundProcessor extends ProcessorBase {
         }
 
         try {
-            final InternalCallContext context = internalCallContextFactory.createInternalCallContext("RefundProcessor", CallOrigin.INTERNAL, UserType.SYSTEM, null);
+            final InternalCallContext context = internalCallContextFactory.createInternalCallContext(tenantContext.getTenantRecordId(), tenantContext.getAccountRecordId(), "RefundProcessor", CallOrigin.INTERNAL, UserType.SYSTEM, null);
             final Account account = accountInternalApi.getAccountById(refundsToBeFixed.iterator().next().getAccountId(), context);
             new WithAccountLock<Void>().processAccountWithLock(locker, account.getExternalKey(), new WithAccountLockCallback<Void>() {
 

diff --git a/usage/src/main/java/com/ning/billing/usage/timeline/aggregator/TimelineAggregator.java b/usage/src/main/java/com/ning/billing/usage/timeline/aggregator/TimelineAggregator.java
index 6d272aa..ec2af28 100644
--- a/usage/src/main/java/com/ning/billing/usage/timeline/aggregator/TimelineAggregator.java
+++ b/usage/src/main/java/com/ning/billing/usage/timeline/aggregator/TimelineAggregator.java
@@ -71,7 +71,7 @@ public class TimelineAggregator {
 
     private final ScheduledExecutorService aggregatorThread = Executors.newSingleThreadScheduledExecutor();
 
-    private Map<String, AtomicLong> aggregatorCounters = new LinkedHashMap<String, AtomicLong>();
+    private final Map<String, AtomicLong> aggregatorCounters = new LinkedHashMap<String, AtomicLong>();
 
     private final AtomicBoolean isAggregating = new AtomicBoolean(false);
 
@@ -426,6 +426,7 @@ public class TimelineAggregator {
     }
 
     private InternalCallContext createCallContext() {
-        return internalCallContextFactory.createInternalCallContext("TimelineAggregator", CallOrigin.INTERNAL, UserType.SYSTEM, null);
+        // TODO add teantRecordId and accountRecordId
+        return internalCallContextFactory.createInternalCallContext(InternalCallContextFactory.INTERNAL_TENANT_RECORD_ID, null, "TimelineAggregator", CallOrigin.INTERNAL, UserType.SYSTEM, null);
     }
 }

diff --git a/usage/src/main/java/com/ning/billing/usage/timeline/BackgroundDBChunkWriter.java b/usage/src/main/java/com/ning/billing/usage/timeline/BackgroundDBChunkWriter.java
index 3c4a786..7a12fa5 100644
--- a/usage/src/main/java/com/ning/billing/usage/timeline/BackgroundDBChunkWriter.java
+++ b/usage/src/main/java/com/ning/billing/usage/timeline/BackgroundDBChunkWriter.java
@@ -67,7 +67,7 @@ public class BackgroundDBChunkWriter {
     private final AtomicBoolean shuttingDown = new AtomicBoolean();
     private List<PendingChunkMap> pendingChunks = new ArrayList<PendingChunkMap>();
     private DateTime lastWriteTime = new DateTime();
-    private AtomicBoolean doingWritesNow = new AtomicBoolean();
+    private final AtomicBoolean doingWritesNow = new AtomicBoolean();
     private final ScheduledExecutorService backgroundWriteThread = Executors.newSingleThreadScheduledExecutor();
 
     private final AtomicLong maybePerformBackgroundWritesCount = new AtomicLong();
@@ -223,6 +223,7 @@ public class BackgroundDBChunkWriter {
     }
 
     private InternalCallContext createCallContext() {
-        return internalCallContextFactory.createInternalCallContext("ChunkWriter", CallOrigin.INTERNAL, UserType.SYSTEM, null);
+        // TODO information about accountRecordId and tenatRecordId
+        return internalCallContextFactory.createInternalCallContext(InternalCallContextFactory.INTERNAL_TENANT_RECORD_ID, null, "ChunkWriter", CallOrigin.INTERNAL, UserType.SYSTEM, null);
     }
 }

diff --git a/util/src/main/java/com/ning/billing/util/bus/PersistentBus.java b/util/src/main/java/com/ning/billing/util/bus/PersistentBus.java
index 44b0110..eab96ab 100644
--- a/util/src/main/java/com/ning/billing/util/bus/PersistentBus.java
+++ b/util/src/main/java/com/ning/billing/util/bus/PersistentBus.java
@@ -1,4 +1,4 @@
-/* 
+/*
  * Copyright 2010-2011 Ning, Inc.
  *
  * Ning licenses this file to you under the Apache License, version 2.0
@@ -108,9 +108,9 @@ public class PersistentBus extends PersistentQueueBase implements Bus {
 
     @Override
     public int doProcessEvents() {
-        // Note: retrieving and clearing bus events is not done per tenant (yet?)
-        final InternalCallContext context = internalCallContextFactory.createInternalCallContext("PersistentBus", CallOrigin.INTERNAL, UserType.SYSTEM, null);
 
+        // TODO Retrieving and clearing bus events is not done per tenant so pass default INTERNAL_TENANT_RECORD_ID; not sure this is something we want to do anyway ?
+        final InternalCallContext context = internalCallContextFactory.createInternalCallContext(InternalCallContextFactory.INTERNAL_TENANT_RECORD_ID, null, "PersistentBus", CallOrigin.INTERNAL, UserType.SYSTEM, null);
         final List<BusEventEntry> events = getNextBusEvent(context);
         if (events.size() == 0) {
             return 0;
@@ -118,11 +118,13 @@ public class PersistentBus extends PersistentQueueBase implements Bus {
 
         int result = 0;
         for (final BusEventEntry cur : events) {
-            final BusInternalEvent evt = deserializeEvent(cur.getBusEventClass(), cur.getBusEventJson());
+            final String jsonWithAccountAndTenantRecorId = tweakJsonToIncludeAccountAndTenantRecordId(cur.getBusEventJson(), cur.getAccountRecordId(), cur.getTenantRecordId());
+            final BusInternalEvent evt = deserializeEvent(cur.getBusEventClass(), jsonWithAccountAndTenantRecorId);
             result++;
             // STEPH exception handling is done by GUAVA-- logged a bug Issue-780
             eventBusDelegate.post(evt);
-            dao.clearBusEvent(cur.getId(), hostname, context);
+            final InternalCallContext rehydratedContext = internalCallContextFactory.createInternalCallContext(cur.getTenantRecordId(), cur.getAccountRecordId(), context);
+            dao.clearBusEvent(cur.getId(), hostname, rehydratedContext);
         }
         return result;
     }
@@ -184,4 +186,17 @@ public class PersistentBus extends PersistentQueueBase implements Bus {
             log.error("Failed to post BusEvent " + event, e);
         }
     }
+
+
+    private String tweakJsonToIncludeAccountAndTenantRecordId(final String input, final Long accountRecordId, final Long tenantRecordId) {
+        int lastIndexPriorFinalBracket = input.lastIndexOf("}");
+        final StringBuilder tmp = new StringBuilder(input.substring(0, lastIndexPriorFinalBracket));
+        tmp.append(",\"accountRecordId\":");
+        tmp.append(accountRecordId);
+        tmp.append(",\"tenantRecordId\":");
+        tmp.append(tenantRecordId);
+        tmp.append("}");
+        return tmp.toString();
+    }
+
 }

diff --git a/util/src/main/java/com/ning/billing/util/callcontext/InternalCallContext.java b/util/src/main/java/com/ning/billing/util/callcontext/InternalCallContext.java
index 1496065..9f6da10 100644
--- a/util/src/main/java/com/ning/billing/util/callcontext/InternalCallContext.java
+++ b/util/src/main/java/com/ning/billing/util/callcontext/InternalCallContext.java
@@ -57,12 +57,13 @@ public class InternalCallContext extends InternalTenantContext {
     }
 
 
-/*
+    // TODO should not be needed if all services are using internal API
+    // Unfortunately not true as some APIs ae hidden in object-- e.g OverdueStateApplicator is doing subscription.cancelWithPolicy(polciy, context);
+    //
     public CallContext toCallContext() {
-        // TODO - can't go back to CallContext (need to extract internal APIs first)
         return new DefaultCallContext(null, userName, callOrigin, userType, reasonCode, comment, userToken, createdDate, updatedDate);
     }
-*/
+
     public UUID getUserToken() {
         return userToken;
     }

diff --git a/util/src/main/java/com/ning/billing/util/callcontext/InternalCallContextFactory.java b/util/src/main/java/com/ning/billing/util/callcontext/InternalCallContextFactory.java
index a9dd53f..950f1e3 100644
--- a/util/src/main/java/com/ning/billing/util/callcontext/InternalCallContextFactory.java
+++ b/util/src/main/java/com/ning/billing/util/callcontext/InternalCallContextFactory.java
@@ -283,10 +283,12 @@ public class InternalCallContextFactory {
         }
         return tenantRecordId;
     }
-/*
-    // TODO - remove
-    public InternalCallContext createInternalCallContext(final String userName, final CallOrigin callOrigin, final UserType userType, final UUID userToken) {
+
+    /*
+    // Needed by PersistentBus to retrieve events
+    public InternalCallContext createInternalCallContextX(final String userName, final CallOrigin callOrigin, final UserType userType, final UUID userToken) {
         return createInternalCallContext(INTERNAL_TENANT_RECORD_ID, null, userName, callOrigin, userType, userToken);
     }
     */
+
 }

diff --git a/util/src/main/java/com/ning/billing/util/events/DefaultBusInternalEvent.java b/util/src/main/java/com/ning/billing/util/events/DefaultBusInternalEvent.java
index 3e6980b..ad450fb 100644
--- a/util/src/main/java/com/ning/billing/util/events/DefaultBusInternalEvent.java
+++ b/util/src/main/java/com/ning/billing/util/events/DefaultBusInternalEvent.java
@@ -17,6 +17,8 @@ package com.ning.billing.util.events;
 
 import java.util.UUID;
 
+import com.fasterxml.jackson.annotation.JsonIgnore;
+
 public abstract class DefaultBusInternalEvent implements BusInternalEvent {
 
     private final UUID userToken;
@@ -34,11 +36,13 @@ public abstract class DefaultBusInternalEvent implements BusInternalEvent {
         return userToken;
     }
 
+    @JsonIgnore
     @Override
     public Long getTenantRecordId() {
         return tenantRecordId;
     }
 
+    @JsonIgnore
     @Override
     public Long getAccountRecordId() {
         return accountRecordId;