diff --git a/grpc-gcp-java/src/main/java/com/google/cloud/grpc/fallback/GcpFallbackChannel.java b/grpc-gcp-java/src/main/java/com/google/cloud/grpc/fallback/GcpFallbackChannel.java index b991234848c1..4d9c7cb40058 100644 --- a/grpc-gcp-java/src/main/java/com/google/cloud/grpc/fallback/GcpFallbackChannel.java +++ b/grpc-gcp-java/src/main/java/com/google/cloud/grpc/fallback/GcpFallbackChannel.java @@ -20,7 +20,6 @@ import static java.util.concurrent.TimeUnit.MILLISECONDS; import static java.util.concurrent.TimeUnit.NANOSECONDS; -import com.google.cloud.grpc.GcpThreadFactory; import com.google.common.annotations.VisibleForTesting; import io.grpc.CallOptions; import io.grpc.Channel; @@ -31,9 +30,10 @@ import io.grpc.ManagedChannelBuilder; import io.grpc.MethodDescriptor; import io.grpc.Status; -import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicLong; import java.util.logging.Logger; import javax.annotation.Nullable; @@ -50,14 +50,19 @@ public class GcpFallbackChannel extends ManagedChannel { private final Channel primaryChannel; // Wrapped fallback channel to be used for RPCs. private final Channel fallbackChannel; - private final AtomicLong primarySuccesses = new AtomicLong(0); - private final AtomicLong primaryFailures = new AtomicLong(0); - private final AtomicLong fallbackSuccesses = new AtomicLong(0); - private final AtomicLong fallbackFailures = new AtomicLong(0); - private boolean inFallbackMode = false; + private final GcpFallbackState fallbackState; + private final boolean ownsFallbackState; private final GcpFallbackOpenTelemetry openTelemetry; + private final AtomicBoolean localInFallbackMode = new AtomicBoolean(false); + private final AtomicLong localProbeSuccesses = new AtomicLong(0); + private final AtomicLong localFirstPrimaryProbeSuccessNanos = new AtomicLong(0); + private final java.util.concurrent.locks.ReentrantLock stateLock = + new java.util.concurrent.locks.ReentrantLock(); + private final ScheduledExecutorService execService; + private volatile ScheduledFuture primaryProbeFuture = null; + private volatile ScheduledFuture fallbackProbeFuture = null; public GcpFallbackChannel( GcpFallbackChannelOptions options, @@ -82,13 +87,18 @@ public GcpFallbackChannel( checkNotNull(options); checkNotNull(primaryChannelBuilder); checkNotNull(fallbackChannelBuilder); - if (execService != null) { - this.execService = execService; + this.options = options; + if (options.getSharedState() != null) { + this.fallbackState = options.getSharedState(); + this.ownsFallbackState = false; } else { - this.execService = - Executors.newScheduledThreadPool(3, GcpThreadFactory.newThreadFactory("gcp-fallback-%d")); + this.fallbackState = + execService != null + ? new GcpFallbackState(execService) + : new GcpFallbackState(); + this.ownsFallbackState = true; } - this.options = options; + this.execService = fallbackState.getOrCreateExecutorService(options); if (options.getGcpOpenTelemetry() != null) { this.openTelemetry = options.getGcpOpenTelemetry(); } else { @@ -149,13 +159,18 @@ public GcpFallbackChannel( checkNotNull(options); checkNotNull(primaryChannel); checkNotNull(fallbackChannel); - if (execService != null) { - this.execService = execService; + this.options = options; + if (options.getSharedState() != null) { + this.fallbackState = options.getSharedState(); + this.ownsFallbackState = false; } else { - this.execService = - Executors.newScheduledThreadPool(3, GcpThreadFactory.newThreadFactory("gcp-fallback-%d")); + this.fallbackState = + execService != null + ? new GcpFallbackState(execService) + : new GcpFallbackState(); + this.ownsFallbackState = true; } - this.options = options; + this.execService = fallbackState.getOrCreateExecutorService(options); if (options.getGcpOpenTelemetry() != null) { this.openTelemetry = options.getGcpOpenTelemetry(); } else { @@ -174,106 +189,154 @@ public GcpFallbackChannel( init(); } + private void syncFallbackModeState(boolean globalFallback) { + if (globalFallback) { + if (!localInFallbackMode.get()) { + stateLock.lock(); + try { + if (localInFallbackMode.compareAndSet(false, true)) { + localProbeSuccesses.set(0); + localFirstPrimaryProbeSuccessNanos.set(0); + } + } finally { + stateLock.unlock(); + } + } + } else if (!options.isEnablePerChannelRecovery()) { + if (localInFallbackMode.get()) { + stateLock.lock(); + try { + if (localInFallbackMode.compareAndSet(true, false)) { + localProbeSuccesses.set(0); + localFirstPrimaryProbeSuccessNanos.set(0); + } + } finally { + stateLock.unlock(); + } + } + } + } + public boolean isInFallbackMode() { - return inFallbackMode || primaryChannel == null; + boolean globalFallback = fallbackState.getInFallbackMode().get(); + syncFallbackModeState(globalFallback); + if (options.isEnablePerChannelRecovery()) { + return (localInFallbackMode.get() && fallbackChannel != null) || primaryChannel == null; + } + return (globalFallback && fallbackChannel != null) || primaryChannel == null; + } + + @VisibleForTesting + GcpFallbackState getFallbackState() { + return fallbackState; + } + + @VisibleForTesting + AtomicBoolean getLocalInFallbackMode() { + return localInFallbackMode; + } + + @VisibleForTesting + AtomicLong getLocalProbeSuccesses() { + return localProbeSuccesses; } private void init() { if (options.getPrimaryProbingFunction() != null) { - execService.scheduleAtFixedRate( - this::probePrimary, - options.getPrimaryProbingInterval().toMillis(), - options.getPrimaryProbingInterval().toMillis(), - MILLISECONDS); + this.primaryProbeFuture = + fallbackState.scheduleTask( + this::probePrimary, + options.getPrimaryProbingInterval().toMillis(), + options.getPrimaryProbingInterval().toMillis(), + MILLISECONDS); } if (options.getFallbackProbingFunction() != null) { - execService.scheduleAtFixedRate( - this::probeFallback, - options.getFallbackProbingInterval().toMillis(), - options.getFallbackProbingInterval().toMillis(), - MILLISECONDS); - } - - if (options.isEnableFallback() - && options.getPeriod() != null - && options.getPeriod().toMillis() > 0) { - execService.scheduleAtFixedRate( - this::checkErrorRates, - options.getPeriod().toMillis(), - options.getPeriod().toMillis(), - MILLISECONDS); + this.fallbackProbeFuture = + fallbackState.scheduleTask( + this::probeFallback, + options.getFallbackProbingInterval().toMillis(), + options.getFallbackProbingInterval().toMillis(), + MILLISECONDS); } - } - private void checkErrorRates() { - long successes = primarySuccesses.getAndSet(0); - long failures = primaryFailures.getAndSet(0); - float errRate = 0f; - if (failures + successes > 0) { - errRate = (float) failures / (failures + successes); - } - // Report primary error rate. - openTelemetry.getModule().reportErrorRate(options.getPrimaryChannelName(), errRate); - - if (!isInFallbackMode() && options.isEnableFallback() && fallbackChannel != null) { - if (failures >= options.getMinFailedCalls() && errRate >= options.getErrorRateThreshold()) { - if (inFallbackMode != true) { - openTelemetry - .getModule() - .reportFallback(options.getPrimaryChannelName(), options.getFallbackChannelName()); - } - inFallbackMode = true; - } - } - successes = fallbackSuccesses.getAndSet(0); - failures = fallbackFailures.getAndSet(0); - errRate = 0f; - if (failures + successes > 0) { - errRate = (float) failures / (failures + successes); - } - // Report fallback error rate. - openTelemetry.getModule().reportErrorRate(options.getFallbackChannelName(), errRate); - - openTelemetry - .getModule() - .reportCurrentChannel(options.getPrimaryChannelName(), inFallbackMode == false); - openTelemetry - .getModule() - .reportCurrentChannel(options.getFallbackChannelName(), inFallbackMode == true); + fallbackState.startPeriodicEvaluation(options); } private void processPrimaryStatusCode(Status.Code statusCode) { if (options.getErroneousStates().contains(statusCode)) { - // Count error. - primaryFailures.incrementAndGet(); + fallbackState.getPrimaryFailures().incrementAndGet(); } else { - // Count success. - primarySuccesses.incrementAndGet(); + fallbackState.getPrimarySuccesses().incrementAndGet(); } - // Report status code. openTelemetry.getModule().reportStatus(options.getPrimaryChannelName(), statusCode); } private void processFallbackStatusCode(Status.Code statusCode) { if (options.getErroneousStates().contains(statusCode)) { - // Count error. - fallbackFailures.incrementAndGet(); + fallbackState.getFallbackFailures().incrementAndGet(); } else { - // Count success. - fallbackSuccesses.incrementAndGet(); + fallbackState.getFallbackSuccesses().incrementAndGet(); } - // Report status code. openTelemetry.getModule().reportStatus(options.getFallbackChannelName(), statusCode); } private void probePrimary() { + boolean globalFallback = fallbackState.getInFallbackMode().get(); + syncFallbackModeState(globalFallback); + boolean inFallback = + options.isEnablePerChannelRecovery() + ? localInFallbackMode.get() + : globalFallback; + if (!inFallback && primaryChannel != null) { + return; + } String result = ""; if (primaryDelegateChannel == null) { result = INIT_FAILURE_REASON; } else { result = options.getPrimaryProbingFunction().apply(primaryDelegateChannel); } + if ("OK".equals(result)) { + stateLock.lock(); + try { + if (localInFallbackMode.get()) { + long nowNanos = System.nanoTime(); + long firstSuccessNanos = + localFirstPrimaryProbeSuccessNanos.updateAndGet(prev -> prev == 0 ? nowNanos : prev); + long primaryProbeSuccessCount = localProbeSuccesses.incrementAndGet(); + + boolean durationSatisfied = true; + if (options.getMinPrimaryProbeSuccessDuration() != null + && !options.getMinPrimaryProbeSuccessDuration().isZero() + && !options.getMinPrimaryProbeSuccessDuration().isNegative()) { + long elapsedNanos = nowNanos - firstSuccessNanos; + durationSatisfied = + elapsedNanos >= options.getMinPrimaryProbeSuccessDuration().toNanos(); + } + + if (options.isEnableRecovery() + && primaryProbeSuccessCount >= options.getMinPrimaryProbeSuccessCount() + && durationSatisfied) { + fallbackState.getInFallbackMode().set(false); + localInFallbackMode.set(false); + localProbeSuccesses.set(0); + localFirstPrimaryProbeSuccessNanos.set(0); + } + } + } finally { + stateLock.unlock(); + } + } else { + stateLock.lock(); + try { + localInFallbackMode.set(true); + localProbeSuccesses.set(0); + localFirstPrimaryProbeSuccessNanos.set(0); + } finally { + stateLock.unlock(); + } + } // Report metric based on result. openTelemetry.getModule().reportProbeResult(options.getPrimaryChannelName(), result); } @@ -308,27 +371,71 @@ public String authority() { return primaryChannel.authority(); } + @Override + public io.grpc.ConnectivityState getState(boolean requestConnection) { + if (isInFallbackMode()) { + if (fallbackDelegateChannel != null) { + return fallbackDelegateChannel.getState(requestConnection); + } + return io.grpc.ConnectivityState.SHUTDOWN; + } + + if (primaryDelegateChannel != null) { + return primaryDelegateChannel.getState(requestConnection); + } + return io.grpc.ConnectivityState.SHUTDOWN; + } + + @Override + public void notifyWhenStateChanged(io.grpc.ConnectivityState source, Runnable callback) { + if (isInFallbackMode()) { + if (fallbackDelegateChannel != null) { + fallbackDelegateChannel.notifyWhenStateChanged(source, callback); + } + } else { + if (primaryDelegateChannel != null) { + primaryDelegateChannel.notifyWhenStateChanged(source, callback); + } + } + } + @Override public ManagedChannel shutdown() { + if (primaryProbeFuture != null) { + primaryProbeFuture.cancel(false); + } + if (fallbackProbeFuture != null) { + fallbackProbeFuture.cancel(false); + } if (primaryDelegateChannel != null) { primaryDelegateChannel.shutdown(); } if (fallbackDelegateChannel != null) { fallbackDelegateChannel.shutdown(); } - execService.shutdown(); + if (ownsFallbackState) { + fallbackState.shutdown(); + } return this; } @Override public ManagedChannel shutdownNow() { + if (primaryProbeFuture != null) { + primaryProbeFuture.cancel(true); + } + if (fallbackProbeFuture != null) { + fallbackProbeFuture.cancel(true); + } if (primaryDelegateChannel != null) { primaryDelegateChannel.shutdownNow(); } if (fallbackDelegateChannel != null) { fallbackDelegateChannel.shutdownNow(); } - execService.shutdownNow(); + if (ownsFallbackState) { + fallbackState.shutdownNow(); + } return this; } @@ -342,7 +449,10 @@ public boolean isShutdown() { return false; } - return execService.isShutdown(); + if (ownsFallbackState && options.getSharedExecutorService() == null) { + return execService.isShutdown(); + } + return true; } @Override @@ -355,7 +465,10 @@ public boolean isTerminated() { return false; } - return execService.isTerminated(); + if (ownsFallbackState && options.getSharedExecutorService() == null) { + return execService.isTerminated(); + } + return true; } @Override @@ -377,6 +490,9 @@ public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedE awaitTimeNanos = endTimeNanos - System.nanoTime(); } - return execService.awaitTermination(awaitTimeNanos, NANOSECONDS); + if (ownsFallbackState && options.getSharedExecutorService() == null) { + return execService.awaitTermination(awaitTimeNanos, NANOSECONDS); + } + return true; } } diff --git a/grpc-gcp-java/src/main/java/com/google/cloud/grpc/fallback/GcpFallbackChannelOptions.java b/grpc-gcp-java/src/main/java/com/google/cloud/grpc/fallback/GcpFallbackChannelOptions.java index 31d5cd981907..0fcc21fd4e34 100644 --- a/grpc-gcp-java/src/main/java/com/google/cloud/grpc/fallback/GcpFallbackChannelOptions.java +++ b/grpc-gcp-java/src/main/java/com/google/cloud/grpc/fallback/GcpFallbackChannelOptions.java @@ -25,6 +25,7 @@ import java.time.Duration; import java.util.EnumSet; import java.util.Set; +import java.util.concurrent.ScheduledExecutorService; import java.util.function.Function; public class GcpFallbackChannelOptions { @@ -37,9 +38,15 @@ public class GcpFallbackChannelOptions { private final Function fallbackProbingFunction; private final Duration primaryProbingInterval; private final Duration fallbackProbingInterval; + private final int minPrimaryProbeSuccessCount; + private final Duration minPrimaryProbeSuccessDuration; + private final boolean enableRecovery; + private final boolean enablePerChannelRecovery; private final String primaryChannelName; private final String fallbackChannelName; private final GcpFallbackOpenTelemetry openTelemetry; + private final ScheduledExecutorService sharedExecutorService; + private final GcpFallbackState sharedState; public GcpFallbackChannelOptions(Builder builder) { this.enableFallback = builder.enableFallback; @@ -51,9 +58,15 @@ public GcpFallbackChannelOptions(Builder builder) { this.fallbackProbingFunction = builder.fallbackProbingFunction; this.primaryProbingInterval = builder.primaryProbingInterval; this.fallbackProbingInterval = builder.fallbackProbingInterval; + this.minPrimaryProbeSuccessCount = builder.minPrimaryProbeSuccessCount; + this.minPrimaryProbeSuccessDuration = builder.minPrimaryProbeSuccessDuration; + this.enableRecovery = builder.enableRecovery; + this.enablePerChannelRecovery = builder.enablePerChannelRecovery; this.primaryChannelName = builder.primaryChannelName; this.fallbackChannelName = builder.fallbackChannelName; this.openTelemetry = builder.openTelemetry; + this.sharedExecutorService = builder.sharedExecutorService; + this.sharedState = builder.sharedState; } public static Builder newBuilder() { @@ -96,6 +109,22 @@ public Duration getFallbackProbingInterval() { return fallbackProbingInterval; } + public int getMinPrimaryProbeSuccessCount() { + return minPrimaryProbeSuccessCount; + } + + public Duration getMinPrimaryProbeSuccessDuration() { + return minPrimaryProbeSuccessDuration; + } + + public boolean isEnableRecovery() { + return enableRecovery; + } + + public boolean isEnablePerChannelRecovery() { + return enablePerChannelRecovery; + } + public String getPrimaryChannelName() { return primaryChannelName; } @@ -108,6 +137,14 @@ public GcpFallbackOpenTelemetry getGcpOpenTelemetry() { return openTelemetry; } + public ScheduledExecutorService getSharedExecutorService() { + return sharedExecutorService; + } + + public GcpFallbackState getSharedState() { + return sharedState; + } + public static class Builder { private boolean enableFallback = true; private float errorRateThreshold = 1f; @@ -122,10 +159,17 @@ public static class Builder { private Duration primaryProbingInterval = Duration.ofMinutes(1); private Duration fallbackProbingInterval = Duration.ofMinutes(15); + private int minPrimaryProbeSuccessCount = 10; + private Duration minPrimaryProbeSuccessDuration = Duration.ZERO; + private boolean enableRecovery = false; + private boolean enablePerChannelRecovery = false; + private String primaryChannelName = "primary"; private String fallbackChannelName = "fallback"; private GcpFallbackOpenTelemetry openTelemetry = null; + private ScheduledExecutorService sharedExecutorService = null; + private GcpFallbackState sharedState = null; public Builder() {} @@ -184,6 +228,26 @@ public Builder setFallbackProbingInterval(Duration fallbackProbingInterval) { return this; } + public Builder setMinPrimaryProbeSuccessCount(int minPrimaryProbeSuccessCount) { + this.minPrimaryProbeSuccessCount = minPrimaryProbeSuccessCount; + return this; + } + + public Builder setMinPrimaryProbeSuccessDuration(Duration minPrimaryProbeSuccessDuration) { + this.minPrimaryProbeSuccessDuration = minPrimaryProbeSuccessDuration; + return this; + } + + public Builder setEnableRecovery(boolean enableRecovery) { + this.enableRecovery = enableRecovery; + return this; + } + + public Builder setEnablePerChannelRecovery(boolean enablePerChannelRecovery) { + this.enablePerChannelRecovery = enablePerChannelRecovery; + return this; + } + public Builder setPrimaryChannelName(String primaryChannelName) { this.primaryChannelName = primaryChannelName; return this; @@ -199,6 +263,16 @@ public Builder setGcpFallbackOpenTelemetry(GcpFallbackOpenTelemetry openTelemetr return this; } + public Builder setSharedExecutorService(ScheduledExecutorService sharedExecutorService) { + this.sharedExecutorService = sharedExecutorService; + return this; + } + + public Builder setSharedState(GcpFallbackState sharedState) { + this.sharedState = sharedState; + return this; + } + public GcpFallbackChannelOptions build() { return new GcpFallbackChannelOptions(this); } diff --git a/grpc-gcp-java/src/main/java/com/google/cloud/grpc/fallback/GcpFallbackState.java b/grpc-gcp-java/src/main/java/com/google/cloud/grpc/fallback/GcpFallbackState.java new file mode 100644 index 000000000000..b6e8fd5099fa --- /dev/null +++ b/grpc-gcp-java/src/main/java/com/google/cloud/grpc/fallback/GcpFallbackState.java @@ -0,0 +1,218 @@ +/* + * Copyright 2026 Google LLC + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package com.google.cloud.grpc.fallback; + +import com.google.cloud.grpc.GcpThreadFactory; +import com.google.common.annotations.VisibleForTesting; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicLong; + +/** + * Shared thread-safe state container for coordinated pool-wide failover, recovery, and background + * tasks. + * + *

All channels in a pool share this state instance and its background executor service, + * consolidating probing and error evaluation threads across the entire channel pool. + */ +public class GcpFallbackState { + private final AtomicLong primarySuccesses = new AtomicLong(0); + private final AtomicLong primaryFailures = new AtomicLong(0); + private final AtomicLong fallbackSuccesses = new AtomicLong(0); + private final AtomicLong fallbackFailures = new AtomicLong(0); + private final AtomicBoolean inFallbackMode = new AtomicBoolean(false); + private final AtomicBoolean evaluationStarted = new AtomicBoolean(false); + + private ScheduledExecutorService execService = null; + private boolean ownsExecutor = false; + private volatile ScheduledFuture scheduledEvaluationFuture = null; + + public GcpFallbackState() {} + + /** + * Constructs a fallback state with an explicit executor service for testing. + * + * @param execService the executor service to use. + */ + @VisibleForTesting + public GcpFallbackState(ScheduledExecutorService execService) { + this.execService = execService; + this.ownsExecutor = true; + } + + public AtomicLong getPrimarySuccesses() { + return primarySuccesses; + } + + public AtomicLong getPrimaryFailures() { + return primaryFailures; + } + + public AtomicLong getFallbackSuccesses() { + return fallbackSuccesses; + } + + public AtomicLong getFallbackFailures() { + return fallbackFailures; + } + + public AtomicBoolean getInFallbackMode() { + return inFallbackMode; + } + + /** + * Retrieves or lazily initializes the background executor service. + * + * @param options optional fallback channel configuration options. + * @return the active ScheduledExecutorService. + */ + public synchronized ScheduledExecutorService getOrCreateExecutorService( + GcpFallbackChannelOptions options) { + if (this.execService != null) { + return this.execService; + } + if (options != null && options.getSharedExecutorService() != null) { + this.execService = options.getSharedExecutorService(); + this.ownsExecutor = false; + } else { + this.execService = + Executors.newScheduledThreadPool( + 3, GcpThreadFactory.newThreadFactory("gcp-fallback-state-%d")); + this.ownsExecutor = true; + } + return this.execService; + } + + /** Schedules a periodic task (e.g., probe) on the shared background executor service. */ + public synchronized ScheduledFuture scheduleTask( + Runnable command, long initialDelay, long period, TimeUnit unit) { + if (this.execService == null || this.execService.isShutdown()) { + return null; + } + return this.execService.scheduleAtFixedRate(command, initialDelay, period, unit); + } + + /** + * Starts the periodic error rate evaluation loop exactly once across all channels sharing this + * state. + * + * @param options the fallback channel configuration options. + */ + public synchronized void startPeriodicEvaluation(GcpFallbackChannelOptions options) { + if (options == null + || !options.isEnableFallback() + || options.getPeriod() == null + || options.getPeriod().toMillis() <= 0) { + return; + } + if (evaluationStarted.compareAndSet(false, true)) { + ScheduledExecutorService executor = getOrCreateExecutorService(options); + if (executor == null || executor.isShutdown()) { + evaluationStarted.set(false); + return; + } + GcpFallbackOpenTelemetry openTelemetry = + options.getGcpOpenTelemetry() != null + ? options.getGcpOpenTelemetry() + : GcpFallbackOpenTelemetry.newBuilder().build(); + + scheduledEvaluationFuture = + executor.scheduleAtFixedRate( + () -> checkErrorRates(options, openTelemetry), + options.getPeriod().toMillis(), + options.getPeriod().toMillis(), + TimeUnit.MILLISECONDS); + } + } + + /** + * Evaluates error rates across all channels sharing this state and updates fallback mode. + * + * @param options the fallback channel configuration options. + * @param openTelemetry telemetry module for recording error metrics. + */ + public void checkErrorRates( + GcpFallbackChannelOptions options, GcpFallbackOpenTelemetry openTelemetry) { + long successes = primarySuccesses.getAndSet(0); + long failures = primaryFailures.getAndSet(0); + float errRate = 0f; + if (failures + successes > 0) { + errRate = (float) failures / (failures + successes); + } + if (openTelemetry != null && openTelemetry.getModule() != null) { + openTelemetry.getModule().reportErrorRate(options.getPrimaryChannelName(), errRate); + } + + if (!inFallbackMode.get() && options.isEnableFallback()) { + if (failures >= options.getMinFailedCalls() && errRate >= options.getErrorRateThreshold()) { + inFallbackMode.set(true); + if (openTelemetry != null && openTelemetry.getModule() != null) { + openTelemetry + .getModule() + .reportFallback(options.getPrimaryChannelName(), options.getFallbackChannelName()); + } + } + } + + successes = fallbackSuccesses.getAndSet(0); + failures = fallbackFailures.getAndSet(0); + errRate = 0f; + if (failures + successes > 0) { + errRate = (float) failures / (failures + successes); + } + if (openTelemetry != null && openTelemetry.getModule() != null) { + openTelemetry.getModule().reportErrorRate(options.getFallbackChannelName(), errRate); + openTelemetry + .getModule() + .reportCurrentChannel(options.getPrimaryChannelName(), !inFallbackMode.get()); + openTelemetry + .getModule() + .reportCurrentChannel(options.getFallbackChannelName(), inFallbackMode.get()); + } + } + + /** Stops any running scheduled evaluation. */ + public synchronized void stopPeriodicEvaluation() { + if (scheduledEvaluationFuture != null) { + scheduledEvaluationFuture.cancel(false); + scheduledEvaluationFuture = null; + } + evaluationStarted.set(false); + } + + /** Shuts down the state, cancelling evaluation and shutting down internal executor if owned. */ + public synchronized void shutdown() { + stopPeriodicEvaluation(); + if (ownsExecutor && execService != null && !execService.isShutdown()) { + execService.shutdown(); + } + } + + /** + * Shuts down the state immediately, cancelling evaluation and terminating internal executor if + * owned. + */ + public synchronized void shutdownNow() { + stopPeriodicEvaluation(); + if (ownsExecutor && execService != null && !execService.isShutdown()) { + execService.shutdownNow(); + } + } +} diff --git a/grpc-gcp-java/src/test/java/com/google/cloud/grpc/fallback/GcpFallbackChannelTest.java b/grpc-gcp-java/src/test/java/com/google/cloud/grpc/fallback/GcpFallbackChannelTest.java index 79ad995121f7..e27bf41d502b 100644 --- a/grpc-gcp-java/src/test/java/com/google/cloud/grpc/fallback/GcpFallbackChannelTest.java +++ b/grpc-gcp-java/src/test/java/com/google/cloud/grpc/fallback/GcpFallbackChannelTest.java @@ -41,6 +41,8 @@ import static org.mockito.ArgumentMatchers.anyLong; import static org.mockito.ArgumentMatchers.eq; import static org.mockito.ArgumentMatchers.isNull; +import static org.mockito.Mockito.atLeast; +import static org.mockito.Mockito.atLeastOnce; import static org.mockito.Mockito.clearInvocations; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.never; @@ -81,6 +83,7 @@ import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicLong; import java.util.function.Function; import javax.annotation.Nonnull; @@ -1081,6 +1084,7 @@ public void testProbingTasksScheduled_ifConfigured() { .build(); initializeChannelAndCaptureTasks(options); + gcpFallbackChannel.getFallbackState().getInFallbackMode().set(true); assertNotNull(primaryProbingTask); assertNotNull(fallbackProbingTask); @@ -1117,6 +1121,7 @@ public void testProbing_reportsMetrics() throws InterruptedException { .build(); initializeChannelAndCaptureTasks(options); + gcpFallbackChannel.getFallbackState().getInFallbackMode().set(true); assertNotNull(primaryProbingTask); assertNotNull(fallbackProbingTask); @@ -1211,6 +1216,7 @@ public void testProbing_reportsInitFailureForFallback() throws InterruptedExcept .build(); initializeChannelWithInvalidFallbackBuilderAndCaptureTasks(options); + gcpFallbackChannel.getFallbackState().getInFallbackMode().set(true); assertNotNull(primaryProbingTask); assertNotNull(fallbackProbingTask); @@ -1285,4 +1291,689 @@ public void testConstructor_failsWhenBothBuildersFail() { new GcpFallbackChannel( getDefaultOptions(), mockPrimaryInvalidBuilder, mockFallbackInvalidBuilder)); } + + @SuppressWarnings({"unchecked"}) + private void simulateCallOnChannel( + GcpFallbackChannel channel, + Status statusToReturn, + ManagedChannel primaryDelegate, + ManagedChannel fallbackDelegate, + ClientCall primaryCall, + ClientCall fallbackCall, + boolean expectFallbackRouting) { + final ClientCall.Listener dummyCallListener = mock(ClientCall.Listener.class); + final Metadata requestHeaders = new Metadata(); + + ClientCall testCall = channel.newCall(methodDescriptor, callOptions); + assertNotNull(testCall); + + ClientCall targetCall; + if (expectFallbackRouting) { + verify(fallbackDelegate).newCall(methodDescriptor, callOptions); + verify(primaryDelegate, never()).newCall(methodDescriptor, callOptions); + targetCall = fallbackCall; + } else { + verify(primaryDelegate).newCall(methodDescriptor, callOptions); + verify(fallbackDelegate, never()).newCall(methodDescriptor, callOptions); + targetCall = primaryCall; + } + + testCall.start(dummyCallListener, requestHeaders); + + ArgumentCaptor> delegateListenerCaptor = + ArgumentCaptor.forClass(ClientCall.Listener.class); + verify(targetCall).start(delegateListenerCaptor.capture(), eq(requestHeaders)); + delegateListenerCaptor.getValue().onClose(statusToReturn, new Metadata()); + + clearInvocations(primaryDelegate, fallbackDelegate, targetCall); + } + + @Test + public void testSharedState_singleEvaluationScheduled() { + ScheduledExecutorService mockExec1 = mock(ScheduledExecutorService.class); + ScheduledExecutorService mockExec2 = mock(ScheduledExecutorService.class); + GcpFallbackState sharedState = new GcpFallbackState(mockExec1); + GcpFallbackChannelOptions options = + getDefaultOptionsBuilder().setSharedState(sharedState).build(); + + GcpFallbackChannel channel1 = + new GcpFallbackChannel(options, mockPrimaryBuilder, mockFallbackBuilder, mockExec1); + GcpFallbackChannel channel2 = + new GcpFallbackChannel(options, mockPrimaryBuilder, mockFallbackBuilder, mockExec2); + + try { + // Periodic evaluation was started on mockExec1 by the shared state + verify(mockExec1) + .scheduleAtFixedRate( + any(Runnable.class), + eq(options.getPeriod().toMillis()), + eq(options.getPeriod().toMillis()), + eq(MILLISECONDS)); + + // Channel 2 sharing the same state did NOT schedule a duplicate evaluation loop + verify(mockExec2, never()) + .scheduleAtFixedRate( + any(Runnable.class), + eq(options.getPeriod().toMillis()), + eq(options.getPeriod().toMillis()), + eq(MILLISECONDS)); + } finally { + channel1.shutdownNow(); + channel2.shutdownNow(); + } + } + + @Test + public void testSharedState_coordinatedFailover() { + ScheduledExecutorService mockExec1 = mock(ScheduledExecutorService.class); + ScheduledExecutorService mockExec2 = mock(ScheduledExecutorService.class); + GcpFallbackState sharedState = new GcpFallbackState(mockExec1); + GcpFallbackChannelOptions options = + getDefaultOptionsBuilder() + .setSharedState(sharedState) + .setMinFailedCalls(3) + .setErrorRateThreshold(0.5f) + .build(); + + ArgumentCaptor taskCaptor = ArgumentCaptor.forClass(Runnable.class); + + GcpFallbackChannel channel1 = + new GcpFallbackChannel(options, mockPrimaryBuilder, mockFallbackBuilder, mockExec1); + GcpFallbackChannel channel2 = + new GcpFallbackChannel(options, mockPrimaryBuilder, mockFallbackBuilder, mockExec2); + + try { + verify(mockExec1) + .scheduleAtFixedRate( + taskCaptor.capture(), + eq(options.getPeriod().toMillis()), + eq(options.getPeriod().toMillis()), + eq(MILLISECONDS)); + Runnable checkErrorRates = taskCaptor.getValue(); + + // Both channels initially in primary mode + assertFalse(channel1.isInFallbackMode()); + assertFalse(channel2.isInFallbackMode()); + + // Channel 1 processes 2 failures, Channel 2 processes 1 failure (total 3 failures on shared + // state) + simulateCallOnChannel( + channel1, + Status.UNAVAILABLE, + mockPrimaryDelegateChannel, + mockFallbackDelegateChannel, + mockPrimaryClientCall, + mockFallbackClientCall, + false); + simulateCallOnChannel( + channel1, + Status.UNAVAILABLE, + mockPrimaryDelegateChannel, + mockFallbackDelegateChannel, + mockPrimaryClientCall, + mockFallbackClientCall, + false); + simulateCallOnChannel( + channel2, + Status.UNAVAILABLE, + mockPrimaryDelegateChannel, + mockFallbackDelegateChannel, + mockPrimaryClientCall, + mockFallbackClientCall, + false); + + // Run checkErrorRates on shared state + checkErrorRates.run(); + + // Both channels must now be in fallback mode + assertTrue(channel1.isInFallbackMode()); + assertTrue(channel2.isInFallbackMode()); + + // Subsequent call on Channel 2 routes to fallback channel + simulateCallOnChannel( + channel2, + Status.OK, + mockPrimaryDelegateChannel, + mockFallbackDelegateChannel, + mockPrimaryClientCall, + mockFallbackClientCall, + true); + } finally { + channel1.shutdownNow(); + channel2.shutdownNow(); + } + } + + @Test + public void testSharedState_channelShutdownLeavesSiblingChannelsFunctional() { + ScheduledExecutorService mockExec1 = mock(ScheduledExecutorService.class); + ScheduledExecutorService mockExec2 = mock(ScheduledExecutorService.class); + GcpFallbackState sharedState = new GcpFallbackState(mockExec1); + GcpFallbackChannelOptions options = + getDefaultOptionsBuilder().setSharedState(sharedState).build(); + + GcpFallbackChannel channel1 = + new GcpFallbackChannel(options, mockPrimaryBuilder, mockFallbackBuilder, mockExec1); + GcpFallbackChannel channel2 = + new GcpFallbackChannel(options, mockPrimaryBuilder, mockFallbackBuilder, mockExec2); + + try { + // Shutting down channel1 does not shut down the shared executor while sibling channels are + // active + channel1.shutdown(); + verify(mockExec1, never()).shutdown(); + + // Channel 2 can still transition and read shared fallback state + sharedState.getInFallbackMode().set(true); + assertTrue(channel2.isInFallbackMode()); + } finally { + channel2.shutdownNow(); + sharedState.shutdown(); + verify(mockExec1).shutdown(); + } + } + + @Test + public void testSharedState_probingRequiresBothCountAndDurationToRecover() + throws InterruptedException { + ScheduledExecutorService mockExec = mock(ScheduledExecutorService.class); + GcpFallbackState sharedState = new GcpFallbackState(mockExec); + sharedState.getInFallbackMode().set(true); + + GcpFallbackChannelOptions options = + getDefaultOptionsBuilder() + .setSharedState(sharedState) + .setEnableRecovery(true) + .setPrimaryProbingFunction(channel -> "OK") + .setMinPrimaryProbeSuccessCount(2) + .setMinPrimaryProbeSuccessDuration(Duration.ofMillis(50)) + .build(); + ArgumentCaptor taskCaptor = ArgumentCaptor.forClass(Runnable.class); + + GcpFallbackChannel channel = + new GcpFallbackChannel(options, mockPrimaryBuilder, mockFallbackBuilder, mockExec); + + try { + verify(mockExec) + .scheduleAtFixedRate( + taskCaptor.capture(), + eq(options.getPrimaryProbingInterval().toMillis()), + eq(options.getPrimaryProbingInterval().toMillis()), + eq(MILLISECONDS)); + Runnable probeTask = taskCaptor.getValue(); + + // Probe 1: Success, but count < 2 and duration not yet met + probeTask.run(); + assertTrue(channel.isInFallbackMode()); + assertEquals(1, channel.getLocalProbeSuccesses().get()); + + // Probe 2 immediately: count == 2, but duration (50ms) not elapsed yet! + probeTask.run(); + assertTrue(channel.isInFallbackMode()); + assertEquals(2, channel.getLocalProbeSuccesses().get()); + + // Wait for duration window to pass + Thread.sleep(60); + + // Probe 3: count >= 2 and duration >= 50ms satisfied -> Recover! + probeTask.run(); + assertFalse(channel.isInFallbackMode()); + assertEquals(0, channel.getLocalProbeSuccesses().get()); + } finally { + channel.shutdownNow(); + sharedState.shutdown(); + } + } + + @Test + public void testSharedState_probingFailureResetsDurationTimer() { + ScheduledExecutorService mockExec = mock(ScheduledExecutorService.class); + GcpFallbackState sharedState = new GcpFallbackState(mockExec); + sharedState.getInFallbackMode().set(true); + + AtomicBoolean probeOk = new AtomicBoolean(true); + GcpFallbackChannelOptions options = + getDefaultOptionsBuilder() + .setSharedState(sharedState) + .setEnableRecovery(true) + .setPrimaryProbingFunction(channel -> probeOk.get() ? "OK" : "UNAVAILABLE") + .setMinPrimaryProbeSuccessCount(5) + .setMinPrimaryProbeSuccessDuration(Duration.ofMinutes(10)) + .build(); + ArgumentCaptor taskCaptor = ArgumentCaptor.forClass(Runnable.class); + + GcpFallbackChannel channel = + new GcpFallbackChannel(options, mockPrimaryBuilder, mockFallbackBuilder, mockExec); + + try { + verify(mockExec) + .scheduleAtFixedRate( + taskCaptor.capture(), + eq(options.getPrimaryProbingInterval().toMillis()), + eq(options.getPrimaryProbingInterval().toMillis()), + eq(MILLISECONDS)); + Runnable probeTask = taskCaptor.getValue(); + + // Successful probe initializes local probe counter + probeTask.run(); + assertEquals(1, channel.getLocalProbeSuccesses().get()); + + // Failing probe resets local probe count to 0 + probeOk.set(false); + probeTask.run(); + assertEquals(0, channel.getLocalProbeSuccesses().get()); + assertTrue(channel.isInFallbackMode()); + } finally { + channel.shutdownNow(); + sharedState.shutdown(); + } + } + + @Test + public void testProbePrimary_skippedWhenNotInFallbackMode() { + ScheduledExecutorService mockExec = mock(ScheduledExecutorService.class); + AtomicLong probeCalls = new AtomicLong(0); + GcpFallbackState sharedState = new GcpFallbackState(mockExec); + sharedState.getInFallbackMode().set(false); + + GcpFallbackChannelOptions options = + getDefaultOptionsBuilder() + .setSharedState(sharedState) + .setPrimaryProbingFunction( + channel -> { + probeCalls.incrementAndGet(); + return "OK"; + }) + .build(); + ArgumentCaptor taskCaptor = ArgumentCaptor.forClass(Runnable.class); + + GcpFallbackChannel channel = + new GcpFallbackChannel(options, mockPrimaryBuilder, mockFallbackBuilder, mockExec); + + try { + verify(mockExec) + .scheduleAtFixedRate( + taskCaptor.capture(), + eq(options.getPrimaryProbingInterval().toMillis()), + eq(options.getPrimaryProbingInterval().toMillis()), + eq(MILLISECONDS)); + Runnable probeTask = taskCaptor.getValue(); + + // Channel is NOT in fallback mode -> probe task should immediately return without probing! + assertFalse(channel.isInFallbackMode()); + probeTask.run(); + assertEquals(0, probeCalls.get()); + assertEquals(0, channel.getLocalProbeSuccesses().get()); + } finally { + channel.shutdownNow(); + sharedState.shutdown(); + } + } + + @Test + public void testPerChannelIndependentRecovery_oneChannelRecoversWhileOtherStaysInFallback() { + ScheduledExecutorService mockExec1 = mock(ScheduledExecutorService.class); + ScheduledExecutorService mockExec2 = mock(ScheduledExecutorService.class); + GcpFallbackState sharedState = new GcpFallbackState(mockExec1); + sharedState.getInFallbackMode().set(true); // Pool-wide fallback active + + GcpFallbackChannelOptions options1 = + getDefaultOptionsBuilder() + .setSharedState(sharedState) + .setEnableRecovery(true) + .setEnablePerChannelRecovery(true) + .setPrimaryProbingFunction(channel -> "OK") + .setMinPrimaryProbeSuccessCount(1) + .setMinPrimaryProbeSuccessDuration(Duration.ZERO) + .build(); + + GcpFallbackChannelOptions options2 = + getDefaultOptionsBuilder() + .setSharedState(sharedState) + .setEnableRecovery(true) + .setEnablePerChannelRecovery(true) + .setPrimaryProbingFunction(channel -> "UNAVAILABLE") + .setMinPrimaryProbeSuccessCount(1) + .setMinPrimaryProbeSuccessDuration(Duration.ZERO) + .build(); + ArgumentCaptor taskCaptor1 = ArgumentCaptor.forClass(Runnable.class); + + GcpFallbackChannel channel1 = + new GcpFallbackChannel(options1, mockPrimaryBuilder, mockFallbackBuilder, mockExec1); + GcpFallbackChannel channel2 = + new GcpFallbackChannel(options2, mockPrimaryBuilder, mockFallbackBuilder, mockExec2); + + try { + verify(mockExec1, atLeastOnce()) + .scheduleAtFixedRate( + taskCaptor1.capture(), + eq(options1.getPrimaryProbingInterval().toMillis()), + eq(options1.getPrimaryProbingInterval().toMillis()), + eq(MILLISECONDS)); + Runnable probeTask1 = taskCaptor1.getAllValues().get(0); + + assertTrue(channel1.isInFallbackMode()); + assertTrue(channel2.isInFallbackMode()); + + // Run probe on channel 1 -> channel 1 recovers to DirectPath + probeTask1.run(); + + assertFalse("Channel 1 should recover to DirectPath", channel1.isInFallbackMode()); + assertTrue("Channel 2 should remain in CloudPath fallback mode", channel2.isInFallbackMode()); + assertFalse("Global fallback should be unlatched", sharedState.getInFallbackMode().get()); + } finally { + channel1.shutdownNow(); + channel2.shutdownNow(); + sharedState.shutdown(); + } + } + + @Test + public void testPoolLevelRecovery_whenPerChannelRecoveryDisabled_allChannelsRecoverTogether() { + ScheduledExecutorService mockExec1 = mock(ScheduledExecutorService.class); + ScheduledExecutorService mockExec2 = mock(ScheduledExecutorService.class); + GcpFallbackState sharedState = new GcpFallbackState(mockExec1); + sharedState.getInFallbackMode().set(true); // Pool-wide fallback active + + GcpFallbackChannelOptions options1 = + getDefaultOptionsBuilder() + .setSharedState(sharedState) + .setEnableRecovery(true) + .setEnablePerChannelRecovery(false) // Pool-level recovery + .setPrimaryProbingFunction(channel -> "OK") + .setMinPrimaryProbeSuccessCount(1) + .setMinPrimaryProbeSuccessDuration(Duration.ZERO) + .build(); + + GcpFallbackChannelOptions options2 = + getDefaultOptionsBuilder() + .setSharedState(sharedState) + .setEnableRecovery(true) + .setEnablePerChannelRecovery(false) // Pool-level recovery + .setPrimaryProbingFunction(channel -> "UNAVAILABLE") + .setMinPrimaryProbeSuccessCount(1) + .setMinPrimaryProbeSuccessDuration(Duration.ZERO) + .build(); + ArgumentCaptor taskCaptor1 = ArgumentCaptor.forClass(Runnable.class); + + GcpFallbackChannel channel1 = + new GcpFallbackChannel(options1, mockPrimaryBuilder, mockFallbackBuilder, mockExec1); + GcpFallbackChannel channel2 = + new GcpFallbackChannel(options2, mockPrimaryBuilder, mockFallbackBuilder, mockExec2); + + try { + verify(mockExec1, atLeastOnce()) + .scheduleAtFixedRate( + taskCaptor1.capture(), + eq(options1.getPrimaryProbingInterval().toMillis()), + eq(options1.getPrimaryProbingInterval().toMillis()), + eq(MILLISECONDS)); + Runnable probeTask1 = taskCaptor1.getAllValues().get(0); + + assertTrue(channel1.isInFallbackMode()); + assertTrue(channel2.isInFallbackMode()); + + // Run probe on channel 1 -> channel 1 recovers to DirectPath and unlatches global fallback + probeTask1.run(); + + // With enablePerChannelRecovery=false, both channel 1 and channel 2 recover to DirectPath + // together + assertFalse("Channel 1 should recover to DirectPath", channel1.isInFallbackMode()); + assertFalse( + "Channel 2 should also recover to DirectPath with pool", channel2.isInFallbackMode()); + assertFalse("Global fallback should be unlatched", sharedState.getInFallbackMode().get()); + } finally { + channel1.shutdownNow(); + channel2.shutdownNow(); + sharedState.shutdown(); + } + } + + @Test + public void testRecoveryDisabled_probingSucceedsButChannelRemainsInFallback() { + ScheduledExecutorService mockExec = mock(ScheduledExecutorService.class); + GcpFallbackState sharedState = new GcpFallbackState(mockExec); + sharedState.getInFallbackMode().set(true); // Pool-wide fallback active + + GcpFallbackChannelOptions options = + getDefaultOptionsBuilder() + .setSharedState(sharedState) + .setEnableRecovery(false) // Recovery disabled by default + .setPrimaryProbingFunction(channel -> "OK") + .setMinPrimaryProbeSuccessCount(1) + .setMinPrimaryProbeSuccessDuration(Duration.ZERO) + .build(); + ArgumentCaptor taskCaptor = ArgumentCaptor.forClass(Runnable.class); + + GcpFallbackChannel channel = + new GcpFallbackChannel(options, mockPrimaryBuilder, mockFallbackBuilder, mockExec); + + try { + verify(mockExec) + .scheduleAtFixedRate( + taskCaptor.capture(), + eq(options.getPrimaryProbingInterval().toMillis()), + eq(options.getPrimaryProbingInterval().toMillis()), + eq(MILLISECONDS)); + Runnable probeTask = taskCaptor.getValue(); + + assertTrue(channel.isInFallbackMode()); + + // Run probe -> probe succeeds, but enableRecovery is false + probeTask.run(); + + // Channel must remain in fallback mode + assertTrue(channel.isInFallbackMode()); + assertTrue(sharedState.getInFallbackMode().get()); + } finally { + channel.shutdownNow(); + sharedState.shutdown(); + } + } + + @Test + public void testPoolLevelRecovery_multipleFailoverCyclesResetProbeStatistics() { + ScheduledExecutorService mockExec = mock(ScheduledExecutorService.class); + GcpFallbackState sharedState = new GcpFallbackState(mockExec); + sharedState.getInFallbackMode().set(true); // Cycle 1: Fallback active + + GcpFallbackChannelOptions options1 = + getDefaultOptionsBuilder() + .setSharedState(sharedState) + .setEnableRecovery(true) + .setEnablePerChannelRecovery(false) + .setPrimaryProbingFunction(channel -> "OK") + .setMinPrimaryProbeSuccessCount(2) + .setMinPrimaryProbeSuccessDuration(Duration.ZERO) + .build(); + + GcpFallbackChannelOptions options2 = + getDefaultOptionsBuilder() + .setSharedState(sharedState) + .setEnableRecovery(true) + .setEnablePerChannelRecovery(false) + .setPrimaryProbingFunction(channel -> "OK") + .setMinPrimaryProbeSuccessCount(2) + .setMinPrimaryProbeSuccessDuration(Duration.ZERO) + .build(); + ArgumentCaptor taskCaptor = ArgumentCaptor.forClass(Runnable.class); + + GcpFallbackChannel channel1 = + new GcpFallbackChannel(options1, mockPrimaryBuilder, mockFallbackBuilder, mockExec); + GcpFallbackChannel channel2 = + new GcpFallbackChannel(options2, mockPrimaryBuilder, mockFallbackBuilder, mockExec); + + try { + verify(mockExec, atLeast(2)) + .scheduleAtFixedRate( + taskCaptor.capture(), + eq(options1.getPrimaryProbingInterval().toMillis()), + eq(options1.getPrimaryProbingInterval().toMillis()), + eq(MILLISECONDS)); + Runnable probeTask1 = taskCaptor.getAllValues().get(0); + Runnable probeTask2 = taskCaptor.getAllValues().get(1); + + // Both channels enter fallback + assertTrue(channel1.isInFallbackMode()); + assertTrue(channel2.isInFallbackMode()); + + // Channel 1 probes twice -> recovers pool + probeTask1.run(); + probeTask1.run(); + assertFalse(channel1.isInFallbackMode()); + assertFalse(channel2.isInFallbackMode()); + + // Now Cycle 2: Incident occurs again, pool enters fallback + sharedState.getInFallbackMode().set(true); + assertTrue(channel2.isInFallbackMode()); + + // Channel 2's localProbeSuccesses must be reset to 0 in new cycle + assertEquals(0, channel2.getLocalProbeSuccesses().get()); + + // Probe once: count is 1 (< 2 required), must still be in fallback + probeTask2.run(); + assertEquals(1, channel2.getLocalProbeSuccesses().get()); + assertTrue(channel2.isInFallbackMode()); + + // Probe second time: count is 2 (>= 2 required) -> recovers! + probeTask2.run(); + assertFalse(channel2.isInFallbackMode()); + assertFalse(sharedState.getInFallbackMode().get()); + } finally { + channel1.shutdownNow(); + channel2.shutdownNow(); + sharedState.shutdown(); + } + } + + @Test + public void testConcurrentIsInFallbackModeDoesNotResetProbeSuccesses() + throws InterruptedException, java.util.concurrent.ExecutionException { + ScheduledExecutorService mockExec = mock(ScheduledExecutorService.class); + GcpFallbackState sharedState = new GcpFallbackState(mockExec); + sharedState.getInFallbackMode().set(true); + + GcpFallbackChannelOptions options = + getDefaultOptionsBuilder() + .setSharedState(sharedState) + .setEnableRecovery(true) + .setEnablePerChannelRecovery(true) + .setPrimaryProbingFunction(channel -> "OK") + .setMinPrimaryProbeSuccessCount(1000) + .setMinPrimaryProbeSuccessDuration(Duration.ofHours(1)) + .build(); + + ArgumentCaptor taskCaptor = ArgumentCaptor.forClass(Runnable.class); + + GcpFallbackChannel channel = + new GcpFallbackChannel(options, mockPrimaryBuilder, mockFallbackBuilder, mockExec); + + try { + verify(mockExec) + .scheduleAtFixedRate( + taskCaptor.capture(), + eq(options.getPrimaryProbingInterval().toMillis()), + eq(options.getPrimaryProbingInterval().toMillis()), + eq(MILLISECONDS)); + Runnable probeTask = taskCaptor.getValue(); + + assertTrue(channel.isInFallbackMode()); + probeTask.run(); + assertEquals(1, channel.getLocalProbeSuccesses().get()); + + int threadCount = 50; + int iterationsPerThread = 200; + java.util.concurrent.ExecutorService threadPool = + java.util.concurrent.Executors.newFixedThreadPool(threadCount); + java.util.concurrent.CountDownLatch startLatch = + new java.util.concurrent.CountDownLatch(1); + java.util.List> futures = new java.util.ArrayList<>(); + + for (int i = 0; i < threadCount; i++) { + futures.add( + threadPool.submit( + () -> { + try { + startLatch.await(); + for (int j = 0; j < iterationsPerThread; j++) { + assertTrue(channel.isInFallbackMode()); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + })); + } + + startLatch.countDown(); + // Run probe while concurrent threads call isInFallbackMode + for (int i = 0; i < 9; i++) { + probeTask.run(); + } + + for (java.util.concurrent.Future future : futures) { + future.get(); + } + threadPool.shutdown(); + + // Ensure that concurrent callers did not reset probe count back to 0 + assertEquals(10, channel.getLocalProbeSuccesses().get()); + assertTrue(channel.isInFallbackMode()); + } finally { + channel.shutdownNow(); + sharedState.shutdown(); + } + } + + @Test + public void testShutdown_whenSuppliedSharedExecutorService_leavesExecutorRunning() { + ScheduledExecutorService sharedExec = mock(ScheduledExecutorService.class); + GcpFallbackChannelOptions options = + getDefaultOptionsBuilder() + .setSharedExecutorService(sharedExec) + .build(); + + GcpFallbackChannel channel = + new GcpFallbackChannel(options, mockPrimaryBuilder, mockFallbackBuilder); + + try { + channel.shutdown(); + verify(sharedExec, never()).shutdown(); + } finally { + channel.shutdownNow(); + verify(sharedExec, never()).shutdownNow(); + } + } + + @Test + public void testSharedState_channelLifecycleIndependentOfSharedExecutor() + throws InterruptedException { + ScheduledExecutorService mockExec = mock(ScheduledExecutorService.class); + GcpFallbackState sharedState = new GcpFallbackState(mockExec); + GcpFallbackChannelOptions options = + getDefaultOptionsBuilder().setSharedState(sharedState).build(); + + when(mockPrimaryDelegateChannel.awaitTermination(anyLong(), any(TimeUnit.class))) + .thenReturn(true); + when(mockFallbackDelegateChannel.awaitTermination(anyLong(), any(TimeUnit.class))) + .thenReturn(true); + when(mockPrimaryDelegateChannel.isShutdown()).thenReturn(true); + when(mockFallbackDelegateChannel.isShutdown()).thenReturn(true); + when(mockPrimaryDelegateChannel.isTerminated()).thenReturn(true); + when(mockFallbackDelegateChannel.isTerminated()).thenReturn(true); + + GcpFallbackChannel channel = + new GcpFallbackChannel(options, mockPrimaryBuilder, mockFallbackBuilder, mockExec); + + try { + channel.shutdown(); + assertTrue(channel.isShutdown()); + assertTrue(channel.isTerminated()); + assertTrue(channel.awaitTermination(1, TimeUnit.SECONDS)); + // Sibling channels / shared state keep executor alive + verify(mockExec, never()).shutdown(); + verify(mockExec, never()).awaitTermination(anyLong(), any(TimeUnit.class)); + } finally { + sharedState.shutdown(); + } + } }