Skip to content

[Java] Fix GRPC channel leak - #39663

Open
shunping wants to merge 1 commit into
apache:masterfrom
shunping:fix-channel-leak
Open

shunping wants to merge 1 commit into
apache:masterfrom
shunping:fix-channel-leak

Conversation

@shunping

@shunping shunping commented Aug 6, 2026 •

Copy link
Copy Markdown
Collaborator

Description

Follow-up to #39390.

While #39390 added BeamFnDataGrpcClient.close() to tear down active BeamFnDataGrpcMultiplexer stream observers on ProcessBundleHandler shutdown, the underlying gRPC ManagedChannels created in the Java SDK harness were still not retained or shut down. When garbage collection runs (for example, in loopback workers started via ExternalWorkerService), gRPC's ManagedChannelOrphanWrapper logs ERROR / SEVERE messages:

~~ Previous channel ManagedChannelImpl{...} was garbage collected without being shut down! ~~ Make sure to call shutdown()/shutdownNow()

Changes

  • ExternalWorkerService: Store the ManagedChannel created for ProvisionEndpoint in startWorker and call channel.shutdown() in a finally block after fetching ProvisionInfo.
  • BeamFnDataGrpcClient: Retain created ManagedChannel instances in channelCache (preventing ManagedChannelOrphanWrapper from being garbage-collected while streams are active) and shut them down in BeamFnDataGrpcClient.close().
  • BeamFnStateGrpcClientCache: Shut down the state ManagedChannel when a GrpcStateClient closes and add BeamFnStateGrpcClientCache.close().
  • FnHarness: Close BeamFnDataGrpcClient, BeamFnStateGrpcClientCache, and shut down the control ManagedChannel during harness shutdown.

Code to reproduce

"""Simple script running an inline YAML pipeline with Java WriteToJson."""

import logging
import os
import tempfile

import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions
from apache_beam.yaml import yaml_transform

PIPELINE_YAML = """
providers:
  - type: beamJar
    config:
      gradle_target: 'sdks:java:extensions:schemaio-expansion-service:shadowJar'
    transforms:
      JavaWriteToJson: 'beam:schematransform:org.apache.beam:json_write:v1'

pipeline:
  type: chain
  transforms:
    - type: Create
      name: CreateData
      config:
        elements:
          - {name: "wx", age: 25}
          - {name: "yz", age: 30}
    - type: JavaWriteToJson
      name: WriteToJson
      config:
        path: "%s"
"""


def run():
  logging.getLogger().setLevel(logging.INFO)
  options = PipelineOptions(runner="DirectRunner")
  with tempfile.TemporaryDirectory() as tmpdir:
    yaml_spec = PIPELINE_YAML % os.path.join(tmpdir, "out.json")
    with beam.Pipeline(options=options) as p:
      yaml_transform.expand_pipeline(p, yaml_spec)


if __name__ == "__main__":
  run()

Error messages (two leaking channels)

