2626import java .util .concurrent .atomic .AtomicLong ;
2727
2828/**
29- * Shared state that coordinates failover across a pool of channels. All channels in a pool share
30- * this instance and its background executor, so probing and error rate evaluation run once for the
31- * whole pool.
32- *
33- * targetFallbackMode: whether channels should route to the fallback channel.
34- * generation: incremented on every pool-wide transition. A channel re-reads targetFallbackMode when
35- * generation advances past the value it last saw.
36- * inFallbackMode: whether all traffic is on the fallback channel. Gates error rate evaluation.
37- *
38- * Failover: the primary error rate crossing the threshold sets targetFallbackMode and increments
39- * generation. All channels switch to the fallback channel.
40- *
41- * Recovery, per-channel disabled: the first channel whose probes succeed clears targetFallbackMode
42- * and increments generation. The whole pool returns to the primary channel.
43- *
44- * Recovery, per-channel enabled: a recovering channel updates only its own state, leaving
45- * targetFallbackMode and generation unchanged. Other channels stay on the fallback channel until
46- * their own probes succeed. inFallbackMode and targetFallbackMode diverge until then.
47- *
48- * A shared instance passed to setSharedState must be shut down by the caller. A state a channel
49- * created for itself is shut down with that channel.
29+ * Shared thread-safe state that coordinates failover, recovery, and periodic error evaluation
30+ * across a pool of GcpFallbackChannel instances.
5031 */
5132public class GcpFallbackState {
5233 private final AtomicLong primarySuccesses = new AtomicLong (0 );
5334 private final AtomicLong primaryFailures = new AtomicLong (0 );
5435 private final AtomicLong fallbackSuccesses = new AtomicLong (0 );
5536 private final AtomicLong fallbackFailures = new AtomicLong (0 );
5637 private final AtomicLong generation = new AtomicLong (0 );
57- private final AtomicBoolean targetFallbackMode = new AtomicBoolean (false );
5838 private final AtomicBoolean inFallbackMode = new AtomicBoolean (false );
5939 private final AtomicBoolean evaluationStarted = new AtomicBoolean (false );
6040
6141 private ScheduledExecutorService execService = null ;
6242 private boolean ownsExecutor = false ;
43+ private boolean isShutdown = false ;
6344 private volatile ScheduledFuture <?> scheduledEvaluationFuture = null ;
6445
6546 public GcpFallbackState () {}
@@ -91,15 +72,7 @@ AtomicLong getFallbackFailures() {
9172 return fallbackFailures ;
9273 }
9374
94- /**
95- * Returns whether periodic error-rate evaluation is armed, i.e. whether any traffic is currently
96- * reaching the primary channel.
97- *
98- * <p>This is <em>not</em> "are the pool's channels routing to the fallback channel". Under
99- * per-channel recovery this returns {@code false} as soon as the first channel recovers, while
100- * other channels may still be in fallback. To ask whether a specific channel is in fallback, call
101- * {@link GcpFallbackChannel#isInFallbackMode()} on that channel.
102- */
75+ /** Returns whether the pool is currently in fallback mode. */
10376 boolean isInFallbackMode () {
10477 return inFallbackMode .get ();
10578 }
@@ -108,34 +81,25 @@ long getGeneration() {
10881 return generation .get ();
10982 }
11083
111- boolean getTargetFallbackMode () {
112- return targetFallbackMode .get ();
113- }
114-
115- /** Bumps the generation counter with a directive to target fallback mode. */
84+ /** Bumps the generation counter and transitions the pool to fallback mode. */
11685 synchronized void triggerFallback () {
11786 inFallbackMode .set (true );
118- targetFallbackMode .set (true );
11987 generation .incrementAndGet ();
12088 }
12189
12290 /**
123- * Records channel recovery by clearing primary error counts and updating fallback mode.
91+ * Records pool recovery by clearing primary error counts and updating fallback mode.
12492 *
12593 * @param expectedGen the generation at which the recovery probe started.
126- * @param perChannelRecovery whether recovery is scoped per channel rather than pool-wide.
12794 * @return the resulting pool generation, or -1 if the pool generation changed concurrently.
12895 */
129- synchronized long recordRecovery (long expectedGen , boolean perChannelRecovery ) {
96+ synchronized long recordRecovery (long expectedGen ) {
13097 if (generation .get () != expectedGen ) {
13198 return -1 ;
13299 }
133- if (inFallbackMode .get ( )) {
100+ if (inFallbackMode .compareAndSet ( true , false )) {
134101 primaryFailures .set (0 );
135102 primarySuccesses .set (0 );
136- inFallbackMode .set (false );
137- }
138- if (!perChannelRecovery && targetFallbackMode .compareAndSet (true , false )) {
139103 generation .incrementAndGet ();
140104 }
141105 return generation .get ();
@@ -167,7 +131,7 @@ synchronized ScheduledExecutorService getOrCreateExecutorService(
167131 /** Schedules a periodic task (e.g., probe) on the shared background executor service. */
168132 synchronized ScheduledFuture <?> scheduleTask (
169133 Runnable command , long initialDelay , long period , TimeUnit unit ) {
170- if (this .execService == null || this .execService .isShutdown ()) {
134+ if (isShutdown || this .execService == null || this .execService .isShutdown ()) {
171135 return null ;
172136 }
173137 return this .execService .scheduleAtFixedRate (command , initialDelay , period , unit );
@@ -180,7 +144,8 @@ synchronized ScheduledFuture<?> scheduleTask(
180144 * @param options the fallback channel configuration options.
181145 */
182146 synchronized void startPeriodicEvaluation (GcpFallbackChannelOptions options ) {
183- if (options == null
147+ if (isShutdown
148+ || options == null
184149 || !options .isEnableFallback ()
185150 || options .getPeriod () == null
186151 || options .getPeriod ().toMillis () <= 0 ) {
@@ -212,8 +177,7 @@ synchronized void startPeriodicEvaluation(GcpFallbackChannelOptions options) {
212177 * @param options the fallback channel configuration options.
213178 * @param openTelemetry telemetry module for recording error metrics.
214179 */
215- void checkErrorRates (
216- GcpFallbackChannelOptions options , GcpFallbackOpenTelemetry openTelemetry ) {
180+ void checkErrorRates (GcpFallbackChannelOptions options , GcpFallbackOpenTelemetry openTelemetry ) {
217181 float primaryErrRate = 0f ;
218182 boolean fallbackTriggered = false ;
219183 boolean currentInFallback ;
@@ -271,6 +235,7 @@ synchronized void stopPeriodicEvaluation() {
271235
272236 /** Shuts down the state, cancelling evaluation and shutting down internal executor if owned. */
273237 public synchronized void shutdown () {
238+ isShutdown = true ;
274239 stopPeriodicEvaluation ();
275240 if (ownsExecutor && execService != null && !execService .isShutdown ()) {
276241 execService .shutdown ();
@@ -282,6 +247,7 @@ public synchronized void shutdown() {
282247 * owned.
283248 */
284249 public synchronized void shutdownNow () {
250+ isShutdown = true ;
285251 stopPeriodicEvaluation ();
286252 if (ownsExecutor && execService != null && !execService .isShutdown ()) {
287253 execService .shutdownNow ();
0 commit comments