From d2bcd3b3ad44301e943d6a9b9b337244f3a48acd Mon Sep 17 00:00:00 2001 From: Suresh Prakash Date: Fri, 23 Oct 2026 15:26:25 +0530 Subject: [PATCH 1/9] Plan registry decorators for deadlines and service circuit breakers. Co-authored-by: Cursor --- .gitignore | 2 + .../2026-10-23-grpc-registry-decorators.md | 332 ++++++++++++++++++ 2 files changed, 334 insertions(+) create mode 100644 docs/superpowers/plans/2026-10-23-grpc-registry-decorators.md diff --git a/.gitignore b/.gitignore index 1f93dfc..fdfba38 100644 --- a/.gitignore +++ b/.gitignore @@ -5,3 +5,5 @@ build .idea + +.superpowers/ diff --git a/docs/superpowers/plans/2026-10-23-grpc-registry-decorators.md b/docs/superpowers/plans/2026-10-23-grpc-registry-decorators.md new file mode 100644 index 0000000..05c7834 --- /dev/null +++ b/docs/superpowers/plans/2026-10-23-grpc-registry-decorators.md @@ -0,0 +1,332 @@ +# gRPC Registry Decorators Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Publish a registry decorator that installs a default deadline, and a service-keyed circuit breaker that clients can install either as that decorator or as a standalone interceptor. + +**Architecture:** `grpc-client-utils` owns `InterceptingChannelRegistry`, a decorator over an already-built `GrpcChannelRegistry`, and `TimeoutChannelRegistry`, which opts in through `grpc.client.timeout`. `grpc-circuitbreaker-utils` owns the service-keyed interceptor and `CircuitBreakerChannelRegistry`, which uses the same decorator. The existing request-keyed interceptor is unchanged. + +**Tech Stack:** Java, Gradle, JUnit 5, Mockito, gRPC, Typesafe Config, Resilience4j 1.7.1 (implementation scope only). + +## Global Constraints + +- Work only in this `java-grpc-utils` worktree. Do not edit `activity-event-service`. Do not push. +- Do not change request-keyed breaker behavior. `ResilienceCircuitBreakerInterceptor` still selects its breaker in `sendMessage()` and still records every non-OK status as a failure. +- Do not change the library slow-call default of 2 seconds on `CircuitBreakerThresholds`. +- Resilience4j stays an `implementation` dependency of `grpc-circuitbreaker-utils`. +- A public method that accepts `com.typesafe.config.Config` requires that dependency to be `api`. +- Comments explain why. Do not add comments that restate what the code does. +- `forAddress` on a decorator must use the overridden plaintext methods. Delegating `forAddress` to the raw registry bypasses interception. `GrpcChannelRegistry.forAddress` already calls `forPlaintextAddress`. +- `shutdown()` and `shutdownNow()` on an intercepted `ManagedChannel` return the wrapper. +- One intercepted channel is cached per delegate channel. +- `grpc.client.timeout` absent, or `enabled = false`, returns the original registry. +- `circuit.breaker.config` absent, or `enabled = false`, returns the original registry from `wrap`. `forService` still returns an interceptor, and that interceptor passes every call through when the root flag is off or the block is absent. +- `failureStatusCodes` defaults to `DEADLINE_EXCEEDED`, `UNAVAILABLE`, and `RESOURCE_EXHAUSTED`. +- The service breaker name is `MethodDescriptor.getServiceName()`. +- An open breaker throws `Status.RESOURCE_EXHAUSTED` before `next.newCall`. +- A status absent from `failureStatusCodes` is recorded with `onSuccess`, including other non-OK statuses. +- An unknown status name fails interceptor construction. +- Quoted service names are read with `ConfigUtil.joinPath`. +- A service is protected only when the root flag is true and that service block is enabled. An omitted service `enabled` inherits true. A missing service block passes through. +- Callers compose decorators. This library does not wrap timeout and circuit breaker together. + +--- + +### Task 1: InterceptingChannelRegistry + +**Files:** +- Create: `grpc-client-utils/src/main/java/org/hypertrace/core/grpcutils/client/InterceptingChannelRegistry.java` +- Create: `grpc-client-utils/src/test/java/org/hypertrace/core/grpcutils/client/InterceptingChannelRegistryTest.java` + +**Interfaces:** +- Consumes: `org.hypertrace.core.grpcutils.client.GrpcChannelRegistry` +- Produces: `public final class InterceptingChannelRegistry extends GrpcChannelRegistry` with `public static GrpcChannelRegistry wrap(GrpcChannelRegistry registry, ClientInterceptor interceptor)` + +- [ ] **Step 1: Write the failing test** + +Create `InterceptingChannelRegistryTest` with tests that fail to compile until the class exists: + +- `wrapReturnsDistinctRegistry` +- `plaintextAndSecureChannelsCarryTheInterceptor` +- `repeatedLookupReturnsTheSameInterceptedChannel` +- `forAddressUsesThePlaintextOverride` +- `shutdownMethodsReturnTheWrapper` + +Use Mockito to mock `GrpcChannelRegistry` and `ManagedChannel`. The interceptor records the `MethodDescriptor` it sees. `forAddress` must verify `delegate.forPlaintextAddress(host, port, config)` and `verify(delegate, never()).forAddress(anyString(), anyInt())`. `shutdown()` and `shutdownNow()` on the returned channel return that same channel instance and call the matching method on the raw channel. + +- [ ] **Step 2: Run test to verify it fails** + +Run: `./gradlew :grpc-client-utils:test --tests org.hypertrace.core.grpcutils.client.InterceptingChannelRegistryTest` + +Expected: compile failure because `InterceptingChannelRegistry` does not exist. + +- [ ] **Step 3: Write minimal implementation** + +`InterceptingChannelRegistry` is a decorator. `@Delegate(excludes = ForAddress.class)` forwards `shutdown()`, `shutdown(Deadline)`, and `isShutdown()` to the delegate. Override all four `forPlaintextAddress` and `forSecureAddress` methods and intercept the delegate channel. Cache with `Collections.synchronizedMap(new IdentityHashMap<>())`. + +`wrap` throws `NullPointerException` when `registry` or `interceptor` is null. + +The inner `InterceptedManagedChannel` uses `@Delegate(excludes = InterceptedCall.class)`. `newCall` and `authority` go through `ClientInterceptors.intercept`. `shutdown` and `shutdownNow` call the delegate and return `this`. + +Keep one why-comment on `ForAddress`: inherited `forAddress` calls `forPlaintextAddress`, and delegating `forAddress` would return the raw channel. Keep one why-comment on `InterceptedCall`: `shutdown` and `shutdownNow` must return this wrapper. + +- [ ] **Step 4: Run test to verify it passes** + +Run: `./gradlew :grpc-client-utils:test --tests org.hypertrace.core.grpcutils.client.InterceptingChannelRegistryTest` + +Expected: PASS. + +- [ ] **Step 5: Commit** + +```bash +git add grpc-client-utils/src/main/java/org/hypertrace/core/grpcutils/client/InterceptingChannelRegistry.java grpc-client-utils/src/test/java/org/hypertrace/core/grpcutils/client/InterceptingChannelRegistryTest.java +git commit -m "$(cat <<'EOF' +Add a channel-registry decorator that installs one client interceptor. + +EOF +)" +``` + +Do not push. + +--- + +### Task 2: TimeoutChannelRegistry + +**Files:** +- Modify: `grpc-client-utils/build.gradle.kts` +- Create: `grpc-client-utils/src/main/java/org/hypertrace/core/grpcutils/client/TimeoutChannelRegistry.java` +- Create: `grpc-client-utils/src/test/java/org/hypertrace/core/grpcutils/client/TimeoutChannelRegistryTest.java` + +**Interfaces:** +- Consumes: `InterceptingChannelRegistry.wrap(GrpcChannelRegistry, ClientInterceptor)` and `DefaultTimeoutClientInterceptor(Duration)` +- Produces: `public final class TimeoutChannelRegistry` with `public static GrpcChannelRegistry wrap(GrpcChannelRegistry registry, Config config)`. Config path is `grpc.client.timeout`. Fields are `enabled` (boolean) and `defaultTimeout` (duration). + +- [ ] **Step 1: Write the failing test** + +Cover these cases: + +- Missing `grpc.client.timeout` returns the same registry instance. +- `enabled = false` returns the same registry instance. +- `enabled = true` without `defaultTimeout` throws `com.typesafe.config.ConfigException`. +- `enabled = true` and `defaultTimeout = 1500ms` returns a channel whose call options contain a deadline about 1500ms out when the incoming `CallOptions` and `Context` have no deadline. +- A call that already has a deadline keeps that deadline. + +Parse config with `ConfigFactory.parseString`. Use a mock delegate registry and a mock raw `ManagedChannel` whose `newCall` captures `CallOptions`. + +- [ ] **Step 2: Run test to verify it fails** + +Run: `./gradlew :grpc-client-utils:test --tests org.hypertrace.core.grpcutils.client.TimeoutChannelRegistryTest` + +Expected: compile failure because `TimeoutChannelRegistry` does not exist. + +- [ ] **Step 3: Write minimal implementation** + +Add `api(commonLibs.typesafe.config)` to `grpc-client-utils/build.gradle.kts`. + +`TimeoutChannelRegistry.wrap` returns `registry` when `registry` is null, `config` is null, the path is absent, or `enabled` is missing or false. When enabled, require `defaultTimeout` via `config.getDuration` on the timeout block and return `InterceptingChannelRegistry.wrap(registry, new DefaultTimeoutClientInterceptor(timeout))`. + +- [ ] **Step 4: Run test to verify it passes** + +Run: `./gradlew :grpc-client-utils:test --tests org.hypertrace.core.grpcutils.client.TimeoutChannelRegistryTest` + +Expected: PASS. + +- [ ] **Step 5: Commit** + +```bash +git add grpc-client-utils/build.gradle.kts grpc-client-utils/src/main/java/org/hypertrace/core/grpcutils/client/TimeoutChannelRegistry.java grpc-client-utils/src/test/java/org/hypertrace/core/grpcutils/client/TimeoutChannelRegistryTest.java +git commit -m "$(cat <<'EOF' +Apply a configured default deadline to every channel from a registry. + +EOF +)" +``` + +Do not push. + +--- + +### Task 3: Configurable failure status codes + +**Files:** +- Modify: `grpc-circuitbreaker-utils/build.gradle.kts` (`implementation(commonLibs.typesafe.config)` becomes `api(commonLibs.typesafe.config)`) +- Modify: `grpc-circuitbreaker-utils/src/main/java/org/hypertrace/circuitbreaker/grpcutils/CircuitBreakerThresholds.java` +- Modify: `grpc-circuitbreaker-utils/src/main/java/org/hypertrace/circuitbreaker/grpcutils/CircuitBreakerConfigParser.java` +- Create: `grpc-circuitbreaker-utils/src/test/java/org/hypertrace/circuitbreaker/grpcutils/CircuitBreakerConfigParserTest.java` + +**Interfaces:** +- Consumes: existing `CircuitBreakerConfigParser.parseConfig(Config)` +- Produces: `CircuitBreakerThresholds.failureStatusCodes`, a `Set` whose builder default is `DEADLINE_EXCEEDED`, `UNAVAILABLE`, and `RESOURCE_EXHAUSTED`. `parseConfig` reads `failureStatusCodes` inside a threshold block. An unknown name throws `com.typesafe.config.ConfigException.BadValue`. + +- [ ] **Step 1: Write the failing test** + +`CircuitBreakerConfigParserTest` asserts: + +- A config with only `enabled = true` yields default thresholds whose `failureStatusCodes` are exactly those three codes and whose `slowCallDurationThreshold` is `Duration.ofSeconds(2)`. +- `defaultThresholds { failureStatusCodes = [UNAVAILABLE] }` replaces the set with only `UNAVAILABLE`. +- `defaultThresholds { failureStatusCodes = [NOT_A_STATUS] }` throws `ConfigException.BadValue`. + +- [ ] **Step 2: Run test to verify it fails** + +Run: `./gradlew :grpc-circuitbreaker-utils:test --tests org.hypertrace.circuitbreaker.grpcutils.CircuitBreakerConfigParserTest` + +Expected: compile failure because `getFailureStatusCodes` does not exist. + +- [ ] **Step 3: Write minimal implementation** + +Add the field and parse `failureStatusCodes` with `config.getStringList` and `Status.Code.valueOf`. Catch `IllegalArgumentException` and throw `new ConfigException.BadValue("failureStatusCodes", name)`. Leave `ResilienceCircuitBreakerConfigConverter` unchanged so the request-keyed interceptor still records every non-OK status. + +Change the Typesafe dependency from `implementation` to `api`. + +- [ ] **Step 4: Run test to verify it passes** + +Run: `./gradlew :grpc-circuitbreaker-utils:test` + +Expected: the new test and the existing resilience tests pass. + +- [ ] **Step 5: Commit** + +```bash +git add grpc-circuitbreaker-utils/build.gradle.kts grpc-circuitbreaker-utils/src/main/java/org/hypertrace/circuitbreaker/grpcutils/CircuitBreakerThresholds.java grpc-circuitbreaker-utils/src/main/java/org/hypertrace/circuitbreaker/grpcutils/CircuitBreakerConfigParser.java grpc-circuitbreaker-utils/src/test/java/org/hypertrace/circuitbreaker/grpcutils/CircuitBreakerConfigParserTest.java +git commit -m "$(cat <<'EOF' +Parse the gRPC statuses that count as circuit-breaker failures. + +EOF +)" +``` + +Do not push. + +--- + +### Task 4: Service-keyed circuit breaker interceptor + +**Files:** +- Create: `grpc-circuitbreaker-utils/src/main/java/org/hypertrace/circuitbreaker/grpcutils/ServiceCircuitBreakerInterceptor.java` +- Modify: `grpc-circuitbreaker-utils/src/main/java/org/hypertrace/circuitbreaker/grpcutils/CircuitBreakerInterceptorFactory.java` +- Create: `grpc-circuitbreaker-utils/src/test/java/org/hypertrace/circuitbreaker/grpcutils/ServiceCircuitBreakerInterceptorTest.java` + +**Interfaces:** +- Consumes: `CircuitBreakerThresholds`, `CircuitBreakerConfigParser.parseConfig`, `CircuitBreakerConfigParser.ENABLED`, `CircuitBreakerConfigParser.DEFAULT_THRESHOLDS` +- Produces: `public static ClientInterceptor CircuitBreakerInterceptorFactory.forService(Config rootConfig)`. The breaker key is `method.getServiceName()`. Config block is `circuit.breaker.config`. + +- [ ] **Step 1: Write the failing test** + +Use an in-process or mocked `Channel`. Drive calls through `interceptCall`. Assert: + +- Absent config and `enabled = false` call the delegate. +- Enabled config with `"ai.traceable.platform.insights.api.v1.InsightsService" { }` calls the delegate for that service. A second service name calls the delegate and does not share failures. +- After enough `UNAVAILABLE` completions to open the breaker (`minimumNumberOfCalls = 1`, `failureRateThreshold = 1`, `slidingWindowSize = 1`, `slidingWindowType = COUNT_BASED`, `waitDurationInOpenState = 60s`), the next call throws `StatusRuntimeException` with `RESOURCE_EXHAUSTED` and the delegate is not called. +- A status absent from `failureStatusCodes`, such as `INTERNAL`, does not open that breaker. Use `minimumNumberOfCalls = 1` and only `INTERNAL` responses. +- `failureStatusCodes = [NOT_A_STATUS]` throws `ConfigException.BadValue` from `forService`. +- An omitted service `enabled` inherits true. `enabled = false` on that service passes through and records nothing. + +Set `slowCallDurationThreshold = 1h` in these tests so latency cannot open the breaker. + +- [ ] **Step 2: Run test to verify it fails** + +Run: `./gradlew :grpc-circuitbreaker-utils:test --tests org.hypertrace.circuitbreaker.grpcutils.ServiceCircuitBreakerInterceptorTest` + +Expected: compile failure because `forService` does not exist. + +- [ ] **Step 3: Write minimal implementation** + +`forService` delegates to `ServiceCircuitBreakerInterceptor.create(rootConfig)`. + +Parse root key `circuit.breaker.config`. When the block is absent, or `enabled` is absent or false, the interceptor calls `next.newCall` immediately. + +For each other root key, read `config.getConfig(ConfigUtil.joinPath(serviceName)).withFallback(defaultThresholds)`. Build default thresholds from `defaultThresholds` with a fallback to `CircuitBreakerThresholds.builder().build()` expressed as config, or by copying builder values. A service whose resolved `enabled` is false is omitted. Store `Map` keyed by the original service name. `recordException` returns true only for that service's `failureStatusCodes`. + +On each call, `circuitBreakerConfigs.get(serviceName)` missing means pass through. Otherwise `CircuitBreakerRegistry.circuitBreaker(serviceName, config)`, then `tryAcquirePermission`. Rejection throws `Status.RESOURCE_EXHAUSTED.withDescription(...)`. If `next.newCall` throws, call `releasePermission` and rethrow. + +The listener records elapsed nanos. `status.isOk()` and statuses outside `failureStatusCodes` call `onSuccess`. Configured failure statuses call `onError` with `status.asRuntimeException()`. + +Keep the why-comment that `getConfig(key)` treats dots as path separators, so service names are read with `joinPath`. + +- [ ] **Step 4: Run test to verify it passes** + +Run: `./gradlew :grpc-circuitbreaker-utils:test --tests org.hypertrace.circuitbreaker.grpcutils.ServiceCircuitBreakerInterceptorTest` + +Expected: PASS. + +- [ ] **Step 5: Commit** + +```bash +git add grpc-circuitbreaker-utils/src/main/java/org/hypertrace/circuitbreaker/grpcutils/ServiceCircuitBreakerInterceptor.java grpc-circuitbreaker-utils/src/main/java/org/hypertrace/circuitbreaker/grpcutils/CircuitBreakerInterceptorFactory.java grpc-circuitbreaker-utils/src/test/java/org/hypertrace/circuitbreaker/grpcutils/ServiceCircuitBreakerInterceptorTest.java +git commit -m "$(cat <<'EOF' +Add a service-keyed gRPC circuit breaker interceptor. + +EOF +)" +``` + +Do not push. + +--- + +### Task 5: CircuitBreakerChannelRegistry + +**Files:** +- Modify: `grpc-circuitbreaker-utils/build.gradle.kts` (add `api(projects.grpcClientUtils)`) +- Create: `grpc-circuitbreaker-utils/src/main/java/org/hypertrace/circuitbreaker/grpcutils/CircuitBreakerChannelRegistry.java` +- Create: `grpc-circuitbreaker-utils/src/test/java/org/hypertrace/circuitbreaker/grpcutils/CircuitBreakerChannelRegistryTest.java` + +**Interfaces:** +- Consumes: `InterceptingChannelRegistry.wrap(GrpcChannelRegistry, ClientInterceptor)` and `CircuitBreakerInterceptorFactory.forService(Config)` +- Produces: `public final class CircuitBreakerChannelRegistry` with `public static GrpcChannelRegistry wrap(GrpcChannelRegistry registry, Config config)` + +- [ ] **Step 1: Write the failing test** + +Assert: + +- Null config, empty config, and `circuit.breaker.config { enabled = false }` return the same registry instance. +- `enabled = true` with a quoted Insights service returns a different registry. A call to that service with `minimumNumberOfCalls = 1` and one `UNAVAILABLE` completion makes the following call throw `RESOURCE_EXHAUSTED` without another delegate `newCall`. +- A second plaintext lookup of the same delegate channel returns the same intercepted channel. + +- [ ] **Step 2: Run test to verify it fails** + +Run: `./gradlew :grpc-circuitbreaker-utils:test --tests org.hypertrace.circuitbreaker.grpcutils.CircuitBreakerChannelRegistryTest` + +Expected: compile failure because `CircuitBreakerChannelRegistry` does not exist. + +- [ ] **Step 3: Write minimal implementation** + +```java +public final class CircuitBreakerChannelRegistry { + static final String CIRCUIT_BREAKER_CONFIG = "circuit.breaker.config"; + + private CircuitBreakerChannelRegistry() {} + + public static GrpcChannelRegistry wrap(GrpcChannelRegistry registry, Config config) { + if (registry == null || config == null || !config.hasPath(CIRCUIT_BREAKER_CONFIG)) { + return registry; + } + Config circuitBreakerConfig = config.getConfig(CIRCUIT_BREAKER_CONFIG); + if (!circuitBreakerConfig.hasPath("enabled") || !circuitBreakerConfig.getBoolean("enabled")) { + return registry; + } + return InterceptingChannelRegistry.wrap( + registry, CircuitBreakerInterceptorFactory.forService(config)); + } +} +``` + +- [ ] **Step 4: Run test to verify it passes** + +Run: `./gradlew :grpc-client-utils:test :grpc-circuitbreaker-utils:test` + +Expected: PASS for both modules. + +- [ ] **Step 5: Commit** + +```bash +git add grpc-circuitbreaker-utils/build.gradle.kts grpc-circuitbreaker-utils/src/main/java/org/hypertrace/circuitbreaker/grpcutils/CircuitBreakerChannelRegistry.java grpc-circuitbreaker-utils/src/test/java/org/hypertrace/circuitbreaker/grpcutils/CircuitBreakerChannelRegistryTest.java +git commit -m "$(cat <<'EOF' +Install the service-keyed circuit breaker from an existing channel registry. + +EOF +)" +``` + +Do not push. From e099cfd1c1f5ca1b5e9dc5d331ae0bede3f5a17b Mon Sep 17 00:00:00 2001 From: Suresh Prakash Date: Fri, 23 Oct 2026 18:34:49 +0530 Subject: [PATCH 2/9] Record the worktree Gradle version required for dependency locking. Co-authored-by: Cursor --- .../2026-10-23-grpc-registry-decorators.md | 21 ++++++++++--------- 1 file changed, 11 insertions(+), 10 deletions(-) diff --git a/docs/superpowers/plans/2026-10-23-grpc-registry-decorators.md b/docs/superpowers/plans/2026-10-23-grpc-registry-decorators.md index 05c7834..881fc78 100644 --- a/docs/superpowers/plans/2026-10-23-grpc-registry-decorators.md +++ b/docs/superpowers/plans/2026-10-23-grpc-registry-decorators.md @@ -29,6 +29,7 @@ - Quoted service names are read with `ConfigUtil.joinPath`. - A service is protected only when the root flag is true and that service block is enabled. An omitted service `enabled` inherits true. A missing service block passes through. - Callers compose decorators. This library does not wrap timeout and circuit breaker together. +- Every Gradle command in this worktree must include `-Pversion=0.13.23`. The linked worktree does not expose git tags to the version plugin. Without that property the BOM replaces `projects.grpcContextUtils` with external `grpc-context-utils:0.13.22`, and dependency locking fails. --- @@ -56,7 +57,7 @@ Use Mockito to mock `GrpcChannelRegistry` and `ManagedChannel`. The interceptor - [ ] **Step 2: Run test to verify it fails** -Run: `./gradlew :grpc-client-utils:test --tests org.hypertrace.core.grpcutils.client.InterceptingChannelRegistryTest` +Run: `./gradlew -Pversion=0.13.23 :grpc-client-utils:test --tests org.hypertrace.core.grpcutils.client.InterceptingChannelRegistryTest` Expected: compile failure because `InterceptingChannelRegistry` does not exist. @@ -72,7 +73,7 @@ Keep one why-comment on `ForAddress`: inherited `forAddress` calls `forPlaintext - [ ] **Step 4: Run test to verify it passes** -Run: `./gradlew :grpc-client-utils:test --tests org.hypertrace.core.grpcutils.client.InterceptingChannelRegistryTest` +Run: `./gradlew -Pversion=0.13.23 :grpc-client-utils:test --tests org.hypertrace.core.grpcutils.client.InterceptingChannelRegistryTest` Expected: PASS. @@ -116,7 +117,7 @@ Parse config with `ConfigFactory.parseString`. Use a mock delegate registry and - [ ] **Step 2: Run test to verify it fails** -Run: `./gradlew :grpc-client-utils:test --tests org.hypertrace.core.grpcutils.client.TimeoutChannelRegistryTest` +Run: `./gradlew -Pversion=0.13.23 :grpc-client-utils:test --tests org.hypertrace.core.grpcutils.client.TimeoutChannelRegistryTest` Expected: compile failure because `TimeoutChannelRegistry` does not exist. @@ -128,7 +129,7 @@ Add `api(commonLibs.typesafe.config)` to `grpc-client-utils/build.gradle.kts`. - [ ] **Step 4: Run test to verify it passes** -Run: `./gradlew :grpc-client-utils:test --tests org.hypertrace.core.grpcutils.client.TimeoutChannelRegistryTest` +Run: `./gradlew -Pversion=0.13.23 :grpc-client-utils:test --tests org.hypertrace.core.grpcutils.client.TimeoutChannelRegistryTest` Expected: PASS. @@ -169,7 +170,7 @@ Do not push. - [ ] **Step 2: Run test to verify it fails** -Run: `./gradlew :grpc-circuitbreaker-utils:test --tests org.hypertrace.circuitbreaker.grpcutils.CircuitBreakerConfigParserTest` +Run: `./gradlew -Pversion=0.13.23 :grpc-circuitbreaker-utils:test --tests org.hypertrace.circuitbreaker.grpcutils.CircuitBreakerConfigParserTest` Expected: compile failure because `getFailureStatusCodes` does not exist. @@ -181,7 +182,7 @@ Change the Typesafe dependency from `implementation` to `api`. - [ ] **Step 4: Run test to verify it passes** -Run: `./gradlew :grpc-circuitbreaker-utils:test` +Run: `./gradlew -Pversion=0.13.23 :grpc-circuitbreaker-utils:test` Expected: the new test and the existing resilience tests pass. @@ -226,7 +227,7 @@ Set `slowCallDurationThreshold = 1h` in these tests so latency cannot open the b - [ ] **Step 2: Run test to verify it fails** -Run: `./gradlew :grpc-circuitbreaker-utils:test --tests org.hypertrace.circuitbreaker.grpcutils.ServiceCircuitBreakerInterceptorTest` +Run: `./gradlew -Pversion=0.13.23 :grpc-circuitbreaker-utils:test --tests org.hypertrace.circuitbreaker.grpcutils.ServiceCircuitBreakerInterceptorTest` Expected: compile failure because `forService` does not exist. @@ -246,7 +247,7 @@ Keep the why-comment that `getConfig(key)` treats dots as path separators, so se - [ ] **Step 4: Run test to verify it passes** -Run: `./gradlew :grpc-circuitbreaker-utils:test --tests org.hypertrace.circuitbreaker.grpcutils.ServiceCircuitBreakerInterceptorTest` +Run: `./gradlew -Pversion=0.13.23 :grpc-circuitbreaker-utils:test --tests org.hypertrace.circuitbreaker.grpcutils.ServiceCircuitBreakerInterceptorTest` Expected: PASS. @@ -286,7 +287,7 @@ Assert: - [ ] **Step 2: Run test to verify it fails** -Run: `./gradlew :grpc-circuitbreaker-utils:test --tests org.hypertrace.circuitbreaker.grpcutils.CircuitBreakerChannelRegistryTest` +Run: `./gradlew -Pversion=0.13.23 :grpc-circuitbreaker-utils:test --tests org.hypertrace.circuitbreaker.grpcutils.CircuitBreakerChannelRegistryTest` Expected: compile failure because `CircuitBreakerChannelRegistry` does not exist. @@ -314,7 +315,7 @@ public final class CircuitBreakerChannelRegistry { - [ ] **Step 4: Run test to verify it passes** -Run: `./gradlew :grpc-client-utils:test :grpc-circuitbreaker-utils:test` +Run: `./gradlew -Pversion=0.13.23 :grpc-client-utils:test :grpc-circuitbreaker-utils:test` Expected: PASS for both modules. From 01e657e2911001d237e3de8e9f19ea18ff7fc432 Mon Sep 17 00:00:00 2001 From: Suresh Prakash Date: Mon, 26 Oct 2026 11:18:28 +0530 Subject: [PATCH 3/9] Add a channel-registry decorator that installs one client interceptor. Co-authored-by: Cursor --- .../client/InterceptingChannelRegistry.java | 121 ++++++++++++++ .../InterceptingChannelRegistryTest.java | 155 ++++++++++++++++++ 2 files changed, 276 insertions(+) create mode 100644 grpc-client-utils/src/main/java/org/hypertrace/core/grpcutils/client/InterceptingChannelRegistry.java create mode 100644 grpc-client-utils/src/test/java/org/hypertrace/core/grpcutils/client/InterceptingChannelRegistryTest.java diff --git a/grpc-client-utils/src/main/java/org/hypertrace/core/grpcutils/client/InterceptingChannelRegistry.java b/grpc-client-utils/src/main/java/org/hypertrace/core/grpcutils/client/InterceptingChannelRegistry.java new file mode 100644 index 0000000..7815ccc --- /dev/null +++ b/grpc-client-utils/src/main/java/org/hypertrace/core/grpcutils/client/InterceptingChannelRegistry.java @@ -0,0 +1,121 @@ +package org.hypertrace.core.grpcutils.client; + +import io.grpc.CallOptions; +import io.grpc.Channel; +import io.grpc.ClientCall; +import io.grpc.ClientInterceptor; +import io.grpc.ClientInterceptors; +import io.grpc.ManagedChannel; +import io.grpc.MethodDescriptor; +import java.util.Collections; +import java.util.IdentityHashMap; +import java.util.Map; +import java.util.Objects; +import lombok.experimental.Delegate; + +public final class InterceptingChannelRegistry extends GrpcChannelRegistry { + + public static GrpcChannelRegistry wrap( + GrpcChannelRegistry registry, ClientInterceptor interceptor) { + return new InterceptingChannelRegistry( + Objects.requireNonNull(registry), Objects.requireNonNull(interceptor)); + } + + @Delegate(excludes = ForAddress.class) + private final GrpcChannelRegistry delegate; + + private final ClientInterceptor interceptor; + private final Map interceptedChannels = + Collections.synchronizedMap(new IdentityHashMap<>()); + + private InterceptingChannelRegistry(GrpcChannelRegistry delegate, ClientInterceptor interceptor) { + this.delegate = delegate; + this.interceptor = interceptor; + } + + @Override + public ManagedChannel forSecureAddress(String host, int port) { + return this.intercept(this.delegate.forSecureAddress(host, port)); + } + + @Override + public ManagedChannel forSecureAddress(String host, int port, GrpcChannelConfig config) { + return this.intercept(this.delegate.forSecureAddress(host, port, config)); + } + + @Override + public ManagedChannel forPlaintextAddress(String host, int port) { + return this.intercept(this.delegate.forPlaintextAddress(host, port)); + } + + @Override + public ManagedChannel forPlaintextAddress(String host, int port, GrpcChannelConfig config) { + return this.intercept(this.delegate.forPlaintextAddress(host, port, config)); + } + + private ManagedChannel intercept(ManagedChannel channel) { + return this.interceptedChannels.computeIfAbsent( + channel, raw -> new InterceptedManagedChannel(raw, this.interceptor)); + } + + // The inherited forAddress calls forPlaintextAddress, so delegating it would return the raw + // channel and skip the interceptor. + private interface ForAddress { + ManagedChannel forAddress(String host, int port); + + ManagedChannel forSecureAddress(String host, int port); + + ManagedChannel forSecureAddress(String host, int port, GrpcChannelConfig config); + + ManagedChannel forPlaintextAddress(String host, int port); + + ManagedChannel forPlaintextAddress(String host, int port, GrpcChannelConfig config); + } + + private static final class InterceptedManagedChannel extends ManagedChannel { + @Delegate(excludes = InterceptedCall.class) + private final ManagedChannel delegate; + + private final Channel interceptedChannel; + + private InterceptedManagedChannel(ManagedChannel delegate, ClientInterceptor interceptor) { + this.delegate = delegate; + this.interceptedChannel = ClientInterceptors.intercept(delegate, interceptor); + } + + @Override + public ClientCall newCall( + MethodDescriptor methodDescriptor, CallOptions callOptions) { + return this.interceptedChannel.newCall(methodDescriptor, callOptions); + } + + @Override + public String authority() { + return this.interceptedChannel.authority(); + } + + @Override + public ManagedChannel shutdown() { + this.delegate.shutdown(); + return this; + } + + @Override + public ManagedChannel shutdownNow() { + this.delegate.shutdownNow(); + return this; + } + } + + // shutdown and shutdownNow must return this wrapper, not the raw delegate channel. + private interface InterceptedCall { + ClientCall newCall( + MethodDescriptor methodDescriptor, CallOptions callOptions); + + String authority(); + + ManagedChannel shutdown(); + + ManagedChannel shutdownNow(); + } +} diff --git a/grpc-client-utils/src/test/java/org/hypertrace/core/grpcutils/client/InterceptingChannelRegistryTest.java b/grpc-client-utils/src/test/java/org/hypertrace/core/grpcutils/client/InterceptingChannelRegistryTest.java new file mode 100644 index 0000000..cfb2907 --- /dev/null +++ b/grpc-client-utils/src/test/java/org/hypertrace/core/grpcutils/client/InterceptingChannelRegistryTest.java @@ -0,0 +1,155 @@ +package org.hypertrace.core.grpcutils.client; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotSame; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.anyInt; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import io.grpc.CallOptions; +import io.grpc.Channel; +import io.grpc.ClientCall; +import io.grpc.ClientInterceptor; +import io.grpc.Deadline; +import io.grpc.ManagedChannel; +import io.grpc.MethodDescriptor; +import io.grpc.MethodDescriptor.Marshaller; +import java.io.InputStream; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.TimeUnit; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +class InterceptingChannelRegistryTest { + + private static final MethodDescriptor METHOD = + MethodDescriptor.newBuilder() + .setType(MethodDescriptor.MethodType.UNARY) + .setFullMethodName("test.Service/Method") + .setRequestMarshaller(new StringMarshaller()) + .setResponseMarshaller(new StringMarshaller()) + .build(); + + private final List> interceptedMethods = new ArrayList<>(); + private final ClientInterceptor recordingInterceptor = + new ClientInterceptor() { + @Override + public ClientCall interceptCall( + MethodDescriptor method, CallOptions callOptions, Channel next) { + interceptedMethods.add(method); + return next.newCall(method, callOptions); + } + }; + + private GrpcChannelRegistry delegate; + private ManagedChannel rawChannel; + private GrpcChannelRegistry wrapper; + + @BeforeEach + void beforeEach() { + this.delegate = mock(GrpcChannelRegistry.class); + this.rawChannel = mock(ManagedChannel.class); + this.wrapper = InterceptingChannelRegistry.wrap(this.delegate, this.recordingInterceptor); + } + + @Test + void wrapReturnsDistinctRegistry() { + assertNotSame(this.delegate, this.wrapper); + assertThrows( + NullPointerException.class, + () -> InterceptingChannelRegistry.wrap(null, this.recordingInterceptor)); + assertThrows( + NullPointerException.class, () -> InterceptingChannelRegistry.wrap(this.delegate, null)); + } + + @Test + void plaintextAndSecureChannelsCarryTheInterceptor() { + GrpcChannelConfig config = GrpcChannelConfig.builder().build(); + ManagedChannel secureRaw = mock(ManagedChannel.class); + when(this.delegate.forPlaintextAddress("host", 1)).thenReturn(this.rawChannel); + when(this.delegate.forPlaintextAddress("host", 2, config)).thenReturn(this.rawChannel); + when(this.delegate.forSecureAddress("host", 3)).thenReturn(secureRaw); + when(this.delegate.forSecureAddress("host", 4, config)).thenReturn(secureRaw); + + List channels = + List.of( + this.wrapper.forPlaintextAddress("host", 1), + this.wrapper.forPlaintextAddress("host", 2, config), + this.wrapper.forSecureAddress("host", 3), + this.wrapper.forSecureAddress("host", 4, config)); + + channels.forEach(channel -> channel.newCall(METHOD, CallOptions.DEFAULT)); + + assertEquals(List.of(METHOD, METHOD, METHOD, METHOD), this.interceptedMethods); + verify(this.rawChannel, times(2)).newCall(METHOD, CallOptions.DEFAULT); + verify(secureRaw, times(2)).newCall(METHOD, CallOptions.DEFAULT); + } + + @Test + void repeatedLookupReturnsTheSameInterceptedChannel() { + when(this.delegate.forPlaintextAddress("host", 1)).thenReturn(this.rawChannel); + + ManagedChannel first = this.wrapper.forPlaintextAddress("host", 1); + + assertNotSame(this.rawChannel, first); + assertSame(first, this.wrapper.forPlaintextAddress("host", 1)); + } + + @Test + void forAddressUsesThePlaintextOverride() { + GrpcChannelConfig config = GrpcChannelConfig.builder().build(); + when(this.delegate.forPlaintextAddress("host", 1, config)).thenReturn(this.rawChannel); + + ManagedChannel channel = this.wrapper.forAddress("host", 1); + channel.newCall(METHOD, CallOptions.DEFAULT); + + verify(this.delegate).forPlaintextAddress("host", 1, config); + verify(this.delegate, never()).forAddress(anyString(), anyInt()); + assertEquals(List.of(METHOD), this.interceptedMethods); + } + + @Test + void shutdownMethodsReturnTheWrapper() { + when(this.delegate.forPlaintextAddress("host", 1)).thenReturn(this.rawChannel); + ManagedChannel channel = this.wrapper.forPlaintextAddress("host", 1); + + assertSame(channel, channel.shutdown()); + assertSame(channel, channel.shutdownNow()); + + verify(this.rawChannel).shutdown(); + verify(this.rawChannel).shutdownNow(); + } + + @Test + void registryShutdownMethodsForwardToTheDelegate() { + Deadline deadline = Deadline.after(1, TimeUnit.SECONDS); + when(this.delegate.isShutdown()).thenReturn(true); + + this.wrapper.shutdown(); + this.wrapper.shutdown(deadline); + + verify(this.delegate).shutdown(); + verify(this.delegate).shutdown(deadline); + assertTrue(this.wrapper.isShutdown()); + } + + private static final class StringMarshaller implements Marshaller { + @Override + public InputStream stream(String value) { + throw new UnsupportedOperationException(); + } + + @Override + public String parse(InputStream stream) { + throw new UnsupportedOperationException(); + } + } +} From fef7b20613d3b2e7853b56f7874d36091cc440e0 Mon Sep 17 00:00:00 2001 From: Suresh Prakash Date: Mon, 26 Oct 2026 12:40:46 +0530 Subject: [PATCH 4/9] Apply a configured default deadline to every channel from a registry. Co-authored-by: Cursor --- grpc-client-utils/build.gradle.kts | 1 + grpc-client-utils/gradle.lockfile | 1 + .../client/TimeoutChannelRegistry.java | 27 ++++ .../client/TimeoutChannelRegistryTest.java | 137 ++++++++++++++++++ 4 files changed, 166 insertions(+) create mode 100644 grpc-client-utils/src/main/java/org/hypertrace/core/grpcutils/client/TimeoutChannelRegistry.java create mode 100644 grpc-client-utils/src/test/java/org/hypertrace/core/grpcutils/client/TimeoutChannelRegistryTest.java diff --git a/grpc-client-utils/build.gradle.kts b/grpc-client-utils/build.gradle.kts index 530a934..c11bcae 100644 --- a/grpc-client-utils/build.gradle.kts +++ b/grpc-client-utils/build.gradle.kts @@ -10,6 +10,7 @@ dependencies { api(commonLibs.grpc.context) api(commonLibs.grpc.api) api(commonLibs.grpc.inprocess) + api(commonLibs.typesafe.config) implementation(projects.grpcContextUtils) implementation(commonLibs.slf4j2.api) diff --git a/grpc-client-utils/gradle.lockfile b/grpc-client-utils/gradle.lockfile index beeec55..ecd50ee 100644 --- a/grpc-client-utils/gradle.lockfile +++ b/grpc-client-utils/gradle.lockfile @@ -21,6 +21,7 @@ com.google.j2objc:j2objc-annotations:2.8=compileClasspath com.google.j2objc:j2objc-annotations:3.0.0=runtimeClasspath,testCompileClasspath,testRuntimeClasspath com.google.protobuf:protobuf-java-util:3.25.8=compileClasspath,runtimeClasspath,testCompileClasspath,testRuntimeClasspath com.google.protobuf:protobuf-java:3.25.8=compileClasspath,runtimeClasspath,testCompileClasspath,testRuntimeClasspath +com.typesafe:config:1.4.2=compileClasspath,runtimeClasspath,testCompileClasspath,testRuntimeClasspath io.grpc:grpc-api:1.75.0=compileClasspath,runtimeClasspath,testCompileClasspath,testRuntimeClasspath io.grpc:grpc-bom:1.75.0=compileClasspath,runtimeClasspath,testCompileClasspath,testRuntimeClasspath io.grpc:grpc-context:1.75.0=compileClasspath,runtimeClasspath,testCompileClasspath,testRuntimeClasspath diff --git a/grpc-client-utils/src/main/java/org/hypertrace/core/grpcutils/client/TimeoutChannelRegistry.java b/grpc-client-utils/src/main/java/org/hypertrace/core/grpcutils/client/TimeoutChannelRegistry.java new file mode 100644 index 0000000..b89b3bc --- /dev/null +++ b/grpc-client-utils/src/main/java/org/hypertrace/core/grpcutils/client/TimeoutChannelRegistry.java @@ -0,0 +1,27 @@ +package org.hypertrace.core.grpcutils.client; + +import com.typesafe.config.Config; +import java.time.Duration; + +public final class TimeoutChannelRegistry { + private static final String TIMEOUT_PATH = "grpc.client.timeout"; + private static final String ENABLED_KEY = "enabled"; + private static final String DEFAULT_TIMEOUT_KEY = "defaultTimeout"; + + public static GrpcChannelRegistry wrap(GrpcChannelRegistry registry, Config config) { + if (registry == null || config == null || !config.hasPath(TIMEOUT_PATH)) { + return registry; + } + + Config timeoutConfig = config.getConfig(TIMEOUT_PATH); + if (!timeoutConfig.hasPath(ENABLED_KEY) || !timeoutConfig.getBoolean(ENABLED_KEY)) { + return registry; + } + + // An enabled block with no timeout must fail at startup rather than silently apply no deadline. + Duration timeout = timeoutConfig.getDuration(DEFAULT_TIMEOUT_KEY); + return InterceptingChannelRegistry.wrap(registry, new DefaultTimeoutClientInterceptor(timeout)); + } + + private TimeoutChannelRegistry() {} +} diff --git a/grpc-client-utils/src/test/java/org/hypertrace/core/grpcutils/client/TimeoutChannelRegistryTest.java b/grpc-client-utils/src/test/java/org/hypertrace/core/grpcutils/client/TimeoutChannelRegistryTest.java new file mode 100644 index 0000000..a340911 --- /dev/null +++ b/grpc-client-utils/src/test/java/org/hypertrace/core/grpcutils/client/TimeoutChannelRegistryTest.java @@ -0,0 +1,137 @@ +package org.hypertrace.core.grpcutils.client; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNotSame; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import com.typesafe.config.Config; +import com.typesafe.config.ConfigException; +import com.typesafe.config.ConfigFactory; +import io.grpc.CallOptions; +import io.grpc.ClientCall; +import io.grpc.Deadline; +import io.grpc.ManagedChannel; +import io.grpc.MethodDescriptor; +import io.grpc.MethodDescriptor.Marshaller; +import java.io.InputStream; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +class TimeoutChannelRegistryTest { + + private static final MethodDescriptor METHOD = + MethodDescriptor.newBuilder() + .setType(MethodDescriptor.MethodType.UNARY) + .setFullMethodName("test.Service/Method") + .setRequestMarshaller(new StringMarshaller()) + .setResponseMarshaller(new StringMarshaller()) + .build(); + + private final AtomicReference capturedOptions = new AtomicReference<>(); + + private GrpcChannelRegistry delegate; + private ManagedChannel rawChannel; + + @BeforeEach + @SuppressWarnings("unchecked") + void beforeEach() { + this.delegate = mock(GrpcChannelRegistry.class); + this.rawChannel = mock(ManagedChannel.class); + when(this.delegate.forPlaintextAddress("host", 1)).thenReturn(this.rawChannel); + when(this.rawChannel.newCall(any(), any())) + .thenAnswer( + invocation -> { + this.capturedOptions.set(invocation.getArgument(1)); + return mock(ClientCall.class); + }); + } + + @Test + void missingTimeoutBlockReturnsTheSameRegistry() { + Config config = ConfigFactory.parseString("grpc.client.other = 1"); + + assertSame(this.delegate, TimeoutChannelRegistry.wrap(this.delegate, config)); + } + + @Test + void disabledReturnsTheSameRegistry() { + Config config = + ConfigFactory.parseString( + "grpc.client.timeout { enabled = false, defaultTimeout = 1500ms }"); + + assertSame(this.delegate, TimeoutChannelRegistry.wrap(this.delegate, config)); + } + + @Test + void missingEnabledReturnsTheSameRegistry() { + Config config = ConfigFactory.parseString("grpc.client.timeout { defaultTimeout = 1500ms }"); + + assertSame(this.delegate, TimeoutChannelRegistry.wrap(this.delegate, config)); + } + + @Test + void nullRegistryOrConfigReturnsTheInput() { + Config config = + ConfigFactory.parseString("grpc.client.timeout { enabled = true, defaultTimeout = 1s }"); + + assertSame(this.delegate, TimeoutChannelRegistry.wrap(this.delegate, null)); + assertEquals(null, TimeoutChannelRegistry.wrap(null, config)); + } + + @Test + void enabledWithoutDefaultTimeoutThrows() { + Config config = ConfigFactory.parseString("grpc.client.timeout { enabled = true }"); + + assertThrows(ConfigException.class, () -> TimeoutChannelRegistry.wrap(this.delegate, config)); + } + + @Test + void enabledAppliesTheDefaultDeadlineWhenNoneIsPresent() { + GrpcChannelRegistry wrapped = TimeoutChannelRegistry.wrap(this.delegate, enabledConfig()); + assertNotSame(this.delegate, wrapped); + + wrapped.forPlaintextAddress("host", 1).newCall(METHOD, CallOptions.DEFAULT); + + Deadline deadline = this.capturedOptions.get().getDeadline(); + assertNotNull(deadline); + long remainingMillis = deadline.timeRemaining(TimeUnit.MILLISECONDS); + assertTrue(remainingMillis > 0 && remainingMillis <= 1500, "remaining: " + remainingMillis); + } + + @Test + void existingDeadlineIsKept() { + GrpcChannelRegistry wrapped = TimeoutChannelRegistry.wrap(this.delegate, enabledConfig()); + Deadline existing = Deadline.after(10, TimeUnit.MINUTES); + + wrapped + .forPlaintextAddress("host", 1) + .newCall(METHOD, CallOptions.DEFAULT.withDeadline(existing)); + + assertSame(existing, this.capturedOptions.get().getDeadline()); + } + + private static Config enabledConfig() { + return ConfigFactory.parseString( + "grpc.client.timeout { enabled = true, defaultTimeout = 1500ms }"); + } + + private static final class StringMarshaller implements Marshaller { + @Override + public InputStream stream(String value) { + throw new UnsupportedOperationException(); + } + + @Override + public String parse(InputStream stream) { + throw new UnsupportedOperationException(); + } + } +} From 8e081c87987b0495566d5f5870b2d25eb9bb9d8c Mon Sep 17 00:00:00 2001 From: Suresh Prakash Date: Mon, 26 Oct 2026 17:14:46 +0530 Subject: [PATCH 5/9] Parse the gRPC statuses that count as circuit-breaker failures. Co-authored-by: Cursor --- grpc-circuitbreaker-utils/build.gradle.kts | 2 +- .../grpcutils/CircuitBreakerConfigParser.java | 23 +++++++++ .../grpcutils/CircuitBreakerThresholds.java | 7 +++ .../CircuitBreakerConfigParserTest.java | 49 +++++++++++++++++++ 4 files changed, 80 insertions(+), 1 deletion(-) create mode 100644 grpc-circuitbreaker-utils/src/test/java/org/hypertrace/circuitbreaker/grpcutils/CircuitBreakerConfigParserTest.java diff --git a/grpc-circuitbreaker-utils/build.gradle.kts b/grpc-circuitbreaker-utils/build.gradle.kts index 58f6201..70c7662 100644 --- a/grpc-circuitbreaker-utils/build.gradle.kts +++ b/grpc-circuitbreaker-utils/build.gradle.kts @@ -11,7 +11,7 @@ dependencies { implementation(commonLibs.slf4j2.api) implementation(localLibs.resilience4j.circuitbreaker) - implementation(commonLibs.typesafe.config) + api(commonLibs.typesafe.config) implementation(commonLibs.guava) implementation(commonLibs.jakarta.inject.api) diff --git a/grpc-circuitbreaker-utils/src/main/java/org/hypertrace/circuitbreaker/grpcutils/CircuitBreakerConfigParser.java b/grpc-circuitbreaker-utils/src/main/java/org/hypertrace/circuitbreaker/grpcutils/CircuitBreakerConfigParser.java index dc4867f..a2e3e95 100644 --- a/grpc-circuitbreaker-utils/src/main/java/org/hypertrace/circuitbreaker/grpcutils/CircuitBreakerConfigParser.java +++ b/grpc-circuitbreaker-utils/src/main/java/org/hypertrace/circuitbreaker/grpcutils/CircuitBreakerConfigParser.java @@ -1,6 +1,10 @@ package org.hypertrace.circuitbreaker.grpcutils; import com.typesafe.config.Config; +import com.typesafe.config.ConfigException; +import io.grpc.Status; +import java.util.EnumSet; +import java.util.List; import java.util.Map; import java.util.Set; import java.util.stream.Collectors; @@ -26,6 +30,7 @@ public class CircuitBreakerConfigParser { private static final String PERMITTED_NUMBER_OF_CALLS_IN_HALF_OPEN_STATE = "permittedNumberOfCallsInHalfOpenState"; private static final String SLIDING_WINDOW_TYPE = "slidingWindowType"; + private static final String FAILURE_STATUS_CODES = "failureStatusCodes"; public static final String ENABLED = "enabled"; public static final String DEFAULT_THRESHOLDS = "defaultThresholds"; private static final Set NON_THRESHOLD_KEYS = Set.of(ENABLED, DEFAULT_THRESHOLDS); @@ -97,9 +102,27 @@ private static CircuitBreakerThresholds buildCircuitBreakerThresholds(Config con builder.enabled(config.getBoolean(ENABLED)); } + if (config.hasPath(FAILURE_STATUS_CODES)) { + builder.failureStatusCodes( + parseFailureStatusCodes(config.getStringList(FAILURE_STATUS_CODES))); + } + return builder.build(); } + // An unknown name must fail startup. Ignoring it would protect the wrong calls. + private static Set parseFailureStatusCodes(List names) { + Set codes = EnumSet.noneOf(Status.Code.class); + for (String name : names) { + try { + codes.add(Status.Code.valueOf(name)); + } catch (IllegalArgumentException exception) { + throw new ConfigException.BadValue(FAILURE_STATUS_CODES, name); + } + } + return Set.copyOf(codes); + } + public static CircuitBreakerThresholds buildCircuitBreakerDefaultThresholds() { return CircuitBreakerThresholds.builder().build(); } diff --git a/grpc-circuitbreaker-utils/src/main/java/org/hypertrace/circuitbreaker/grpcutils/CircuitBreakerThresholds.java b/grpc-circuitbreaker-utils/src/main/java/org/hypertrace/circuitbreaker/grpcutils/CircuitBreakerThresholds.java index 3b7f89e..aedacf3 100644 --- a/grpc-circuitbreaker-utils/src/main/java/org/hypertrace/circuitbreaker/grpcutils/CircuitBreakerThresholds.java +++ b/grpc-circuitbreaker-utils/src/main/java/org/hypertrace/circuitbreaker/grpcutils/CircuitBreakerThresholds.java @@ -1,6 +1,8 @@ package org.hypertrace.circuitbreaker.grpcutils; +import io.grpc.Status; import java.time.Duration; +import java.util.Set; import lombok.Builder; import lombok.Value; @@ -25,6 +27,11 @@ public class CircuitBreakerThresholds { @Builder.Default int permittedNumberOfCallsInHalfOpenState = 5; @Builder.Default boolean enabled = true; + @Builder.Default + Set failureStatusCodes = + Set.of( + Status.Code.DEADLINE_EXCEEDED, Status.Code.UNAVAILABLE, Status.Code.RESOURCE_EXHAUSTED); + public enum SlidingWindowType { COUNT_BASED, TIME_BASED diff --git a/grpc-circuitbreaker-utils/src/test/java/org/hypertrace/circuitbreaker/grpcutils/CircuitBreakerConfigParserTest.java b/grpc-circuitbreaker-utils/src/test/java/org/hypertrace/circuitbreaker/grpcutils/CircuitBreakerConfigParserTest.java new file mode 100644 index 0000000..1d7dfcb --- /dev/null +++ b/grpc-circuitbreaker-utils/src/test/java/org/hypertrace/circuitbreaker/grpcutils/CircuitBreakerConfigParserTest.java @@ -0,0 +1,49 @@ +package org.hypertrace.circuitbreaker.grpcutils; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; + +import com.typesafe.config.ConfigException; +import com.typesafe.config.ConfigFactory; +import io.grpc.Status.Code; +import java.time.Duration; +import java.util.Set; +import org.junit.jupiter.api.Test; + +class CircuitBreakerConfigParserTest { + + @Test + void omittedFailureStatusesUseTheLibraryDefault() { + CircuitBreakerThresholds thresholds = + CircuitBreakerConfigParser.parseConfig(ConfigFactory.parseString("enabled = true")) + .build() + .getDefaultThresholds(); + + assertEquals( + Set.of(Code.DEADLINE_EXCEEDED, Code.UNAVAILABLE, Code.RESOURCE_EXHAUSTED), + thresholds.getFailureStatusCodes()); + assertEquals(Duration.ofSeconds(2), thresholds.getSlowCallDurationThreshold()); + } + + @Test + void defaultThresholdsReplaceFailureStatuses() { + CircuitBreakerThresholds thresholds = + CircuitBreakerConfigParser.parseConfig( + ConfigFactory.parseString( + "defaultThresholds { failureStatusCodes = [UNAVAILABLE] }")) + .build() + .getDefaultThresholds(); + + assertEquals(Set.of(Code.UNAVAILABLE), thresholds.getFailureStatusCodes()); + } + + @Test + void unknownFailureStatusFailsParsing() { + assertThrows( + ConfigException.BadValue.class, + () -> + CircuitBreakerConfigParser.parseConfig( + ConfigFactory.parseString( + "defaultThresholds { failureStatusCodes = [NOT_A_STATUS] }"))); + } +} From a5cdb78391cd1c84208267690be860ea0e57ac13 Mon Sep 17 00:00:00 2001 From: Suresh Prakash Date: Mon, 26 Oct 2026 18:23:40 +0530 Subject: [PATCH 6/9] Add a service-keyed gRPC circuit breaker interceptor. Co-authored-by: Cursor --- .../CircuitBreakerInterceptorFactory.java | 6 + .../ServiceCircuitBreakerInterceptor.java | 218 +++++++++++++++++ .../ServiceCircuitBreakerInterceptorTest.java | 220 ++++++++++++++++++ 3 files changed, 444 insertions(+) create mode 100644 grpc-circuitbreaker-utils/src/main/java/org/hypertrace/circuitbreaker/grpcutils/ServiceCircuitBreakerInterceptor.java create mode 100644 grpc-circuitbreaker-utils/src/test/java/org/hypertrace/circuitbreaker/grpcutils/ServiceCircuitBreakerInterceptorTest.java diff --git a/grpc-circuitbreaker-utils/src/main/java/org/hypertrace/circuitbreaker/grpcutils/CircuitBreakerInterceptorFactory.java b/grpc-circuitbreaker-utils/src/main/java/org/hypertrace/circuitbreaker/grpcutils/CircuitBreakerInterceptorFactory.java index a353357..9f16be6 100644 --- a/grpc-circuitbreaker-utils/src/main/java/org/hypertrace/circuitbreaker/grpcutils/CircuitBreakerInterceptorFactory.java +++ b/grpc-circuitbreaker-utils/src/main/java/org/hypertrace/circuitbreaker/grpcutils/CircuitBreakerInterceptorFactory.java @@ -1,5 +1,7 @@ package org.hypertrace.circuitbreaker.grpcutils; +import com.typesafe.config.Config; +import io.grpc.ClientInterceptor; import jakarta.inject.Inject; import java.time.Clock; import lombok.AllArgsConstructor; @@ -9,6 +11,10 @@ public class CircuitBreakerInterceptorFactory { private final Clock clock; + public static ClientInterceptor forService(Config rootConfig) { + return ServiceCircuitBreakerInterceptor.create(rootConfig); + } + public CircuitBreakerInterceptor buildInterceptor(CircuitBreakerConfiguration configuration) { return ResilienceCircuitBreakerFactory.getCircuitBreakerInterceptor(configuration, clock); } diff --git a/grpc-circuitbreaker-utils/src/main/java/org/hypertrace/circuitbreaker/grpcutils/ServiceCircuitBreakerInterceptor.java b/grpc-circuitbreaker-utils/src/main/java/org/hypertrace/circuitbreaker/grpcutils/ServiceCircuitBreakerInterceptor.java new file mode 100644 index 0000000..e30615a --- /dev/null +++ b/grpc-circuitbreaker-utils/src/main/java/org/hypertrace/circuitbreaker/grpcutils/ServiceCircuitBreakerInterceptor.java @@ -0,0 +1,218 @@ +package org.hypertrace.circuitbreaker.grpcutils; + +import com.typesafe.config.Config; +import com.typesafe.config.ConfigFactory; +import com.typesafe.config.ConfigUtil; +import io.github.resilience4j.circuitbreaker.CircuitBreaker; +import io.github.resilience4j.circuitbreaker.CircuitBreakerConfig; +import io.github.resilience4j.circuitbreaker.CircuitBreakerRegistry; +import io.grpc.CallOptions; +import io.grpc.Channel; +import io.grpc.ClientCall; +import io.grpc.ClientInterceptor; +import io.grpc.ForwardingClientCall; +import io.grpc.ForwardingClientCallListener; +import io.grpc.Metadata; +import io.grpc.MethodDescriptor; +import io.grpc.Status; +import io.grpc.StatusRuntimeException; +import java.util.HashMap; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.TimeUnit; +import java.util.stream.Collectors; + +public final class ServiceCircuitBreakerInterceptor implements ClientInterceptor { + + private static final String CONFIG_PATH = "circuit.breaker.config"; + private static final String SYNTHETIC_SERVICE_KEY = "service"; + + private final Map circuitBreakerConfigs; + private final Map> failureStatusCodesByService; + private final CircuitBreakerRegistry circuitBreakerRegistry; + + static ClientInterceptor create(Config rootConfig) { + if (!rootConfig.hasPath(CONFIG_PATH)) { + return passthrough(); + } + Config config = rootConfig.getConfig(CONFIG_PATH); + if (!config.hasPath(CircuitBreakerConfigParser.ENABLED) + || !config.getBoolean(CircuitBreakerConfigParser.ENABLED)) { + return passthrough(); + } + return configured(config); + } + + private static ServiceCircuitBreakerInterceptor passthrough() { + return new ServiceCircuitBreakerInterceptor(Map.of(), Map.of()); + } + + private static ServiceCircuitBreakerInterceptor configured(Config config) { + Config defaults = libraryDefaults(); + if (config.hasPath(CircuitBreakerConfigParser.DEFAULT_THRESHOLDS)) { + defaults = + config.getConfig(CircuitBreakerConfigParser.DEFAULT_THRESHOLDS).withFallback(defaults); + } + + Map circuitBreakerConfigs = new HashMap<>(); + Map> failureStatusCodesByService = new HashMap<>(); + for (String serviceName : config.root().keySet()) { + if (CircuitBreakerConfigParser.ENABLED.equals(serviceName) + || CircuitBreakerConfigParser.DEFAULT_THRESHOLDS.equals(serviceName)) { + continue; + } + // getConfig(key) treats dots as path separators, so service names are read with joinPath. + Config resolved = config.getConfig(ConfigUtil.joinPath(serviceName)).withFallback(defaults); + CircuitBreakerThresholds thresholds = parseServiceThresholds(resolved); + if (!thresholds.isEnabled()) { + continue; + } + circuitBreakerConfigs.put(serviceName, toCircuitBreakerConfig(thresholds)); + failureStatusCodesByService.put(serviceName, thresholds.getFailureStatusCodes()); + } + return new ServiceCircuitBreakerInterceptor( + Map.copyOf(circuitBreakerConfigs), Map.copyOf(failureStatusCodesByService)); + } + + private static Config libraryDefaults() { + CircuitBreakerThresholds defaults = CircuitBreakerThresholds.builder().build(); + String failureStatuses = + defaults.getFailureStatusCodes().stream() + .map(Status.Code::name) + .collect(Collectors.joining(", ")); + return ConfigFactory.parseString( + "failureRateThreshold = " + + defaults.getFailureRateThreshold() + + "\n" + + "slowCallRateThreshold = " + + defaults.getSlowCallRateThreshold() + + "\n" + + "slowCallDurationThreshold = " + + defaults.getSlowCallDurationThreshold().toMillis() + + "ms\n" + + "slidingWindowType = " + + defaults.getSlidingWindowType().name() + + "\n" + + "slidingWindowSize = " + + defaults.getSlidingWindowSize() + + "\n" + + "waitDurationInOpenState = " + + defaults.getWaitDurationInOpenState().toMillis() + + "ms\n" + + "minimumNumberOfCalls = " + + defaults.getMinimumNumberOfCalls() + + "\n" + + "permittedNumberOfCallsInHalfOpenState = " + + defaults.getPermittedNumberOfCallsInHalfOpenState() + + "\n" + + "enabled = " + + defaults.isEnabled() + + "\n" + + "failureStatusCodes = [" + + failureStatuses + + "]\n"); + } + + private static CircuitBreakerThresholds parseServiceThresholds(Config resolvedServiceConfig) { + return CircuitBreakerConfigParser.parseConfig( + resolvedServiceConfig.atKey(SYNTHETIC_SERVICE_KEY)) + .build() + .getCircuitBreakerThresholdsMap() + .get(SYNTHETIC_SERVICE_KEY); + } + + // onError is counted as a success when this predicate returns false. + private static CircuitBreakerConfig toCircuitBreakerConfig(CircuitBreakerThresholds thresholds) { + Set failureStatusCodes = thresholds.getFailureStatusCodes(); + return CircuitBreakerConfig.custom() + .failureRateThreshold(thresholds.getFailureRateThreshold()) + .slowCallRateThreshold(thresholds.getSlowCallRateThreshold()) + .slowCallDurationThreshold(thresholds.getSlowCallDurationThreshold()) + .slidingWindowType(slidingWindowType(thresholds.getSlidingWindowType())) + .slidingWindowSize(thresholds.getSlidingWindowSize()) + .waitDurationInOpenState(thresholds.getWaitDurationInOpenState()) + .permittedNumberOfCallsInHalfOpenState( + thresholds.getPermittedNumberOfCallsInHalfOpenState()) + .minimumNumberOfCalls(thresholds.getMinimumNumberOfCalls()) + .recordException( + throwable -> + throwable instanceof StatusRuntimeException + && failureStatusCodes.contains( + ((StatusRuntimeException) throwable).getStatus().getCode())) + .build(); + } + + private static CircuitBreakerConfig.SlidingWindowType slidingWindowType( + CircuitBreakerThresholds.SlidingWindowType slidingWindowType) { + switch (slidingWindowType) { + case COUNT_BASED: + return CircuitBreakerConfig.SlidingWindowType.COUNT_BASED; + case TIME_BASED: + default: + return CircuitBreakerConfig.SlidingWindowType.TIME_BASED; + } + } + + private ServiceCircuitBreakerInterceptor( + Map circuitBreakerConfigs, + Map> failureStatusCodesByService) { + this.circuitBreakerConfigs = circuitBreakerConfigs; + this.failureStatusCodesByService = failureStatusCodesByService; + this.circuitBreakerRegistry = CircuitBreakerRegistry.ofDefaults(); + } + + @Override + public ClientCall interceptCall( + MethodDescriptor method, CallOptions callOptions, Channel next) { + String serviceName = method.getServiceName(); + CircuitBreakerConfig circuitBreakerConfig = circuitBreakerConfigs.get(serviceName); + if (circuitBreakerConfig == null) { + return next.newCall(method, callOptions); + } + + CircuitBreaker circuitBreaker = + circuitBreakerRegistry.circuitBreaker(serviceName, circuitBreakerConfig); + if (!circuitBreaker.tryAcquirePermission()) { + throw Status.RESOURCE_EXHAUSTED + .withDescription("Circuit breaker is open for " + serviceName) + .asRuntimeException(); + } + + ClientCall delegate; + try { + delegate = next.newCall(method, callOptions); + } catch (RuntimeException exception) { + circuitBreaker.releasePermission(); + throw exception; + } + return recordingCall(delegate, circuitBreaker, failureStatusCodesByService.get(serviceName)); + } + + private ClientCall recordingCall( + ClientCall delegate, + CircuitBreaker circuitBreaker, + Set failureStatusCodes) { + return new ForwardingClientCall.SimpleForwardingClientCall<>(delegate) { + @Override + public void start(Listener responseListener, Metadata headers) { + long startedAtNanos = System.nanoTime(); + super.start( + new ForwardingClientCallListener.SimpleForwardingClientCallListener<>( + responseListener) { + @Override + public void onClose(Status status, Metadata trailers) { + long elapsedNanos = System.nanoTime() - startedAtNanos; + if (status.isOk() || !failureStatusCodes.contains(status.getCode())) { + circuitBreaker.onSuccess(elapsedNanos, TimeUnit.NANOSECONDS); + } else { + circuitBreaker.onError( + elapsedNanos, TimeUnit.NANOSECONDS, status.asRuntimeException()); + } + super.onClose(status, trailers); + } + }, + headers); + } + }; + } +} diff --git a/grpc-circuitbreaker-utils/src/test/java/org/hypertrace/circuitbreaker/grpcutils/ServiceCircuitBreakerInterceptorTest.java b/grpc-circuitbreaker-utils/src/test/java/org/hypertrace/circuitbreaker/grpcutils/ServiceCircuitBreakerInterceptorTest.java new file mode 100644 index 0000000..cb51954 --- /dev/null +++ b/grpc-circuitbreaker-utils/src/test/java/org/hypertrace/circuitbreaker/grpcutils/ServiceCircuitBreakerInterceptorTest.java @@ -0,0 +1,220 @@ +package org.hypertrace.circuitbreaker.grpcutils; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import com.typesafe.config.Config; +import com.typesafe.config.ConfigException; +import com.typesafe.config.ConfigFactory; +import io.grpc.CallOptions; +import io.grpc.Channel; +import io.grpc.ClientCall; +import io.grpc.ClientInterceptor; +import io.grpc.Metadata; +import io.grpc.MethodDescriptor; +import io.grpc.Status; +import io.grpc.StatusRuntimeException; +import org.junit.jupiter.api.Test; +import org.mockito.ArgumentCaptor; + +class ServiceCircuitBreakerInterceptorTest { + + private static final String INSIGHTS_SERVICE = + "ai.traceable.platform.insights.api.v1.InsightsService"; + private static final String OTHER_SERVICE = "ai.traceable.platform.other.OtherService"; + private static final String OPEN_QUICKLY = + "slowCallDurationThreshold = 1h\n" + + "minimumNumberOfCalls = 1\n" + + "failureRateThreshold = 1\n" + + "slidingWindowSize = 1\n" + + "slidingWindowType = COUNT_BASED\n" + + "waitDurationInOpenState = 60s\n"; + + @Test + void absentConfigCallsTheDelegate() { + Channel channel = mock(Channel.class); + ClientInterceptor interceptor = + CircuitBreakerInterceptorFactory.forService(ConfigFactory.empty()); + + complete(interceptor, channel, INSIGHTS_SERVICE, Status.UNAVAILABLE); + complete(interceptor, channel, INSIGHTS_SERVICE, Status.UNAVAILABLE); + verify(channel, times(2)).newCall(any(), any()); + } + + @Test + void missingRootEnabledCallsTheDelegate() { + Channel channel = mock(Channel.class); + ClientInterceptor interceptor = + CircuitBreakerInterceptorFactory.forService( + circuitBreaker(serviceBlock(INSIGHTS_SERVICE, OPEN_QUICKLY))); + + complete(interceptor, channel, INSIGHTS_SERVICE, Status.UNAVAILABLE); + complete(interceptor, channel, INSIGHTS_SERVICE, Status.UNAVAILABLE); + verify(channel, times(2)).newCall(any(), any()); + } + + @Test + void rootEnabledFalseCallsTheDelegate() { + Channel channel = mock(Channel.class); + ClientInterceptor interceptor = + CircuitBreakerInterceptorFactory.forService( + circuitBreaker("enabled = false\n" + serviceBlock(INSIGHTS_SERVICE, OPEN_QUICKLY))); + + complete(interceptor, channel, INSIGHTS_SERVICE, Status.UNAVAILABLE); + complete(interceptor, channel, INSIGHTS_SERVICE, Status.UNAVAILABLE); + verify(channel, times(2)).newCall(any(), any()); + } + + @Test + void enabledServiceCallsTheDelegate() { + Channel channel = mock(Channel.class); + ClientInterceptor interceptor = + CircuitBreakerInterceptorFactory.forService( + circuitBreaker( + "enabled = true\n" + + "defaultThresholds { slowCallDurationThreshold = 1h }\n" + + serviceBlock(INSIGHTS_SERVICE, ""))); + + complete(interceptor, channel, INSIGHTS_SERVICE, Status.OK); + verify(channel).newCall(any(), any()); + } + + @Test + void serviceFailuresDoNotTripAnotherService() { + Channel channel = mock(Channel.class); + ClientInterceptor interceptor = + CircuitBreakerInterceptorFactory.forService( + circuitBreaker( + "enabled = true\n" + + serviceBlock(INSIGHTS_SERVICE, OPEN_QUICKLY) + + serviceBlock(OTHER_SERVICE, OPEN_QUICKLY))); + + complete(interceptor, channel, INSIGHTS_SERVICE, Status.UNAVAILABLE); + complete(interceptor, channel, OTHER_SERVICE, Status.OK); + verify(channel, times(2)).newCall(any(), any()); + + StatusRuntimeException rejected = reject(interceptor, channel, INSIGHTS_SERVICE); + assertEquals(Status.Code.RESOURCE_EXHAUSTED, rejected.getStatus().getCode()); + verify(channel, times(2)).newCall(any(), any()); + } + + @Test + void unavailableOpensTheBreakerWithoutCallingTheDelegate() { + Channel channel = mock(Channel.class); + ClientInterceptor interceptor = + CircuitBreakerInterceptorFactory.forService( + circuitBreaker("enabled = true\n" + serviceBlock(INSIGHTS_SERVICE, OPEN_QUICKLY))); + + complete(interceptor, channel, INSIGHTS_SERVICE, Status.UNAVAILABLE); + + StatusRuntimeException rejected = reject(interceptor, channel, INSIGHTS_SERVICE); + assertEquals(Status.Code.RESOURCE_EXHAUSTED, rejected.getStatus().getCode()); + assertNotNull(rejected.getStatus().getDescription()); + assertFalse(rejected.getStatus().getDescription().isEmpty()); + verify(channel, times(1)).newCall(any(), any()); + } + + @Test + void internalDoesNotOpenTheBreakerForTheDefaultFailureStatuses() { + Channel channel = mock(Channel.class); + ClientInterceptor interceptor = + CircuitBreakerInterceptorFactory.forService( + circuitBreaker("enabled = true\n" + serviceBlock(INSIGHTS_SERVICE, OPEN_QUICKLY))); + + complete(interceptor, channel, INSIGHTS_SERVICE, Status.INTERNAL); + complete(interceptor, channel, INSIGHTS_SERVICE, Status.OK); + verify(channel, times(2)).newCall(any(), any()); + } + + @Test + void onlyConfiguredUnavailableCountsAsFailure() { + Channel channel = mock(Channel.class); + ClientInterceptor interceptor = + CircuitBreakerInterceptorFactory.forService( + circuitBreaker( + "enabled = true\n" + + "defaultThresholds { failureStatusCodes = [UNAVAILABLE] }\n" + + serviceBlock(INSIGHTS_SERVICE, OPEN_QUICKLY))); + + complete(interceptor, channel, INSIGHTS_SERVICE, Status.INTERNAL); + complete(interceptor, channel, INSIGHTS_SERVICE, Status.DEADLINE_EXCEEDED); + complete(interceptor, channel, INSIGHTS_SERVICE, Status.UNAVAILABLE); + + StatusRuntimeException rejected = reject(interceptor, channel, INSIGHTS_SERVICE); + assertEquals(Status.Code.RESOURCE_EXHAUSTED, rejected.getStatus().getCode()); + verify(channel, times(3)).newCall(any(), any()); + } + + @Test + void unknownFailureStatusFailsConstruction() { + Config config = + circuitBreaker( + "enabled = true\n" + + serviceBlock(INSIGHTS_SERVICE, "failureStatusCodes = [NOT_A_STATUS]")); + + assertThrows( + ConfigException.BadValue.class, () -> CircuitBreakerInterceptorFactory.forService(config)); + } + + @Test + void disabledServicePassesThrough() { + Channel channel = mock(Channel.class); + ClientInterceptor interceptor = + CircuitBreakerInterceptorFactory.forService( + circuitBreaker( + "enabled = true\n" + + serviceBlock(INSIGHTS_SERVICE, "enabled = false\n" + OPEN_QUICKLY))); + + complete(interceptor, channel, INSIGHTS_SERVICE, Status.UNAVAILABLE); + complete(interceptor, channel, INSIGHTS_SERVICE, Status.UNAVAILABLE); + verify(channel, times(2)).newCall(any(), any()); + } + + private static Config circuitBreaker(String body) { + return ConfigFactory.parseString("circuit.breaker.config {\n" + body + "\n}"); + } + + private static String serviceBlock(String serviceName, String body) { + return "\"" + serviceName + "\" {\n" + body + "\n}\n"; + } + + @SuppressWarnings("unchecked") + private static void complete( + ClientInterceptor interceptor, Channel channel, String serviceName, Status status) { + ClientCall delegate = mock(ClientCall.class); + when(channel.newCall(any(), any())).thenReturn(delegate); + ClientCall.Listener listener = mock(ClientCall.Listener.class); + Metadata headers = new Metadata(); + ClientCall call = + interceptor.interceptCall(method(serviceName), CallOptions.DEFAULT, channel); + call.start(listener, headers); + + ArgumentCaptor> captor = + ArgumentCaptor.forClass(ClientCall.Listener.class); + verify(delegate).start(captor.capture(), any()); + Metadata trailers = new Metadata(); + captor.getValue().onClose(status, trailers); + verify(listener).onClose(status, trailers); + } + + private static StatusRuntimeException reject( + ClientInterceptor interceptor, Channel channel, String serviceName) { + return assertThrows( + StatusRuntimeException.class, + () -> interceptor.interceptCall(method(serviceName), CallOptions.DEFAULT, channel)); + } + + @SuppressWarnings("unchecked") + private static MethodDescriptor method(String serviceName) { + MethodDescriptor method = mock(MethodDescriptor.class); + when(method.getServiceName()).thenReturn(serviceName); + return method; + } +} From f8a38442f678adaeef2ad46a6de767ab514654fb Mon Sep 17 00:00:00 2001 From: Suresh Prakash Date: Wed, 28 Oct 2026 08:31:19 +0530 Subject: [PATCH 7/9] Install the service-keyed circuit breaker from an existing channel registry. Co-authored-by: Cursor --- grpc-circuitbreaker-utils/build.gradle.kts | 3 +- grpc-circuitbreaker-utils/gradle.lockfile | 5 +- .../CircuitBreakerChannelRegistry.java | 24 ++++ .../CircuitBreakerChannelRegistryTest.java | 133 ++++++++++++++++++ 4 files changed, 163 insertions(+), 2 deletions(-) create mode 100644 grpc-circuitbreaker-utils/src/main/java/org/hypertrace/circuitbreaker/grpcutils/CircuitBreakerChannelRegistry.java create mode 100644 grpc-circuitbreaker-utils/src/test/java/org/hypertrace/circuitbreaker/grpcutils/CircuitBreakerChannelRegistryTest.java diff --git a/grpc-circuitbreaker-utils/build.gradle.kts b/grpc-circuitbreaker-utils/build.gradle.kts index 70c7662..cbae719 100644 --- a/grpc-circuitbreaker-utils/build.gradle.kts +++ b/grpc-circuitbreaker-utils/build.gradle.kts @@ -7,11 +7,12 @@ plugins { dependencies { api(commonLibs.grpc.api) + api(commonLibs.typesafe.config) api(projects.grpcContextUtils) + api(projects.grpcClientUtils) implementation(commonLibs.slf4j2.api) implementation(localLibs.resilience4j.circuitbreaker) - api(commonLibs.typesafe.config) implementation(commonLibs.guava) implementation(commonLibs.jakarta.inject.api) diff --git a/grpc-circuitbreaker-utils/gradle.lockfile b/grpc-circuitbreaker-utils/gradle.lockfile index 7c54541..f768438 100644 --- a/grpc-circuitbreaker-utils/gradle.lockfile +++ b/grpc-circuitbreaker-utils/gradle.lockfile @@ -20,13 +20,16 @@ com.google.guava:guava:33.3.1-android=runtimeClasspath,testRuntimeClasspath com.google.guava:listenablefuture:9999.0-empty-to-avoid-conflict-with-guava=compileClasspath,runtimeClasspath,testCompileClasspath,testRuntimeClasspath com.google.j2objc:j2objc-annotations:2.8=compileClasspath,testCompileClasspath com.google.j2objc:j2objc-annotations:3.0.0=runtimeClasspath,testRuntimeClasspath +com.google.protobuf:protobuf-java-util:3.25.8=testRuntimeClasspath +com.google.protobuf:protobuf-java:3.25.8=testRuntimeClasspath com.typesafe:config:1.4.2=compileClasspath,runtimeClasspath,testCompileClasspath,testRuntimeClasspath io.github.resilience4j:resilience4j-circuitbreaker:1.7.1=compileClasspath,runtimeClasspath,testCompileClasspath,testRuntimeClasspath io.github.resilience4j:resilience4j-core:1.7.1=compileClasspath,runtimeClasspath,testCompileClasspath,testRuntimeClasspath io.grpc:grpc-api:1.75.0=compileClasspath,runtimeClasspath,testCompileClasspath,testRuntimeClasspath io.grpc:grpc-bom:1.75.0=compileClasspath,runtimeClasspath,testCompileClasspath,testRuntimeClasspath -io.grpc:grpc-context:1.75.0=runtimeClasspath,testRuntimeClasspath +io.grpc:grpc-context:1.75.0=compileClasspath,runtimeClasspath,testCompileClasspath,testRuntimeClasspath io.grpc:grpc-core:1.75.0=runtimeClasspath,testRuntimeClasspath +io.grpc:grpc-inprocess:1.75.0=compileClasspath,runtimeClasspath,testCompileClasspath,testRuntimeClasspath io.netty:netty-bom:4.1.132.Final=compileClasspath,runtimeClasspath,testCompileClasspath,testRuntimeClasspath io.perfmark:perfmark-api:0.27.0=runtimeClasspath,testRuntimeClasspath io.vavr:vavr-match:0.10.2=compileClasspath,runtimeClasspath,testCompileClasspath,testRuntimeClasspath diff --git a/grpc-circuitbreaker-utils/src/main/java/org/hypertrace/circuitbreaker/grpcutils/CircuitBreakerChannelRegistry.java b/grpc-circuitbreaker-utils/src/main/java/org/hypertrace/circuitbreaker/grpcutils/CircuitBreakerChannelRegistry.java new file mode 100644 index 0000000..9ee46a5 --- /dev/null +++ b/grpc-circuitbreaker-utils/src/main/java/org/hypertrace/circuitbreaker/grpcutils/CircuitBreakerChannelRegistry.java @@ -0,0 +1,24 @@ +package org.hypertrace.circuitbreaker.grpcutils; + +import com.typesafe.config.Config; +import org.hypertrace.core.grpcutils.client.GrpcChannelRegistry; +import org.hypertrace.core.grpcutils.client.InterceptingChannelRegistry; + +public final class CircuitBreakerChannelRegistry { + + static final String CIRCUIT_BREAKER_CONFIG = "circuit.breaker.config"; + + private CircuitBreakerChannelRegistry() {} + + public static GrpcChannelRegistry wrap(GrpcChannelRegistry registry, Config config) { + if (registry == null || config == null || !config.hasPath(CIRCUIT_BREAKER_CONFIG)) { + return registry; + } + Config circuitBreakerConfig = config.getConfig(CIRCUIT_BREAKER_CONFIG); + if (!circuitBreakerConfig.hasPath("enabled") || !circuitBreakerConfig.getBoolean("enabled")) { + return registry; + } + return InterceptingChannelRegistry.wrap( + registry, CircuitBreakerInterceptorFactory.forService(config)); + } +} diff --git a/grpc-circuitbreaker-utils/src/test/java/org/hypertrace/circuitbreaker/grpcutils/CircuitBreakerChannelRegistryTest.java b/grpc-circuitbreaker-utils/src/test/java/org/hypertrace/circuitbreaker/grpcutils/CircuitBreakerChannelRegistryTest.java new file mode 100644 index 0000000..bebc5c1 --- /dev/null +++ b/grpc-circuitbreaker-utils/src/test/java/org/hypertrace/circuitbreaker/grpcutils/CircuitBreakerChannelRegistryTest.java @@ -0,0 +1,133 @@ +package org.hypertrace.circuitbreaker.grpcutils; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotSame; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import com.typesafe.config.Config; +import com.typesafe.config.ConfigFactory; +import io.grpc.CallOptions; +import io.grpc.ClientCall; +import io.grpc.ManagedChannel; +import io.grpc.Metadata; +import io.grpc.MethodDescriptor; +import io.grpc.Status; +import io.grpc.StatusRuntimeException; +import org.hypertrace.core.grpcutils.client.GrpcChannelRegistry; +import org.junit.jupiter.api.Test; +import org.mockito.ArgumentCaptor; + +class CircuitBreakerChannelRegistryTest { + + private static final String INSIGHTS_SERVICE = + "ai.traceable.platform.insights.api.v1.InsightsService"; + private static final String ENABLED_INSIGHTS = + "circuit.breaker.config {\n" + + " enabled = true\n" + + " \"" + + INSIGHTS_SERVICE + + "\" {\n" + + " slowCallDurationThreshold = 1h\n" + + " minimumNumberOfCalls = 1\n" + + " failureRateThreshold = 1\n" + + " slidingWindowSize = 1\n" + + " slidingWindowType = COUNT_BASED\n" + + " waitDurationInOpenState = 60s\n" + + " }\n" + + "}\n"; + + @Test + void nullConfigReturnsTheSameRegistry() { + GrpcChannelRegistry registry = mock(GrpcChannelRegistry.class); + + assertSame(registry, CircuitBreakerChannelRegistry.wrap(registry, null)); + } + + @Test + void emptyConfigReturnsTheSameRegistry() { + GrpcChannelRegistry registry = mock(GrpcChannelRegistry.class); + + assertSame(registry, CircuitBreakerChannelRegistry.wrap(registry, ConfigFactory.empty())); + } + + @Test + void disabledConfigReturnsTheSameRegistry() { + GrpcChannelRegistry registry = mock(GrpcChannelRegistry.class); + Config config = ConfigFactory.parseString("circuit.breaker.config { enabled = false }"); + + assertSame(registry, CircuitBreakerChannelRegistry.wrap(registry, config)); + } + + @Test + void missingEnabledReturnsTheSameRegistry() { + GrpcChannelRegistry registry = mock(GrpcChannelRegistry.class); + Config config = ConfigFactory.parseString("circuit.breaker.config { }"); + + assertSame(registry, CircuitBreakerChannelRegistry.wrap(registry, config)); + } + + @Test + void nullRegistryIsReturnedUnchanged() { + assertSame(null, CircuitBreakerChannelRegistry.wrap(null, ConfigFactory.empty())); + } + + @Test + void enabledConfigOpensTheBreakerOnUnavailable() { + GrpcChannelRegistry registry = mock(GrpcChannelRegistry.class); + ManagedChannel rawChannel = mock(ManagedChannel.class); + when(registry.forPlaintextAddress("host", 80)).thenReturn(rawChannel); + + GrpcChannelRegistry wrapped = + CircuitBreakerChannelRegistry.wrap(registry, ConfigFactory.parseString(ENABLED_INSIGHTS)); + assertNotSame(registry, wrapped); + ManagedChannel channel = wrapped.forPlaintextAddress("host", 80); + + completeCall(channel, rawChannel, Status.UNAVAILABLE); + + StatusRuntimeException rejected = + assertThrows( + StatusRuntimeException.class, () -> channel.newCall(method(), CallOptions.DEFAULT)); + assertEquals(Status.Code.RESOURCE_EXHAUSTED, rejected.getStatus().getCode()); + verify(rawChannel, times(1)).newCall(any(), any()); + } + + @Test + void repeatedLookupReturnsTheSameInterceptedChannel() { + GrpcChannelRegistry registry = mock(GrpcChannelRegistry.class); + ManagedChannel rawChannel = mock(ManagedChannel.class); + when(registry.forPlaintextAddress("host", 80)).thenReturn(rawChannel); + + GrpcChannelRegistry wrapped = + CircuitBreakerChannelRegistry.wrap(registry, ConfigFactory.parseString(ENABLED_INSIGHTS)); + + assertSame(wrapped.forPlaintextAddress("host", 80), wrapped.forPlaintextAddress("host", 80)); + } + + @SuppressWarnings("unchecked") + private static void completeCall(ManagedChannel channel, ManagedChannel raw, Status status) { + ClientCall delegateCall = mock(ClientCall.class); + when(raw.newCall(any(), any())).thenReturn((ClientCall) delegateCall); + ClientCall.Listener listener = mock(ClientCall.Listener.class); + + ClientCall call = channel.newCall(method(), CallOptions.DEFAULT); + call.start(listener, new Metadata()); + + ArgumentCaptor> captor = + ArgumentCaptor.forClass(ClientCall.Listener.class); + verify(delegateCall).start(captor.capture(), any()); + captor.getValue().onClose(status, new Metadata()); + } + + @SuppressWarnings("unchecked") + private static MethodDescriptor method() { + MethodDescriptor method = mock(MethodDescriptor.class); + when(method.getServiceName()).thenReturn(INSIGHTS_SERVICE); + return method; + } +} From 6c644d13312bfbccf6a85d4ba75dc34dc4898bf0 Mon Sep 17 00:00:00 2001 From: Suresh Prakash Date: Thu, 29 Oct 2026 12:30:39 +0530 Subject: [PATCH 8/9] Address review notes on the registry decorators. Parse service breakers into CircuitBreakerThresholds instead of rebuilding defaults as HOCON, and log when a decorator is skipped or a call is rejected. Lombok still excludes forAddress because that method is inherited, not overridden. Co-authored-by: Cursor --- .gitignore | 1 + .../2026-10-23-grpc-registry-decorators.md | 333 ------------------ .../CircuitBreakerChannelRegistry.java | 15 +- .../grpcutils/CircuitBreakerConfigParser.java | 52 ++- .../CircuitBreakerInterceptorFactory.java | 8 +- .../grpcutils/CircuitBreakerThresholds.java | 2 +- .../ServiceCircuitBreakerInterceptor.java | 124 ++----- .../client/InterceptingChannelRegistry.java | 31 +- .../client/TimeoutChannelRegistry.java | 15 +- 9 files changed, 117 insertions(+), 464 deletions(-) delete mode 100644 docs/superpowers/plans/2026-10-23-grpc-registry-decorators.md diff --git a/.gitignore b/.gitignore index fdfba38..f47b1ba 100644 --- a/.gitignore +++ b/.gitignore @@ -7,3 +7,4 @@ build .idea .superpowers/ +docs/superpowers/ diff --git a/docs/superpowers/plans/2026-10-23-grpc-registry-decorators.md b/docs/superpowers/plans/2026-10-23-grpc-registry-decorators.md deleted file mode 100644 index 881fc78..0000000 --- a/docs/superpowers/plans/2026-10-23-grpc-registry-decorators.md +++ /dev/null @@ -1,333 +0,0 @@ -# gRPC Registry Decorators Implementation Plan - -> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. - -**Goal:** Publish a registry decorator that installs a default deadline, and a service-keyed circuit breaker that clients can install either as that decorator or as a standalone interceptor. - -**Architecture:** `grpc-client-utils` owns `InterceptingChannelRegistry`, a decorator over an already-built `GrpcChannelRegistry`, and `TimeoutChannelRegistry`, which opts in through `grpc.client.timeout`. `grpc-circuitbreaker-utils` owns the service-keyed interceptor and `CircuitBreakerChannelRegistry`, which uses the same decorator. The existing request-keyed interceptor is unchanged. - -**Tech Stack:** Java, Gradle, JUnit 5, Mockito, gRPC, Typesafe Config, Resilience4j 1.7.1 (implementation scope only). - -## Global Constraints - -- Work only in this `java-grpc-utils` worktree. Do not edit `activity-event-service`. Do not push. -- Do not change request-keyed breaker behavior. `ResilienceCircuitBreakerInterceptor` still selects its breaker in `sendMessage()` and still records every non-OK status as a failure. -- Do not change the library slow-call default of 2 seconds on `CircuitBreakerThresholds`. -- Resilience4j stays an `implementation` dependency of `grpc-circuitbreaker-utils`. -- A public method that accepts `com.typesafe.config.Config` requires that dependency to be `api`. -- Comments explain why. Do not add comments that restate what the code does. -- `forAddress` on a decorator must use the overridden plaintext methods. Delegating `forAddress` to the raw registry bypasses interception. `GrpcChannelRegistry.forAddress` already calls `forPlaintextAddress`. -- `shutdown()` and `shutdownNow()` on an intercepted `ManagedChannel` return the wrapper. -- One intercepted channel is cached per delegate channel. -- `grpc.client.timeout` absent, or `enabled = false`, returns the original registry. -- `circuit.breaker.config` absent, or `enabled = false`, returns the original registry from `wrap`. `forService` still returns an interceptor, and that interceptor passes every call through when the root flag is off or the block is absent. -- `failureStatusCodes` defaults to `DEADLINE_EXCEEDED`, `UNAVAILABLE`, and `RESOURCE_EXHAUSTED`. -- The service breaker name is `MethodDescriptor.getServiceName()`. -- An open breaker throws `Status.RESOURCE_EXHAUSTED` before `next.newCall`. -- A status absent from `failureStatusCodes` is recorded with `onSuccess`, including other non-OK statuses. -- An unknown status name fails interceptor construction. -- Quoted service names are read with `ConfigUtil.joinPath`. -- A service is protected only when the root flag is true and that service block is enabled. An omitted service `enabled` inherits true. A missing service block passes through. -- Callers compose decorators. This library does not wrap timeout and circuit breaker together. -- Every Gradle command in this worktree must include `-Pversion=0.13.23`. The linked worktree does not expose git tags to the version plugin. Without that property the BOM replaces `projects.grpcContextUtils` with external `grpc-context-utils:0.13.22`, and dependency locking fails. - ---- - -### Task 1: InterceptingChannelRegistry - -**Files:** -- Create: `grpc-client-utils/src/main/java/org/hypertrace/core/grpcutils/client/InterceptingChannelRegistry.java` -- Create: `grpc-client-utils/src/test/java/org/hypertrace/core/grpcutils/client/InterceptingChannelRegistryTest.java` - -**Interfaces:** -- Consumes: `org.hypertrace.core.grpcutils.client.GrpcChannelRegistry` -- Produces: `public final class InterceptingChannelRegistry extends GrpcChannelRegistry` with `public static GrpcChannelRegistry wrap(GrpcChannelRegistry registry, ClientInterceptor interceptor)` - -- [ ] **Step 1: Write the failing test** - -Create `InterceptingChannelRegistryTest` with tests that fail to compile until the class exists: - -- `wrapReturnsDistinctRegistry` -- `plaintextAndSecureChannelsCarryTheInterceptor` -- `repeatedLookupReturnsTheSameInterceptedChannel` -- `forAddressUsesThePlaintextOverride` -- `shutdownMethodsReturnTheWrapper` - -Use Mockito to mock `GrpcChannelRegistry` and `ManagedChannel`. The interceptor records the `MethodDescriptor` it sees. `forAddress` must verify `delegate.forPlaintextAddress(host, port, config)` and `verify(delegate, never()).forAddress(anyString(), anyInt())`. `shutdown()` and `shutdownNow()` on the returned channel return that same channel instance and call the matching method on the raw channel. - -- [ ] **Step 2: Run test to verify it fails** - -Run: `./gradlew -Pversion=0.13.23 :grpc-client-utils:test --tests org.hypertrace.core.grpcutils.client.InterceptingChannelRegistryTest` - -Expected: compile failure because `InterceptingChannelRegistry` does not exist. - -- [ ] **Step 3: Write minimal implementation** - -`InterceptingChannelRegistry` is a decorator. `@Delegate(excludes = ForAddress.class)` forwards `shutdown()`, `shutdown(Deadline)`, and `isShutdown()` to the delegate. Override all four `forPlaintextAddress` and `forSecureAddress` methods and intercept the delegate channel. Cache with `Collections.synchronizedMap(new IdentityHashMap<>())`. - -`wrap` throws `NullPointerException` when `registry` or `interceptor` is null. - -The inner `InterceptedManagedChannel` uses `@Delegate(excludes = InterceptedCall.class)`. `newCall` and `authority` go through `ClientInterceptors.intercept`. `shutdown` and `shutdownNow` call the delegate and return `this`. - -Keep one why-comment on `ForAddress`: inherited `forAddress` calls `forPlaintextAddress`, and delegating `forAddress` would return the raw channel. Keep one why-comment on `InterceptedCall`: `shutdown` and `shutdownNow` must return this wrapper. - -- [ ] **Step 4: Run test to verify it passes** - -Run: `./gradlew -Pversion=0.13.23 :grpc-client-utils:test --tests org.hypertrace.core.grpcutils.client.InterceptingChannelRegistryTest` - -Expected: PASS. - -- [ ] **Step 5: Commit** - -```bash -git add grpc-client-utils/src/main/java/org/hypertrace/core/grpcutils/client/InterceptingChannelRegistry.java grpc-client-utils/src/test/java/org/hypertrace/core/grpcutils/client/InterceptingChannelRegistryTest.java -git commit -m "$(cat <<'EOF' -Add a channel-registry decorator that installs one client interceptor. - -EOF -)" -``` - -Do not push. - ---- - -### Task 2: TimeoutChannelRegistry - -**Files:** -- Modify: `grpc-client-utils/build.gradle.kts` -- Create: `grpc-client-utils/src/main/java/org/hypertrace/core/grpcutils/client/TimeoutChannelRegistry.java` -- Create: `grpc-client-utils/src/test/java/org/hypertrace/core/grpcutils/client/TimeoutChannelRegistryTest.java` - -**Interfaces:** -- Consumes: `InterceptingChannelRegistry.wrap(GrpcChannelRegistry, ClientInterceptor)` and `DefaultTimeoutClientInterceptor(Duration)` -- Produces: `public final class TimeoutChannelRegistry` with `public static GrpcChannelRegistry wrap(GrpcChannelRegistry registry, Config config)`. Config path is `grpc.client.timeout`. Fields are `enabled` (boolean) and `defaultTimeout` (duration). - -- [ ] **Step 1: Write the failing test** - -Cover these cases: - -- Missing `grpc.client.timeout` returns the same registry instance. -- `enabled = false` returns the same registry instance. -- `enabled = true` without `defaultTimeout` throws `com.typesafe.config.ConfigException`. -- `enabled = true` and `defaultTimeout = 1500ms` returns a channel whose call options contain a deadline about 1500ms out when the incoming `CallOptions` and `Context` have no deadline. -- A call that already has a deadline keeps that deadline. - -Parse config with `ConfigFactory.parseString`. Use a mock delegate registry and a mock raw `ManagedChannel` whose `newCall` captures `CallOptions`. - -- [ ] **Step 2: Run test to verify it fails** - -Run: `./gradlew -Pversion=0.13.23 :grpc-client-utils:test --tests org.hypertrace.core.grpcutils.client.TimeoutChannelRegistryTest` - -Expected: compile failure because `TimeoutChannelRegistry` does not exist. - -- [ ] **Step 3: Write minimal implementation** - -Add `api(commonLibs.typesafe.config)` to `grpc-client-utils/build.gradle.kts`. - -`TimeoutChannelRegistry.wrap` returns `registry` when `registry` is null, `config` is null, the path is absent, or `enabled` is missing or false. When enabled, require `defaultTimeout` via `config.getDuration` on the timeout block and return `InterceptingChannelRegistry.wrap(registry, new DefaultTimeoutClientInterceptor(timeout))`. - -- [ ] **Step 4: Run test to verify it passes** - -Run: `./gradlew -Pversion=0.13.23 :grpc-client-utils:test --tests org.hypertrace.core.grpcutils.client.TimeoutChannelRegistryTest` - -Expected: PASS. - -- [ ] **Step 5: Commit** - -```bash -git add grpc-client-utils/build.gradle.kts grpc-client-utils/src/main/java/org/hypertrace/core/grpcutils/client/TimeoutChannelRegistry.java grpc-client-utils/src/test/java/org/hypertrace/core/grpcutils/client/TimeoutChannelRegistryTest.java -git commit -m "$(cat <<'EOF' -Apply a configured default deadline to every channel from a registry. - -EOF -)" -``` - -Do not push. - ---- - -### Task 3: Configurable failure status codes - -**Files:** -- Modify: `grpc-circuitbreaker-utils/build.gradle.kts` (`implementation(commonLibs.typesafe.config)` becomes `api(commonLibs.typesafe.config)`) -- Modify: `grpc-circuitbreaker-utils/src/main/java/org/hypertrace/circuitbreaker/grpcutils/CircuitBreakerThresholds.java` -- Modify: `grpc-circuitbreaker-utils/src/main/java/org/hypertrace/circuitbreaker/grpcutils/CircuitBreakerConfigParser.java` -- Create: `grpc-circuitbreaker-utils/src/test/java/org/hypertrace/circuitbreaker/grpcutils/CircuitBreakerConfigParserTest.java` - -**Interfaces:** -- Consumes: existing `CircuitBreakerConfigParser.parseConfig(Config)` -- Produces: `CircuitBreakerThresholds.failureStatusCodes`, a `Set` whose builder default is `DEADLINE_EXCEEDED`, `UNAVAILABLE`, and `RESOURCE_EXHAUSTED`. `parseConfig` reads `failureStatusCodes` inside a threshold block. An unknown name throws `com.typesafe.config.ConfigException.BadValue`. - -- [ ] **Step 1: Write the failing test** - -`CircuitBreakerConfigParserTest` asserts: - -- A config with only `enabled = true` yields default thresholds whose `failureStatusCodes` are exactly those three codes and whose `slowCallDurationThreshold` is `Duration.ofSeconds(2)`. -- `defaultThresholds { failureStatusCodes = [UNAVAILABLE] }` replaces the set with only `UNAVAILABLE`. -- `defaultThresholds { failureStatusCodes = [NOT_A_STATUS] }` throws `ConfigException.BadValue`. - -- [ ] **Step 2: Run test to verify it fails** - -Run: `./gradlew -Pversion=0.13.23 :grpc-circuitbreaker-utils:test --tests org.hypertrace.circuitbreaker.grpcutils.CircuitBreakerConfigParserTest` - -Expected: compile failure because `getFailureStatusCodes` does not exist. - -- [ ] **Step 3: Write minimal implementation** - -Add the field and parse `failureStatusCodes` with `config.getStringList` and `Status.Code.valueOf`. Catch `IllegalArgumentException` and throw `new ConfigException.BadValue("failureStatusCodes", name)`. Leave `ResilienceCircuitBreakerConfigConverter` unchanged so the request-keyed interceptor still records every non-OK status. - -Change the Typesafe dependency from `implementation` to `api`. - -- [ ] **Step 4: Run test to verify it passes** - -Run: `./gradlew -Pversion=0.13.23 :grpc-circuitbreaker-utils:test` - -Expected: the new test and the existing resilience tests pass. - -- [ ] **Step 5: Commit** - -```bash -git add grpc-circuitbreaker-utils/build.gradle.kts grpc-circuitbreaker-utils/src/main/java/org/hypertrace/circuitbreaker/grpcutils/CircuitBreakerThresholds.java grpc-circuitbreaker-utils/src/main/java/org/hypertrace/circuitbreaker/grpcutils/CircuitBreakerConfigParser.java grpc-circuitbreaker-utils/src/test/java/org/hypertrace/circuitbreaker/grpcutils/CircuitBreakerConfigParserTest.java -git commit -m "$(cat <<'EOF' -Parse the gRPC statuses that count as circuit-breaker failures. - -EOF -)" -``` - -Do not push. - ---- - -### Task 4: Service-keyed circuit breaker interceptor - -**Files:** -- Create: `grpc-circuitbreaker-utils/src/main/java/org/hypertrace/circuitbreaker/grpcutils/ServiceCircuitBreakerInterceptor.java` -- Modify: `grpc-circuitbreaker-utils/src/main/java/org/hypertrace/circuitbreaker/grpcutils/CircuitBreakerInterceptorFactory.java` -- Create: `grpc-circuitbreaker-utils/src/test/java/org/hypertrace/circuitbreaker/grpcutils/ServiceCircuitBreakerInterceptorTest.java` - -**Interfaces:** -- Consumes: `CircuitBreakerThresholds`, `CircuitBreakerConfigParser.parseConfig`, `CircuitBreakerConfigParser.ENABLED`, `CircuitBreakerConfigParser.DEFAULT_THRESHOLDS` -- Produces: `public static ClientInterceptor CircuitBreakerInterceptorFactory.forService(Config rootConfig)`. The breaker key is `method.getServiceName()`. Config block is `circuit.breaker.config`. - -- [ ] **Step 1: Write the failing test** - -Use an in-process or mocked `Channel`. Drive calls through `interceptCall`. Assert: - -- Absent config and `enabled = false` call the delegate. -- Enabled config with `"ai.traceable.platform.insights.api.v1.InsightsService" { }` calls the delegate for that service. A second service name calls the delegate and does not share failures. -- After enough `UNAVAILABLE` completions to open the breaker (`minimumNumberOfCalls = 1`, `failureRateThreshold = 1`, `slidingWindowSize = 1`, `slidingWindowType = COUNT_BASED`, `waitDurationInOpenState = 60s`), the next call throws `StatusRuntimeException` with `RESOURCE_EXHAUSTED` and the delegate is not called. -- A status absent from `failureStatusCodes`, such as `INTERNAL`, does not open that breaker. Use `minimumNumberOfCalls = 1` and only `INTERNAL` responses. -- `failureStatusCodes = [NOT_A_STATUS]` throws `ConfigException.BadValue` from `forService`. -- An omitted service `enabled` inherits true. `enabled = false` on that service passes through and records nothing. - -Set `slowCallDurationThreshold = 1h` in these tests so latency cannot open the breaker. - -- [ ] **Step 2: Run test to verify it fails** - -Run: `./gradlew -Pversion=0.13.23 :grpc-circuitbreaker-utils:test --tests org.hypertrace.circuitbreaker.grpcutils.ServiceCircuitBreakerInterceptorTest` - -Expected: compile failure because `forService` does not exist. - -- [ ] **Step 3: Write minimal implementation** - -`forService` delegates to `ServiceCircuitBreakerInterceptor.create(rootConfig)`. - -Parse root key `circuit.breaker.config`. When the block is absent, or `enabled` is absent or false, the interceptor calls `next.newCall` immediately. - -For each other root key, read `config.getConfig(ConfigUtil.joinPath(serviceName)).withFallback(defaultThresholds)`. Build default thresholds from `defaultThresholds` with a fallback to `CircuitBreakerThresholds.builder().build()` expressed as config, or by copying builder values. A service whose resolved `enabled` is false is omitted. Store `Map` keyed by the original service name. `recordException` returns true only for that service's `failureStatusCodes`. - -On each call, `circuitBreakerConfigs.get(serviceName)` missing means pass through. Otherwise `CircuitBreakerRegistry.circuitBreaker(serviceName, config)`, then `tryAcquirePermission`. Rejection throws `Status.RESOURCE_EXHAUSTED.withDescription(...)`. If `next.newCall` throws, call `releasePermission` and rethrow. - -The listener records elapsed nanos. `status.isOk()` and statuses outside `failureStatusCodes` call `onSuccess`. Configured failure statuses call `onError` with `status.asRuntimeException()`. - -Keep the why-comment that `getConfig(key)` treats dots as path separators, so service names are read with `joinPath`. - -- [ ] **Step 4: Run test to verify it passes** - -Run: `./gradlew -Pversion=0.13.23 :grpc-circuitbreaker-utils:test --tests org.hypertrace.circuitbreaker.grpcutils.ServiceCircuitBreakerInterceptorTest` - -Expected: PASS. - -- [ ] **Step 5: Commit** - -```bash -git add grpc-circuitbreaker-utils/src/main/java/org/hypertrace/circuitbreaker/grpcutils/ServiceCircuitBreakerInterceptor.java grpc-circuitbreaker-utils/src/main/java/org/hypertrace/circuitbreaker/grpcutils/CircuitBreakerInterceptorFactory.java grpc-circuitbreaker-utils/src/test/java/org/hypertrace/circuitbreaker/grpcutils/ServiceCircuitBreakerInterceptorTest.java -git commit -m "$(cat <<'EOF' -Add a service-keyed gRPC circuit breaker interceptor. - -EOF -)" -``` - -Do not push. - ---- - -### Task 5: CircuitBreakerChannelRegistry - -**Files:** -- Modify: `grpc-circuitbreaker-utils/build.gradle.kts` (add `api(projects.grpcClientUtils)`) -- Create: `grpc-circuitbreaker-utils/src/main/java/org/hypertrace/circuitbreaker/grpcutils/CircuitBreakerChannelRegistry.java` -- Create: `grpc-circuitbreaker-utils/src/test/java/org/hypertrace/circuitbreaker/grpcutils/CircuitBreakerChannelRegistryTest.java` - -**Interfaces:** -- Consumes: `InterceptingChannelRegistry.wrap(GrpcChannelRegistry, ClientInterceptor)` and `CircuitBreakerInterceptorFactory.forService(Config)` -- Produces: `public final class CircuitBreakerChannelRegistry` with `public static GrpcChannelRegistry wrap(GrpcChannelRegistry registry, Config config)` - -- [ ] **Step 1: Write the failing test** - -Assert: - -- Null config, empty config, and `circuit.breaker.config { enabled = false }` return the same registry instance. -- `enabled = true` with a quoted Insights service returns a different registry. A call to that service with `minimumNumberOfCalls = 1` and one `UNAVAILABLE` completion makes the following call throw `RESOURCE_EXHAUSTED` without another delegate `newCall`. -- A second plaintext lookup of the same delegate channel returns the same intercepted channel. - -- [ ] **Step 2: Run test to verify it fails** - -Run: `./gradlew -Pversion=0.13.23 :grpc-circuitbreaker-utils:test --tests org.hypertrace.circuitbreaker.grpcutils.CircuitBreakerChannelRegistryTest` - -Expected: compile failure because `CircuitBreakerChannelRegistry` does not exist. - -- [ ] **Step 3: Write minimal implementation** - -```java -public final class CircuitBreakerChannelRegistry { - static final String CIRCUIT_BREAKER_CONFIG = "circuit.breaker.config"; - - private CircuitBreakerChannelRegistry() {} - - public static GrpcChannelRegistry wrap(GrpcChannelRegistry registry, Config config) { - if (registry == null || config == null || !config.hasPath(CIRCUIT_BREAKER_CONFIG)) { - return registry; - } - Config circuitBreakerConfig = config.getConfig(CIRCUIT_BREAKER_CONFIG); - if (!circuitBreakerConfig.hasPath("enabled") || !circuitBreakerConfig.getBoolean("enabled")) { - return registry; - } - return InterceptingChannelRegistry.wrap( - registry, CircuitBreakerInterceptorFactory.forService(config)); - } -} -``` - -- [ ] **Step 4: Run test to verify it passes** - -Run: `./gradlew -Pversion=0.13.23 :grpc-client-utils:test :grpc-circuitbreaker-utils:test` - -Expected: PASS for both modules. - -- [ ] **Step 5: Commit** - -```bash -git add grpc-circuitbreaker-utils/build.gradle.kts grpc-circuitbreaker-utils/src/main/java/org/hypertrace/circuitbreaker/grpcutils/CircuitBreakerChannelRegistry.java grpc-circuitbreaker-utils/src/test/java/org/hypertrace/circuitbreaker/grpcutils/CircuitBreakerChannelRegistryTest.java -git commit -m "$(cat <<'EOF' -Install the service-keyed circuit breaker from an existing channel registry. - -EOF -)" -``` - -Do not push. diff --git a/grpc-circuitbreaker-utils/src/main/java/org/hypertrace/circuitbreaker/grpcutils/CircuitBreakerChannelRegistry.java b/grpc-circuitbreaker-utils/src/main/java/org/hypertrace/circuitbreaker/grpcutils/CircuitBreakerChannelRegistry.java index 9ee46a5..7b89e46 100644 --- a/grpc-circuitbreaker-utils/src/main/java/org/hypertrace/circuitbreaker/grpcutils/CircuitBreakerChannelRegistry.java +++ b/grpc-circuitbreaker-utils/src/main/java/org/hypertrace/circuitbreaker/grpcutils/CircuitBreakerChannelRegistry.java @@ -1,21 +1,26 @@ package org.hypertrace.circuitbreaker.grpcutils; import com.typesafe.config.Config; +import lombok.AccessLevel; +import lombok.NoArgsConstructor; +import lombok.extern.slf4j.Slf4j; import org.hypertrace.core.grpcutils.client.GrpcChannelRegistry; import org.hypertrace.core.grpcutils.client.InterceptingChannelRegistry; +@Slf4j +@NoArgsConstructor(access = AccessLevel.PRIVATE) public final class CircuitBreakerChannelRegistry { - static final String CIRCUIT_BREAKER_CONFIG = "circuit.breaker.config"; + static final String CIRCUIT_BREAKER_CONFIG = CircuitBreakerConfigParser.CIRCUIT_BREAKER_CONFIG; - private CircuitBreakerChannelRegistry() {} - - public static GrpcChannelRegistry wrap(GrpcChannelRegistry registry, Config config) { + public static GrpcChannelRegistry wrap(final GrpcChannelRegistry registry, final Config config) { if (registry == null || config == null || !config.hasPath(CIRCUIT_BREAKER_CONFIG)) { + log.debug("circuit.breaker.config is absent; leaving the channel registry unchanged"); return registry; } - Config circuitBreakerConfig = config.getConfig(CIRCUIT_BREAKER_CONFIG); + final Config circuitBreakerConfig = config.getConfig(CIRCUIT_BREAKER_CONFIG); if (!circuitBreakerConfig.hasPath("enabled") || !circuitBreakerConfig.getBoolean("enabled")) { + log.debug("circuit.breaker.config is disabled; leaving the channel registry unchanged"); return registry; } return InterceptingChannelRegistry.wrap( diff --git a/grpc-circuitbreaker-utils/src/main/java/org/hypertrace/circuitbreaker/grpcutils/CircuitBreakerConfigParser.java b/grpc-circuitbreaker-utils/src/main/java/org/hypertrace/circuitbreaker/grpcutils/CircuitBreakerConfigParser.java index a2e3e95..d42ac31 100644 --- a/grpc-circuitbreaker-utils/src/main/java/org/hypertrace/circuitbreaker/grpcutils/CircuitBreakerConfigParser.java +++ b/grpc-circuitbreaker-utils/src/main/java/org/hypertrace/circuitbreaker/grpcutils/CircuitBreakerConfigParser.java @@ -2,8 +2,10 @@ import com.typesafe.config.Config; import com.typesafe.config.ConfigException; +import com.typesafe.config.ConfigUtil; import io.grpc.Status; import java.util.EnumSet; +import java.util.LinkedHashMap; import java.util.List; import java.util.Map; import java.util.Set; @@ -31,6 +33,7 @@ public class CircuitBreakerConfigParser { "permittedNumberOfCallsInHalfOpenState"; private static final String SLIDING_WINDOW_TYPE = "slidingWindowType"; private static final String FAILURE_STATUS_CODES = "failureStatusCodes"; + static final String CIRCUIT_BREAKER_CONFIG = "circuit.breaker.config"; public static final String ENABLED = "enabled"; public static final String DEFAULT_THRESHOLDS = "defaultThresholds"; private static final Set NON_THRESHOLD_KEYS = Set.of(ENABLED, DEFAULT_THRESHOLDS); @@ -61,9 +64,52 @@ public static CircuitBreakerConfiguration.CircuitBreakerConfigurationBuilder return builder; } + /** + * Service-keyed thresholds for {@code circuit.breaker.config}. An absent or disabled block is an + * empty map. Defaults come from {@link CircuitBreakerThresholds}; config only overlays fields + * that are present. + */ + public static Map parseServiceThresholds(Config rootConfig) { + if (rootConfig == null || !rootConfig.hasPath(CIRCUIT_BREAKER_CONFIG)) { + log.debug("circuit.breaker.config is absent; no service circuit breakers installed"); + return Map.of(); + } + Config config = rootConfig.getConfig(CIRCUIT_BREAKER_CONFIG); + if (!config.hasPath(ENABLED) || !config.getBoolean(ENABLED)) { + log.debug("circuit.breaker.config is disabled; no service circuit breakers installed"); + return Map.of(); + } + CircuitBreakerThresholds defaults = + config.hasPath(DEFAULT_THRESHOLDS) + ? overlay(buildCircuitBreakerDefaultThresholds(), config.getConfig(DEFAULT_THRESHOLDS)) + : buildCircuitBreakerDefaultThresholds(); + Map thresholdsByService = new LinkedHashMap<>(); + for (String serviceName : config.root().keySet()) { + if (ENABLED.equals(serviceName) || DEFAULT_THRESHOLDS.equals(serviceName)) { + continue; + } + // getConfig(key) treats dots as path separators, so service names are read with joinPath. + CircuitBreakerThresholds thresholds = + overlay(defaults, config.getConfig(ConfigUtil.joinPath(serviceName))); + if (!thresholds.isEnabled()) { + log.debug("Service circuit breaker disabled for {}", serviceName); + continue; + } + thresholdsByService.put(serviceName, thresholds); + } + return Map.copyOf(thresholdsByService); + } + + public static CircuitBreakerThresholds overlay(CircuitBreakerThresholds base, Config overrides) { + return applyOverrides(base.toBuilder(), overrides).build(); + } + private static CircuitBreakerThresholds buildCircuitBreakerThresholds(Config config) { - CircuitBreakerThresholds.CircuitBreakerThresholdsBuilder builder = - CircuitBreakerThresholds.builder(); + return overlay(buildCircuitBreakerDefaultThresholds(), config); + } + + private static CircuitBreakerThresholds.CircuitBreakerThresholdsBuilder applyOverrides( + CircuitBreakerThresholds.CircuitBreakerThresholdsBuilder builder, Config config) { if (config.hasPath(FAILURE_RATE_THRESHOLD)) { builder.failureRateThreshold((float) config.getDouble(FAILURE_RATE_THRESHOLD)); @@ -107,7 +153,7 @@ private static CircuitBreakerThresholds buildCircuitBreakerThresholds(Config con parseFailureStatusCodes(config.getStringList(FAILURE_STATUS_CODES))); } - return builder.build(); + return builder; } // An unknown name must fail startup. Ignoring it would protect the wrong calls. diff --git a/grpc-circuitbreaker-utils/src/main/java/org/hypertrace/circuitbreaker/grpcutils/CircuitBreakerInterceptorFactory.java b/grpc-circuitbreaker-utils/src/main/java/org/hypertrace/circuitbreaker/grpcutils/CircuitBreakerInterceptorFactory.java index 9f16be6..aea7c5e 100644 --- a/grpc-circuitbreaker-utils/src/main/java/org/hypertrace/circuitbreaker/grpcutils/CircuitBreakerInterceptorFactory.java +++ b/grpc-circuitbreaker-utils/src/main/java/org/hypertrace/circuitbreaker/grpcutils/CircuitBreakerInterceptorFactory.java @@ -4,6 +4,7 @@ import io.grpc.ClientInterceptor; import jakarta.inject.Inject; import java.time.Clock; +import java.util.Map; import lombok.AllArgsConstructor; import org.hypertrace.circuitbreaker.grpcutils.resilience.ResilienceCircuitBreakerFactory; @@ -12,7 +13,12 @@ public class CircuitBreakerInterceptorFactory { private final Clock clock; public static ClientInterceptor forService(Config rootConfig) { - return ServiceCircuitBreakerInterceptor.create(rootConfig); + return forService(CircuitBreakerConfigParser.parseServiceThresholds(rootConfig)); + } + + public static ClientInterceptor forService( + Map thresholdsByService) { + return ServiceCircuitBreakerInterceptor.create(thresholdsByService); } public CircuitBreakerInterceptor buildInterceptor(CircuitBreakerConfiguration configuration) { diff --git a/grpc-circuitbreaker-utils/src/main/java/org/hypertrace/circuitbreaker/grpcutils/CircuitBreakerThresholds.java b/grpc-circuitbreaker-utils/src/main/java/org/hypertrace/circuitbreaker/grpcutils/CircuitBreakerThresholds.java index aedacf3..1bbeafd 100644 --- a/grpc-circuitbreaker-utils/src/main/java/org/hypertrace/circuitbreaker/grpcutils/CircuitBreakerThresholds.java +++ b/grpc-circuitbreaker-utils/src/main/java/org/hypertrace/circuitbreaker/grpcutils/CircuitBreakerThresholds.java @@ -7,7 +7,7 @@ import lombok.Value; @Value -@Builder +@Builder(toBuilder = true) public class CircuitBreakerThresholds { // Percentage of failures to trigger OPEN state @Builder.Default float failureRateThreshold = 50f; diff --git a/grpc-circuitbreaker-utils/src/main/java/org/hypertrace/circuitbreaker/grpcutils/ServiceCircuitBreakerInterceptor.java b/grpc-circuitbreaker-utils/src/main/java/org/hypertrace/circuitbreaker/grpcutils/ServiceCircuitBreakerInterceptor.java index e30615a..856e861 100644 --- a/grpc-circuitbreaker-utils/src/main/java/org/hypertrace/circuitbreaker/grpcutils/ServiceCircuitBreakerInterceptor.java +++ b/grpc-circuitbreaker-utils/src/main/java/org/hypertrace/circuitbreaker/grpcutils/ServiceCircuitBreakerInterceptor.java @@ -1,8 +1,5 @@ package org.hypertrace.circuitbreaker.grpcutils; -import com.typesafe.config.Config; -import com.typesafe.config.ConfigFactory; -import com.typesafe.config.ConfigUtil; import io.github.resilience4j.circuitbreaker.CircuitBreaker; import io.github.resilience4j.circuitbreaker.CircuitBreakerConfig; import io.github.resilience4j.circuitbreaker.CircuitBreakerRegistry; @@ -20,110 +17,44 @@ import java.util.Map; import java.util.Set; import java.util.concurrent.TimeUnit; -import java.util.stream.Collectors; +import lombok.extern.slf4j.Slf4j; +@Slf4j public final class ServiceCircuitBreakerInterceptor implements ClientInterceptor { - private static final String CONFIG_PATH = "circuit.breaker.config"; - private static final String SYNTHETIC_SERVICE_KEY = "service"; - private final Map circuitBreakerConfigs; private final Map> failureStatusCodesByService; private final CircuitBreakerRegistry circuitBreakerRegistry; - static ClientInterceptor create(Config rootConfig) { - if (!rootConfig.hasPath(CONFIG_PATH)) { - return passthrough(); - } - Config config = rootConfig.getConfig(CONFIG_PATH); - if (!config.hasPath(CircuitBreakerConfigParser.ENABLED) - || !config.getBoolean(CircuitBreakerConfigParser.ENABLED)) { + static ClientInterceptor create(Map thresholdsByService) { + if (thresholdsByService == null || thresholdsByService.isEmpty()) { return passthrough(); } - return configured(config); - } - - private static ServiceCircuitBreakerInterceptor passthrough() { - return new ServiceCircuitBreakerInterceptor(Map.of(), Map.of()); - } - - private static ServiceCircuitBreakerInterceptor configured(Config config) { - Config defaults = libraryDefaults(); - if (config.hasPath(CircuitBreakerConfigParser.DEFAULT_THRESHOLDS)) { - defaults = - config.getConfig(CircuitBreakerConfigParser.DEFAULT_THRESHOLDS).withFallback(defaults); - } - Map circuitBreakerConfigs = new HashMap<>(); Map> failureStatusCodesByService = new HashMap<>(); - for (String serviceName : config.root().keySet()) { - if (CircuitBreakerConfigParser.ENABLED.equals(serviceName) - || CircuitBreakerConfigParser.DEFAULT_THRESHOLDS.equals(serviceName)) { - continue; - } - // getConfig(key) treats dots as path separators, so service names are read with joinPath. - Config resolved = config.getConfig(ConfigUtil.joinPath(serviceName)).withFallback(defaults); - CircuitBreakerThresholds thresholds = parseServiceThresholds(resolved); + for (Map.Entry entry : thresholdsByService.entrySet()) { + CircuitBreakerThresholds thresholds = entry.getValue(); if (!thresholds.isEnabled()) { + log.debug("Service circuit breaker disabled for {}", entry.getKey()); continue; } - circuitBreakerConfigs.put(serviceName, toCircuitBreakerConfig(thresholds)); - failureStatusCodesByService.put(serviceName, thresholds.getFailureStatusCodes()); + circuitBreakerConfigs.put(entry.getKey(), toCircuitBreakerConfig(thresholds)); + failureStatusCodesByService.put(entry.getKey(), thresholds.getFailureStatusCodes()); + } + if (circuitBreakerConfigs.isEmpty()) { + return passthrough(); } return new ServiceCircuitBreakerInterceptor( Map.copyOf(circuitBreakerConfigs), Map.copyOf(failureStatusCodesByService)); } - private static Config libraryDefaults() { - CircuitBreakerThresholds defaults = CircuitBreakerThresholds.builder().build(); - String failureStatuses = - defaults.getFailureStatusCodes().stream() - .map(Status.Code::name) - .collect(Collectors.joining(", ")); - return ConfigFactory.parseString( - "failureRateThreshold = " - + defaults.getFailureRateThreshold() - + "\n" - + "slowCallRateThreshold = " - + defaults.getSlowCallRateThreshold() - + "\n" - + "slowCallDurationThreshold = " - + defaults.getSlowCallDurationThreshold().toMillis() - + "ms\n" - + "slidingWindowType = " - + defaults.getSlidingWindowType().name() - + "\n" - + "slidingWindowSize = " - + defaults.getSlidingWindowSize() - + "\n" - + "waitDurationInOpenState = " - + defaults.getWaitDurationInOpenState().toMillis() - + "ms\n" - + "minimumNumberOfCalls = " - + defaults.getMinimumNumberOfCalls() - + "\n" - + "permittedNumberOfCallsInHalfOpenState = " - + defaults.getPermittedNumberOfCallsInHalfOpenState() - + "\n" - + "enabled = " - + defaults.isEnabled() - + "\n" - + "failureStatusCodes = [" - + failureStatuses - + "]\n"); - } - - private static CircuitBreakerThresholds parseServiceThresholds(Config resolvedServiceConfig) { - return CircuitBreakerConfigParser.parseConfig( - resolvedServiceConfig.atKey(SYNTHETIC_SERVICE_KEY)) - .build() - .getCircuitBreakerThresholdsMap() - .get(SYNTHETIC_SERVICE_KEY); + private static ServiceCircuitBreakerInterceptor passthrough() { + return new ServiceCircuitBreakerInterceptor(Map.of(), Map.of()); } - // onError is counted as a success when this predicate returns false. + // Resilience4j records onError as a success when recordException returns false. private static CircuitBreakerConfig toCircuitBreakerConfig(CircuitBreakerThresholds thresholds) { - Set failureStatusCodes = thresholds.getFailureStatusCodes(); + final Set failureStatusCodes = thresholds.getFailureStatusCodes(); return CircuitBreakerConfig.custom() .failureRateThreshold(thresholds.getFailureRateThreshold()) .slowCallRateThreshold(thresholds.getSlowCallRateThreshold()) @@ -164,21 +95,22 @@ private ServiceCircuitBreakerInterceptor( @Override public ClientCall interceptCall( MethodDescriptor method, CallOptions callOptions, Channel next) { - String serviceName = method.getServiceName(); - CircuitBreakerConfig circuitBreakerConfig = circuitBreakerConfigs.get(serviceName); + final String serviceName = method.getServiceName(); + final CircuitBreakerConfig circuitBreakerConfig = circuitBreakerConfigs.get(serviceName); if (circuitBreakerConfig == null) { return next.newCall(method, callOptions); } - CircuitBreaker circuitBreaker = + final CircuitBreaker circuitBreaker = circuitBreakerRegistry.circuitBreaker(serviceName, circuitBreakerConfig); if (!circuitBreaker.tryAcquirePermission()) { + log.debug("Circuit breaker is open for {}; rejecting the call", serviceName); throw Status.RESOURCE_EXHAUSTED .withDescription("Circuit breaker is open for " + serviceName) .asRuntimeException(); } - ClientCall delegate; + final ClientCall delegate; try { delegate = next.newCall(method, callOptions); } catch (RuntimeException exception) { @@ -195,16 +127,26 @@ private ClientCall recordingCall( return new ForwardingClientCall.SimpleForwardingClientCall<>(delegate) { @Override public void start(Listener responseListener, Metadata headers) { - long startedAtNanos = System.nanoTime(); + final long startedAtNanos = System.nanoTime(); super.start( new ForwardingClientCallListener.SimpleForwardingClientCallListener<>( responseListener) { @Override public void onClose(Status status, Metadata trailers) { - long elapsedNanos = System.nanoTime() - startedAtNanos; + final long elapsedNanos = System.nanoTime() - startedAtNanos; if (status.isOk() || !failureStatusCodes.contains(status.getCode())) { + log.trace( + "Recording success for {} after {} ns with status {}", + circuitBreaker.getName(), + elapsedNanos, + status.getCode()); circuitBreaker.onSuccess(elapsedNanos, TimeUnit.NANOSECONDS); } else { + log.debug( + "Recording failure for {} after {} ns with status {}", + circuitBreaker.getName(), + elapsedNanos, + status.getCode()); circuitBreaker.onError( elapsedNanos, TimeUnit.NANOSECONDS, status.asRuntimeException()); } diff --git a/grpc-client-utils/src/main/java/org/hypertrace/core/grpcutils/client/InterceptingChannelRegistry.java b/grpc-client-utils/src/main/java/org/hypertrace/core/grpcutils/client/InterceptingChannelRegistry.java index 7815ccc..5421bb6 100644 --- a/grpc-client-utils/src/main/java/org/hypertrace/core/grpcutils/client/InterceptingChannelRegistry.java +++ b/grpc-client-utils/src/main/java/org/hypertrace/core/grpcutils/client/InterceptingChannelRegistry.java @@ -16,11 +16,14 @@ public final class InterceptingChannelRegistry extends GrpcChannelRegistry { public static GrpcChannelRegistry wrap( - GrpcChannelRegistry registry, ClientInterceptor interceptor) { + final GrpcChannelRegistry registry, final ClientInterceptor interceptor) { return new InterceptingChannelRegistry( Objects.requireNonNull(registry), Objects.requireNonNull(interceptor)); } + // Lombok skips a delegate method only when this class overrides it. forAddress is not + // overridden: the superclass implementation calls forPlaintextAddress. Delegating it would + // return the raw channel and skip the interceptor. @Delegate(excludes = ForAddress.class) private final GrpcChannelRegistry delegate; @@ -58,23 +61,13 @@ private ManagedChannel intercept(ManagedChannel channel) { channel, raw -> new InterceptedManagedChannel(raw, this.interceptor)); } - // The inherited forAddress calls forPlaintextAddress, so delegating it would return the raw - // channel and skip the interceptor. private interface ForAddress { ManagedChannel forAddress(String host, int port); - - ManagedChannel forSecureAddress(String host, int port); - - ManagedChannel forSecureAddress(String host, int port, GrpcChannelConfig config); - - ManagedChannel forPlaintextAddress(String host, int port); - - ManagedChannel forPlaintextAddress(String host, int port, GrpcChannelConfig config); } private static final class InterceptedManagedChannel extends ManagedChannel { - @Delegate(excludes = InterceptedCall.class) - private final ManagedChannel delegate; + // Overridden methods are not delegated. shutdown and shutdownNow return this wrapper. + @Delegate private final ManagedChannel delegate; private final Channel interceptedChannel; @@ -106,16 +99,4 @@ public ManagedChannel shutdownNow() { return this; } } - - // shutdown and shutdownNow must return this wrapper, not the raw delegate channel. - private interface InterceptedCall { - ClientCall newCall( - MethodDescriptor methodDescriptor, CallOptions callOptions); - - String authority(); - - ManagedChannel shutdown(); - - ManagedChannel shutdownNow(); - } } diff --git a/grpc-client-utils/src/main/java/org/hypertrace/core/grpcutils/client/TimeoutChannelRegistry.java b/grpc-client-utils/src/main/java/org/hypertrace/core/grpcutils/client/TimeoutChannelRegistry.java index b89b3bc..e862045 100644 --- a/grpc-client-utils/src/main/java/org/hypertrace/core/grpcutils/client/TimeoutChannelRegistry.java +++ b/grpc-client-utils/src/main/java/org/hypertrace/core/grpcutils/client/TimeoutChannelRegistry.java @@ -2,26 +2,31 @@ import com.typesafe.config.Config; import java.time.Duration; +import lombok.AccessLevel; +import lombok.NoArgsConstructor; +import lombok.extern.slf4j.Slf4j; +@Slf4j +@NoArgsConstructor(access = AccessLevel.PRIVATE) public final class TimeoutChannelRegistry { private static final String TIMEOUT_PATH = "grpc.client.timeout"; private static final String ENABLED_KEY = "enabled"; private static final String DEFAULT_TIMEOUT_KEY = "defaultTimeout"; - public static GrpcChannelRegistry wrap(GrpcChannelRegistry registry, Config config) { + public static GrpcChannelRegistry wrap(final GrpcChannelRegistry registry, final Config config) { if (registry == null || config == null || !config.hasPath(TIMEOUT_PATH)) { + log.debug("grpc.client.timeout is absent; leaving the channel registry unchanged"); return registry; } - Config timeoutConfig = config.getConfig(TIMEOUT_PATH); + final Config timeoutConfig = config.getConfig(TIMEOUT_PATH); if (!timeoutConfig.hasPath(ENABLED_KEY) || !timeoutConfig.getBoolean(ENABLED_KEY)) { + log.debug("grpc.client.timeout is disabled; leaving the channel registry unchanged"); return registry; } // An enabled block with no timeout must fail at startup rather than silently apply no deadline. - Duration timeout = timeoutConfig.getDuration(DEFAULT_TIMEOUT_KEY); + final Duration timeout = timeoutConfig.getDuration(DEFAULT_TIMEOUT_KEY); return InterceptingChannelRegistry.wrap(registry, new DefaultTimeoutClientInterceptor(timeout)); } - - private TimeoutChannelRegistry() {} } From 68ee8a54183f3ff880157a9761ede0d27b4e2b92 Mon Sep 17 00:00:00 2001 From: Suresh Prakash Date: Thu, 29 Oct 2026 13:57:15 +0530 Subject: [PATCH 9/9] Let registry wrap take service thresholds directly. Callers can install the breaker from CircuitBreakerThresholds without passing Typesafe Config. Parameters in the new code are final, and a breaker logs when it changes state. Co-authored-by: Cursor --- .../CircuitBreakerChannelRegistry.java | 18 ++++--- .../grpcutils/CircuitBreakerConfigParser.java | 18 ++++--- .../CircuitBreakerInterceptorFactory.java | 4 +- .../ServiceCircuitBreakerInterceptor.java | 51 +++++++++++++------ .../CircuitBreakerChannelRegistryTest.java | 2 +- .../client/InterceptingChannelRegistry.java | 20 +++++--- 6 files changed, 70 insertions(+), 43 deletions(-) diff --git a/grpc-circuitbreaker-utils/src/main/java/org/hypertrace/circuitbreaker/grpcutils/CircuitBreakerChannelRegistry.java b/grpc-circuitbreaker-utils/src/main/java/org/hypertrace/circuitbreaker/grpcutils/CircuitBreakerChannelRegistry.java index 7b89e46..2fea39e 100644 --- a/grpc-circuitbreaker-utils/src/main/java/org/hypertrace/circuitbreaker/grpcutils/CircuitBreakerChannelRegistry.java +++ b/grpc-circuitbreaker-utils/src/main/java/org/hypertrace/circuitbreaker/grpcutils/CircuitBreakerChannelRegistry.java @@ -1,6 +1,7 @@ package org.hypertrace.circuitbreaker.grpcutils; import com.typesafe.config.Config; +import java.util.Map; import lombok.AccessLevel; import lombok.NoArgsConstructor; import lombok.extern.slf4j.Slf4j; @@ -14,16 +15,17 @@ public final class CircuitBreakerChannelRegistry { static final String CIRCUIT_BREAKER_CONFIG = CircuitBreakerConfigParser.CIRCUIT_BREAKER_CONFIG; public static GrpcChannelRegistry wrap(final GrpcChannelRegistry registry, final Config config) { - if (registry == null || config == null || !config.hasPath(CIRCUIT_BREAKER_CONFIG)) { - log.debug("circuit.breaker.config is absent; leaving the channel registry unchanged"); - return registry; - } - final Config circuitBreakerConfig = config.getConfig(CIRCUIT_BREAKER_CONFIG); - if (!circuitBreakerConfig.hasPath("enabled") || !circuitBreakerConfig.getBoolean("enabled")) { - log.debug("circuit.breaker.config is disabled; leaving the channel registry unchanged"); + return wrap(registry, CircuitBreakerConfigParser.parseServiceThresholds(config)); + } + + public static GrpcChannelRegistry wrap( + final GrpcChannelRegistry registry, + final Map thresholdsByService) { + if (registry == null || thresholdsByService == null || thresholdsByService.isEmpty()) { + log.debug("No enabled service circuit breakers; leaving the channel registry unchanged"); return registry; } return InterceptingChannelRegistry.wrap( - registry, CircuitBreakerInterceptorFactory.forService(config)); + registry, CircuitBreakerInterceptorFactory.forService(thresholdsByService)); } } diff --git a/grpc-circuitbreaker-utils/src/main/java/org/hypertrace/circuitbreaker/grpcutils/CircuitBreakerConfigParser.java b/grpc-circuitbreaker-utils/src/main/java/org/hypertrace/circuitbreaker/grpcutils/CircuitBreakerConfigParser.java index d42ac31..e7f9f25 100644 --- a/grpc-circuitbreaker-utils/src/main/java/org/hypertrace/circuitbreaker/grpcutils/CircuitBreakerConfigParser.java +++ b/grpc-circuitbreaker-utils/src/main/java/org/hypertrace/circuitbreaker/grpcutils/CircuitBreakerConfigParser.java @@ -69,27 +69,28 @@ public static CircuitBreakerConfiguration.CircuitBreakerConfigurationBuilder * empty map. Defaults come from {@link CircuitBreakerThresholds}; config only overlays fields * that are present. */ - public static Map parseServiceThresholds(Config rootConfig) { + public static Map parseServiceThresholds( + final Config rootConfig) { if (rootConfig == null || !rootConfig.hasPath(CIRCUIT_BREAKER_CONFIG)) { log.debug("circuit.breaker.config is absent; no service circuit breakers installed"); return Map.of(); } - Config config = rootConfig.getConfig(CIRCUIT_BREAKER_CONFIG); + final Config config = rootConfig.getConfig(CIRCUIT_BREAKER_CONFIG); if (!config.hasPath(ENABLED) || !config.getBoolean(ENABLED)) { log.debug("circuit.breaker.config is disabled; no service circuit breakers installed"); return Map.of(); } - CircuitBreakerThresholds defaults = + final CircuitBreakerThresholds defaults = config.hasPath(DEFAULT_THRESHOLDS) ? overlay(buildCircuitBreakerDefaultThresholds(), config.getConfig(DEFAULT_THRESHOLDS)) : buildCircuitBreakerDefaultThresholds(); - Map thresholdsByService = new LinkedHashMap<>(); - for (String serviceName : config.root().keySet()) { + final Map thresholdsByService = new LinkedHashMap<>(); + for (final String serviceName : config.root().keySet()) { if (ENABLED.equals(serviceName) || DEFAULT_THRESHOLDS.equals(serviceName)) { continue; } // getConfig(key) treats dots as path separators, so service names are read with joinPath. - CircuitBreakerThresholds thresholds = + final CircuitBreakerThresholds thresholds = overlay(defaults, config.getConfig(ConfigUtil.joinPath(serviceName))); if (!thresholds.isEnabled()) { log.debug("Service circuit breaker disabled for {}", serviceName); @@ -100,7 +101,8 @@ public static Map parseServiceThresholds(Confi return Map.copyOf(thresholdsByService); } - public static CircuitBreakerThresholds overlay(CircuitBreakerThresholds base, Config overrides) { + public static CircuitBreakerThresholds overlay( + final CircuitBreakerThresholds base, final Config overrides) { return applyOverrides(base.toBuilder(), overrides).build(); } @@ -109,7 +111,7 @@ private static CircuitBreakerThresholds buildCircuitBreakerThresholds(Config con } private static CircuitBreakerThresholds.CircuitBreakerThresholdsBuilder applyOverrides( - CircuitBreakerThresholds.CircuitBreakerThresholdsBuilder builder, Config config) { + final CircuitBreakerThresholds.CircuitBreakerThresholdsBuilder builder, final Config config) { if (config.hasPath(FAILURE_RATE_THRESHOLD)) { builder.failureRateThreshold((float) config.getDouble(FAILURE_RATE_THRESHOLD)); diff --git a/grpc-circuitbreaker-utils/src/main/java/org/hypertrace/circuitbreaker/grpcutils/CircuitBreakerInterceptorFactory.java b/grpc-circuitbreaker-utils/src/main/java/org/hypertrace/circuitbreaker/grpcutils/CircuitBreakerInterceptorFactory.java index aea7c5e..93df720 100644 --- a/grpc-circuitbreaker-utils/src/main/java/org/hypertrace/circuitbreaker/grpcutils/CircuitBreakerInterceptorFactory.java +++ b/grpc-circuitbreaker-utils/src/main/java/org/hypertrace/circuitbreaker/grpcutils/CircuitBreakerInterceptorFactory.java @@ -12,12 +12,12 @@ public class CircuitBreakerInterceptorFactory { private final Clock clock; - public static ClientInterceptor forService(Config rootConfig) { + public static ClientInterceptor forService(final Config rootConfig) { return forService(CircuitBreakerConfigParser.parseServiceThresholds(rootConfig)); } public static ClientInterceptor forService( - Map thresholdsByService) { + final Map thresholdsByService) { return ServiceCircuitBreakerInterceptor.create(thresholdsByService); } diff --git a/grpc-circuitbreaker-utils/src/main/java/org/hypertrace/circuitbreaker/grpcutils/ServiceCircuitBreakerInterceptor.java b/grpc-circuitbreaker-utils/src/main/java/org/hypertrace/circuitbreaker/grpcutils/ServiceCircuitBreakerInterceptor.java index 856e861..355b282 100644 --- a/grpc-circuitbreaker-utils/src/main/java/org/hypertrace/circuitbreaker/grpcutils/ServiceCircuitBreakerInterceptor.java +++ b/grpc-circuitbreaker-utils/src/main/java/org/hypertrace/circuitbreaker/grpcutils/ServiceCircuitBreakerInterceptor.java @@ -26,14 +26,15 @@ public final class ServiceCircuitBreakerInterceptor implements ClientInterceptor private final Map> failureStatusCodesByService; private final CircuitBreakerRegistry circuitBreakerRegistry; - static ClientInterceptor create(Map thresholdsByService) { + static ClientInterceptor create(final Map thresholdsByService) { if (thresholdsByService == null || thresholdsByService.isEmpty()) { + log.debug("No service circuit breakers configured"); return passthrough(); } - Map circuitBreakerConfigs = new HashMap<>(); - Map> failureStatusCodesByService = new HashMap<>(); - for (Map.Entry entry : thresholdsByService.entrySet()) { - CircuitBreakerThresholds thresholds = entry.getValue(); + final Map circuitBreakerConfigs = new HashMap<>(); + final Map> failureStatusCodesByService = new HashMap<>(); + for (final Map.Entry entry : thresholdsByService.entrySet()) { + final CircuitBreakerThresholds thresholds = entry.getValue(); if (!thresholds.isEnabled()) { log.debug("Service circuit breaker disabled for {}", entry.getKey()); continue; @@ -52,8 +53,8 @@ private static ServiceCircuitBreakerInterceptor passthrough() { return new ServiceCircuitBreakerInterceptor(Map.of(), Map.of()); } - // Resilience4j records onError as a success when recordException returns false. - private static CircuitBreakerConfig toCircuitBreakerConfig(CircuitBreakerThresholds thresholds) { + private static CircuitBreakerConfig toCircuitBreakerConfig( + final CircuitBreakerThresholds thresholds) { final Set failureStatusCodes = thresholds.getFailureStatusCodes(); return CircuitBreakerConfig.custom() .failureRateThreshold(thresholds.getFailureRateThreshold()) @@ -65,6 +66,7 @@ private static CircuitBreakerConfig toCircuitBreakerConfig(CircuitBreakerThresho .permittedNumberOfCallsInHalfOpenState( thresholds.getPermittedNumberOfCallsInHalfOpenState()) .minimumNumberOfCalls(thresholds.getMinimumNumberOfCalls()) + // Returning false makes Resilience4j count this onError as a success. .recordException( throwable -> throwable instanceof StatusRuntimeException @@ -74,7 +76,7 @@ private static CircuitBreakerConfig toCircuitBreakerConfig(CircuitBreakerThresho } private static CircuitBreakerConfig.SlidingWindowType slidingWindowType( - CircuitBreakerThresholds.SlidingWindowType slidingWindowType) { + final CircuitBreakerThresholds.SlidingWindowType slidingWindowType) { switch (slidingWindowType) { case COUNT_BASED: return CircuitBreakerConfig.SlidingWindowType.COUNT_BASED; @@ -85,16 +87,33 @@ private static CircuitBreakerConfig.SlidingWindowType slidingWindowType( } private ServiceCircuitBreakerInterceptor( - Map circuitBreakerConfigs, - Map> failureStatusCodesByService) { + final Map circuitBreakerConfigs, + final Map> failureStatusCodesByService) { this.circuitBreakerConfigs = circuitBreakerConfigs; this.failureStatusCodesByService = failureStatusCodesByService; this.circuitBreakerRegistry = CircuitBreakerRegistry.ofDefaults(); + circuitBreakerConfigs.forEach(this::registerStateChangeLog); + } + + private void registerStateChangeLog(final String serviceName, final CircuitBreakerConfig config) { + final CircuitBreaker circuitBreaker = + circuitBreakerRegistry.circuitBreaker(serviceName, config); + circuitBreaker + .getEventPublisher() + .onStateTransition( + event -> + log.debug( + "Circuit breaker {} changed from {} to {}", + event.getCircuitBreakerName(), + event.getStateTransition().getFromState(), + event.getStateTransition().getToState())); } @Override public ClientCall interceptCall( - MethodDescriptor method, CallOptions callOptions, Channel next) { + final MethodDescriptor method, + final CallOptions callOptions, + final Channel next) { final String serviceName = method.getServiceName(); final CircuitBreakerConfig circuitBreakerConfig = circuitBreakerConfigs.get(serviceName); if (circuitBreakerConfig == null) { @@ -121,18 +140,18 @@ public ClientCall interceptCall( } private ClientCall recordingCall( - ClientCall delegate, - CircuitBreaker circuitBreaker, - Set failureStatusCodes) { + final ClientCall delegate, + final CircuitBreaker circuitBreaker, + final Set failureStatusCodes) { return new ForwardingClientCall.SimpleForwardingClientCall<>(delegate) { @Override - public void start(Listener responseListener, Metadata headers) { + public void start(final Listener responseListener, final Metadata headers) { final long startedAtNanos = System.nanoTime(); super.start( new ForwardingClientCallListener.SimpleForwardingClientCallListener<>( responseListener) { @Override - public void onClose(Status status, Metadata trailers) { + public void onClose(final Status status, final Metadata trailers) { final long elapsedNanos = System.nanoTime() - startedAtNanos; if (status.isOk() || !failureStatusCodes.contains(status.getCode())) { log.trace( diff --git a/grpc-circuitbreaker-utils/src/test/java/org/hypertrace/circuitbreaker/grpcutils/CircuitBreakerChannelRegistryTest.java b/grpc-circuitbreaker-utils/src/test/java/org/hypertrace/circuitbreaker/grpcutils/CircuitBreakerChannelRegistryTest.java index bebc5c1..38e599b 100644 --- a/grpc-circuitbreaker-utils/src/test/java/org/hypertrace/circuitbreaker/grpcutils/CircuitBreakerChannelRegistryTest.java +++ b/grpc-circuitbreaker-utils/src/test/java/org/hypertrace/circuitbreaker/grpcutils/CircuitBreakerChannelRegistryTest.java @@ -46,7 +46,7 @@ class CircuitBreakerChannelRegistryTest { void nullConfigReturnsTheSameRegistry() { GrpcChannelRegistry registry = mock(GrpcChannelRegistry.class); - assertSame(registry, CircuitBreakerChannelRegistry.wrap(registry, null)); + assertSame(registry, CircuitBreakerChannelRegistry.wrap(registry, (Config) null)); } @Test diff --git a/grpc-client-utils/src/main/java/org/hypertrace/core/grpcutils/client/InterceptingChannelRegistry.java b/grpc-client-utils/src/main/java/org/hypertrace/core/grpcutils/client/InterceptingChannelRegistry.java index 5421bb6..830535e 100644 --- a/grpc-client-utils/src/main/java/org/hypertrace/core/grpcutils/client/InterceptingChannelRegistry.java +++ b/grpc-client-utils/src/main/java/org/hypertrace/core/grpcutils/client/InterceptingChannelRegistry.java @@ -31,32 +31,35 @@ public static GrpcChannelRegistry wrap( private final Map interceptedChannels = Collections.synchronizedMap(new IdentityHashMap<>()); - private InterceptingChannelRegistry(GrpcChannelRegistry delegate, ClientInterceptor interceptor) { + private InterceptingChannelRegistry( + final GrpcChannelRegistry delegate, final ClientInterceptor interceptor) { this.delegate = delegate; this.interceptor = interceptor; } @Override - public ManagedChannel forSecureAddress(String host, int port) { + public ManagedChannel forSecureAddress(final String host, final int port) { return this.intercept(this.delegate.forSecureAddress(host, port)); } @Override - public ManagedChannel forSecureAddress(String host, int port, GrpcChannelConfig config) { + public ManagedChannel forSecureAddress( + final String host, final int port, final GrpcChannelConfig config) { return this.intercept(this.delegate.forSecureAddress(host, port, config)); } @Override - public ManagedChannel forPlaintextAddress(String host, int port) { + public ManagedChannel forPlaintextAddress(final String host, final int port) { return this.intercept(this.delegate.forPlaintextAddress(host, port)); } @Override - public ManagedChannel forPlaintextAddress(String host, int port, GrpcChannelConfig config) { + public ManagedChannel forPlaintextAddress( + final String host, final int port, final GrpcChannelConfig config) { return this.intercept(this.delegate.forPlaintextAddress(host, port, config)); } - private ManagedChannel intercept(ManagedChannel channel) { + private ManagedChannel intercept(final ManagedChannel channel) { return this.interceptedChannels.computeIfAbsent( channel, raw -> new InterceptedManagedChannel(raw, this.interceptor)); } @@ -71,14 +74,15 @@ private static final class InterceptedManagedChannel extends ManagedChannel { private final Channel interceptedChannel; - private InterceptedManagedChannel(ManagedChannel delegate, ClientInterceptor interceptor) { + private InterceptedManagedChannel( + final ManagedChannel delegate, final ClientInterceptor interceptor) { this.delegate = delegate; this.interceptedChannel = ClientInterceptors.intercept(delegate, interceptor); } @Override public ClientCall newCall( - MethodDescriptor methodDescriptor, CallOptions callOptions) { + final MethodDescriptor methodDescriptor, final CallOptions callOptions) { return this.interceptedChannel.newCall(methodDescriptor, callOptions); }