Skip to content

Add a FailoverChannel wrapper on top of IsolationChannel to maintain a set of primary and failover channel. #37840

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
parveensania wants to merge 9 commits into
base: master
Choose a base branch
from
+1,137 −76
Open

Add a FailoverChannel wrapper on top of IsolationChannel to maintain a set of primary and failover channel. #37840

Show file tree
Hide file tree
Changes from 5 commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
848f343
Add a wrapper on top of IsolationChannel to maintain a set of primary…
parveensania Mar 12, 2026
234a533
Merge branch 'master' into failover-channel
parveensania Mar 12, 2026
42ec76c
Removing check to verify isolation channel option is set
parveensania Mar 16, 2026
b64de5c
Merge branch 'failover-channel' of github.com:parveensania/beam-dp in…
parveensania Mar 16, 2026
d3f0069
Adding failover mode based on connection status
parveensania Mar 19, 2026
03dcb31
Change to wrap FailoverChannel by IsolationChannel
parveensania Mar 24, 2026
e3ab71e
Adding channel id in logs + addressing other review comments
parveensania Mar 26, 2026
0a6604c
Removing getState from newCall
parveensania Mar 26, 2026
3debb37
Initializaing fallback channel lazily
parveensania Mar 31, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
44 changes: 33 additions & 11 deletions ...worker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingDataflowWorker.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -88,8 +88,10 @@
import org.apache.beam.runners.dataflow.worker.windmill.client.grpc.GrpcDispatcherClient;
import org.apache.beam.runners.dataflow.worker.windmill.client.grpc.GrpcWindmillServer;
import org.apache.beam.runners.dataflow.worker.windmill.client.grpc.GrpcWindmillStreamFactory;
import org.apache.beam.runners.dataflow.worker.windmill.client.grpc.auth.VendoredCredentialsAdapter;
import org.apache.beam.runners.dataflow.worker.windmill.client.grpc.stubs.ChannelCache;
import org.apache.beam.runners.dataflow.worker.windmill.client.grpc.stubs.ChannelCachingRemoteStubFactory;
import org.apache.beam.runners.dataflow.worker.windmill.client.grpc.stubs.FailoverChannel;
import org.apache.beam.runners.dataflow.worker.windmill.client.grpc.stubs.IsolationChannel;
import org.apache.beam.runners.dataflow.worker.windmill.client.grpc.stubs.WindmillStubFactoryFactory;
import org.apache.beam.runners.dataflow.worker.windmill.client.grpc.stubs.WindmillStubFactoryFactoryImpl;
Expand All @@ -114,6 +116,8 @@
import org.apache.beam.sdk.metrics.MetricsEnvironment;
import org.apache.beam.sdk.util.construction.CoderTranslation;
import org.apache.beam.sdk.values.WindowedValues;
import org.apache.beam.vendor.grpc.v1p69p0.io.grpc.ManagedChannel;
import org.apache.beam.vendor.grpc.v1p69p0.io.grpc.auth.MoreCallCredentials;
import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.annotations.VisibleForTesting;
import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions;
import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.cache.CacheStats;
Expand Down Expand Up @@ -381,7 +385,8 @@ private StreamingWorkerHarnessFactoryOutput createFanOutStreamingEngineWorkerHar
MemoryMonitor memoryMonitor,
GrpcDispatcherClient dispatcherClient) {
WeightedSemaphore<Commit> maxCommitByteSemaphore = Commits.maxCommitByteSemaphore();
ChannelCache channelCache = createChannelCache(options, checkNotNull(configFetcher));
ChannelCache channelCache =
createChannelCache(options, checkNotNull(configFetcher), dispatcherClient);
@SuppressWarnings("methodref.receiver.bound")
FanOutStreamingEngineWorkerHarness fanOutStreamingEngineWorkerHarness =
FanOutStreamingEngineWorkerHarness.create(
Expand Down Expand Up @@ -804,20 +809,37 @@ private static void validateWorkerOptions(DataflowWorkerHarnessOptions options)
}

private static ChannelCache createChannelCache(
DataflowWorkerHarnessOptions workerOptions, ComputationConfig.Fetcher configFetcher) {
DataflowWorkerHarnessOptions workerOptions,
ComputationConfig.Fetcher configFetcher,
GrpcDispatcherClient dispatcherClient) {
ChannelCache channelCache =
ChannelCache.create(
(currentFlowControlSettings, serviceAddress) -> {
// IsolationChannel will create and manage separate RPC channels to the same
// serviceAddress.
return IsolationChannel.create(
() ->
remoteChannel(
serviceAddress,
workerOptions.getWindmillServiceRpcChannelAliveTimeoutSec(),
currentFlowControlSettings),
currentFlowControlSettings.getOnReadyThresholdBytes());
ManagedChannel primaryChannel =
IsolationChannel.create(
Copy link
Copy Markdown
Contributor

@scwhittle scwhittle Mar 19, 2026

Choose a reason for hiding this comment

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

since it's being setup this way IsolationChannel connectivity callbacks are going to be what is used. I'm not sure how that will work since it internally has multiple channels. Looking it seems just has the default ManagedChannel implementation which throws unimplemented exception.

What about having IsolationChannel on top of fallback channels? That seems simpler to me since IsolationChannel just internally creates the separate channels and otherwise doesn't do much than forward things on.

It would be good to have a unit test of whatever setup we do use so that we flush out the issues there instead of requiring an integration test.

Copy link
Copy Markdown
Contributor Author

@parveensania parveensania Mar 24, 2026

Choose a reason for hiding this comment

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

Addressed this. IsolationChannel now wraps FailoverChannel which creates two channels per active RPC.

The original intent was to keep IsolationChannel unmodified (since it is used by dispatcher client) and handle fallback at per-worker level. The new ordering (IsolationChannel over FailoverChannel) changes the semantic to per-RPC failover. Which means in case of connectivity issues, each RPC would independently discover the failure and switch at different times, rather than switching together in a coordinated way.

I do agree managing state at per RPC level seems to be less error prone, but would like to callout this semantic change.

() ->
remoteChannel(
serviceAddress,
workerOptions.getWindmillServiceRpcChannelAliveTimeoutSec(),
currentFlowControlSettings),
currentFlowControlSettings.getOnReadyThresholdBytes());
// Create an isolated fallback channel from dispatcher endpoints.
// This ensures both primary and fallback use separate isolated channels.
ManagedChannel fallbackChannel =
IsolationChannel.create(
() ->
remoteChannel(
dispatcherClient.getDispatcherEndpoints().iterator().next(),
workerOptions.getWindmillServiceRpcChannelAliveTimeoutSec(),
currentFlowControlSettings),
currentFlowControlSettings.getOnReadyThresholdBytes());
return FailoverChannel.create(
primaryChannel,
fallbackChannel,
MoreCallCredentials.from(
new VendoredCredentialsAdapter(workerOptions.getGcpCredential())));
});

configFetcher
.getGlobalConfigHandle()
.registerConfigObserver(
Expand Down
15 changes: 9 additions & 6 deletions ...he/beam/runners/dataflow/worker/streaming/harness/FanOutStreamingEngineWorkerHarness.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -410,15 +410,18 @@ private GlobalDataStreamSender getOrCreateGlobalDataSteam(
}

private WindmillStreamSender createAndStartWindmillStreamSender(Endpoint endpoint) {
GetWorkRequest.Builder getWorkRequestBuilder =
GetWorkRequest.newBuilder()
.setClientId(jobHeader.getClientId())
.setJobId(jobHeader.getJobId())
.setProjectId(jobHeader.getProjectId())
.setWorkerId(jobHeader.getWorkerId());
endpoint.workerToken().ifPresent(getWorkRequestBuilder::setBackendWorkerToken);

WindmillStreamSender windmillStreamSender =
WindmillStreamSender.create(
WindmillConnection.from(endpoint, this::createWindmillStub),
GetWorkRequest.newBuilder()
.setClientId(jobHeader.getClientId())
.setJobId(jobHeader.getJobId())
.setProjectId(jobHeader.getProjectId())
.setWorkerId(jobHeader.getWorkerId())
.build(),
getWorkRequestBuilder.build(),
GetWorkBudget.noBudget(),
streamFactory,
workItemScheduler,
Expand Down
2 changes: 1 addition & 1 deletion ...va/org/apache/beam/runners/dataflow/worker/windmill/client/grpc/GrpcDispatcherClient.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -128,7 +128,7 @@ public CloudWindmillServiceV1Alpha1Stub getWindmillServiceStub() {
: randomlySelectNextStub(windmillServiceStubs));
}

ImmutableSet<HostAndPort> getDispatcherEndpoints() {
public ImmutableSet<HostAndPort> getDispatcherEndpoints() {
return dispatcherStubs.get().dispatcherEndpoints();
}

Expand Down
Loading
Loading