INFO:apache_beam.runners.worker.statecache:Creating state cache with size 104857600
INFO:apache_beam.runners.portability.fn_api_runner.worker_handlers:starting control server on port 57739
INFO:apache_beam.runners.portability.fn_api_runner.worker_handlers:starting data server on port 57740
INFO:apache_beam.runners.portability.fn_api_runner.worker_handlers:starting state server on port 57741
INFO:apache_beam.runners.portability.fn_api_runner.worker_handlers:starting logging server on port 57742
INFO:apache_beam.runners.portability.fn_api_runner.worker_handlers:Requesting worker at localhost:57732
INFO:apache_beam.runners.portability.fn_api_runner.worker_handlers:self.control_address: localhost:57739
INFO:ExpansionService:Oct 13, 2026 11:15:41 AM org.apache.beam.fn.harness.ExternalWorkerService startWorker
INFO:ExpansionService:INFO: Starting worker worker_1 pointing at localhost:57739.
INFO:ExpansionService:Oct 13, 2026 11:15:41 AM org.apache.beam.vendor.grpc.v1p69p0.io.grpc.internal.ManagedChannelOrphanWrapper$ManagedChannelReference cleanQueue
INFO:ExpansionService:SEVERE: *~*~*~ Previous channel ManagedChannelImpl{logId=8, target=localhost:57739} was garbage collected without being shut down! ~*~*~*
INFO:ExpansionService:    Make sure to call shutdown()/shutdownNow()
INFO:ExpansionService:java.lang.RuntimeException: ManagedChannel allocation site
INFO:ExpansionService:  at org.apache.beam.vendor.grpc.v1p69p0.io.grpc.internal.ManagedChannelOrphanWrapper$ManagedChannelReference.<init>(ManagedChannelOrphanWrapper.java:102)
INFO:ExpansionService:  at org.apache.beam.vendor.grpc.v1p69p0.io.grpc.internal.ManagedChannelOrphanWrapper.<init>(ManagedChannelOrphanWrapper.java:60)
INFO:ExpansionService:  at org.apache.beam.vendor.grpc.v1p69p0.io.grpc.internal.ManagedChannelOrphanWrapper.<init>(ManagedChannelOrphanWrapper.java:51)
INFO:ExpansionService:  at org.apache.beam.vendor.grpc.v1p69p0.io.grpc.internal.ManagedChannelImplBuilder.build(ManagedChannelImplBuilder.java:710)
INFO:ExpansionService:  at org.apache.beam.vendor.grpc.v1p69p0.io.grpc.ForwardingChannelBuilder2.build(ForwardingChannelBuilder2.java:272)
INFO:ExpansionService:  at org.apache.beam.sdk.fn.channel.ManagedChannelFactory.forDescriptor(ManagedChannelFactory.java:101)
INFO:ExpansionService:  at org.apache.beam.fn.harness.ExternalWorkerService.startWorker(ExternalWorkerService.java:83)
INFO:ExpansionService:  at org.apache.beam.model.fnexecution.v1.BeamFnExternalWorkerPoolGrpc$MethodHandlers.invoke(BeamFnExternalWorkerPoolGrpc.java:326)
INFO:ExpansionService:  at org.apache.beam.vendor.grpc.v1p69p0.io.grpc.stub.ServerCalls$UnaryServerCallHandler$UnaryServerCallListener.onHalfClose(ServerCalls.java:182)
INFO:ExpansionService:  at org.apache.beam.vendor.grpc.v1p69p0.io.grpc.internal.ServerCallImpl$ServerStreamListenerImpl.halfClosed(ServerCallImpl.java:356)
INFO:ExpansionService:  at org.apache.beam.vendor.grpc.v1p69p0.io.grpc.internal.ServerImpl$JumpToApplicationThreadServerStreamListener$1HalfClosed.runInContext(ServerImpl.java:861)
INFO:ExpansionService:  at org.apache.beam.vendor.grpc.v1p69p0.io.grpc.internal.ContextRunnable.run(ContextRunnable.java:37)
INFO:ExpansionService:  at org.apache.beam.vendor.grpc.v1p69p0.io.grpc.internal.SerializingExecutor.run(SerializingExecutor.java:133)
INFO:ExpansionService:  at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1144)
INFO:ExpansionService:  at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:642)
INFO:ExpansionService:  at java.base/java.lang.Thread.run(Thread.java:1583)
INFO:ExpansionService:
INFO:root:severity: INFO
timestamp {
  seconds: 1791904541
  nanos: 472000000
}
message: "Fn Harness started"
log_location: "org.apache.beam.fn.harness.FnHarness"
thread: "41"

...

INFO:root:severity: INFO
timestamp {
  seconds: 1791904541
  nanos: 865000000
}
message: "Hanged up for url: \"localhost:57740\"\n."
log_location: "org.apache.beam.sdk.fn.data.BeamFnDataGrpcMultiplexer"
thread: "41"

INFO:root:severity: INFO
timestamp {
  seconds: 1791904541
  nanos: 865000000
}
message: "Hanged up for url: \"localhost:57740\"\n."
log_location: "org.apache.beam.sdk.fn.data.BeamFnDataGrpcMultiplexer"
thread: "35"

