When '_read_from_pubsub' (runnners/direct/transform_evaluator.py:448) is being invoked, it will create a Pubsub SubscriberClient. This will create a grpc channel which is not being closed. When you run a pipeline with the direct runner which has a pubsub source in it, it will run eventually into an error: "too many open files". This is because these grpc channels are never being closed.
The suggestion is to wrap the code after the creation of the client in a try/finally block and close the channel (exposed on sub_client.api.transport._channel) in the finally block.
Some extra info about the issue: https://github.com/googleapis/google-cloud-python/issues/5523