Skip to content

feat(pubsub): add publish telemetry headers for publish attempt observability - #14338

Open
tonyyyycui wants to merge 50 commits into
googleapis:mainfrom
tonyyyycui:publish-telemetry-header
Open

feat(pubsub): add publish telemetry headers for publish attempt observability#14338
tonyyyycui wants to merge 50 commits into
googleapis:mainfrom
tonyyyycui:publish-telemetry-header

Conversation

@tonyyyycui

@tonyyyycui tonyyyycui commented Sep 9, 2026

Copy link
Copy Markdown

Adds client-side publish telemetry headers to provide observability into publish latency and hedging attempts.

NOTE: Stacked on #13735. Review commit 885792e for the isolated diff.

Tony Cui and others added 30 commits July 13, 2026 15:16
…r for hedgeTokenBucket. Also added HEDGE_TOKEN_SCALE.
…le-cloud-java into publish-hedging-settings

# Please enter a commit message to explain why this merge is necessary,
# especially if it merges an updated upstream into a topic branch.
#
# Lines starting with '#' will be ignored, and an empty message aborts
# the commit.
@tonyyyycui
tonyyyycui requested review from a team as code owners September 9, 2026 20:29

@gemini-code-assist gemini-code-assist Bot left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Code Review

This pull request introduces publish hedging for the Google Cloud Pub/Sub Java client, adding configuration settings (HedgingSettings), a coordinator (CancellationSharer), and queue management (HedgedRequest) to manage and rate-limit hedged attempts. Feedback on these changes highlights several critical issues, including a race condition in scheduleQueueProcessing that could leave tasks unscheduled, a logic error in CancellationSharer where a failure of the original attempt prematurely cancels active hedged attempts, a thread-safety issue with queueProcessingFuture lacking volatile visibility, and a hardcoded 10-second timeout cap that ignores the user's configured maximum RPC timeout.

Comment on lines +757 to +778
private void scheduleQueueProcessing() {
if (isQueueProcessingScheduled.compareAndSet(false, true)) {
HedgedRequest nextItem = hedgingQueue.peek();
if (nextItem == null) {
isQueueProcessingScheduled.set(false);
return;
}

long delay = Math.max(0, nextItem.getSendAfterMs() - clock.millisTime());

queueProcessingFuture =
executor.schedule(
new Runnable() {
@Override
public void run() {
processQueue();
}
},
delay,
TimeUnit.MILLISECONDS);
}
}

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

high

There is a race condition in scheduleQueueProcessing when scheduling a new hedged request. If hedgingQueue.peek() returns null (because the queue was empty at that moment), but another thread adds an item to the queue and calls scheduleQueueProcessing right before isQueueProcessingScheduled.set(false) is executed, the compareAndSet in the second thread will fail (since it is still true). The first thread will then set the flag to false and return. As a result, the newly added item will remain in the queue without any scheduled task to process it. Using a loop to re-check the queue after setting the flag to false resolves this issue.

  private void scheduleQueueProcessing() {
    while (true) {
      if (!isQueueProcessingScheduled.compareAndSet(false, true)) {
        return;
      }
      HedgedRequest nextItem = hedgingQueue.peek();
      if (nextItem == null) {
        isQueueProcessingScheduled.set(false);
        if (!hedgingQueue.isEmpty()) {
          continue;
        }
        return;
      }

      long delay = Math.max(0, nextItem.getSendAfterMs() - clock.millisTime());

      queueProcessingFuture =
          executor.schedule(
              new Runnable() {
                @Override
                public void run() {
                  processQueue();
                }
              },
              delay,
              TimeUnit.MILLISECONDS);
      return;
    }
  }


private final ConcurrentLinkedQueue<HedgedRequest> hedgingQueue;
private final AtomicBoolean isQueueProcessingScheduled;
private ScheduledFuture<?> queueProcessingFuture;

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

The queueProcessingFuture field is written to by both the publisher thread (via startHedgedCall) and the executor thread (via processQueue) without any synchronization or volatile visibility. This can lead to data races and visibility issues, particularly during shutdown where queueProcessingFuture is read and cancelled. Marking this field as volatile ensures proper thread visibility.

Suggested change
private ScheduledFuture<?> queueProcessingFuture;
private volatile ScheduledFuture<?> queueProcessingFuture;

if (remainingTimeoutMs <= 0) {
continue;
}
long attemptTimeoutMs = Math.min(10000, remainingTimeoutMs);

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

Capping the attempt timeout at a hardcoded 10000 (10 seconds) ignores the user's configured maxRpcTimeoutDuration in retrySettings. It should instead be capped at retrySettings.getMaxRpcTimeoutDuration().toMillis() to respect the publisher's configuration.

Suggested change
long attemptTimeoutMs = Math.min(10000, remainingTimeoutMs);
long attemptTimeoutMs = Math.min(retrySettings.getMaxRpcTimeoutDuration().toMillis(), remainingTimeoutMs);

@googleapis googleapis deleted a comment from gemini-code-assist Bot Sep 9, 2026
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant