refactor(flagd): Replace Thread.sleep with ScheduledExecutorService in SyncStreamQueueSource

[Copilot]

SyncStreamQueueSource.observeSyncStream() uses Thread.sleep() for retry backoff, blocking the thread and limiting scalability. This replaces it with non-blocking scheduled execution.

Changes

• Add ScheduledExecutorService for retry scheduling with named daemon thread (flagd-sync-retry-scheduler)
• Replace blocking sleep with scheduler.schedule() that re-invokes observeSyncStream() after backoff delay
• Handle shutdown race conditions via RejectedExecutionException catch and proper awaitTermination() with interrupt handling

// Before: blocking
if (shouldThrottle.getAndSet(false)) {
    Thread.sleep(this.maxBackoffMs);
}

// After: non-blocking
if (shouldThrottle.getAndSet(false)) {
    scheduleRetry();
    return;
}

private void scheduleRetry() {
    if (shutdown.get()) return;
    try {
        scheduler.schedule(this::observeSyncStream, this.maxBackoffMs, TimeUnit.MILLISECONDS);
    } catch (RejectedExecutionException e) {
        log.debug("Retry scheduling rejected, scheduler is shut down", e);
    }
}

Follows the same pattern already used in FlagdProvider.java for error task scheduling.

Warning

Firewall rules blocked me from connecting to one or more addresses (expand for details)

I tried to connect to the following addresses, but was blocked by firewall rules:

• checkstyle.org
  • Triggering command: /opt/hostedtoolcache/CodeQL/2.23.3/x64/codeql/tools/linux64/java/bin/java /opt/hostedtoolcache/CodeQL/2.23.3/x64/codeql/tools/linux64/java/bin/java -jar /opt/hostedtoolcache/CodeQL/2.23.3/x64/codeql/xml/tools/xml-extractor.jar --fileList=/home/REDACTED/work/java-sdk-contrib/.codeql-scratch/dbs/java/working/files-to-index13214004798705728053.list --sourceArchiveDir=/home/REDACTED/work/java-sdk-contrib/.codeql-scratch/dbs/java/src --outputDir=/home/REDACTED/work/java-sdk-contrib/.codeql-scratch/dbs/java/trap/java (dns block)

If you need me to access, download, or install something from one of these locations, you can either:

• Configure Actions setup steps to set up my environment, which run before the firewall is enabled
• Add the appropriate URLs or hosts to the custom allowlist in this repository's Copilot coding agent settings (admins only)

Original prompt

---

This section details on the original issue you should resolve

<issue_title>[flagd] Use a scheduler instead of Thread.sleep for retry/backoff in SyncStreamQueueSource</issue_title>
<issue_description>## Problem
The current implementation of SyncStreamQueueSource in the flagd provider core connection logic uses Thread.sleep() for retry backoff when re-establishing the stream connection after an error. This usage is found in the observeSyncStream() method (line 126) and can be undesirable because Thread.sleep blocks the thread, limiting scalability and responsiveness, especially in environments that favor non-blocking or asynchronous IO.

Suggested Improvement

It would be better to refactor this retry/backoff mechanism to utilize a scheduled executor (such as ScheduledExecutorService) or similar scheduling abstraction. This would allow the retry logic to be non-blocking and would enable easier integration with asynchronous patterns. Additionally, this could simplify future improvements to connection logic, such as jitter, exponential backoff, or cancellation support.

References

Example code location:

• SyncStreamQueueSource.java
• The logic is around line 126, in the observeSyncStream() method.

Proposal

• Replace usages of Thread.sleep(this.maxBackoffMs); in observeSyncStream() and any related retry logic with non-blocking scheduling logic (e.g., via ScheduledExecutorService).
• Consider further enhancements such as custom retry policies or graceful cancellation.

Example: Using ScheduledExecutorService for Non-blocking Retry

// Define the scheduler (typically a field in your class)
private final ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();

// Instead of blocking with Thread.sleep:
if (shouldThrottle.getAndSet(false)) {
    scheduler.schedule(
        this::retrySyncStream,
        this.maxBackoffMs,
        TimeUnit.MILLISECONDS
    );
    return;
}

// Add a method to handle the retry logic:
private void retrySyncStream() {
    if (shutdown.get()) return;
    observeSyncStream(); // or the desired reconnection logic
}

Key Points:

• Using a scheduler avoids blocking the thread.
• You can extend this pattern for exponential backoff, jitter, or cancellation.
• Ensure proper shutdown of ScheduledExecutorService in your cleanup logic.

Benefits

• Improves scalability and responsiveness.
• Aligns with modern Java practices for IO and connection management.
• Simplifies enhancement of retry logic in future.
  </issue_description>

Comments on the Issue (you are @copilot in this section)

@aepfli @chrfwow do you think this is feasible - I really hate our thread.sleeps @chrfwow I do think it is feasible. But I am not sure if it is worth it if it's not easily and cleanly doable

• Fixes [flagd] Use a scheduler instead of Thread.sleep for retry/backoff in SyncStreamQueueSource #1659

---

💡 You can make Copilot smarter by setting up custom instructions, customizing its development environment and configuring Model Context Protocol (MCP) servers. Learn more Copilot coding agent tips in the docs.

[aepfli]

/gemini review

[gemini-code-assist]

Code Review

This pull request effectively refactors the SyncStreamQueueSource to use a ScheduledExecutorService for non-blocking retry backoff, replacing the blocking Thread.sleep(). This is a solid improvement for scalability and responsiveness. The implementation of the scheduler and its shutdown sequence is well-handled. However, I've identified a few areas for improvement. There is some duplicated code for the scheduler initialization which could be refactored for better maintainability. More importantly, the refactoring has removed specific handling for InterruptedException, which is now caught by a general Exception handler. This is a regression that could affect graceful shutdown, and I've provided suggestions to restore correct interruption handling. I also found a minor issue with a potentially confusing log message in the new retry logic.

[chrfwow]

Don't we have a helper method for that?

[aepfli]

do we? and where and why?

[chrfwow]

Nvm, I thought about this one: https://github.com/open-feature/java-sdk/pull/1704/files#diff-b4678dafe534f1165ab3932bd9549589a219976b3fd8ee35f20a9df4743cd46b