From 61d0b608427a53832b47318b6d076bee32e11719 Mon Sep 17 00:00:00 2001 From: Tony Cui Date: Mon, 13 Jul 2026 15:14:59 +0000 Subject: [PATCH 1/6] Add hedgeSettings with configurable hedgeDelay --- .../google/cloud/pubsub/v1/HedgeSettings.java | 77 +++++++++++++++++++ .../cloud/pubsub/v1/HedgeSettingsTest.java | 68 ++++++++++++++++ 2 files changed, 145 insertions(+) create mode 100644 java-pubsub/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/HedgeSettings.java create mode 100644 java-pubsub/google-cloud-pubsub/src/test/java/com/google/cloud/pubsub/v1/HedgeSettingsTest.java diff --git a/java-pubsub/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/HedgeSettings.java b/java-pubsub/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/HedgeSettings.java new file mode 100644 index 000000000000..9a2dbf66c9dd --- /dev/null +++ b/java-pubsub/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/HedgeSettings.java @@ -0,0 +1,77 @@ +/* + * Copyright 2026 Google LLC + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package com.google.cloud.pubsub.v1; + +import com.google.common.base.Preconditions; +import java.time.Duration; + +/** Settings for configuring publish hedging. */ +public final class HedgeSettings { + private static final Duration DEFAULT_DELAY = Duration.ofMillis(100); + private static final int DEFAULT_MAX_TOKENS = 100; + private static final float DEFAULT_REFILL = 0.1f; + + private final Duration hedgeDelay; + private final int maxTokens; + private final float refill; + + private HedgeSettings(Builder builder) { + this.hedgeDelay = builder.hedgeDelay; + this.maxTokens = builder.maxTokens; + this.refill = builder.refill; + } + + /** Returns the configured hedging delay. */ + public Duration getHedgeDelay() { + return hedgeDelay; + } + + int getMaxTokens() { + return maxTokens; + } + + float getRefill() { + return refill; + } + + /** Returns a new builder for {@code HedgeSettings}. */ + public static Builder newBuilder() { + return new Builder(); + } + + /** Builder for {@code HedgeSettings}. */ + public static final class Builder { + private Duration hedgeDelay = DEFAULT_DELAY; + private int maxTokens = DEFAULT_MAX_TOKENS; + private float refill = DEFAULT_REFILL; + + private Builder() {} + + /** Allows hedging delay to be configurable. */ + public Builder setHedgeDelay(Duration hedgeDelay) { + Preconditions.checkNotNull(hedgeDelay); + Preconditions.checkArgument(!hedgeDelay.isNegative() && !hedgeDelay.isZero(), "hedgeDelay must be positive"); + this.hedgeDelay = hedgeDelay; + return this; + } + + /** Builds an instance of {@code HedgeSettings}. */ + public HedgeSettings build() { + return new HedgeSettings(this); + } + } +} diff --git a/java-pubsub/google-cloud-pubsub/src/test/java/com/google/cloud/pubsub/v1/HedgeSettingsTest.java b/java-pubsub/google-cloud-pubsub/src/test/java/com/google/cloud/pubsub/v1/HedgeSettingsTest.java new file mode 100644 index 000000000000..2a01408b7bdf --- /dev/null +++ b/java-pubsub/google-cloud-pubsub/src/test/java/com/google/cloud/pubsub/v1/HedgeSettingsTest.java @@ -0,0 +1,68 @@ +/* + * Copyright 2026 Google LLC + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * 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 com.google.cloud.pubsub.v1; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertThrows; + +import java.time.Duration; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.junit.runners.JUnit4; + +@RunWith(JUnit4.class) +public class HedgeSettingsTest { + + @Test + public void testDefaultSettings() { + HedgeSettings settings = HedgeSettings.newBuilder().build(); + assertNotNull(settings); + assertEquals(Duration.ofMillis(100), settings.getHedgeDelay()); + assertEquals(100, settings.getMaxTokens()); + assertEquals(0.1f, settings.getRefill(), 0.0001f); + } + + @Test + public void testCustomDelay() { + Duration customDelay = Duration.ofMillis(50); + HedgeSettings settings = HedgeSettings.newBuilder().setHedgeDelay(customDelay).build(); + assertNotNull(settings); + assertEquals(customDelay, settings.getHedgeDelay()); + } + + @Test + public void testNegativeDelayThrows() { + assertThrows( + IllegalArgumentException.class, + () -> HedgeSettings.newBuilder().setHedgeDelay(Duration.ofMillis(-10))); + } + + @Test + public void testZeroDelayThrows() { + assertThrows( + IllegalArgumentException.class, + () -> HedgeSettings.newBuilder().setHedgeDelay(Duration.ZERO)); + } + + @Test + public void testNullDelayThrows() { + assertThrows( + NullPointerException.class, + () -> HedgeSettings.newBuilder().setHedgeDelay(null)); + } +} From 19afd68374fbd8899e1f5ecd4ad3e15ee2a001d1 Mon Sep 17 00:00:00 2001 From: Tony Cui Date: Mon, 13 Jul 2026 16:08:47 +0000 Subject: [PATCH 2/6] Add HedgeTokenBucket class to allow for token operations --- .../cloud/pubsub/v1/HedgeTokenBucket.java | 65 +++++++++++++++++ .../cloud/pubsub/v1/HedgeTokenBucketTest.java | 71 +++++++++++++++++++ 2 files changed, 136 insertions(+) create mode 100644 java-pubsub/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/HedgeTokenBucket.java create mode 100644 java-pubsub/google-cloud-pubsub/src/test/java/com/google/cloud/pubsub/v1/HedgeTokenBucketTest.java diff --git a/java-pubsub/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/HedgeTokenBucket.java b/java-pubsub/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/HedgeTokenBucket.java new file mode 100644 index 000000000000..5378478cb429 --- /dev/null +++ b/java-pubsub/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/HedgeTokenBucket.java @@ -0,0 +1,65 @@ +/* + * Copyright 2026 Google LLC + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package com.google.cloud.pubsub.v1; + +import com.google.common.annotations.VisibleForTesting; +import javax.annotation.concurrent.GuardedBy; + +/** Token bucket for limiting hedged requests. Thread-safe. */ +final class HedgeTokenBucket { + private final int maxTokens; + private final float refillAmount; + + @GuardedBy("this") + private float tokens; + + HedgeTokenBucket(HedgeSettings settings) { + this.maxTokens = settings.getMaxTokens(); + this.refillAmount = settings.getRefill(); + this.tokens = maxTokens; + } + + @VisibleForTesting + HedgeTokenBucket(int maxTokens, float refillAmount) { + this.maxTokens = maxTokens; + this.refillAmount = refillAmount; + this.tokens = maxTokens; + } + + /** + * Attempts to acquire a token for a hedged request. + * + * @return {@code true} if a token was acquired, {@code false} otherwise. + */ + synchronized boolean tryAcquire() { + if (tokens >= 1.0f) { + tokens -= 1.0f; + return true; + } + return false; + } + + /** Refills the bucket by the configured refill amount, capped at max tokens. */ + synchronized void refill() { + tokens = Math.min(maxTokens, tokens + refillAmount); + } + + @VisibleForTesting + synchronized float getTokens() { + return tokens; + } +} diff --git a/java-pubsub/google-cloud-pubsub/src/test/java/com/google/cloud/pubsub/v1/HedgeTokenBucketTest.java b/java-pubsub/google-cloud-pubsub/src/test/java/com/google/cloud/pubsub/v1/HedgeTokenBucketTest.java new file mode 100644 index 000000000000..e92ecd9a6ce2 --- /dev/null +++ b/java-pubsub/google-cloud-pubsub/src/test/java/com/google/cloud/pubsub/v1/HedgeTokenBucketTest.java @@ -0,0 +1,71 @@ +/* + * Copyright 2026 Google LLC + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * 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 com.google.cloud.pubsub.v1; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; + +import org.junit.Test; +import org.junit.runner.RunWith; +import org.junit.runners.JUnit4; + +@RunWith(JUnit4.class) +public class HedgeTokenBucketTest { + + @Test + public void testInitialState() { + HedgeTokenBucket bucket = new HedgeTokenBucket(10, 0.5f); + assertEquals(10.0f, bucket.getTokens(), 0.0001f); + } + + @Test + public void testAcquire() { + HedgeTokenBucket bucket = new HedgeTokenBucket(10, 0.5f); + assertTrue(bucket.tryAcquire()); + assertEquals(9.0f, bucket.getTokens(), 0.0001f); + } + + @Test + public void testAcquireUntilEmpty() { + HedgeTokenBucket bucket = new HedgeTokenBucket(2, 0.5f); + assertTrue(bucket.tryAcquire()); + assertTrue(bucket.tryAcquire()); + assertFalse(bucket.tryAcquire()); + assertEquals(0.0f, bucket.getTokens(), 0.0001f); + } + + @Test + public void testRefill() { + HedgeTokenBucket bucket = new HedgeTokenBucket(2, 0.5f); + assertTrue(bucket.tryAcquire()); // 1.0 left + assertTrue(bucket.tryAcquire()); // 0.0 left + + bucket.refill(); // 0.5 tokens + assertFalse(bucket.tryAcquire()); // needs 1.0 + + bucket.refill(); // 1.0 tokens + assertTrue(bucket.tryAcquire()); // succeeds, 0.0 left + } + + @Test + public void testRefillCap() { + HedgeTokenBucket bucket = new HedgeTokenBucket(2, 0.5f); + bucket.refill(); // already at max (2) + assertEquals(2.0f, bucket.getTokens(), 0.0001f); + } +} From 49426a9b490abb6a6ecb7d0f78cf26e66f1a4816 Mon Sep 17 00:00:00 2001 From: Tony Cui Date: Mon, 13 Jul 2026 16:43:24 +0000 Subject: [PATCH 3/6] Add publisher integration with HedgeSettings --- .../com/google/cloud/pubsub/v1/Publisher.java | 22 +++++++++ .../cloud/pubsub/v1/PublisherImplTest.java | 48 ++++++++++++++----- 2 files changed, 59 insertions(+), 11 deletions(-) diff --git a/java-pubsub/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/Publisher.java b/java-pubsub/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/Publisher.java index 56c920bcfdc1..1654cb364418 100644 --- a/java-pubsub/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/Publisher.java +++ b/java-pubsub/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/Publisher.java @@ -138,6 +138,9 @@ public class Publisher implements PublisherInterface { private final OpenTelemetry openTelemetry; private OpenTelemetryPubsubTracer tracer = new OpenTelemetryPubsubTracer(null, false); + private final HedgeSettings hedgeSettings; + private final HedgeTokenBucket hedgeTokenBucket; + /** The maximum number of messages in one request. Defined by the API. */ public static long getApiMaxRequestElementCount() { return 1000L; @@ -230,6 +233,9 @@ private Publisher(Builder builder) throws IOException { backgroundResources = new BackgroundResourceAggregation(backgroundResourceList); shutdown = new AtomicBoolean(false); messagesWaiter = new Waiter(); + this.hedgeSettings = builder.hedgeSettings; + this.hedgeTokenBucket = + this.hedgeSettings != null ? new HedgeTokenBucket(this.hedgeSettings) : null; this.publishContext = GrpcCallContext.createDefault(); this.publishContextWithCompression = GrpcCallContext.createDefault() @@ -246,6 +252,15 @@ public String getTopicNameString() { return topicName; } + /** Returns the configured hedging settings, or null if hedging is disabled. */ + public HedgeSettings getHedgeSettings() { + return hedgeSettings; + } + + HedgeTokenBucket getHedgeTokenBucket() { + return hedgeTokenBucket; + } + /** * Schedules the publishing of a message. The publishing of the message may occur immediately or * be delayed based on the publisher batching options. @@ -814,6 +829,7 @@ public PubsubMessage apply(PubsubMessage input) { private boolean enableOpenTelemetryTracing = false; private OpenTelemetry openTelemetry = null; + private HedgeSettings hedgeSettings = null; private Builder(String topic) { this.topicName = Preconditions.checkNotNull(topic); @@ -966,6 +982,12 @@ public Builder setOpenTelemetry(OpenTelemetry openTelemetry) { return this; } + /** Configures the Publisher's hedging parameters. */ + public Builder setHedgeSettings(HedgeSettings hedgeSettings) { + this.hedgeSettings = hedgeSettings; + return this; + } + /** Returns the default BatchingSettings used by the client if settings are not provided. */ public static BatchingSettings getDefaultBatchingSettings() { return DEFAULT_BATCHING_SETTINGS; diff --git a/java-pubsub/google-cloud-pubsub/src/test/java/com/google/cloud/pubsub/v1/PublisherImplTest.java b/java-pubsub/google-cloud-pubsub/src/test/java/com/google/cloud/pubsub/v1/PublisherImplTest.java index 8e6efaf372c9..a5bd92c2aa0d 100644 --- a/java-pubsub/google-cloud-pubsub/src/test/java/com/google/cloud/pubsub/v1/PublisherImplTest.java +++ b/java-pubsub/google-cloud-pubsub/src/test/java/com/google/cloud/pubsub/v1/PublisherImplTest.java @@ -64,7 +64,6 @@ import org.easymock.EasyMock; import org.junit.After; import org.junit.Before; -import org.junit.Ignore; import org.junit.Test; import org.junit.runner.RunWith; import org.junit.runners.JUnit4; @@ -513,20 +512,21 @@ public void testEnableMessageOrdering_overwritesMaxAttempts() throws Exception { *
  • publish with key orderA, which should now succeed * */ - @Ignore("https://github.com/googleapis/google-cloud-java/issues/13394") + /* + Temporarily disabled due to https://github.com/googleapis/java-pubsub/issues/1861. + TODO(maitrimangal): Enable once resolved. @Test public void testResumePublish() throws Exception { Publisher publisher = getTestPublisherBuilder() .setBatchingSettings( - Publisher.Builder.DEFAULT_BATCHING_SETTINGS.toBuilder() + Publisher.Builder.DEFAULT_BATCHING_SETTINGS + .toBuilder() .setElementCountThreshold(2L) .build()) .setEnableMessageOrdering(true) .build(); - testPublisherServiceImpl.setExecutor(fakeExecutor); - ApiFuture future1 = sendTestMessageWithOrderingKey(publisher, "m1", "orderA"); ApiFuture future2 = sendTestMessageWithOrderingKey(publisher, "m2", "orderA"); @@ -536,6 +536,7 @@ public void testResumePublish() throws Exception { // This exception should stop future publishing to the same key testPublisherServiceImpl.addPublishError(new StatusException(Status.INVALID_ARGUMENT)); + fakeExecutor.advanceTime(Duration.ZERO); try { @@ -575,8 +576,8 @@ public void testResumePublish() throws Exception { testPublisherServiceImpl.addPublishResponse( PublishResponse.newBuilder().addMessageIds("5").addMessageIds("6")); - assertEquals("5", future5.get()); - assertEquals("6", future6.get()); + Assert.assertEquals("5", future5.get()); + Assert.assertEquals("6", future6.get()); // Resume publishing of "orderA", which should now succeed publisher.resumePublish("orderA"); @@ -587,19 +588,20 @@ public void testResumePublish() throws Exception { testPublisherServiceImpl.addPublishResponse( PublishResponse.newBuilder().addMessageIds("7").addMessageIds("8")); - assertEquals("7", future7.get()); - assertEquals("8", future8.get()); + Assert.assertEquals("7", future7.get()); + Assert.assertEquals("8", future8.get()); shutdownTestPublisher(publisher); } - @Ignore("https://github.com/googleapis/google-cloud-java/issues/13394") @Test public void testPublishThrowExceptionForUnsubmittedOrderingKeyMessage() throws Exception { Publisher publisher = getTestPublisherBuilder() + .setExecutorProvider(SINGLE_THREAD_EXECUTOR) .setBatchingSettings( - Publisher.Builder.DEFAULT_BATCHING_SETTINGS.toBuilder() + Publisher.Builder.DEFAULT_BATCHING_SETTINGS + .toBuilder() .setElementCountThreshold(2L) .setDelayThresholdDuration(Duration.ofSeconds(500)) .build()) @@ -642,6 +644,7 @@ public void testPublishThrowExceptionForUnsubmittedOrderingKeyMessage() throws E assertEquals(SequentialExecutorService.CallbackExecutor.CANCELLATION_EXCEPTION, e.getCause()); } } + */ private ApiFuture sendTestMessageWithOrderingKey( Publisher publisher, String data, String orderingKey) { @@ -1339,6 +1342,29 @@ public void testPublishOpenTelemetryTracing() throws Exception { .hasEnded(); } + @Test + public void testPublisherWithHedgeSettings() throws Exception { + HedgeSettings hedgeSettings = + HedgeSettings.newBuilder().setHedgeDelay(Duration.ofMillis(50)).build(); + Publisher publisher = getTestPublisherBuilder().setHedgeSettings(hedgeSettings).build(); + + assertThat(publisher.getHedgeSettings()).isEqualTo(hedgeSettings); + assertThat(publisher.getHedgeTokenBucket()).isNotNull(); + assertThat(publisher.getHedgeTokenBucket().getTokens()).isWithin(0.0001f).of(100.0f); + + shutdownTestPublisher(publisher); + } + + @Test + public void testPublisherWithoutHedgeSettings() throws Exception { + Publisher publisher = getTestPublisherBuilder().build(); + + assertThat(publisher.getHedgeSettings()).isNull(); + assertThat(publisher.getHedgeTokenBucket()).isNull(); + + shutdownTestPublisher(publisher); + } + private Builder getTestPublisherBuilder() { return Publisher.newBuilder(TEST_TOPIC) .setExecutorProvider(FixedExecutorProvider.create(fakeExecutor)) From 22417d156d31e10398f225c6f98ccebae9f93595 Mon Sep 17 00:00:00 2001 From: Tony Cui Date: Mon, 13 Jul 2026 16:51:32 +0000 Subject: [PATCH 4/6] Restore original @Ignore annotations and code in PublisherImplTest --- .../cloud/pubsub/v1/PublisherImplTest.java | 48 +++++-------------- 1 file changed, 11 insertions(+), 37 deletions(-) diff --git a/java-pubsub/google-cloud-pubsub/src/test/java/com/google/cloud/pubsub/v1/PublisherImplTest.java b/java-pubsub/google-cloud-pubsub/src/test/java/com/google/cloud/pubsub/v1/PublisherImplTest.java index a5bd92c2aa0d..8e6efaf372c9 100644 --- a/java-pubsub/google-cloud-pubsub/src/test/java/com/google/cloud/pubsub/v1/PublisherImplTest.java +++ b/java-pubsub/google-cloud-pubsub/src/test/java/com/google/cloud/pubsub/v1/PublisherImplTest.java @@ -64,6 +64,7 @@ import org.easymock.EasyMock; import org.junit.After; import org.junit.Before; +import org.junit.Ignore; import org.junit.Test; import org.junit.runner.RunWith; import org.junit.runners.JUnit4; @@ -512,21 +513,20 @@ public void testEnableMessageOrdering_overwritesMaxAttempts() throws Exception { *
  • publish with key orderA, which should now succeed * */ - /* - Temporarily disabled due to https://github.com/googleapis/java-pubsub/issues/1861. - TODO(maitrimangal): Enable once resolved. + @Ignore("https://github.com/googleapis/google-cloud-java/issues/13394") @Test public void testResumePublish() throws Exception { Publisher publisher = getTestPublisherBuilder() .setBatchingSettings( - Publisher.Builder.DEFAULT_BATCHING_SETTINGS - .toBuilder() + Publisher.Builder.DEFAULT_BATCHING_SETTINGS.toBuilder() .setElementCountThreshold(2L) .build()) .setEnableMessageOrdering(true) .build(); + testPublisherServiceImpl.setExecutor(fakeExecutor); + ApiFuture future1 = sendTestMessageWithOrderingKey(publisher, "m1", "orderA"); ApiFuture future2 = sendTestMessageWithOrderingKey(publisher, "m2", "orderA"); @@ -536,7 +536,6 @@ public void testResumePublish() throws Exception { // This exception should stop future publishing to the same key testPublisherServiceImpl.addPublishError(new StatusException(Status.INVALID_ARGUMENT)); - fakeExecutor.advanceTime(Duration.ZERO); try { @@ -576,8 +575,8 @@ public void testResumePublish() throws Exception { testPublisherServiceImpl.addPublishResponse( PublishResponse.newBuilder().addMessageIds("5").addMessageIds("6")); - Assert.assertEquals("5", future5.get()); - Assert.assertEquals("6", future6.get()); + assertEquals("5", future5.get()); + assertEquals("6", future6.get()); // Resume publishing of "orderA", which should now succeed publisher.resumePublish("orderA"); @@ -588,20 +587,19 @@ public void testResumePublish() throws Exception { testPublisherServiceImpl.addPublishResponse( PublishResponse.newBuilder().addMessageIds("7").addMessageIds("8")); - Assert.assertEquals("7", future7.get()); - Assert.assertEquals("8", future8.get()); + assertEquals("7", future7.get()); + assertEquals("8", future8.get()); shutdownTestPublisher(publisher); } + @Ignore("https://github.com/googleapis/google-cloud-java/issues/13394") @Test public void testPublishThrowExceptionForUnsubmittedOrderingKeyMessage() throws Exception { Publisher publisher = getTestPublisherBuilder() - .setExecutorProvider(SINGLE_THREAD_EXECUTOR) .setBatchingSettings( - Publisher.Builder.DEFAULT_BATCHING_SETTINGS - .toBuilder() + Publisher.Builder.DEFAULT_BATCHING_SETTINGS.toBuilder() .setElementCountThreshold(2L) .setDelayThresholdDuration(Duration.ofSeconds(500)) .build()) @@ -644,7 +642,6 @@ public void testPublishThrowExceptionForUnsubmittedOrderingKeyMessage() throws E assertEquals(SequentialExecutorService.CallbackExecutor.CANCELLATION_EXCEPTION, e.getCause()); } } - */ private ApiFuture sendTestMessageWithOrderingKey( Publisher publisher, String data, String orderingKey) { @@ -1342,29 +1339,6 @@ public void testPublishOpenTelemetryTracing() throws Exception { .hasEnded(); } - @Test - public void testPublisherWithHedgeSettings() throws Exception { - HedgeSettings hedgeSettings = - HedgeSettings.newBuilder().setHedgeDelay(Duration.ofMillis(50)).build(); - Publisher publisher = getTestPublisherBuilder().setHedgeSettings(hedgeSettings).build(); - - assertThat(publisher.getHedgeSettings()).isEqualTo(hedgeSettings); - assertThat(publisher.getHedgeTokenBucket()).isNotNull(); - assertThat(publisher.getHedgeTokenBucket().getTokens()).isWithin(0.0001f).of(100.0f); - - shutdownTestPublisher(publisher); - } - - @Test - public void testPublisherWithoutHedgeSettings() throws Exception { - Publisher publisher = getTestPublisherBuilder().build(); - - assertThat(publisher.getHedgeSettings()).isNull(); - assertThat(publisher.getHedgeTokenBucket()).isNull(); - - shutdownTestPublisher(publisher); - } - private Builder getTestPublisherBuilder() { return Publisher.newBuilder(TEST_TOPIC) .setExecutorProvider(FixedExecutorProvider.create(fakeExecutor)) From 91706a2a212828ba92fb22dc9aae51b67bc9f6a6 Mon Sep 17 00:00:00 2001 From: Tony Cui Date: Mon, 13 Jul 2026 17:39:56 +0000 Subject: [PATCH 5/6] style(pubsub): fix checkstyle violations in HedgeSettings --- .../google/cloud/pubsub/v1/HedgeSettings.java | 48 +++++++++++++++---- 1 file changed, 38 insertions(+), 10 deletions(-) diff --git a/java-pubsub/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/HedgeSettings.java b/java-pubsub/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/HedgeSettings.java index 9a2dbf66c9dd..4641346fca31 100644 --- a/java-pubsub/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/HedgeSettings.java +++ b/java-pubsub/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/HedgeSettings.java @@ -21,21 +21,31 @@ /** Settings for configuring publish hedging. */ public final class HedgeSettings { + /** Default hedging delay. */ private static final Duration DEFAULT_DELAY = Duration.ofMillis(100); + /** Default maximum number of tokens in the bucket. */ private static final int DEFAULT_MAX_TOKENS = 100; + /** Default refill rate (tokens per successful request). */ private static final float DEFAULT_REFILL = 0.1f; + /** Hedging delay. */ private final Duration hedgeDelay; + /** Maximum tokens. */ private final int maxTokens; + /** Refill rate. */ private final float refill; - private HedgeSettings(Builder builder) { + private HedgeSettings(final Builder builder) { this.hedgeDelay = builder.hedgeDelay; this.maxTokens = builder.maxTokens; this.refill = builder.refill; } - /** Returns the configured hedging delay. */ + /** + * Returns the configured hedging delay. + * + * @return the hedging delay. + */ public Duration getHedgeDelay() { return hedgeDelay; } @@ -48,28 +58,46 @@ float getRefill() { return refill; } - /** Returns a new builder for {@code HedgeSettings}. */ + /** + * Returns a new builder for {@code HedgeSettings}. + * + * @return a new builder. + */ public static Builder newBuilder() { return new Builder(); } /** Builder for {@code HedgeSettings}. */ public static final class Builder { + /** Hedging delay. */ private Duration hedgeDelay = DEFAULT_DELAY; + /** Maximum tokens. */ private int maxTokens = DEFAULT_MAX_TOKENS; + /** Refill rate. */ private float refill = DEFAULT_REFILL; - private Builder() {} + private Builder() { } - /** Allows hedging delay to be configurable. */ - public Builder setHedgeDelay(Duration hedgeDelay) { - Preconditions.checkNotNull(hedgeDelay); - Preconditions.checkArgument(!hedgeDelay.isNegative() && !hedgeDelay.isZero(), "hedgeDelay must be positive"); - this.hedgeDelay = hedgeDelay; + /** + * Allows hedging delay to be configurable. + * + * @param delay the hedging delay, must be positive. + * @return this builder. + */ + public Builder setHedgeDelay(final Duration delay) { + Preconditions.checkNotNull(delay); + Preconditions.checkArgument( + !delay.isNegative() && !delay.isZero(), + "hedgeDelay must be positive"); + this.hedgeDelay = delay; return this; } - /** Builds an instance of {@code HedgeSettings}. */ + /** + * Builds an instance of {@code HedgeSettings}. + * + * @return the built {@code HedgeSettings} instance. + */ public HedgeSettings build() { return new HedgeSettings(this); } From bd76ede662b5a951773c34cab4f4bd79b7c9373e Mon Sep 17 00:00:00 2001 From: Tony Cui Date: Mon, 13 Jul 2026 17:53:42 +0000 Subject: [PATCH 6/6] style(pubsub): fix formatting and checkstyle violations in HedgeSettings and tests --- .../google/cloud/pubsub/v1/HedgeSettings.java | 16 +++++++++---- .../cloud/pubsub/v1/HedgeSettingsTest.java | 4 +--- .../cloud/pubsub/v1/PublisherImplTest.java | 23 +++++++++++++++++++ 3 files changed, 36 insertions(+), 7 deletions(-) diff --git a/java-pubsub/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/HedgeSettings.java b/java-pubsub/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/HedgeSettings.java index 4641346fca31..ffa6daed1d91 100644 --- a/java-pubsub/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/HedgeSettings.java +++ b/java-pubsub/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/HedgeSettings.java @@ -23,15 +23,19 @@ public final class HedgeSettings { /** Default hedging delay. */ private static final Duration DEFAULT_DELAY = Duration.ofMillis(100); + /** Default maximum number of tokens in the bucket. */ private static final int DEFAULT_MAX_TOKENS = 100; + /** Default refill rate (tokens per successful request). */ private static final float DEFAULT_REFILL = 0.1f; /** Hedging delay. */ private final Duration hedgeDelay; + /** Maximum tokens. */ private final int maxTokens; + /** Refill rate. */ private final float refill; @@ -71,12 +75,16 @@ public static Builder newBuilder() { public static final class Builder { /** Hedging delay. */ private Duration hedgeDelay = DEFAULT_DELAY; + /** Maximum tokens. */ private int maxTokens = DEFAULT_MAX_TOKENS; + /** Refill rate. */ private float refill = DEFAULT_REFILL; - private Builder() { } + private Builder() { + super(); + } /** * Allows hedging delay to be configurable. @@ -86,9 +94,9 @@ private Builder() { } */ public Builder setHedgeDelay(final Duration delay) { Preconditions.checkNotNull(delay); - Preconditions.checkArgument( - !delay.isNegative() && !delay.isZero(), - "hedgeDelay must be positive"); + if (delay.isNegative() || delay.isZero()) { + throw new IllegalArgumentException("delay must be positive"); + } this.hedgeDelay = delay; return this; } diff --git a/java-pubsub/google-cloud-pubsub/src/test/java/com/google/cloud/pubsub/v1/HedgeSettingsTest.java b/java-pubsub/google-cloud-pubsub/src/test/java/com/google/cloud/pubsub/v1/HedgeSettingsTest.java index 2a01408b7bdf..f3b26aedb3c3 100644 --- a/java-pubsub/google-cloud-pubsub/src/test/java/com/google/cloud/pubsub/v1/HedgeSettingsTest.java +++ b/java-pubsub/google-cloud-pubsub/src/test/java/com/google/cloud/pubsub/v1/HedgeSettingsTest.java @@ -61,8 +61,6 @@ public void testZeroDelayThrows() { @Test public void testNullDelayThrows() { - assertThrows( - NullPointerException.class, - () -> HedgeSettings.newBuilder().setHedgeDelay(null)); + assertThrows(NullPointerException.class, () -> HedgeSettings.newBuilder().setHedgeDelay(null)); } } diff --git a/java-pubsub/google-cloud-pubsub/src/test/java/com/google/cloud/pubsub/v1/PublisherImplTest.java b/java-pubsub/google-cloud-pubsub/src/test/java/com/google/cloud/pubsub/v1/PublisherImplTest.java index 8e6efaf372c9..95b19d6b0108 100644 --- a/java-pubsub/google-cloud-pubsub/src/test/java/com/google/cloud/pubsub/v1/PublisherImplTest.java +++ b/java-pubsub/google-cloud-pubsub/src/test/java/com/google/cloud/pubsub/v1/PublisherImplTest.java @@ -1339,6 +1339,29 @@ public void testPublishOpenTelemetryTracing() throws Exception { .hasEnded(); } + @Test + public void testPublisherWithHedgeSettings() throws Exception { + HedgeSettings hedgeSettings = + HedgeSettings.newBuilder().setHedgeDelay(Duration.ofMillis(50)).build(); + Publisher publisher = getTestPublisherBuilder().setHedgeSettings(hedgeSettings).build(); + + assertThat(publisher.getHedgeSettings()).isEqualTo(hedgeSettings); + assertThat(publisher.getHedgeTokenBucket()).isNotNull(); + assertThat(publisher.getHedgeTokenBucket().getTokens()).isWithin(0.0001f).of(100.0f); + + shutdownTestPublisher(publisher); + } + + @Test + public void testPublisherWithoutHedgeSettings() throws Exception { + Publisher publisher = getTestPublisherBuilder().build(); + + assertThat(publisher.getHedgeSettings()).isNull(); + assertThat(publisher.getHedgeTokenBucket()).isNull(); + + shutdownTestPublisher(publisher); + } + private Builder getTestPublisherBuilder() { return Publisher.newBuilder(TEST_TOPIC) .setExecutorProvider(FixedExecutorProvider.create(fakeExecutor))