INFO:ExpansionService:Oct 13, 2026 11:15:42 AM org.apache.beam.fn.harness.logging.BeamFnLoggingClient flushFinalLogs
INFO:ExpansionService:SEVERE: *~*~*~ Previous channel ManagedChannelImpl{logId=24, target=localhost:57740} was garbage collected without being shut down! ~*~*~*
INFO:ExpansionService:    Make sure to call shutdown()/shutdownNow()
INFO:ExpansionService:java.lang.Throwable: java.lang.RuntimeException: ManagedChannel allocation site
INFO:ExpansionService:  at org.apache.beam.vendor.grpc.v1p69p0.io.grpc.internal.ManagedChannelOrphanWrapper$ManagedChannelReference.<init>(ManagedChannelOrphanWrapper.java:102)
INFO:ExpansionService:  at org.apache.beam.vendor.grpc.v1p69p0.io.grpc.internal.ManagedChannelOrphanWrapper.<init>(ManagedChannelOrphanWrapper.java:60)
INFO:ExpansionService:  at org.apache.beam.vendor.grpc.v1p69p0.io.grpc.internal.ManagedChannelOrphanWrapper.<init>(ManagedChannelOrphanWrapper.java:51)
INFO:ExpansionService:  at org.apache.beam.vendor.grpc.v1p69p0.io.grpc.internal.ManagedChannelImplBuilder.build(ManagedChannelImplBuilder.java:710)
INFO:ExpansionService:  at org.apache.beam.vendor.grpc.v1p69p0.io.grpc.ForwardingChannelBuilder2.build(ForwardingChannelBuilder2.java:272)
INFO:ExpansionService:  at org.apache.beam.sdk.fn.channel.ManagedChannelFactory.forDescriptor(ManagedChannelFactory.java:101)
INFO:ExpansionService:  at org.apache.beam.fn.harness.data.BeamFnDataGrpcClient.lambda$getMultiplexer$0(BeamFnDataGrpcClient.java:148)
INFO:ExpansionService:  at org.apache.beam.sdk.fn.stream.OutboundObserverFactory$DirectClient.outboundObserverFor(OutboundObserverFactory.java:97)
INFO:ExpansionService:  at org.apache.beam.sdk.fn.data.BeamFnDataGrpcMultiplexer.<init>(BeamFnDataGrpcMultiplexer.java:84)
INFO:ExpansionService:  at org.apache.beam.fn.harness.data.BeamFnDataGrpcClient.lambda$getMultiplexer$1(BeamFnDataGrpcClient.java:158)
INFO:ExpansionService:  at java.base/java.util.concurrent.ConcurrentHashMap.computeIfAbsent(ConcurrentHashMap.java:1708)
INFO:ExpansionService:  at org.apache.beam.fn.harness.data.BeamFnDataGrpcClient.getMultiplexer(BeamFnDataGrpcClient.java:142)
INFO:ExpansionService:  at org.apache.beam.fn.harness.data.BeamFnDataGrpcClient.getOutboundObserver(BeamFnDataGrpcClient.java:136)
INFO:ExpansionService:  at org.apache.beam.fn.harness.control.ProcessBundleHandler.processBundle(ProcessBundleHandler.java:519)
INFO:ExpansionService:  at org.apache.beam.fn.harness.control.BeamFnControlClient.delegateOnInstructionRequestType(BeamFnControlClient.java:151)
INFO:ExpansionService:  at org.apache.beam.fn.harness.control.BeamFnControlClient$InboundObserver.lambda$onNext$0(BeamFnControlClient.java:116)
INFO:ExpansionService:  at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:572)
INFO:ExpansionService:  at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:317)
INFO:ExpansionService:  at org.apache.beam.sdk.util.UnboundedScheduledExecutorService$ScheduledFutureTask.run(UnboundedScheduledExecutorService.java:164)
INFO:ExpansionService:  at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1144)
INFO:ExpansionService:  at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:642)
INFO:ExpansionService:  at java.base/java.lang.Thread.run(Thread.java:1583)
INFO:ExpansionService:
INFO:ExpansionService:  at org.apache.beam.fn.harness.logging.BeamFnLoggingClient.flushFinalLogs(BeamFnLoggingClient.java:390)
INFO:ExpansionService:  at org.apache.beam.fn.harness.logging.BeamFnLoggingClient.lambda$new$0(BeamFnLoggingClient.java:172)
INFO:ExpansionService:  at java.base/java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java:1768)
INFO:ExpansionService:  at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:572)
INFO:ExpansionService:  at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:317)
INFO:ExpansionService:  at org.apache.beam.sdk.util.UnboundedScheduledExecutorService$ScheduledFutureTask.run(UnboundedScheduledExecutorService.java:164)
INFO:ExpansionService:  at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1144)
INFO:ExpansionService:  at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:642)
INFO:ExpansionService:  at java.base/java.lang.Thread.run(Thread.java:1583)
INFO:ExpansionService:
INFO:ExpansionService:Oct 13, 2026 11:15:42 AM org.apache.beam.fn.harness.FnHarness main
INFO:ExpansionService:INFO: Shutting SDK harness down.

With the fix in this PR, the above error messages are gone.

@github-actions

github-actions Bot commented Oct 6, 2026

