[Bug]: Go harness leaks gRPC connections and can deadlock when closing streams #40260

package harness

import (
    "context"
    "fmt"
    "io"
    "runtime"
    "testing"
    "time"

    fnpb "github.com/apache/beam/sdks/v2/go/pkg/beam/model/fnexecution_v1"
)

type blockRecvClient struct{ block chan struct{} }
func (c *blockRecvClient) Recv() (*fnpb.Elements, error) { <-c.block; return nil, io.EOF }
func (c *blockRecvClient) Send(*fnpb.Elements) error     { return nil }

func TestDataChannelManagerLeak(t *testing.T) {
    const jobs = 50
    block := make(chan struct{})
    defer close(block)

    runtime.GC()
    time.Sleep(50 * time.Millisecond)
    before := runtime.NumGoroutine()

    for i := 0; i < jobs; i++ {
        // One harness per job. Control EOFs; MainWithOptions returns without closing the data manager.
        m := &DataChannelManager{}
        ch := makeDataChannel(context.Background(), "data", &blockRecvClient{block: block}, func() {})
        m.ports = map[string]*DataChannel{"data": ch}
        _ = m
    }

    time.Sleep(50 * time.Millisecond)
    after := runtime.NumGoroutine()
    t.Logf("jobs=%d goroutines before=%d after=%d delta=%d", jobs, before, after, after-before)
    if after-before < jobs/2 {
        t.Fatalf("did not reproduce leak: before=%d after=%d", before, after)
    }
}