Fix memory leak where TimerReference was not getting cleared on lost-…

…race request-cache insert

hystrix-core/src/main/java/com/netflix/hystrix/AbstractCommand.java

@@ -54,7 +54,6 @@
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
-import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.AtomicReference;
 
 /* package */abstract class AbstractCommand<R> implements HystrixInvokableInfo<R>, HystrixObservable<R> {
@@ -95,7 +94,6 @@ protected static enum TimedOutStatus {
     protected final AtomicReference<Reference<TimerListener>> timeoutTimer = new AtomicReference<Reference<TimerListener>>();
 
     protected final AtomicBoolean commandStarted = new AtomicBoolean();
-    protected volatile boolean executionStarted = false;
     protected volatile boolean isExecutionComplete = false;
 
     /*
@@ -361,8 +359,7 @@ public Observable<R> toObservable() {
             throw new IllegalStateException("This instance can only be executed once. Please instantiate a new instance.");
         }
 
-        final long startTimestamp = System.currentTimeMillis();
-        commandStartTimestamp = startTimestamp;
+        commandStartTimestamp = System.currentTimeMillis();
 
         if (properties.requestLogEnabled().get()) {
             // log this command execution regardless of what happened
@@ -449,21 +446,7 @@ public Observable<R> call() {
             .doOnUnsubscribe(unsubscribeCommandCleanup); // perform cleanup once
     }
 
-    private void handleCommandEnd() {
-        Reference<TimerListener> tl = timeoutTimer.get();
-        if (tl != null) {
-            tl.clear();
-        }
 
-        long userThreadLatency = System.currentTimeMillis() - commandStartTimestamp;
-        executionResult = executionResult.markUserThreadCompletion((int) userThreadLatency);
-        ExecutionResult cancelled = executionResultAtTimeOfCancellation;
-        if (cancelled == null) {
-            metrics.markCommandDone(executionResult, commandKey, threadPoolKey, executionStarted);
-        } else {
-            metrics.markCommandDone(cancelled, commandKey, threadPoolKey, executionStarted);
-        }
-    }
 
     private Observable<R> applyHystrixSemantics(final AbstractCommand<R> _cmd) {
         // mark that we're starting execution on the ExecutionHook
@@ -601,7 +584,7 @@ private Observable<R> executeCommandWithSpecifiedIsolation(final AbstractCommand
             return Observable.defer(new Func0<Observable<R>>() {
                 @Override
                 public Observable<R> call() {
-                    executionStarted = true;
+                    executionResult = executionResult.setExecutionOccurred();
                     metrics.markCommandStart(commandKey, threadPoolKey, ExecutionIsolationStrategy.THREAD);
 
                     if (isCommandTimedOut.get() == TimedOutStatus.TIMED_OUT) {
@@ -639,7 +622,7 @@ public Boolean call() {
             return Observable.defer(new Func0<Observable<R>>() {
                 @Override
                 public Observable<R> call() {
-                    executionStarted = true;
+                    executionResult = executionResult.setExecutionOccurred();
                     metrics.markCommandStart(commandKey, threadPoolKey, ExecutionIsolationStrategy.SEMAPHORE);
                     // semaphore isolated
                     // store the command that is being run
@@ -849,17 +832,38 @@ public void call() {
     }
 
     private void cleanUpAfterResponseFromCache() {
+        Reference<TimerListener> tl = timeoutTimer.get();
+        if (tl != null) {
+            tl.clear();
+        }
+
         final long latency = System.currentTimeMillis() - commandStartTimestamp;
         executionResult = executionResult
                 .addEvent(-1, HystrixEventType.RESPONSE_FROM_CACHE)
                 .markUserThreadCompletion(latency)
                 .setNotExecutedInThread();
         ExecutionResult cacheOnlyForMetrics = ExecutionResult.from(HystrixEventType.RESPONSE_FROM_CACHE)
                 .markUserThreadCompletion(latency);
-        metrics.markCommandDone(cacheOnlyForMetrics, commandKey, threadPoolKey, executionStarted);
+        metrics.markCommandDone(cacheOnlyForMetrics, commandKey, threadPoolKey);
         eventNotifier.markEvent(HystrixEventType.RESPONSE_FROM_CACHE, commandKey);
     }
 
+    private void handleCommandEnd() {
+        Reference<TimerListener> tl = timeoutTimer.get();
+        if (tl != null) {
+            tl.clear();
+        }
+
+        long userThreadLatency = System.currentTimeMillis() - commandStartTimestamp;
+        executionResult = executionResult.markUserThreadCompletion((int) userThreadLatency);
+        ExecutionResult cancelled = executionResultAtTimeOfCancellation;
+        if (cancelled == null) {
+            metrics.markCommandDone(executionResult, commandKey, threadPoolKey);
+        } else {
+            metrics.markCommandDone(cancelled, commandKey, threadPoolKey);
+        }
+    }
+
     private Observable<R> handleSemaphoreRejectionViaFallback() {
         Exception semaphoreRejectionException = new RuntimeException("could not acquire a semaphore for execution");
         executionResult = executionResult.setExecutionException(semaphoreRejectionException);

hystrix-core/src/main/java/com/netflix/hystrix/ExecutionResult.java

@@ -219,6 +219,11 @@ private static boolean didExecutionOccur(HystrixEventType eventType) {
         }
     }
 
+    public ExecutionResult setExecutionOccurred() {
+        return new ExecutionResult(eventCounts, startTimestamp, executionLatency, userThreadLatency,
+                failedExecutionException, executionException, true, isExecutedInThread, collapserKey);
+    }
+
     public ExecutionResult setExecutionLatency(int executionLatency) {
         return new ExecutionResult(eventCounts, startTimestamp, executionLatency, userThreadLatency,
                 failedExecutionException, executionException, executionOccurred, isExecutedInThread, collapserKey);
@@ -273,14 +278,14 @@ public ExecutionResult markUserThreadCompletion(long userThreadLatency) {
     public ExecutionResult addEvent(HystrixEventType eventType) {
         return new ExecutionResult(eventCounts.plus(eventType), startTimestamp, executionLatency,
                 userThreadLatency, failedExecutionException, executionException,
-                executionOccurred ? executionOccurred : didExecutionOccur(eventType), isExecutedInThread, collapserKey);
+                executionOccurred, isExecutedInThread, collapserKey);
     }
 
     public ExecutionResult addEvent(int executionLatency, HystrixEventType eventType) {
         if (startTimestamp >= 0 && !isResponseRejected()) {
             return new ExecutionResult(eventCounts.plus(eventType), startTimestamp, executionLatency,
                     userThreadLatency, failedExecutionException, executionException,
-                    executionOccurred ? executionOccurred : didExecutionOccur(eventType), isExecutedInThread, collapserKey);
+                    executionOccurred, isExecutedInThread, collapserKey);
         } else {
             return addEvent(eventType);
         }

hystrix-core/src/main/java/com/netflix/hystrix/HystrixCachedObservable.java

@@ -84,11 +84,6 @@ public void call() {
                             completionLogicRun.set(true);
                         }
                     }
-                })
-                .doOnSubscribe(new Action0() {
-                    @Override
-                    public void call() {
-                    }
                 });
     }
 

hystrix-core/src/main/java/com/netflix/hystrix/HystrixCommandMetrics.java

@@ -343,9 +343,9 @@ public int getCurrentConcurrentExecutionCount() {
         HystrixThreadEventStream.getInstance().commandExecutionStarted(commandKey, threadPoolKey, isolationStrategy, currentCount);
     }
 
-    /* package-private */ void markCommandDone(ExecutionResult executionResult, HystrixCommandKey commandKey, HystrixThreadPoolKey threadPoolKey, boolean executionStarted) {
+    /* package-private */ void markCommandDone(ExecutionResult executionResult, HystrixCommandKey commandKey, HystrixThreadPoolKey threadPoolKey) {
         HystrixThreadEventStream.getInstance().executionDone(executionResult, commandKey, threadPoolKey);
-        if (executionStarted) {
+        if (executionResult.executionOccurred()) {
             concurrentExecutionCount.decrementAndGet();
         }
     }