Copy link
Copy Markdown
Contributor

This pull request has been marked as stale due to 60 days of inactivity. It will be closed in 1 week if no further activity occurs. If you think that’s incorrect or this pull request requires a review, please simply write any comment. If closed, you can revive the PR at any time and @mention a reviewer or discuss it on the dev@beam.apache.org list. Thank you for your contributions.

@github-actions github-actions Bot added the stale label Oct 6, 2026
@github-actions

Copy link
Copy Markdown
Contributor

This pull request has been closed due to lack of activity. If you think that is incorrect, or the pull request requires review, you can revive the PR at any time.

@github-actions github-actions Bot closed this Oct 13, 2026
@shunping shunping reopened this Oct 13, 2026
@shunping
shunping marked this pull request as ready for review October 13, 2026 15:27
@shunping shunping changed the title Fix GRPC channel leak [Java] Fix GRPC channel leak Oct 13, 2026
@shunping
shunping requested a review from Abacn October 13, 2026 15:27
@shunping
shunping marked this pull request as draft October 13, 2026 15:35
@shunping
shunping removed the request for review from Abacn October 13, 2026 15:35
@shunping

Copy link
Copy Markdown
Collaborator Author

Converting it back to draft since some tests are failing.

…Service

Prevent leaked gRPC ManagedChannel errors ("Previous channel ... was
garbage collected without being shut down!") in the Java SDK harness:
- Shut down the provision ManagedChannel in ExternalWorkerService.startWorker
  after fetching ProvisionInfo.
- Retain created ManagedChannels in BeamFnDataGrpcClient and shut them
  down when BeamFnDataGrpcClient.close() is called.
- Shut down state ManagedChannels in BeamFnStateGrpcClientCache on close.
- Close BeamFnDataGrpcClient, BeamFnStateGrpcClientCache, and the control
  ManagedChannel during FnHarness shutdown.
@codecov

codecov Bot commented Oct 13, 2026 •

Copy link
Copy Markdown

Codecov Report

❌ Patch coverage is 42.42424% with 19 lines in your changes missing coverage. Please review.
✅ Project coverage is 65.05%. Comparing base (76f6e6d) to head (98687da).
⚠️ Report is 156 commits behind head on master.

Files with missing lines Patch % Lines
.../apache/beam/fn/harness/ExternalWorkerService.java 0.00% 12 Missing ⚠️
...m/fn/harness/state/BeamFnStateGrpcClientCache.java 60.00% 3 Missing and 1 partial ⚠️
...che/beam/fn/harness/data/BeamFnDataGrpcClient.java 62.50% 2 Missing and 1 partial ⚠️
Additional details and impacted files
@@             Coverage Diff              @@
##             master   #39663      +/-   ##
============================================
+ Coverage     56.16%   65.05%   +8.88%     
- Complexity     2288    13977   +11689     
============================================
  Files          1124     1729     +605     
  Lines        178336   108487   -69849     
  Branches       1489    11307    +9818     
============================================
- Hits         100170    70571   -29599     
+ Misses        75645    33707   -41938     
- Partials       2521     4209    +1688     
Flag Coverage Δ
go ?
java 65.05% <42.42%> (-10.74%) ⬇️
python ?

Flags with carried forward coverage won't be shown. Click here to find out more.

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:
  • ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
  • 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

@shunping
shunping marked this pull request as ready for review October 13, 2026 17:56
@shunping
shunping requested a review from Abacn October 13, 2026 17:56
beamFnStatusClient.close();
}
processBundleHandler.shutdown();
beamFnDataMultiplexer.close();

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.

processBundleHandler.shutdown() already calls beamFnStateGrpcClientCache.close(); and beamFnDataClient.close();

This is redundant though probably harmless, however we should close client in the place where it owns it in general, to avoid potential pre-mature channel break

@github-actions

Copy link
Copy Markdown
Contributor

Assigning reviewers:

R: @Abacn for label java.
Backup reviewer: ahmedabu98 (if the main reviewer is unresponsive, you can escalate to them by tagging their username)

This pull request likely touches a core component ("core" label). Please review with scrutiny.

Note: If you would like to opt out of this review, comment assign to next reviewer.

Available commands:

  • stop reviewer notifications - opt out of the automated review tooling
  • remind me after tests pass - tag the comment author after tests pass
  • waiting on author - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)

The PR bot will only process comments in the main thread (not review comments).

@github-actions
github-actions Bot requested a review from Abacn October 15, 2026 15:42
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants