feat(pubsub): implement HedgeSettings and integrate with Publisher

[tonyyyycui]

This PR implements a data class hedgeSettings and a method class hedgeTokenBucket used for publish hedging functionality. Publisher is modified to integrate the new hedgeSettings class.

Within hedgeSettings, hedgeDelay is a user-configurable field used to represent the duration (in ms) before a hedged publish request is sent. All other fields are abstracted away from the user.

The hedgeTokenBucket class implements a token-based method to throttle the number of hedged publish requests on the client side.

[gemini-code-assist]

Code Review

This pull request introduces a hedging mechanism for the Pub/Sub Publisher, adding HedgeSettings for configuration and a thread-safe HedgeTokenBucket to limit hedged requests. The feedback suggests replacing the synchronized methods in HedgeTokenBucket with explicit ReentrantLock to reduce lock contention and improve performance on the critical path.

[gemini-code-assist]

In performance-sensitive code like the Pub/Sub Publisher, using synchronized methods on the critical path can introduce significant lock contention and performance overhead. To protect shared state while ensuring thread safety and visibility, prefer using explicit locks (such as ReentrantLock) over the synchronized keyword.

import com.google.common.annotations.VisibleForTesting;
import java.util.concurrent.locks.ReentrantLock;

/** Token bucket for limiting hedged requests. Thread-safe. */
final class HedgeTokenBucket {
  private final ReentrantLock lock = new ReentrantLock();
  private final int maxTokens;
  private final float refillAmount;
  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.
   */
  boolean tryAcquire() {
    lock.lock();
    try {
      if (tokens < 1.0f) {
        return false;
      }
      tokens -= 1.0f;
      return true;
    } finally {
      lock.unlock();
    }
  }

  /** Refills the bucket by the configured refill amount, capped at max tokens. */
  void refill() {
    lock.lock();
    try {
      tokens = Math.min((float) maxTokens, tokens + refillAmount);
    } finally {
      lock.unlock();
    }
  }

  @VisibleForTesting
  float getTokens() {
    lock.lock();
    try {
      return tokens;
    } finally {
      lock.unlock();
    }
  }

References

1. In performance-sensitive code, prefer using explicit locks over the 'synchronized' keyword to protect shared state while ensuring thread safety and visibility.

[michaelpri10]

I think this is a valid concern regarding high contention from the use of synchronized for lock synchronization in this class. I'll take a deeper look on Monday at what better options would be.

[michaelpri10]

At a higher level, I think the full implementation should be one PR (instead of the , given any submitted changes become public. It would be unexpected for customers to be able to set HedgingSettings without anything actually happening, so the final PR should have everything needed for the implementation included.

[michaelpri10]

As a note, this value will be changing (potentially other default values as well). Leaving this as a placeholder comment to ensure these meet the agreed upon values before merging.

[michaelpri10]

This method should enforce the constraint that ordering is not enabled.

[michaelpri10]

I don't think super() is needed here.

[michaelpri10]

nit: Does this method need to be public? I think leaving it without an access modifier (i.e., making it package-private) should be sufficient.

[michaelpri10]

I think this is a valid concern regarding high contention from the use of synchronized for lock synchronization in this class. I'll take a deeper look on Monday at what better options would be.