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..da84f3cdd741 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,8 +30,8 @@ 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.AtomicLong; import java.util.logging.Logger; @@ -50,14 +49,17 @@ 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 AtomicLong localProbeGeneration = new AtomicLong(0); + private final AtomicLong localProbeSuccesses = new AtomicLong(0); + private final AtomicLong localFirstPrimaryProbeSuccessNanos = new AtomicLong(0); + private final ScheduledExecutorService execService; + private volatile ScheduledFuture primaryProbeFuture = null; + private volatile ScheduledFuture fallbackProbeFuture = null; public GcpFallbackChannel( GcpFallbackChannelOptions options, @@ -82,13 +84,16 @@ 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 +154,16 @@ 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 +182,108 @@ public GcpFallbackChannel( init(); } + private void resetProbeStatsIfGenerationChanged() { + long currentGen = fallbackState.getGeneration(); + if (localProbeGeneration.getAndSet(currentGen) != currentGen) { + localProbeSuccesses.set(0); + localFirstPrimaryProbeSuccessNanos.set(0); + } + } + public boolean isInFallbackMode() { - return inFallbackMode || primaryChannel == null; + return (fallbackState.isInFallbackMode() && fallbackChannel != null) || primaryChannel == null; + } + + @VisibleForTesting + GcpFallbackState getFallbackState() { + return fallbackState; + } + + @VisibleForTesting + AtomicLong getLocalProbeSuccesses() { + resetProbeStatsIfGenerationChanged(); + return localProbeSuccesses; } private void init() { + localProbeGeneration.set(fallbackState.getGeneration()); 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(); - } else { - // Count success. - primarySuccesses.incrementAndGet(); + if (fallbackChannel != null && !fallbackState.isInFallbackMode()) { + if (options.getErroneousStates().contains(statusCode)) { + fallbackState.getPrimaryFailures().incrementAndGet(); + } else { + 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() { + if (!fallbackState.isInFallbackMode() && primaryChannel != null) { + return; + } + resetProbeStatsIfGenerationChanged(); + long probeStartGen = localProbeGeneration.get(); String result = ""; if (primaryDelegateChannel == null) { result = INIT_FAILURE_REASON; } else { result = options.getPrimaryProbingFunction().apply(primaryDelegateChannel); } + if ("".equals(result) && fallbackState.getGeneration() == probeStartGen) { + if (options.isEnableRecovery() && fallbackChannel != null) { + 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 (primaryProbeSuccessCount >= options.getMinPrimaryProbeSuccessCount() + && durationSatisfied) { + fallbackState.recordRecovery(probeStartGen); + } + } + } else { + localProbeSuccesses.set(0); + localFirstPrimaryProbeSuccessNanos.set(0); + } // Report metric based on result. openTelemetry.getModule().reportProbeResult(options.getPrimaryChannelName(), result); } @@ -308,27 +318,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 +396,10 @@ public boolean isShutdown() { return false; } - return execService.isShutdown(); + if (ownsFallbackState && options.getSharedExecutorService() == null) { + return execService.isShutdown(); + } + return true; } @Override @@ -355,7 +412,10 @@ public boolean isTerminated() { return false; } - return execService.isTerminated(); + if (ownsFallbackState && options.getSharedExecutorService() == null) { + return execService.isTerminated(); + } + return true; } @Override @@ -377,6 +437,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..8232c0782835 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,14 @@ 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 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 +57,14 @@ 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.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 +107,18 @@ public Duration getFallbackProbingInterval() { return fallbackProbingInterval; } + public int getMinPrimaryProbeSuccessCount() { + return minPrimaryProbeSuccessCount; + } + + public Duration getMinPrimaryProbeSuccessDuration() { + return minPrimaryProbeSuccessDuration; + } + + public boolean isEnableRecovery() { + return enableRecovery; + } + public String getPrimaryChannelName() { return primaryChannelName; } @@ -108,6 +131,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 +153,16 @@ 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 String primaryChannelName = "primary"; private String fallbackChannelName = "fallback"; private GcpFallbackOpenTelemetry openTelemetry = null; + private ScheduledExecutorService sharedExecutorService = null; + private GcpFallbackState sharedState = null; public Builder() {} @@ -184,6 +221,21 @@ 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 setPrimaryChannelName(String primaryChannelName) { this.primaryChannelName = primaryChannelName; return this; @@ -199,6 +251,20 @@ public Builder setGcpFallbackOpenTelemetry(GcpFallbackOpenTelemetry openTelemetr return this; } + public Builder setSharedExecutorService(ScheduledExecutorService sharedExecutorService) { + this.sharedExecutorService = sharedExecutorService; + return this; + } + + /** + * Sets the shared fallback state across channels in a pool. Channels sharing this state should + * use consistent fallback evaluation options. + */ + 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..ed87db98914a --- /dev/null +++ b/grpc-gcp-java/src/main/java/com/google/cloud/grpc/fallback/GcpFallbackState.java @@ -0,0 +1,262 @@ +/* + * 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 that coordinates failover, recovery, and periodic error evaluation + * across a pool of GcpFallbackChannel instances. + */ +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 AtomicLong generation = 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 boolean isShutdown = 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 + GcpFallbackState(ScheduledExecutorService execService) { + this.execService = execService; + this.ownsExecutor = true; + } + + AtomicLong getPrimarySuccesses() { + return primarySuccesses; + } + + AtomicLong getPrimaryFailures() { + return primaryFailures; + } + + AtomicLong getFallbackSuccesses() { + return fallbackSuccesses; + } + + AtomicLong getFallbackFailures() { + return fallbackFailures; + } + + /** Returns whether the pool is currently in fallback mode. */ + boolean isInFallbackMode() { + return inFallbackMode.get(); + } + + long getGeneration() { + return generation.get(); + } + + /** Bumps the generation counter and transitions the pool to fallback mode. */ + synchronized void triggerFallback() { + inFallbackMode.set(true); + generation.incrementAndGet(); + } + + /** + * Records pool recovery by clearing primary error counts and updating fallback mode. + * + * @param expectedGen the generation at which the recovery probe started. + * @return the resulting pool generation, or -1 if the pool generation changed concurrently. + */ + synchronized long recordRecovery(long expectedGen) { + if (generation.get() != expectedGen) { + return -1; + } + if (inFallbackMode.compareAndSet(true, false)) { + primaryFailures.set(0); + primarySuccesses.set(0); + generation.incrementAndGet(); + } + return generation.get(); + } + + /** + * Retrieves or lazily initializes the background executor service. + * + * @param options optional fallback channel configuration options. + * @return the active ScheduledExecutorService. + */ + 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. */ + synchronized ScheduledFuture scheduleTask( + Runnable command, long initialDelay, long period, TimeUnit unit) { + if (isShutdown || 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. Channels sharing this state should use consistent evaluation options, as the first + * channel to start evaluation configures the shared loop. + * + * @param options the fallback channel configuration options. + */ + synchronized void startPeriodicEvaluation(GcpFallbackChannelOptions options) { + if (isShutdown + || 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(); + + try { + scheduledEvaluationFuture = + executor.scheduleAtFixedRate( + () -> checkErrorRates(options, openTelemetry), + options.getPeriod().toMillis(), + options.getPeriod().toMillis(), + TimeUnit.MILLISECONDS); + } catch (RuntimeException e) { + evaluationStarted.set(false); + throw e; + } + } + } + + /** + * 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. + */ + void checkErrorRates(GcpFallbackChannelOptions options, GcpFallbackOpenTelemetry openTelemetry) { + float primaryErrRate = 0f; + boolean fallbackTriggered = false; + boolean currentInFallback; + synchronized (this) { + boolean wasInFallback = inFallbackMode.get(); + long successes = primarySuccesses.getAndSet(0); + long failures = primaryFailures.getAndSet(0); + if (failures + successes > 0) { + primaryErrRate = (float) failures / (failures + successes); + } + if (!wasInFallback && options.isEnableFallback()) { + if (failures >= options.getMinFailedCalls() + && primaryErrRate >= options.getErrorRateThreshold()) { + triggerFallback(); + fallbackTriggered = true; + } + } + currentInFallback = inFallbackMode.get(); + } + + if (openTelemetry != null && openTelemetry.getModule() != null) { + openTelemetry.getModule().reportErrorRate(options.getPrimaryChannelName(), primaryErrRate); + if (fallbackTriggered) { + openTelemetry + .getModule() + .reportFallback(options.getPrimaryChannelName(), options.getFallbackChannelName()); + } + } + + long fallbackSucc = fallbackSuccesses.getAndSet(0); + long fallbackFail = fallbackFailures.getAndSet(0); + float fallbackErrRate = 0f; + if (fallbackFail + fallbackSucc > 0) { + fallbackErrRate = (float) fallbackFail / (fallbackFail + fallbackSucc); + } + if (openTelemetry != null && openTelemetry.getModule() != null) { + openTelemetry.getModule().reportErrorRate(options.getFallbackChannelName(), fallbackErrRate); + openTelemetry + .getModule() + .reportCurrentChannel(options.getPrimaryChannelName(), !currentInFallback); + openTelemetry + .getModule() + .reportCurrentChannel(options.getFallbackChannelName(), currentInFallback); + } + } + + /** Stops any running scheduled evaluation. */ + 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() { + isShutdown = true; + 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() { + isShutdown = true; + 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..26e985323b1e 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; @@ -718,8 +721,13 @@ public void testNoFallback_minCallsNotMet() { @Test public void testNoFallback_fallbackChannelBuildFails() { + GcpFallbackState sharedState = new GcpFallbackState(mockScheduledExecutorService); GcpFallbackChannelOptions options = - getDefaultOptionsBuilder().setMinFailedCalls(1).setErrorRateThreshold(0.1f).build(); + getDefaultOptionsBuilder() + .setMinFailedCalls(1) + .setErrorRateThreshold(0.1f) + .setSharedState(sharedState) + .build(); initializeChannelWithInvalidFallbackBuilderAndCaptureTasks(options); assertFalse("Should not be in fallback mode initially.", gcpFallbackChannel.isInFallbackMode()); @@ -735,6 +743,10 @@ public void testNoFallback_fallbackChannelBuildFails() { checkErrorRatesTask.run(); assertFalse("Should not be in fallback mode.", gcpFallbackChannel.isInFallbackMode()); + assertFalse( + "Shared state should not trip fallback when fallback channel is null.", + sharedState.isInFallbackMode()); + assertEquals(0L, sharedState.getGeneration()); assertEquals(primaryAuthority, gcpFallbackChannel.authority()); } @@ -1081,6 +1093,7 @@ public void testProbingTasksScheduled_ifConfigured() { .build(); initializeChannelAndCaptureTasks(options); + gcpFallbackChannel.getFallbackState().triggerFallback(); assertNotNull(primaryProbingTask); assertNotNull(fallbackProbingTask); @@ -1117,6 +1130,7 @@ public void testProbing_reportsMetrics() throws InterruptedException { .build(); initializeChannelAndCaptureTasks(options); + gcpFallbackChannel.getFallbackState().triggerFallback(); assertNotNull(primaryProbingTask); assertNotNull(fallbackProbingTask); @@ -1211,6 +1225,7 @@ public void testProbing_reportsInitFailureForFallback() throws InterruptedExcept .build(); initializeChannelWithInvalidFallbackBuilderAndCaptureTasks(options); + gcpFallbackChannel.getFallbackState().triggerFallback(); assertNotNull(primaryProbingTask); assertNotNull(fallbackProbingTask); @@ -1285,4 +1300,645 @@ 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.triggerFallback(); + 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.triggerFallback(); + + GcpFallbackChannelOptions options = + getDefaultOptionsBuilder() + .setSharedState(sharedState) + .setEnableRecovery(true) + .setPrimaryProbingFunction(channel -> "") + .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.triggerFallback(); + + AtomicBoolean probeOk = new AtomicBoolean(true); + GcpFallbackChannelOptions options = + getDefaultOptionsBuilder() + .setSharedState(sharedState) + .setEnableRecovery(true) + .setPrimaryProbingFunction(channel -> probeOk.get() ? "" : "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); + + GcpFallbackChannelOptions options = + getDefaultOptionsBuilder() + .setSharedState(sharedState) + .setPrimaryProbingFunction( + channel -> { + probeCalls.incrementAndGet(); + return ""; + }) + .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 testPoolLevelRecovery_allChannelsRecoverTogether() { + ScheduledExecutorService mockExec1 = mock(ScheduledExecutorService.class); + ScheduledExecutorService mockExec2 = mock(ScheduledExecutorService.class); + GcpFallbackState sharedState = new GcpFallbackState(mockExec1); + sharedState.triggerFallback(); // Pool-wide fallback active + + GcpFallbackChannelOptions options1 = + getDefaultOptionsBuilder() + .setSharedState(sharedState) + .setEnableRecovery(true) + .setPrimaryProbingFunction(channel -> "") + .setMinPrimaryProbeSuccessCount(1) + .setMinPrimaryProbeSuccessDuration(Duration.ZERO) + .build(); + + GcpFallbackChannelOptions options2 = + getDefaultOptionsBuilder() + .setSharedState(sharedState) + .setEnableRecovery(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 and unlatches global fallback + probeTask1.run(); + + // 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.isInFallbackMode()); + } finally { + channel1.shutdownNow(); + channel2.shutdownNow(); + sharedState.shutdown(); + } + } + + @Test + public void testRecoveryDisabled_probingSucceedsButChannelRemainsInFallback() { + ScheduledExecutorService mockExec = mock(ScheduledExecutorService.class); + GcpFallbackState sharedState = new GcpFallbackState(mockExec); + sharedState.triggerFallback(); // Pool-wide fallback active + + GcpFallbackChannelOptions options = + getDefaultOptionsBuilder() + .setSharedState(sharedState) + .setEnableRecovery(false) // Recovery disabled by default + .setPrimaryProbingFunction(channel -> "") + .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.isInFallbackMode()); + } finally { + channel.shutdownNow(); + sharedState.shutdown(); + } + } + + @Test + public void testPoolLevelRecovery_multipleFailoverCyclesResetProbeStatistics() { + ScheduledExecutorService mockExec = mock(ScheduledExecutorService.class); + GcpFallbackState sharedState = new GcpFallbackState(mockExec); + sharedState.triggerFallback(); // Cycle 1: Fallback active + + GcpFallbackChannelOptions options1 = + getDefaultOptionsBuilder() + .setSharedState(sharedState) + .setEnableRecovery(true) + .setPrimaryProbingFunction(channel -> "") + .setMinPrimaryProbeSuccessCount(2) + .setMinPrimaryProbeSuccessDuration(Duration.ZERO) + .build(); + + GcpFallbackChannelOptions options2 = + getDefaultOptionsBuilder() + .setSharedState(sharedState) + .setEnableRecovery(true) + .setPrimaryProbingFunction(channel -> "") + .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.triggerFallback(); + 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.isInFallbackMode()); + } 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.triggerFallback(); + + GcpFallbackChannelOptions options = + getDefaultOptionsBuilder() + .setSharedState(sharedState) + .setEnableRecovery(true) + .setPrimaryProbingFunction(channel -> "") + .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(); + } + } + + @Test + public void testSharedState_shutdownPreventsFurtherTaskSchedulingWithExternalExecutor() { + ScheduledExecutorService externalExec = mock(ScheduledExecutorService.class); + GcpFallbackState sharedState = new GcpFallbackState(); + GcpFallbackChannelOptions options = + getDefaultOptionsBuilder() + .setSharedState(sharedState) + .setSharedExecutorService(externalExec) + .build(); + + // Initialize executor via options + sharedState.getOrCreateExecutorService(options); + + // Shutdown the shared state (external executor is NOT shut down since it's not owned) + sharedState.shutdown(); + verify(externalExec, never()).shutdown(); + + // Subsequent task scheduling or periodic evaluation on the shut-down state must be rejected + assertNull(sharedState.scheduleTask(() -> {}, 1, 1, TimeUnit.SECONDS)); + sharedState.startPeriodicEvaluation(options); + verify(externalExec, never()) + .scheduleAtFixedRate(any(Runnable.class), anyLong(), anyLong(), any(TimeUnit.class)); + } }