Hystrix: Race condition in HystrixCircuitBreaker in version 1.5.13

[tjuszczyk]

In Hystrix 1.5.13 there's a race condition, which makes it possible that HystrixCircuitBreaker object will end up in inconsistent state: status==OPEN and circuitOpened==-1. This causes isOpen() method to answer false and circuit will never be closed again, because of:

if (status.compareAndSet(Status.CLOSED, Status.OPEN)) {
  circuitOpened.set(System.currentTimeMillis());
}

Race condition is happening between:

 @Override
public void markSuccess() {      
  if (status.compareAndSet(Status.HALF_OPEN, Status.CLOSED)) {
    //This thread wins the race to close the circuit - it resets the stream to start it over from 0
    metrics.resetStream();
    Subscription previousSubscription = activeSubscription.get();
    if (previousSubscription != null) {
      previousSubscription.unsubscribe();
    }
    Subscription newSubscription = subscribeToStream();
    activeSubscription.set(newSubscription);
    circuitOpened.set(-1L);
  }
}

and onNext in HealthCountStream subscriber:

@Override
public void onNext(HealthCounts hc) {
// check if we are past the statisticalWindowVolumeThreshold
   if (hc.getTotalRequests() < properties.circuitBreakerRequestVolumeThreshold().get()) {
    // we are not past the minimum volume threshold for the stat window,
    // so no change to circuit status.
    // if it was CLOSED, it stays CLOSED
    // if it was open, we need to wait for sleep window to elapse
  } else {
      if (hc.getErrorPercentage() < properties.circuitBreakerErrorThresholdPercentage().get()) {
      //we are not past the minimum error threshold for the stat window,
      // so no change to circuit status.
      // if it was CLOSED, it stays CLOSED
      // if it was half-open, we need to wait for a successful command execution
      // if it was open, we need to wait for sleep window to elapse
    } else {
       // our failure rate is too high, we need to set the state to OPEN
        if (status.compareAndSet(Status.CLOSED, Status.OPEN)) {
            circuitOpened.set(System.currentTimeMillis());
        }
     }
  }
}

Please find the test below, which reproduce the issue:

public class HystrixCircuitBreakerRaceConditionTest {

    private static final org.slf4j.Logger LOG = LoggerFactory.getLogger(HystrixCircuitBreakerRaceConditionTest.class);

    private String groupKey;
    private static final int VERY_SMALL_SLEEP_MILLIS = 5;

    @Before
    public void setUp() {
        groupKey = UUID.randomUUID().toString();
        ConfigurationManager.getConfigInstance().clear();

        ConfigurationManager.getConfigInstance().setProperty(String.format("hystrix.threadpool.%s.coreSize", groupKey), String.valueOf(1));
        Hystrix.reset();
    }

    @After
    public void tearDown() {
        ConfigurationManager.getConfigInstance().clear();
    }


    @Test
    public void requestVolumeIsResetToZeroAfterTheCircuitIsClosed() throws Exception {
        //given
        ConfigurationManager.getConfigInstance().setProperty("hystrix.command.default.execution.timeout.enabled", "false");

        ConfigurationManager.getConfigInstance().setProperty("hystrix.command." + groupKey + "-raceCondition.circuitBreaker.enabled", "true");
        ConfigurationManager.getConfigInstance().setProperty("hystrix.command." + groupKey + "-raceCondition.circuitBreaker.requestVolumeThreshold", "2");
        ConfigurationManager.getConfigInstance().setProperty("hystrix.command." + groupKey + "-raceCondition.circuitBreaker.errorThresholdPercentage", "50");
        ConfigurationManager.getConfigInstance().setProperty("hystrix.command." + groupKey + "-raceCondition.circuitBreaker.sleepWindowInMilliseconds", "500");
        ConfigurationManager.getConfigInstance().setProperty("hystrix.command." + groupKey + "-raceCondition.metrics.healthSnapshot.intervalInMilliseconds", "10");

        AtomicInteger doRunCount = new AtomicInteger(0);
        AtomicInteger doFallbackCount = new AtomicInteger(0);
        AtomicInteger circuitOpenCount = new AtomicInteger(0);

        //when
        // First successful request (doRunCount++)
        LOG.info("Request 1:");
        new CommandWithExecutionCounters(groupKey, "raceCondition", doRunCount, doFallbackCount, circuitOpenCount, true).execute();
        Thread.sleep(20); // we always sleep 50mills so that hystrix recalculates it's metrics based on intervalInMillliseconds

        // First Failure - circuit is broken after this point because we have hit 50% and we have had 2 requests. (doRun++, doFallBack++, circuit is opened after)
        LOG.info("Request 2:");
        new CommandWithExecutionCounters(groupKey, "raceCondition", doRunCount, doFallbackCount, circuitOpenCount, false).execute();
        Thread.sleep(20);

        // Second Failure - circuit was already broken before this request, so we go straight to fallback (doFallBackCount++, circuitOpen++)
        LOG.info("Request 3:");
        new CommandWithExecutionCounters(groupKey, "raceCondition", doRunCount, doFallbackCount, circuitOpenCount, true).execute();
        Thread.sleep(20);

        LOG.info("....waiting for sleep window:");
        // waiting for sleepWindowInMilliseconds to pass, so that hystrix will try to execute doRun in the next request.
        Thread.sleep(505);

        // Second successful request - this request closes the circuit, but this request is not considered in hystrix's following metrics calculations (doRun++)
        LOG.info("Request 4:");
        new CommandWithExecutionCounters(groupKey, "raceCondition", doRunCount, doFallbackCount, circuitOpenCount, true).execute();
        Thread.sleep(20);

        // Third successful request - this request is the first request considered in hystrix metrics calculations use for tripping the circuit (doRun++)
        LOG.info("Request 5:");
        new CommandWithExecutionCounters(groupKey, "raceCondition", doRunCount, doFallbackCount, circuitOpenCount, true).execute();
        Thread.sleep(20);

        // First failure request after circuit was closed -  circuit is broken after this point because we have hit 50% and we have had 2 requests. (doRun++, doFallBack++)
        LOG.info("Request 6:");
        new CommandWithExecutionCounters(groupKey, "raceCondition", doRunCount, doFallbackCount, circuitOpenCount, false).execute();
        Thread.sleep(20);

        // Second Failure after circuit was closed - circuit was already broken before this request, so we go straight to fallback (doFallBack++, circuit open)
        LOG.info("Request 7:");
        new CommandWithExecutionCounters(groupKey, "raceCondition", doRunCount, doFallbackCount, circuitOpenCount, false).execute();
        Thread.sleep(20);

        // Failure after circuit was closed - circuit was already broken before this request, so we go straight to fallback (doFallBack++, circuitOpen
        //++)
        LOG.info("Request 8:");
        new CommandWithExecutionCounters(groupKey, "raceCondition", doRunCount, doFallbackCount, circuitOpenCount, false).execute();
        Thread.sleep(20);

        // Second Failure after circuit was closed - circuit was already broken before this request, so we go straight to fallback (doFallBack++, circuitOpen++)
        LOG.info("Request 9:");
        new CommandWithExecutionCounters(groupKey, "raceCondition", doRunCount, doFallbackCount, circuitOpenCount, false).execute();
        Thread.sleep(20);

        //circuit should be opened now
        LOG.info("Request 10+(opened circuit):");
        new CommandWithExecutionCounters(groupKey, "raceCondition", doRunCount, doFallbackCount, circuitOpenCount, false).execute();
        Thread.sleep(20);
        new CommandWithExecutionCounters(groupKey, "raceCondition", doRunCount, doFallbackCount, circuitOpenCount, false).execute();
        Thread.sleep(20);
        new CommandWithExecutionCounters(groupKey, "raceCondition", doRunCount, doFallbackCount, circuitOpenCount, false).execute();
        Thread.sleep(20);
        new CommandWithExecutionCounters(groupKey, "raceCondition", doRunCount, doFallbackCount, circuitOpenCount, false).execute();
        Thread.sleep(20);

        //then
        assertThat(circuitOpenCount.get()).as("circuit broken test does not match").isEqualTo(7);
        assertThat(doRunCount.get()).as("Do run count does not match").isEqualTo(6);
        assertThat(doFallbackCount.get()).as("Do fallback count does not match").isEqualTo(10);

    }

