Go中使用Sarama测试Kafka客户端遇超时问题求助

问题背景 #

使用Sarama 1.38.0版本测试自定义Kafka消息库客户端时，遇到客户端初始化超时错误。直接复制Sarama官方测试代码后，问题依旧存在，最终通过完善MockBroker的请求处理逻辑解决了超时问题。

失败的测试代码对比 #

自定义测试代码 #

func TestSimpleClient(t *testing.T) {
    seedBroker := sarama.NewMockBroker(t, 1)

    seedBroker.Returns(new(sarama.MetadataResponse))

    client, err := NewClient([]string{seedBroker.Addr()}, WithSaramaConfig(mocks.NewTestConfig()))
    if err != nil {
        t.Fatal(err)
    }

    seedBroker.Close()
    safeClose(t, client)
}

复制的Sarama官方测试代码 #

func TestSimpleClient(t *testing.T) {
    seedBroker := NewMockBroker(t, 1)

    seedBroker.Returns(new(MetadataResponse))

    client, err := NewClient([]string{seedBroker.Addr()}, NewTestConfig())
    if err != nil {
        t.Fatal(err)
    }

    seedBroker.Close()
    safeClose(t, client)
}

修复后可正常运行的代码 #

func initSimpleBroker(t *testing.T) *sarama.MockBroker {
    topics := []string{"test.topic"}
    mockBroker := sarama.NewMockBroker(t, 0)

    mockBroker.SetHandlerByMap(map[string]sarama.MockResponse{
        "MetadataRequest": sarama.NewMockMetadataResponse(t).
            SetBroker(mockBroker.Addr(), mockBroker.BrokerID()).
            SetLeader(topics[0], 0, mockBroker.BrokerID()).
            SetController(mockBroker.BrokerID()),
        "ApiVersionsRequest": sarama.NewMockApiVersionsResponse(t),
        "OffsetRequest": sarama.NewMockOffsetResponse(t).
            SetOffset(topics[0], 0, sarama.OffsetOldest, 0).
            SetOffset(topics[0], 0, sarama.OffsetNewest, 1),
    })
    return mockBroker
}

func TestNewClient_Success(t *testing.T) {
    mockBroker := initSimpleBroker(t)
    config := mocks.NewTestConfig()

    kc, err := NewClient([]string{mockBroker.Addr()}, WithSaramaConfig(config))

    assert.NoError(t, err)
    assert.NotNil(t, kc)

    mockBroker.Close()
    safeClose(t, kc)

    assert.Nil(t, kc.client)
}

超时原因分析 #

1. MockBroker需显式处理关键请求
   Sarama的MockBroker不会自动处理所有Kafka协议请求，必须为客户端初始化过程中发送的每个请求配置对应响应。旧代码仅返回空的MetadataResponse，无法覆盖客户端初始化的完整流程。

2. 缺少ApiVersionsRequest响应
   Sarama 1.30+版本后，客户端初始化时会先发送ApiVersionsRequest协商API版本。如果MockBroker未配置该请求的响应，客户端会持续等待，最终触发超时。

3. 空MetadataResponse缺失核心信息
   旧代码中的空MetadataResponse未包含Broker自身地址、控制器ID等关键数据，客户端无法确认可用节点，后续流程无法推进导致超时。修复后的MetadataResponse通过SetBroker、SetController等方法补充了这些必要信息，让客户端能完成节点识别。

4. 官方测试的隐含上下文
   Sarama官方测试代码可能在内部测试框架中预设了部分请求的处理逻辑，或依赖特定配置上下文，直接复制到自定义测试环境时，这些隐含条件缺失，导致代码失效。

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