# Implement Circuit Breaker (/learn/rpc/how-to/circuit-breaker)



## The Problem [#the-problem]

When a responder is down or slow, every RPC request blocks until the timeout expires. This cascades through the system — the sender's threads or goroutines are tied up waiting, eventually causing the sender itself to become unresponsive.

## Circuit Breaker Pattern [#circuit-breaker-pattern]

A circuit breaker tracks failures and short-circuits requests when a threshold is reached, returning a fallback immediately instead of waiting for a timeout.

<Mermaid
  chart="stateDiagram-v2
    [*] --> Closed
    Closed --> Open: N consecutive failures
    Open --> HalfOpen: Reset timeout expires
    HalfOpen --> Closed: Probe succeeds
    HalfOpen --> Open: Probe fails"
/>

*The three circuit states and the transitions between them.*

| State         | Behavior                                                               |
| ------------- | ---------------------------------------------------------------------- |
| **Closed**    | Normal operation — requests pass through to KubeMQ                     |
| **Open**      | Fail fast — return fallback immediately without calling KubeMQ         |
| **Half-Open** | Probe — allow one test request to check if the responder has recovered |

## Implementation [#implementation]

<Steps>
  <Step>
    ### Define the Circuit Breaker [#define-the-circuit-breaker]

    <Tabs groupId="language" items="['Go', 'Python', 'Node.js', 'Java', 'C#', 'Kotlin', 'C++', 'Rust', 'Ruby', 'Elixir']">
      <Tab value="Go">
        ```go title="circuit_breaker.go"
        type CircuitBreaker struct {
            mu              sync.Mutex
            failures        int
            threshold       int
            state           string // "closed", "open", "half-open"
            lastFailureTime time.Time
            resetTimeout    time.Duration
        }

        func NewCircuitBreaker(threshold int, resetTimeout time.Duration) *CircuitBreaker {
            return &CircuitBreaker{
                threshold:    threshold,
                state:        "closed",
                resetTimeout: resetTimeout,
            }
        }
        ```
      </Tab>

      <Tab value="Python">
        ```python title="circuit_breaker.py"
        import time
        import threading

        class CircuitBreaker:
            def __init__(self, threshold=5, reset_timeout=30):
                self.threshold = threshold
                self.reset_timeout = reset_timeout
                self.failures = 0
                self.state = "closed"
                self.last_failure_time = 0
                self._lock = threading.Lock()
        ```
      </Tab>

      <Tab value="Node.js">
        ```javascript title="circuit_breaker.js"
        class CircuitBreaker {
          constructor(threshold = 5, resetTimeout = 30000) {
            this.threshold = threshold;
            this.resetTimeout = resetTimeout;
            this.failures = 0;
            this.state = "closed";
            this.lastFailureTime = 0;
          }
        }
        ```
      </Tab>

      <Tab value="Java">
        ```java title="CircuitBreaker.java"
        public class CircuitBreaker {
            private final int threshold;
            private final long resetTimeoutMs;
            private int failures = 0;
            private String state = "closed";
            private long lastFailureTime = 0;

            public CircuitBreaker(int threshold, long resetTimeoutMs) {
                this.threshold = threshold;
                this.resetTimeoutMs = resetTimeoutMs;
            }
        }
        ```
      </Tab>

      <Tab value="C#">
        ```csharp title="CircuitBreaker.cs"
        public class CircuitBreaker
        {
            private readonly int _threshold;
            private readonly TimeSpan _resetTimeout;
            private int _failures;
            private string _state = "closed";
            private DateTime _lastFailureTime;
            private readonly object _lock = new();

            public CircuitBreaker(int threshold = 5, TimeSpan? resetTimeout = null)
            {
                _threshold = threshold;
                _resetTimeout = resetTimeout ?? TimeSpan.FromSeconds(30);
            }
        }
        ```
      </Tab>

      <Tab value="Kotlin">
        ```kotlin title="CircuitBreaker.kt"
        class CircuitBreaker(
            private val threshold: Int = 5,
            private val resetTimeout: Long = 30000
        ) {
            private var failures = 0
            private var state = "closed"
            private var lastFailureTime = 0L
            private val lock = Any()
        }
        ```
      </Tab>

      <Tab value="C++">
        ```cpp title="circuit_breaker.cpp"
        class CircuitBreaker {
            int threshold;
            int resetTimeout;
            int failures = 0;
            std::string state = "closed";
            std::chrono::steady_clock::time_point lastFailureTime;
            std::mutex mtx;

        public:
            CircuitBreaker(int threshold = 5, int resetTimeoutSec = 30)
                : threshold(threshold), resetTimeout(resetTimeoutSec) {}
        };
        ```
      </Tab>

      <Tab value="Rust">
        ```rust title="circuit_breaker.rs"
        use std::time::{Duration, Instant};

        #[derive(Clone, Copy, PartialEq)]
        enum State {
            Closed,
            Open,
            HalfOpen,
        }

        struct CircuitBreaker {
            threshold: u32,
            reset_timeout: Duration,
            failures: u32,
            state: State,
            last_failure: Option<Instant>,
        }

        impl CircuitBreaker {
            fn new(threshold: u32, reset_timeout: Duration) -> Self {
                Self {
                    threshold,
                    reset_timeout,
                    failures: 0,
                    state: State::Closed,
                    last_failure: None,
                }
            }
        }
        ```
      </Tab>

      <Tab value="Ruby">
        ```ruby title="circuit_breaker.rb"
        class CircuitBreaker
          def initialize(threshold: 5, reset_timeout: 30)
            @threshold = threshold
            @reset_timeout = reset_timeout
            @failures = 0
            @state = :closed
            @last_failure_time = nil
            @mutex = Mutex.new
          end
        end
        ```
      </Tab>

      <Tab value="Elixir">
        ```elixir title="circuit_breaker.ex"
        defmodule CircuitBreaker do
          # Backed by an Agent so the state survives across calls.
          defstruct threshold: 5,
                    reset_timeout_ms: 30_000,
                    failures: 0,
                    state: :closed,
                    last_failure_ms: nil

          def start_link(threshold \\ 5, reset_timeout_ms \\ 30_000) do
            Agent.start_link(fn ->
              %CircuitBreaker{threshold: threshold, reset_timeout_ms: reset_timeout_ms}
            end)
          end
        end
        ```
      </Tab>
    </Tabs>
  </Step>

  <Step>
    ### Track Failures and Open the Circuit [#track-failures-and-open-the-circuit]

    Record each failure. When consecutive failures reach the threshold, open the circuit.

    <Tabs groupId="language" items="['Go', 'Python', 'Node.js', 'Java', 'C#', 'Kotlin', 'C++', 'Rust', 'Ruby', 'Elixir']">
      <Tab value="Go">
        ```go title="track_failures.go"
        func (cb *CircuitBreaker) RecordFailure() {
            cb.mu.Lock()
            defer cb.mu.Unlock()
            cb.failures++
            cb.lastFailureTime = time.Now()
            if cb.failures >= cb.threshold {
                cb.state = "open"
                log.Printf("Circuit OPEN after %d failures", cb.failures)
            }
        }

        func (cb *CircuitBreaker) RecordSuccess() {
            cb.mu.Lock()
            defer cb.mu.Unlock()
            cb.failures = 0
            cb.state = "closed"
        }
        ```
      </Tab>

      <Tab value="Python">
        ```python title="track_failures.py"
        def record_failure(self):
            with self._lock:
                self.failures += 1
                self.last_failure_time = time.time()
                if self.failures >= self.threshold:
                    self.state = "open"
                    print(f"Circuit OPEN after {self.failures} failures")

        def record_success(self):
            with self._lock:
                self.failures = 0
                self.state = "closed"
        ```
      </Tab>

      <Tab value="Node.js">
        ```javascript title="track_failures.js"
        recordFailure() {
          this.failures++;
          this.lastFailureTime = Date.now();
          if (this.failures >= this.threshold) {
            this.state = "open";
            console.log(`Circuit OPEN after ${this.failures} failures`);
          }
        }

        recordSuccess() {
          this.failures = 0;
          this.state = "closed";
        }
        ```
      </Tab>

      <Tab value="Java">
        ```java title="TrackFailures.java"
        public synchronized void recordFailure() {
            failures++;
            lastFailureTime = System.currentTimeMillis();
            if (failures >= threshold) {
                state = "open";
                System.out.printf("Circuit OPEN after %d failures%n", failures);
            }
        }

        public synchronized void recordSuccess() {
            failures = 0;
            state = "closed";
        }
        ```
      </Tab>

      <Tab value="C#">
        ```csharp title="TrackFailures.cs"
        public void RecordFailure()
        {
            lock (_lock)
            {
                _failures++;
                _lastFailureTime = DateTime.UtcNow;
                if (_failures >= _threshold)
                {
                    _state = "open";
                    Console.WriteLine($"Circuit OPEN after {_failures} failures");
                }
            }
        }

        public void RecordSuccess()
        {
            lock (_lock) { _failures = 0; _state = "closed"; }
        }
        ```
      </Tab>

      <Tab value="Kotlin">
        ```kotlin title="TrackFailures.kt"
        fun recordFailure() = synchronized(lock) {
            failures++
            lastFailureTime = System.currentTimeMillis()
            if (failures >= threshold) {
                state = "open"
                println("Circuit OPEN after $failures failures")
            }
        }

        fun recordSuccess() = synchronized(lock) {
            failures = 0
            state = "closed"
        }
        ```
      </Tab>

      <Tab value="C++">
        ```cpp title="track_failures.cpp"
        void recordFailure() {
            std::lock_guard<std::mutex> lock(mtx);
            failures++;
            lastFailureTime = std::chrono::steady_clock::now();
            if (failures >= threshold) {
                state = "open";
                std::cout << "Circuit OPEN after " << failures << " failures" << std::endl;
            }
        }

        void recordSuccess() {
            std::lock_guard<std::mutex> lock(mtx);
            failures = 0;
            state = "closed";
        }
        ```
      </Tab>

      <Tab value="Rust">
        ```rust title="track_failures.rs"
        impl CircuitBreaker {
            fn record_failure(&mut self) {
                self.failures += 1;
                self.last_failure = Some(Instant::now());
                if self.failures >= self.threshold {
                    self.state = State::Open;
                    println!("Circuit OPEN after {} failures", self.failures);
                }
            }

            fn record_success(&mut self) {
                self.failures = 0;
                self.state = State::Closed;
            }
        }
        ```
      </Tab>

      <Tab value="Ruby">
        ```ruby title="track_failures.rb"
        def record_failure
          @mutex.synchronize do
            @failures += 1
            @last_failure_time = Time.now
            if @failures >= @threshold
              @state = :open
              puts "Circuit OPEN after #{@failures} failures"
            end
          end
        end

        def record_success
          @mutex.synchronize do
            @failures = 0
            @state = :closed
          end
        end
        ```
      </Tab>

      <Tab value="Elixir">
        ```elixir title="track_failures.ex"
        def record_failure(cb) do
          Agent.update(cb, fn s ->
            failures = s.failures + 1
            state = if failures >= s.threshold, do: :open, else: s.state
            if state == :open, do: IO.puts("Circuit OPEN after #{failures} failures")
            %{s | failures: failures, state: state, last_failure_ms: System.monotonic_time(:millisecond)}
          end)
        end

        def record_success(cb) do
          Agent.update(cb, fn s -> %{s | failures: 0, state: :closed} end)
        end
        ```
      </Tab>
    </Tabs>
  </Step>

  <Step>
    ### Check State Before Sending [#check-state-before-sending]

    Before each RPC call, check the circuit state. If open, check whether the reset timeout has expired to transition to half-open.

    <Tabs groupId="language" items="['Go', 'Python', 'Node.js', 'Java', 'C#', 'Kotlin', 'C++', 'Rust', 'Ruby', 'Elixir']">
      <Tab value="Go">
        ```go title="check_state.go"
        func (cb *CircuitBreaker) AllowRequest() bool {
            cb.mu.Lock()
            defer cb.mu.Unlock()

            switch cb.state {
            case "closed":
                return true
            case "open":
                if time.Since(cb.lastFailureTime) > cb.resetTimeout {
                    cb.state = "half-open"
                    log.Println("Circuit HALF-OPEN — probing...")
                    return true
                }
                return false
            case "half-open":
                return true
            }
            return false
        }
        ```
      </Tab>

      <Tab value="Python">
        ```python title="check_state.py"
        def allow_request(self):
            with self._lock:
                if self.state == "closed":
                    return True
                if self.state == "open":
                    if time.time() - self.last_failure_time > self.reset_timeout:
                        self.state = "half-open"
                        print("Circuit HALF-OPEN — probing...")
                        return True
                    return False
                return True  # half-open allows one probe
        ```
      </Tab>

      <Tab value="Node.js">
        ```javascript title="check_state.js"
        allowRequest() {
          if (this.state === "closed") return true;
          if (this.state === "open") {
            if (Date.now() - this.lastFailureTime > this.resetTimeout) {
              this.state = "half-open";
              console.log("Circuit HALF-OPEN — probing...");
              return true;
            }
            return false;
          }
          return true; // half-open allows one probe
        }
        ```
      </Tab>

      <Tab value="Java">
        ```java title="CheckState.java"
        public synchronized boolean allowRequest() {
            if ("closed".equals(state)) return true;
            if ("open".equals(state)) {
                if (System.currentTimeMillis() - lastFailureTime > resetTimeoutMs) {
                    state = "half-open";
                    System.out.println("Circuit HALF-OPEN — probing...");
                    return true;
                }
                return false;
            }
            return true;
        }
        ```
      </Tab>

      <Tab value="C#">
        ```csharp title="CheckState.cs"
        public bool AllowRequest()
        {
            lock (_lock)
            {
                if (_state == "closed") return true;
                if (_state == "open")
                {
                    if (DateTime.UtcNow - _lastFailureTime > _resetTimeout)
                    {
                        _state = "half-open";
                        Console.WriteLine("Circuit HALF-OPEN — probing...");
                        return true;
                    }
                    return false;
                }
                return true;
            }
        }
        ```
      </Tab>

      <Tab value="Kotlin">
        ```kotlin title="CheckState.kt"
        fun allowRequest(): Boolean = synchronized(lock) {
            when (state) {
                "closed" -> true
                "open" -> {
                    if (System.currentTimeMillis() - lastFailureTime > resetTimeout) {
                        state = "half-open"
                        println("Circuit HALF-OPEN — probing...")
                        true
                    } else false
                }
                else -> true
            }
        }
        ```
      </Tab>

      <Tab value="C++">
        ```cpp title="check_state.cpp"
        bool allowRequest() {
            std::lock_guard<std::mutex> lock(mtx);
            if (state == "closed") return true;
            if (state == "open") {
                auto elapsed = std::chrono::steady_clock::now() - lastFailureTime;
                if (elapsed > std::chrono::seconds(resetTimeout)) {
                    state = "half-open";
                    return true;
                }
                return false;
            }
            return true;
        }
        ```
      </Tab>

      <Tab value="Rust">
        ```rust title="check_state.rs"
        impl CircuitBreaker {
            fn allow_request(&mut self) -> bool {
                match self.state {
                    State::Closed => true,
                    State::Open => {
                        let expired = self
                            .last_failure
                            .map(|t| t.elapsed() > self.reset_timeout)
                            .unwrap_or(true);
                        if expired {
                            self.state = State::HalfOpen;
                            println!("Circuit HALF-OPEN — probing...");
                            true
                        } else {
                            false
                        }
                    }
                    State::HalfOpen => true, // allow one probe
                }
            }
        }
        ```
      </Tab>

      <Tab value="Ruby">
        ```ruby title="check_state.rb"
        def allow_request?
          @mutex.synchronize do
            case @state
            when :closed
              true
            when :open
              if Time.now - @last_failure_time > @reset_timeout
                @state = :half_open
                puts "Circuit HALF-OPEN — probing..."
                true
              else
                false
              end
            else
              true # half-open allows one probe
            end
          end
        end
        ```
      </Tab>

      <Tab value="Elixir">
        ```elixir title="check_state.ex"
        def allow_request?(cb) do
          Agent.get_and_update(cb, fn s ->
            case s.state do
              :closed ->
                {true, s}

              :open ->
                now = System.monotonic_time(:millisecond)
                if now - (s.last_failure_ms || 0) > s.reset_timeout_ms do
                  IO.puts("Circuit HALF-OPEN — probing...")
                  {true, %{s | state: :half_open}}
                else
                  {false, s}
                end

              :half_open ->
                {true, s} # allow one probe
            end
          end)
        end
        ```
      </Tab>
    </Tabs>
  </Step>

  <Step>
    ### Use the Circuit Breaker [#use-the-circuit-breaker]

    Wrap your RPC calls with the circuit breaker.

    <Tabs groupId="language" items="['Go', 'Python', 'Node.js', 'Java', 'C#', 'Kotlin', 'C++', 'Rust', 'Ruby', 'Elixir']">
      <Tab value="Go">
        ```go title="usage.go"
        cb := NewCircuitBreaker(5, 30*time.Second)

        func sendCommand(ctx context.Context, client *kubemq.Client,
            body []byte) (*kubemq.CommandResponse, error) {
            if !cb.AllowRequest() {
                return nil, fmt.Errorf("circuit open — service unavailable")
            }

            resp, err := client.SendCommand(ctx, kubemq.NewCommand().
                SetChannel("orders.process").
                SetBody(body).
                SetTimeout(5 * time.Second))

            if err != nil || !resp.Executed {
                cb.RecordFailure()
                return resp, err
            }

            cb.RecordSuccess()
            return resp, nil
        }
        ```
      </Tab>

      <Tab value="Python">
        ```python title="usage.py"
        cb = CircuitBreaker(threshold=5, reset_timeout=30)

        def send_command(client, body):
            if not cb.allow_request():
                raise RuntimeError("circuit open — service unavailable")

            try:
                response = client.send_command(CommandMessage(
                    channel="orders.process", body=body, timeout_in_seconds=5))
                if response.is_executed:
                    cb.record_success()
                    return response
                cb.record_failure()
                return response
            except Exception:
                cb.record_failure()
                raise
        ```
      </Tab>

      <Tab value="Node.js">
        ```javascript title="usage.js"
        const cb = new CircuitBreaker(5, 30000);

        async function sendCommand(client, body) {
          if (!cb.allowRequest()) {
            throw new Error("circuit open — service unavailable");
          }
          try {
            const response = await client.sendCommand({
              channel: "orders.process",
              body: Buffer.from(body),
              timeoutInSeconds: 5,
            });
            if (response.isExecuted) {
              cb.recordSuccess();
            } else {
              cb.recordFailure();
            }
            return response;
          } catch (err) {
            cb.recordFailure();
            throw err;
          }
        }
        ```
      </Tab>

      <Tab value="Java">
        ```java title="Usage.java"
        CircuitBreaker cb = new CircuitBreaker(5, 30000);

        CommandResponseMessage sendCommand(CQClient client, byte[] body) {
            if (!cb.allowRequest()) {
                throw new RuntimeException("circuit open — service unavailable");
            }
            try {
                var resp = client.sendCommandRequest(CommandMessage.builder()
                    .channel("orders.process").body(body).timeout(5000).build());
                if (resp.isExecuted()) cb.recordSuccess();
                else cb.recordFailure();
                return resp;
            } catch (Exception e) {
                cb.recordFailure();
                throw e;
            }
        }
        ```
      </Tab>

      <Tab value="C#">
        ```csharp title="Usage.cs"
        var cb = new CircuitBreaker(5, TimeSpan.FromSeconds(30));

        async Task<CommandResponse> SendCommand(KubeMQClient client, byte[] body)
        {
            if (!cb.AllowRequest())
                throw new InvalidOperationException("circuit open — service unavailable");

            try
            {
                var resp = await client.SendCommandAsync(new CommandMessage
                {
                    Channel = "orders.process", Body = body, Timeout = TimeSpan.FromSeconds(5)
                });
                if (resp.IsExecuted) cb.RecordSuccess();
                else cb.RecordFailure();
                return resp;
            }
            catch
            {
                cb.RecordFailure();
                throw;
            }
        }
        ```
      </Tab>

      <Tab value="Kotlin">
        ```kotlin title="Usage.kt"
        val cb = CircuitBreaker(threshold = 5, resetTimeout = 30000)

        fun sendCommand(client: CQClient, body: ByteArray): CommandResponse {
            if (!cb.allowRequest())
                throw RuntimeException("circuit open — service unavailable")

            return try {
                val resp = client.sendCommand(CommandMessage(
                    channel = "orders.process", body = body, timeout = 5000))
                if (resp.isExecuted) cb.recordSuccess() else cb.recordFailure()
                resp
            } catch (e: Exception) {
                cb.recordFailure()
                throw e
            }
        }
        ```
      </Tab>

      <Tab value="C++">
        ```cpp title="usage.cpp"
        CircuitBreaker cb(5, 30);

        kubemq::CommandResponse sendCommand(kubemq::CQClient& client,
            const std::string& body) {
            if (!cb.allowRequest()) {
                throw std::runtime_error("circuit open — service unavailable");
            }
            try {
                kubemq::CommandMessage cmd;
                cmd.channel = "orders.process";
                cmd.body = body;
                cmd.timeout = 5000;
                auto resp = client.sendCommand(cmd);
                if (resp.isExecuted) cb.recordSuccess();
                else cb.recordFailure();
                return resp;
            } catch (...) {
                cb.recordFailure();
                throw;
            }
        }
        ```
      </Tab>

      <Tab value="Rust">
        ```rust title="usage.rs"
        use kubemq::prelude::*;
        use kubemq::CommandBuilder;
        use std::error::Error;
        use std::time::Duration;

        async fn send_command(
            client: &KubemqClient,
            cb: &mut CircuitBreaker,
            body: Vec<u8>,
        ) -> Result<(), Box<dyn Error>> {
            if !cb.allow_request() {
                return Err("circuit open — service unavailable".into());
            }

            let command = CommandBuilder::new()
                .channel("orders.process")
                .body(body)
                .timeout(Duration::from_secs(5))
                .build();

            match client.send_command(command).await {
                Ok(resp) if resp.executed => {
                    cb.record_success();
                    Ok(())
                }
                Ok(_) => {
                    cb.record_failure();
                    Ok(())
                }
                Err(e) => {
                    cb.record_failure();
                    Err(Box::new(e))
                }
            }
        }
        ```
      </Tab>

      <Tab value="Ruby">
        ```ruby title="usage.rb"
        require 'kubemq'

        cb = CircuitBreaker.new(threshold: 5, reset_timeout: 30)

        def send_command(client, cb, body)
          raise 'circuit open — service unavailable' unless cb.allow_request?

          begin
            msg = KubeMQ::CQ::CommandMessage.new(
              channel: 'orders.process', body: body, timeout: 5000
            )
            result = client.send_command(msg)
            if result.executed
              cb.record_success
            else
              cb.record_failure
            end
            result
          rescue KubeMQ::Error
            cb.record_failure
            raise
          end
        end
        ```
      </Tab>

      <Tab value="Elixir">
        ```elixir title="usage.ex"
        def send_command(client, cb, body) do
          unless CircuitBreaker.allow_request?(cb) do
            {:error, :circuit_open}
          else
            command = KubeMQ.Command.new(
              channel: "orders.process",
              body: body,
              timeout: 5_000
            )

            case KubeMQ.Client.send_command(client, command) do
              {:ok, %{executed: true} = resp} ->
                CircuitBreaker.record_success(cb)
                {:ok, resp}

              {:ok, resp} ->
                CircuitBreaker.record_failure(cb)
                {:ok, resp}

              {:error, err} ->
                CircuitBreaker.record_failure(cb)
                {:error, err}
            end
          end
        end
        ```
      </Tab>
    </Tabs>
  </Step>
</Steps>

## Configuration [#configuration]

| Parameter             | Description                                     | Recommended   |
| --------------------- | ----------------------------------------------- | ------------- |
| **Failure threshold** | Consecutive failures before opening the circuit | 3–5           |
| **Reset timeout**     | How long the circuit stays open before probing  | 15–60 seconds |
| **Half-open probes**  | Number of test requests before closing          | 1–3           |

<Callout type="info">
  Combine circuit breakers with [timeout configuration](/learn/rpc/how-to/timeout-configuration) and [load balancing](/learn/rpc/how-to/load-balancing) for a resilient RPC setup.
</Callout>