    private static class CommandWithExecutionCounters extends HystrixCommand<String> {

        private static final org.slf4j.Logger LOG = LoggerFactory.getLogger(CommandWithExecutionCounters.class);

        private final AtomicInteger doRunCount;
        private final AtomicInteger doFallbackCount;
        private final AtomicInteger circuitOpenCount;
        private final boolean successful;

        public CommandWithExecutionCounters(String groupKey, String commandName, AtomicInteger doRunCount, AtomicInteger doFallbackCount, AtomicInteger circuitBrokenCount, boolean isSuccessful) {
            super(Setter.withGroupKey(HystrixCommandGroupKey.Factory.asKey(groupKey))
                    .andCommandKey(HystrixCommandKey.Factory.asKey(groupKey + "-" + commandName)));
            this.doRunCount = doRunCount;
            this.doFallbackCount = doFallbackCount;
            this.circuitOpenCount = circuitBrokenCount;
            this.successful = isSuccessful;
        }

        @Override
        protected final String getFallback() {
            Uninterruptibles.sleepUninterruptibly(VERY_SMALL_SLEEP_MILLIS, TimeUnit.MILLISECONDS);
            LOG.info("::in doFallback: circuitOpen:{}", isCircuitBreakerOpen());
            if (isCircuitBreakerOpen()) {
                circuitOpenCount.incrementAndGet();
            }
            doFallbackCount.incrementAndGet();
            return "DO_FALLBACK_RESULT";
        }

        @Override
        protected String run() throws Exception {
            LOG.info("::in doRun:");
            LOG.warn("::::::Circuit Breaket status::::::{}/{}[{}] - status:{} ", getMetrics().getHealthCounts().getTotalRequests(), getMetrics().getHealthCounts().getErrorCount(),
                    getMetrics().getHealthCounts().getErrorPercentage(), isCircuitBreakerOpen());

            doRunCount.incrementAndGet();
            if (successful) return "DO_RUN_RESULT";
            throw new RuntimeException("Exception during execution");

        }
    }
}

[worldtiki]

This also affects 1.5.12

[tandonraghav]

Facing same issue in 1.5.12. Is there any tentative timeline for this Fix?
@mattrjacobs Please let me know if we can use this in Production?

We have temporarily fixed the module like this:-

````
if (status.compareAndSet(Status.CLOSED, Status.OPEN)) {
synchronized (hystrixCommandKey){
if(status.get().equals(Status.OPEN)){
circuitOpened.set(System.currentTimeMillis());
}
}
}

   @Override
    public void markSuccess() {
        if (status.compareAndSet(Status.HALF_OPEN, Status.CLOSED)) {
            //This thread wins the race to close the circuit - it resets the stream to start it over from 0
            metrics.resetStream();
            Subscription previousSubscription = activeSubscription.get();
            if (previousSubscription != null) {
                previousSubscription.unsubscribe();
            }
            Subscription newSubscription = subscribeToStream();
            activeSubscription.set(newSubscription);
            synchronized (hystrixCommandKey){
                if (status.get().equals(Status.CLOSED)){
                    circuitOpened.set(-1L);
                }
            }
        }
    }