From 3983ec92205221ec78eeefc2284bcd39ddd500a6 Mon Sep 17 00:00:00 2001 From: Quinn Klassen Date: Wed, 25 Sep 2024 08:22:53 -0700 Subject: [PATCH 1/5] Add get system info timeout --- .../WorkflowServiceStubsOptions.java | 37 ++++++++++++++++++- 1 file changed, 36 insertions(+), 1 deletion(-) diff --git a/temporal-serviceclient/src/main/java/io/temporal/serviceclient/WorkflowServiceStubsOptions.java b/temporal-serviceclient/src/main/java/io/temporal/serviceclient/WorkflowServiceStubsOptions.java index 7b9b5e092..9c4301e5f 100644 --- a/temporal-serviceclient/src/main/java/io/temporal/serviceclient/WorkflowServiceStubsOptions.java +++ b/temporal-serviceclient/src/main/java/io/temporal/serviceclient/WorkflowServiceStubsOptions.java @@ -37,6 +37,8 @@ public final class WorkflowServiceStubsOptions extends ServiceStubsOptions { /** Default RPC timeout for workflow queries */ public static final Duration DEFAULT_QUERY_RPC_TIMEOUT = Duration.ofSeconds(10); + public static final Duration DEFAULT_SYSTEM_INFO_TIMEOUT = Duration.ofSeconds(5); + private static final WorkflowServiceStubsOptions DEFAULT_INSTANCE = newBuilder().validateAndBuildWithDefaults(); @@ -54,6 +56,8 @@ public final class WorkflowServiceStubsOptions extends ServiceStubsOptions { /** Retry options for outgoing RPC calls */ private final RpcRetryOptions rpcRetryOptions; + /** Timeout for the RPC made by the client to fetch server capabilities. */ + private final Duration systemInfoTimeout; public static Builder newBuilder() { return new Builder(); @@ -72,11 +76,13 @@ private WorkflowServiceStubsOptions( boolean disableHealthCheck, Duration rpcLongPollTimeout, Duration rpcQueryTimeout, + Duration systemInfoTimeout, RpcRetryOptions rpcRetryOptions) { super(serviceStubsOptions); this.disableHealthCheck = disableHealthCheck; this.rpcLongPollTimeout = rpcLongPollTimeout; this.rpcQueryTimeout = rpcQueryTimeout; + this.systemInfoTimeout = systemInfoTimeout; this.rpcRetryOptions = rpcRetryOptions; } @@ -115,6 +121,13 @@ public RpcRetryOptions getRpcRetryOptions() { return rpcRetryOptions; } + /** + * SystemInfoTimeout is the timeout for the RPC made by the client to fetch server capabilities. + */ + public Duration getSystemInfoTimeout() { + return systemInfoTimeout; + } + @Override public boolean equals(Object o) { if (this == o) return true; @@ -123,12 +136,18 @@ public boolean equals(Object o) { return disableHealthCheck == that.disableHealthCheck && Objects.equals(rpcLongPollTimeout, that.rpcLongPollTimeout) && Objects.equals(rpcQueryTimeout, that.rpcQueryTimeout) + && Objects.equals(systemInfoTimeout, that.systemInfoTimeout) && Objects.equals(rpcRetryOptions, that.rpcRetryOptions); } @Override public int hashCode() { - return Objects.hash(disableHealthCheck, rpcLongPollTimeout, rpcQueryTimeout, rpcRetryOptions); + return Objects.hash( + disableHealthCheck, + rpcLongPollTimeout, + rpcQueryTimeout, + systemInfoTimeout, + rpcRetryOptions); } @Override @@ -140,6 +159,8 @@ public String toString() { + rpcLongPollTimeout + ", rpcQueryTimeout=" + rpcQueryTimeout + + ", systemInfoTimeout=" + + systemInfoTimeout + ", rpcRetryOptions=" + rpcRetryOptions + '}'; @@ -150,6 +171,7 @@ public static class Builder extends ServiceStubsOptions.Builder { private boolean disableHealthCheck = true; private Duration rpcLongPollTimeout = DEFAULT_POLL_RPC_TIMEOUT; private Duration rpcQueryTimeout = DEFAULT_QUERY_RPC_TIMEOUT; + private Duration systemInfoTimeout = DEFAULT_SYSTEM_INFO_TIMEOUT; private RpcRetryOptions rpcRetryOptions = DefaultStubServiceOperationRpcRetryOptions.INSTANCE; private Builder() {} @@ -160,6 +182,7 @@ private Builder(ServiceStubsOptions options) { WorkflowServiceStubsOptions castedOptions = (WorkflowServiceStubsOptions) options; this.rpcLongPollTimeout = castedOptions.rpcLongPollTimeout; this.rpcQueryTimeout = castedOptions.rpcQueryTimeout; + this.systemInfoTimeout = castedOptions.systemInfoTimeout; this.rpcRetryOptions = castedOptions.rpcRetryOptions; } } @@ -233,6 +256,16 @@ public Builder setRpcQueryTimeout(Duration rpcQueryTimeout) { return this; } + /** + * Sets the rpc timeout value RPC call to fetch server capabilities. + * + * @param timeout timeout. + */ + public Builder setSystemInfoTimeout(Duration timeout) { + this.systemInfoTimeout = Objects.requireNonNull(timeout); + return this; + } + /** * Allows customization of retry options for the outgoing RPC calls to temporal service. * @@ -280,6 +313,7 @@ public WorkflowServiceStubsOptions build() { this.disableHealthCheck, this.rpcLongPollTimeout, this.rpcQueryTimeout, + this.systemInfoTimeout, this.rpcRetryOptions); } @@ -292,6 +326,7 @@ public WorkflowServiceStubsOptions validateAndBuildWithDefaults() { this.disableHealthCheck, this.rpcLongPollTimeout, this.rpcQueryTimeout, + this.systemInfoTimeout, retryOptions); } } From 653b1e4d7093b1559b5c1107164ae3810f8dfc7e Mon Sep 17 00:00:00 2001 From: Quinn Klassen Date: Wed, 25 Sep 2024 08:50:05 -0700 Subject: [PATCH 2/5] Add systemInfoTimeout --- .../serviceclient/ChannelManager.java | 2 +- .../serviceclient/ServiceStubsOptions.java | 37 +++++++++++++++++++ .../serviceclient/SystemInfoInterceptor.java | 5 +-- .../serviceclient/ChannelManagerTest.java | 18 ++++----- 4 files changed, 49 insertions(+), 13 deletions(-) diff --git a/temporal-serviceclient/src/main/java/io/temporal/serviceclient/ChannelManager.java b/temporal-serviceclient/src/main/java/io/temporal/serviceclient/ChannelManager.java index bca4113a3..9c16d0d0e 100644 --- a/temporal-serviceclient/src/main/java/io/temporal/serviceclient/ChannelManager.java +++ b/temporal-serviceclient/src/main/java/io/temporal/serviceclient/ChannelManager.java @@ -346,7 +346,7 @@ public Supplier getServerCapabilities() { SystemInfoInterceptor.getServerCapabilitiesWithRetryOrThrow( serverCapabilitiesFuture, interceptedChannel, - deadlineFrom(options.getHealthCheckAttemptTimeout())); + deadlineFrom(options.getSystemInfoTimeout())); } private static Deadline deadlineFrom(Duration duration) { diff --git a/temporal-serviceclient/src/main/java/io/temporal/serviceclient/ServiceStubsOptions.java b/temporal-serviceclient/src/main/java/io/temporal/serviceclient/ServiceStubsOptions.java index 5ec5b9055..dd6a7497e 100644 --- a/temporal-serviceclient/src/main/java/io/temporal/serviceclient/ServiceStubsOptions.java +++ b/temporal-serviceclient/src/main/java/io/temporal/serviceclient/ServiceStubsOptions.java @@ -71,6 +71,12 @@ public class ServiceStubsOptions { */ protected final Duration healthCheckAttemptTimeout; + /** + * SystemInfoTimeout specifies how to long to wait for service response on each health check + * attempt. Default: 5s. + */ + protected final Duration systemInfoTimeout; + /** * HealthCheckTimeout defines how long client should be sending health check requests to the * server before concluding that it is unavailable. Defaults to 10s. @@ -128,6 +134,7 @@ public class ServiceStubsOptions { this.enableHttps = that.enableHttps; this.sslContext = that.sslContext; this.healthCheckAttemptTimeout = that.healthCheckAttemptTimeout; + this.systemInfoTimeout = that.systemInfoTimeout; this.healthCheckTimeout = that.healthCheckTimeout; this.enableKeepAlive = that.enableKeepAlive; this.keepAliveTime = that.keepAliveTime; @@ -150,6 +157,7 @@ public class ServiceStubsOptions { SslContext sslContext, Duration healthCheckAttemptTimeout, Duration healthCheckTimeout, + Duration systemInfoTimeout, boolean enableKeepAlive, Duration keepAliveTime, Duration keepAliveTimeout, @@ -168,6 +176,7 @@ public class ServiceStubsOptions { this.sslContext = sslContext; this.healthCheckAttemptTimeout = healthCheckAttemptTimeout; this.healthCheckTimeout = healthCheckTimeout; + this.systemInfoTimeout = systemInfoTimeout; this.enableKeepAlive = enableKeepAlive; this.keepAliveTime = keepAliveTime; this.keepAliveTimeout = keepAliveTimeout; @@ -233,6 +242,13 @@ public Duration getHealthCheckAttemptTimeout() { return healthCheckAttemptTimeout; } + /** + * @return The timeout for the RPC made by the client to fetch server capabilities. + */ + public Duration getSystemInfoTimeout() { + return systemInfoTimeout; + } + /** * @return duration of time to wait while checking server connection when creating new client */ @@ -337,6 +353,7 @@ public boolean equals(Object o) { && Objects.equals(sslContext, that.sslContext) && Objects.equals(healthCheckAttemptTimeout, that.healthCheckAttemptTimeout) && Objects.equals(healthCheckTimeout, that.healthCheckTimeout) + && Objects.equals(systemInfoTimeout, that.systemInfoTimeout) && Objects.equals(keepAliveTime, that.keepAliveTime) && Objects.equals(keepAliveTimeout, that.keepAliveTimeout) && Objects.equals(rpcTimeout, that.rpcTimeout) @@ -358,6 +375,7 @@ public int hashCode() { sslContext, healthCheckAttemptTimeout, healthCheckTimeout, + systemInfoTimeout, enableKeepAlive, keepAliveTime, keepAliveTimeout, @@ -389,6 +407,8 @@ public String toString() { + healthCheckAttemptTimeout + ", healthCheckTimeout=" + healthCheckTimeout + + ", systemInfoTimeout=" + + systemInfoTimeout + ", enableKeepAlive=" + enableKeepAlive + ", keepAliveTime=" @@ -421,6 +441,7 @@ public static class Builder> { private String target; private Consumer> channelInitializer; private Duration healthCheckAttemptTimeout; + private Duration systemInfoTimeout; private Duration healthCheckTimeout; private boolean enableKeepAlive = true; private Duration keepAliveTime = Duration.ofSeconds(30); @@ -444,6 +465,7 @@ protected Builder(ServiceStubsOptions options) { this.sslContext = options.sslContext; this.healthCheckAttemptTimeout = options.healthCheckAttemptTimeout; this.healthCheckTimeout = options.healthCheckTimeout; + this.systemInfoTimeout = options.systemInfoTimeout; this.enableKeepAlive = options.enableKeepAlive; this.keepAliveTime = options.keepAliveTime; this.keepAliveTimeout = options.keepAliveTimeout; @@ -713,6 +735,17 @@ public T setHealthCheckTimeout(Duration healthCheckTimeout) { return self(); } + /** + * Set a SystemInfoTimeout that specifies how long the client tries to fetch server + * capabilities. + * + * @return {@code this} + */ + public T setSystemInfoTimeout(Duration systemInfoTimeout) { + this.systemInfoTimeout = systemInfoTimeout; + return self(); + } + /** * Enables keep alive ping from client to the server, which can help drop abruptly closed * connections faster. @@ -796,6 +829,7 @@ public ServiceStubsOptions build() { this.sslContext, this.healthCheckAttemptTimeout, this.healthCheckTimeout, + this.systemInfoTimeout, this.enableKeepAlive, this.keepAliveTime, this.keepAliveTimeout, @@ -847,6 +881,8 @@ public ServiceStubsOptions validateAndBuildWithDefaults() { Duration healthCheckTimeout = this.healthCheckTimeout != null ? this.healthCheckTimeout : Duration.ofSeconds(10); + Duration systemInfoTimeout = + this.systemInfoTimeout != null ? this.systemInfoTimeout : Duration.ofSeconds(5); return new ServiceStubsOptions( this.channel, target, @@ -855,6 +891,7 @@ public ServiceStubsOptions validateAndBuildWithDefaults() { this.sslContext, healthCheckAttemptTimeout, healthCheckTimeout, + systemInfoTimeout, this.enableKeepAlive, this.keepAliveTime, this.keepAliveTimeout, diff --git a/temporal-serviceclient/src/main/java/io/temporal/serviceclient/SystemInfoInterceptor.java b/temporal-serviceclient/src/main/java/io/temporal/serviceclient/SystemInfoInterceptor.java index 3f65cf2a3..ca7aa45fe 100644 --- a/temporal-serviceclient/src/main/java/io/temporal/serviceclient/SystemInfoInterceptor.java +++ b/temporal-serviceclient/src/main/java/io/temporal/serviceclient/SystemInfoInterceptor.java @@ -102,7 +102,7 @@ public static Capabilities getServerCapabilitiesWithRetryOrThrow( capabilities = future.getNow(null); if (capabilities == null) { if (deadline == null) { - deadline = Deadline.after(30, TimeUnit.SECONDS); + deadline = Deadline.after(10, TimeUnit.MINUTES); } Deadline computedDeadline = deadline; RpcRetryOptions rpcRetryOptions = @@ -110,8 +110,7 @@ public static Capabilities getServerCapabilitiesWithRetryOrThrow( .setExpiration( Duration.ofMillis(computedDeadline.timeRemaining(TimeUnit.MILLISECONDS))) .validateBuildWithDefaults(); - GrpcRetryerOptions grpcRetryerOptions = - new GrpcRetryerOptions(rpcRetryOptions, computedDeadline); + GrpcRetryerOptions grpcRetryerOptions = new GrpcRetryerOptions(rpcRetryOptions, deadline); capabilities = new GrpcRetryer(Capabilities::getDefaultInstance) .retryWithResult( diff --git a/temporal-serviceclient/src/test/java/io/temporal/serviceclient/ChannelManagerTest.java b/temporal-serviceclient/src/test/java/io/temporal/serviceclient/ChannelManagerTest.java index 362b1d9e9..1433fab79 100644 --- a/temporal-serviceclient/src/test/java/io/temporal/serviceclient/ChannelManagerTest.java +++ b/temporal-serviceclient/src/test/java/io/temporal/serviceclient/ChannelManagerTest.java @@ -134,14 +134,14 @@ public void setUp() throws Exception { } @After - public void tearDown() throws Exception { + public void tearDown() { if (channelManager != null) { channelManager.shutdownNow(); } } @Test - public void testGetServerCapabilities() throws Exception { + public void testGetServerCapabilities() { Capabilities capabilities = channelManager.getServerCapabilities().get(); assertEquals(CAPABILITIES, capabilities); assertEquals(1, getSystemInfoCount.get()); @@ -150,7 +150,7 @@ public void testGetServerCapabilities() throws Exception { } @Test - public void testGetServerCapabilitiesRetry() throws Exception { + public void testGetServerCapabilitiesRetry() { getSystemInfoUnavailable.set(2); Capabilities capabilities = channelManager.getServerCapabilities().get(); assertEquals(CAPABILITIES, capabilities); @@ -160,7 +160,7 @@ public void testGetServerCapabilitiesRetry() throws Exception { } @Test - public void testGetServerCapabilitiesUnavailable() throws Exception { + public void testGetServerCapabilitiesUnavailable() { getSystemInfoUnavailable.set(Integer.MAX_VALUE); try { Capabilities unused = channelManager.getServerCapabilities().get(); @@ -174,7 +174,7 @@ public void testGetServerCapabilitiesUnavailable() throws Exception { } @Test - public void testGetServerCapabilitiesUnimplemented() throws Exception { + public void testGetServerCapabilitiesUnimplemented() { getSystemInfoUnimplemented.set(1); Capabilities capabilities = channelManager.getServerCapabilities().get(); assertEquals(Capabilities.getDefaultInstance(), capabilities); @@ -184,7 +184,7 @@ public void testGetServerCapabilitiesUnimplemented() throws Exception { } @Test - public void testGetServerCapabilitiesWithConnect() throws Exception { + public void testGetServerCapabilitiesWithConnect() { channelManager.connect(HEALTH_CHECK_NAME, Duration.ofMillis(100)); Capabilities capabilities = channelManager.getServerCapabilities().get(); assertEquals(CAPABILITIES, capabilities); @@ -194,7 +194,7 @@ public void testGetServerCapabilitiesWithConnect() throws Exception { } @Test - public void testGetServerCapabilitiesRetryWithConnect() throws Exception { + public void testGetServerCapabilitiesRetryWithConnect() { getSystemInfoUnavailable.set(2); channelManager.connect(HEALTH_CHECK_NAME, Duration.ofMillis(100)); Capabilities capabilities = channelManager.getServerCapabilities().get(); @@ -205,7 +205,7 @@ public void testGetServerCapabilitiesRetryWithConnect() throws Exception { } @Test - public void testGetServerCapabilitiesUnavailableWithConnect() throws Exception { + public void testGetServerCapabilitiesUnavailableWithConnect() { getSystemInfoUnavailable.set(Integer.MAX_VALUE); try { channelManager.connect(HEALTH_CHECK_NAME, Duration.ofMillis(100)); @@ -220,7 +220,7 @@ public void testGetServerCapabilitiesUnavailableWithConnect() throws Exception { } @Test - public void testGetServerCapabilitiesUnimplementedWithConnect() throws Exception { + public void testGetServerCapabilitiesUnimplementedWithConnect() { getSystemInfoUnimplemented.set(1); channelManager.connect(HEALTH_CHECK_NAME, Duration.ofMillis(100)); Capabilities capabilities = channelManager.getServerCapabilities().get(); From 18ce3a64793d92bb66862b7bbf428752ac500eee Mon Sep 17 00:00:00 2001 From: Quinn Klassen Date: Wed, 25 Sep 2024 09:59:20 -0700 Subject: [PATCH 3/5] Add SystemInfoTimeoutTest --- .../serviceclient/SystemInfoTimeoutTest.java | 155 ++++++++++++++++++ 1 file changed, 155 insertions(+) create mode 100644 temporal-serviceclient/src/test/java/io/temporal/serviceclient/SystemInfoTimeoutTest.java diff --git a/temporal-serviceclient/src/test/java/io/temporal/serviceclient/SystemInfoTimeoutTest.java b/temporal-serviceclient/src/test/java/io/temporal/serviceclient/SystemInfoTimeoutTest.java new file mode 100644 index 000000000..8c423ad2c --- /dev/null +++ b/temporal-serviceclient/src/test/java/io/temporal/serviceclient/SystemInfoTimeoutTest.java @@ -0,0 +1,155 @@ +package io.temporal.serviceclient; + +import static org.junit.Assert.assertEquals; + +import io.grpc.ClientInterceptor; +import io.grpc.ManagedChannel; +import io.grpc.Status; +import io.grpc.StatusRuntimeException; +import io.grpc.inprocess.InProcessChannelBuilder; +import io.grpc.inprocess.InProcessServerBuilder; +import io.grpc.stub.StreamObserver; +import io.grpc.testing.GrpcCleanupRule; +import io.temporal.api.workflowservice.v1.GetSystemInfoRequest; +import io.temporal.api.workflowservice.v1.GetSystemInfoResponse; +import io.temporal.api.workflowservice.v1.WorkflowServiceGrpc; +import java.time.Duration; +import java.util.*; +import java.util.concurrent.ArrayBlockingQueue; +import java.util.concurrent.atomic.AtomicInteger; +import org.junit.*; + +public class SystemInfoTimeoutTest { + + private static final GetSystemInfoResponse.Capabilities CAPABILITIES = + GetSystemInfoResponse.Capabilities.newBuilder().setInternalErrorDifferentiation(true).build(); + + private static final GetSystemInfoResponse GET_SYSTEM_INFO_RESPONSE = + GetSystemInfoResponse.newBuilder().setCapabilities(CAPABILITIES).build(); + + private static final RpcRetryOptions RPC_RETRY_OPTIONS = + RpcRetryOptions.newBuilder() + .setInitialInterval(Duration.ofMillis(10)) + .setBackoffCoefficient(1.0) + .setMaximumAttempts(3) + .setExpiration(Duration.ofMillis(100)) + .validateBuildWithDefaults(); + + @Rule public final GrpcCleanupRule grpcCleanupRule = new GrpcCleanupRule(); + private final AtomicInteger getSystemInfoCount = new AtomicInteger(0); + private final AbstractQueue getSystemInfoTimeout = new ArrayBlockingQueue(10); + + private final WorkflowServiceGrpc.WorkflowServiceImplBase workflowImpl = + new WorkflowServiceGrpc.WorkflowServiceImplBase() { + @Override + public void getSystemInfo( + GetSystemInfoRequest request, StreamObserver responseObserver) { + Duration timeout = getSystemInfoTimeout.poll(); + if (timeout != null) { + try { + Thread.sleep(timeout.toMillis()); + } catch (InterruptedException e) { + throw new RuntimeException(e); + } + } + getSystemInfoCount.getAndIncrement(); + responseObserver.onNext(GET_SYSTEM_INFO_RESPONSE); + responseObserver.onCompleted(); + } + }; + + private ManagedChannel managedChannel; + + @Before + public void setUp() throws Exception { + getSystemInfoCount.set(0); + String serverName = InProcessServerBuilder.generateName(); + grpcCleanupRule.register( + InProcessServerBuilder.forName(serverName) + .directExecutor() + .addService(workflowImpl) + .build() + .start()); + managedChannel = + grpcCleanupRule.register( + InProcessChannelBuilder.forName(serverName).directExecutor().build()); + } + + @Test + public void testGetServerCapabilitiesTimeoutExceeded() { + WorkflowServiceStubsOptions serviceStubsOptions = + WorkflowServiceStubsOptions.newBuilder() + .setChannel(managedChannel) + .setRpcRetryOptions(RPC_RETRY_OPTIONS) + .setSystemInfoTimeout(Duration.ofSeconds(1)) + .validateAndBuildWithDefaults(); + + ClientInterceptor deadlineInterceptor = + new GrpcDeadlineInterceptor( + serviceStubsOptions.getRpcTimeout(), + serviceStubsOptions.getRpcLongPollTimeout(), + serviceStubsOptions.getRpcQueryTimeout()); + + ChannelManager channelManager = + new ChannelManager(serviceStubsOptions, Collections.singletonList(deadlineInterceptor)); + + getSystemInfoTimeout.add(Duration.ofSeconds(2)); + + StatusRuntimeException sre = + Assert.assertThrows( + StatusRuntimeException.class, () -> channelManager.getServerCapabilities().get()); + assertEquals(Status.Code.DEADLINE_EXCEEDED, sre.getStatus().getCode()); + } + + @Test + public void testGetServerCapabilitiesRetry() { + WorkflowServiceStubsOptions serviceStubsOptions = + WorkflowServiceStubsOptions.newBuilder() + .setChannel(managedChannel) + .setRpcRetryOptions(RPC_RETRY_OPTIONS) + .setRpcTimeout(Duration.ofMillis(500)) + .setSystemInfoTimeout(Duration.ofSeconds(5)) + .validateAndBuildWithDefaults(); + + ClientInterceptor deadlineInterceptor = + new GrpcDeadlineInterceptor( + serviceStubsOptions.getRpcTimeout(), + serviceStubsOptions.getRpcLongPollTimeout(), + serviceStubsOptions.getRpcQueryTimeout()); + + ChannelManager channelManager = + new ChannelManager(serviceStubsOptions, Collections.singletonList(deadlineInterceptor)); + + getSystemInfoTimeout.add(Duration.ofSeconds(1)); + getSystemInfoTimeout.add(Duration.ofSeconds(1)); + + GetSystemInfoResponse.Capabilities capabilities = channelManager.getServerCapabilities().get(); + assertEquals(CAPABILITIES, capabilities); + assertEquals(3, getSystemInfoCount.get()); + } + + @Test + public void testGetServerCapabilitiesTimeout() { + WorkflowServiceStubsOptions serviceStubsOptions = + WorkflowServiceStubsOptions.newBuilder() + .setChannel(managedChannel) + .setRpcRetryOptions(RPC_RETRY_OPTIONS) + .setSystemInfoTimeout(Duration.ofSeconds(10)) + .validateAndBuildWithDefaults(); + + ClientInterceptor deadlineInterceptor = + new GrpcDeadlineInterceptor( + serviceStubsOptions.getRpcTimeout(), + serviceStubsOptions.getRpcLongPollTimeout(), + serviceStubsOptions.getRpcQueryTimeout()); + + ChannelManager channelManager = + new ChannelManager(serviceStubsOptions, Collections.singletonList(deadlineInterceptor)); + + getSystemInfoTimeout.add(Duration.ofSeconds(6)); + + GetSystemInfoResponse.Capabilities capabilities = channelManager.getServerCapabilities().get(); + assertEquals(CAPABILITIES, capabilities); + assertEquals(1, getSystemInfoCount.get()); + } +} From 4be842038d0347aecb5ba8122ca3207c56bde679 Mon Sep 17 00:00:00 2001 From: Quinn Klassen Date: Wed, 25 Sep 2024 10:03:30 -0700 Subject: [PATCH 4/5] cleanup --- .../serviceclient/SystemInfoInterceptor.java | 5 ++- .../WorkflowServiceStubsOptions.java | 37 +------------------ 2 files changed, 4 insertions(+), 38 deletions(-) diff --git a/temporal-serviceclient/src/main/java/io/temporal/serviceclient/SystemInfoInterceptor.java b/temporal-serviceclient/src/main/java/io/temporal/serviceclient/SystemInfoInterceptor.java index ca7aa45fe..3f65cf2a3 100644 --- a/temporal-serviceclient/src/main/java/io/temporal/serviceclient/SystemInfoInterceptor.java +++ b/temporal-serviceclient/src/main/java/io/temporal/serviceclient/SystemInfoInterceptor.java @@ -102,7 +102,7 @@ public static Capabilities getServerCapabilitiesWithRetryOrThrow( capabilities = future.getNow(null); if (capabilities == null) { if (deadline == null) { - deadline = Deadline.after(10, TimeUnit.MINUTES); + deadline = Deadline.after(30, TimeUnit.SECONDS); } Deadline computedDeadline = deadline; RpcRetryOptions rpcRetryOptions = @@ -110,7 +110,8 @@ public static Capabilities getServerCapabilitiesWithRetryOrThrow( .setExpiration( Duration.ofMillis(computedDeadline.timeRemaining(TimeUnit.MILLISECONDS))) .validateBuildWithDefaults(); - GrpcRetryerOptions grpcRetryerOptions = new GrpcRetryerOptions(rpcRetryOptions, deadline); + GrpcRetryerOptions grpcRetryerOptions = + new GrpcRetryerOptions(rpcRetryOptions, computedDeadline); capabilities = new GrpcRetryer(Capabilities::getDefaultInstance) .retryWithResult( diff --git a/temporal-serviceclient/src/main/java/io/temporal/serviceclient/WorkflowServiceStubsOptions.java b/temporal-serviceclient/src/main/java/io/temporal/serviceclient/WorkflowServiceStubsOptions.java index 9c4301e5f..7b9b5e092 100644 --- a/temporal-serviceclient/src/main/java/io/temporal/serviceclient/WorkflowServiceStubsOptions.java +++ b/temporal-serviceclient/src/main/java/io/temporal/serviceclient/WorkflowServiceStubsOptions.java @@ -37,8 +37,6 @@ public final class WorkflowServiceStubsOptions extends ServiceStubsOptions { /** Default RPC timeout for workflow queries */ public static final Duration DEFAULT_QUERY_RPC_TIMEOUT = Duration.ofSeconds(10); - public static final Duration DEFAULT_SYSTEM_INFO_TIMEOUT = Duration.ofSeconds(5); - private static final WorkflowServiceStubsOptions DEFAULT_INSTANCE = newBuilder().validateAndBuildWithDefaults(); @@ -56,8 +54,6 @@ public final class WorkflowServiceStubsOptions extends ServiceStubsOptions { /** Retry options for outgoing RPC calls */ private final RpcRetryOptions rpcRetryOptions; - /** Timeout for the RPC made by the client to fetch server capabilities. */ - private final Duration systemInfoTimeout; public static Builder newBuilder() { return new Builder(); @@ -76,13 +72,11 @@ private WorkflowServiceStubsOptions( boolean disableHealthCheck, Duration rpcLongPollTimeout, Duration rpcQueryTimeout, - Duration systemInfoTimeout, RpcRetryOptions rpcRetryOptions) { super(serviceStubsOptions); this.disableHealthCheck = disableHealthCheck; this.rpcLongPollTimeout = rpcLongPollTimeout; this.rpcQueryTimeout = rpcQueryTimeout; - this.systemInfoTimeout = systemInfoTimeout; this.rpcRetryOptions = rpcRetryOptions; } @@ -121,13 +115,6 @@ public RpcRetryOptions getRpcRetryOptions() { return rpcRetryOptions; } - /** - * SystemInfoTimeout is the timeout for the RPC made by the client to fetch server capabilities. - */ - public Duration getSystemInfoTimeout() { - return systemInfoTimeout; - } - @Override public boolean equals(Object o) { if (this == o) return true; @@ -136,18 +123,12 @@ public boolean equals(Object o) { return disableHealthCheck == that.disableHealthCheck && Objects.equals(rpcLongPollTimeout, that.rpcLongPollTimeout) && Objects.equals(rpcQueryTimeout, that.rpcQueryTimeout) - && Objects.equals(systemInfoTimeout, that.systemInfoTimeout) && Objects.equals(rpcRetryOptions, that.rpcRetryOptions); } @Override public int hashCode() { - return Objects.hash( - disableHealthCheck, - rpcLongPollTimeout, - rpcQueryTimeout, - systemInfoTimeout, - rpcRetryOptions); + return Objects.hash(disableHealthCheck, rpcLongPollTimeout, rpcQueryTimeout, rpcRetryOptions); } @Override @@ -159,8 +140,6 @@ public String toString() { + rpcLongPollTimeout + ", rpcQueryTimeout=" + rpcQueryTimeout - + ", systemInfoTimeout=" - + systemInfoTimeout + ", rpcRetryOptions=" + rpcRetryOptions + '}'; @@ -171,7 +150,6 @@ public static class Builder extends ServiceStubsOptions.Builder { private boolean disableHealthCheck = true; private Duration rpcLongPollTimeout = DEFAULT_POLL_RPC_TIMEOUT; private Duration rpcQueryTimeout = DEFAULT_QUERY_RPC_TIMEOUT; - private Duration systemInfoTimeout = DEFAULT_SYSTEM_INFO_TIMEOUT; private RpcRetryOptions rpcRetryOptions = DefaultStubServiceOperationRpcRetryOptions.INSTANCE; private Builder() {} @@ -182,7 +160,6 @@ private Builder(ServiceStubsOptions options) { WorkflowServiceStubsOptions castedOptions = (WorkflowServiceStubsOptions) options; this.rpcLongPollTimeout = castedOptions.rpcLongPollTimeout; this.rpcQueryTimeout = castedOptions.rpcQueryTimeout; - this.systemInfoTimeout = castedOptions.systemInfoTimeout; this.rpcRetryOptions = castedOptions.rpcRetryOptions; } } @@ -256,16 +233,6 @@ public Builder setRpcQueryTimeout(Duration rpcQueryTimeout) { return this; } - /** - * Sets the rpc timeout value RPC call to fetch server capabilities. - * - * @param timeout timeout. - */ - public Builder setSystemInfoTimeout(Duration timeout) { - this.systemInfoTimeout = Objects.requireNonNull(timeout); - return this; - } - /** * Allows customization of retry options for the outgoing RPC calls to temporal service. * @@ -313,7 +280,6 @@ public WorkflowServiceStubsOptions build() { this.disableHealthCheck, this.rpcLongPollTimeout, this.rpcQueryTimeout, - this.systemInfoTimeout, this.rpcRetryOptions); } @@ -326,7 +292,6 @@ public WorkflowServiceStubsOptions validateAndBuildWithDefaults() { this.disableHealthCheck, this.rpcLongPollTimeout, this.rpcQueryTimeout, - this.systemInfoTimeout, retryOptions); } } From 0ff1e306b7882c6b13c10a0a5af553acf9975b9b Mon Sep 17 00:00:00 2001 From: Quinn Klassen Date: Wed, 25 Sep 2024 12:38:03 -0700 Subject: [PATCH 5/5] add license --- .../serviceclient/SystemInfoTimeoutTest.java | 20 +++++++++++++++++++ 1 file changed, 20 insertions(+) diff --git a/temporal-serviceclient/src/test/java/io/temporal/serviceclient/SystemInfoTimeoutTest.java b/temporal-serviceclient/src/test/java/io/temporal/serviceclient/SystemInfoTimeoutTest.java index 8c423ad2c..050aa07f8 100644 --- a/temporal-serviceclient/src/test/java/io/temporal/serviceclient/SystemInfoTimeoutTest.java +++ b/temporal-serviceclient/src/test/java/io/temporal/serviceclient/SystemInfoTimeoutTest.java @@ -1,3 +1,23 @@ +/* + * Copyright (C) 2022 Temporal Technologies, Inc. All Rights Reserved. + * + * Copyright (C) 2012-2016 Amazon.com, Inc. or its affiliates. All Rights Reserved. + * + * Modifications copyright (C) 2017 Uber Technologies, Inc. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this material except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + package io.temporal.serviceclient; import static org.junit.Assert.assertEquals;