Apache Beam Go SDK随机访问侧输入遇类型错误（Direct Runner）

以下是一段简单的Apache Beam Go程序：

package main

import (
    "context"
    "log"

    "github.com/apache/beam/sdks/v2/go/pkg/beam"
    "github.com/apache/beam/sdks/v2/go/pkg/beam/register"
    _ "github.com/apache/beam/sdks/v2/go/pkg/beam/runners/direct"
    "github.com/apache/beam/sdks/v2/go/pkg/beam/transforms/stats"
    "github.com/apache/beam/sdks/v2/go/pkg/beam/x/beamx"
)

func init() {
    register.Function2x0(fn)
}

func main() {
    beam.Init()

    p := beam.NewPipeline()
    s := p.Root()

    rows := beam.Create(s, "1", "1", "2")
    counts := stats.Count(s, rows)

    beam.ParDo0(s, fn, rows, beam.SideInput{Input: counts})

    if err := beamx.Run(context.Background(), p); err != nil {
        log.Fatal(err)
    }
}

func fn(in string, lookup func(string) func(*int) bool) {
    var count int
    lookup("1")(&count)
}

使用Direct Runner运行时出现如下错误：

2022/09/08 22:09:36 Executing pipeline with the direct runner.
2022/09/08 22:09:36 Pipeline:
2022/09/08 22:09:36 Nodes: {1: []uint8/bytes GLO}
{2: string/string GLO}
{3: KV<string,int>/KV<string,int[varintz]> GLO}
{4: CoGBK<string,int>/CoGBK<string,int[varintz]> GLO}
{5: KV<string,int>/KV<string,int[varintz]> GLO}
Edges: 1: Impulse [] -> [Out: []uint8 -> {1: []uint8/bytes GLO}]
2: ParDo [In(Main): []uint8 <- {1: []uint8/bytes GLO}] -> [Out: T -> {2: string/string GLO}]
3: ParDo [In(Main): T <- {2: string/string GLO}] -> [Out: KV<T,int> -> {3: KV<string,int>/KV<string,int[varintz]> GLO}]
4: CoGBK [In(Main): KV<string,int> <- {3: KV<string,int>/KV<string,int[varintz]> GLO}] -> [Out: CoGBK<string,int> -> {4: CoGBK<string,int>/CoGBK<string,int[varintz]> GLO}]
5: Combine [In(Main): int <- {4: CoGBK<string,int>/CoGBK<string,int[varintz]> GLO}] -> [Out: KV<string,int> -> {5: KV<string,int>/KV<string,int[varintz]> GLO}]
6: ParDo [In(Main): string <- {2: string/string GLO} In(MultiMap): KV<string,int> <- {5: KV<string,int>/KV<string,int[varintz]> GLO}] -> []
2022/09/08 22:09:36 Plan[plan]:
10: Impulse[0]
1: ParDo[main.fn] Out:[]
2: wait[1] Out:1
3: buffer[3]. wait:2 Out:1
4: Combine[stats.sumIntFn] Keyed:false Out:3
5: CoGBK. Out:4
6: Inject[0]. Out:5
7: ParDo[stats.keyedCountFn] Out:[6]
8: Multiplex. Out:[7 2]
9: ParDo[beam.createFn] Out:[8]
2022/09/08 22:09:36 wait[2] unblocked w/ 3 [false]
2022/09/08 22:09:36 panic: interface conversion: interface {} is string, not int
Full error:
while executing FinishBundle for Plan[plan]:
10: Impulse[0]
1: ParDo[main.fn] Out:[]
2: wait[1] Out:1
3: buffer[3]. wait:2 Out:1
4: Combine[stats.sumIntFn] Keyed:false Out:3
5: CoGBK. Out:4
6: Inject[0]. Out:5
7: ParDo[stats.keyedCountFn] Out:[6]
8: Multiplex. Out:[7 2]
9: ParDo[beam.createFn] Out:[8]
        caused by:
panic: interface conversion: interface {} is string, not int

明明已成功注册KV<string,int>签名的侧输入函数，为何类型被推断为string？

更新：该问题属于Direct Runner的bug，在Dataflow上运行正常。

内容的提问来源于stack exchange，提问作者salient