突然无法获取collections列表

[elpsycongroo]

部署：2.4.5 k8s集群 外置kafka/etcd/minio

现象： 突然无法获取collections
querynode日志：

[2024/12/13 07:05:20.165 +00:00] [WARN] [kafka/kafka_consumer.go:138] ["consume msg failed"] [topic=by-dev-rootcoord-dml_1] [groupID=querynode-64-by-dev-rootcoord-dml_1_454555578458527444v0-true] [error="Local: Timed out"]
[2024/12/13 07:05:20.518 +00:00] [WARN] [kafka/kafka_consumer.go:138] ["consume msg failed"] [topic=by-dev-rootcoord-dml_0] [groupID=querynode-64-by-dev-rootcoord-dml_0_451699909218184751v0-true] [error="Local: Timed out"]

已经确定目前kafka正常，重启pod无效，依旧timed out，kafka端显示无消息生产，milvus消费端也消费不到数据...

[elpsycongroo]

补充proxy日志：

[2024/12/13 07:14:39.453 +00:00] [INFO] [proxy/meta_cache.go:928] ["no shard cache for collection, try to get shard leaders from QueryCoord"] [traceID=1f4c7dac1db96420ffad892a84c2f53c] [collectionName=Vector_index_387927b4_c233_4c86_9556_852c9a6e9f28_Node] [collectionID=451699909167843048]
[2024/12/13 07:14:39.455 +00:00] [WARN] [proxy/lb_policy.go:203] ["failed to get shards"] [traceID=1f4c7dac1db96420ffad892a84c2f53c] [error="channel not subscribed: channel lacks[channel=by-dev-rootcoord-dml_2_451699909167843048v0]: channel not available[channel=by-dev-rootcoord-dml_2_451699909167843048v0]"]
[2024/12/13 07:14:39.455 +00:00] [WARN] [proxy/task_search.go:511] ["search execute failed"] [traceID=1f4c7dac1db96420ffad892a84c2f53c] [nq=1] [error="channel not subscribed: channel lacks[channel=by-dev-rootcoord-dml_2_451699909167843048v0]: channel not available[channel=by-dev-rootcoord-dml_2_451699909167843048v0]"]
[2024/12/13 07:14:39.455 +00:00] [DEBUG] [proxy/task_search.go:512] ["tr/proxy execute search 454576853308887485"] [traceID=1f4c7dac1db96420ffad892a84c2f53c] [msg=done] [duration=1.882624ms]
[2024/12/13 07:14:39.455 +00:00] [WARN] [proxy/task_scheduler.go:469] ["Failed to execute task: "] [traceID=1f4c7dac1db96420ffad892a84c2f53c] [error="failed to search: channel not subscribed: channel lacks[channel=by-dev-rootcoord-dml_2_451699909167843048v0]: channel not available[channel=by-dev-rootcoord-dml_2_451699909167843048v0]"] [errorVerbose="failed to search: channel not subscribed: channel lacks[channel=by-dev-rootcoord-dml_2_451699909167843048v0]: channel not available[channel=by-dev-rootcoord-dml_2_451699909167843048v0]\n(1) attached stack trace\n  -- stack trace:\n  | github.com/milvus-io/milvus/internal/proxy.(*searchTask).Execute\n  | \t/go/src/github.com/milvus-io/milvus/internal/proxy/task_search.go:512\n  | github.com/milvus-io/milvus/internal/proxy.(*taskScheduler).processTask\n  | \t/go/src/github.com/milvus-io/milvus/internal/proxy/task_scheduler.go:466\n  | github.com/milvus-io/milvus/internal/proxy.(*taskScheduler).queryLoop.func1\n  | \t/go/src/github.com/milvus-io/milvus/internal/proxy/task_scheduler.go:545\n  | github.com/milvus-io/milvus/pkg/util/conc.(*Pool[...]).Submit.func1\n  | \t/go/src/github.com/milvus-io/milvus/pkg/util/conc/pool.go:81\n  | github.com/panjf2000/ants/v2.(*goWorker).run.func1\n  | \t/go/pkg/mod/github.com/panjf2000/ants/v2@v2.7.2/worker.go:67\n  | runtime.goexit\n  | \t/usr/local/go/src/runtime/asm_amd64.s:1598\nWraps: (2) failed to search\nWraps: (3) channel not subscribed: channel lacks[channel=by-dev-rootcoord-dml_2_451699909167843048v0]: channel not available[channel=by-dev-rootcoord-dml_2_451699909167843048v0]\nError types: (1) *withstack.withStack (2) *errutil.withPrefix (3) merr.milvusError"]
[2024/12/13 07:14:39.455 +00:00] [WARN] [proxy/impl.go:3006] ["Search failed to WaitToFinish"] [traceID=1f4c7dac1db96420ffad892a84c2f53c] [role=proxy] [db=llm] [collection=Vector_index_387927b4_c233_4c86_9556_852c9a6e9f28_Node] [partitions="[]"] [dsl=] [len(PlaceholderGroup)=4108] [OutputFields="[page_content,metadata]"] [search_params="[{\"key\":\"topk\",\"value\":\"1\"},{\"key\":\"params\",\"value\":\"{}\"},{\"key\":\"round_decimal\",\"value\":\"-1\"},{\"key\":\"ignore_growing\",\"value\":\"False\"}]"] [guarantee_timestamp=1] [useDefaultConsistency=true] [nq=1] [error="failed to search: channel not subscribed: channel lacks[channel=by-dev-rootcoord-dml_2_451699909167843048v0]: channel not available[channel=by-dev-rootcoord-dml_2_451699909167843048v0]"] [errorVerbose="failed to search: channel not subscribed: channel lacks[channel=by-dev-rootcoord-dml_2_451699909167843048v0]: channel not available[channel=by-dev-rootcoord-dml_2_451699909167843048v0]\n(1) attached stack trace\n  -- stack trace:\n  | github.com/milvus-io/milvus/internal/proxy.(*searchTask).Execute\n  | \t/go/src/github.com/milvus-io/milvus/internal/proxy/task_search.go:512\n  | github.com/milvus-io/milvus/internal/proxy.(*taskScheduler).processTask\n  | \t/go/src/github.com/milvus-io/milvus/internal/proxy/task_scheduler.go:466\n  | github.com/milvus-io/milvus/internal/proxy.(*taskScheduler).queryLoop.func1\n  | \t/go/src/github.com/milvus-io/milvus/internal/proxy/task_scheduler.go:545\n  | github.com/milvus-io/milvus/pkg/util/conc.(*Pool[...]).Submit.func1\n  | \t/go/src/github.com/milvus-io/milvus/pkg/util/conc/pool.go:81\n  | github.com/panjf2000/ants/v2.(*goWorker).run.func1\n  | \t/go/pkg/mod/github.com/panjf2000/ants/v2@v2.7.2/worker.go:67\n  | runtime.goexit\n  | \t/usr/local/go/src/runtime/asm_amd64.s:1598\nWraps: (2) failed to search\nWraps: (3) channel not subscribed: channel lacks[channel=by-dev-rootcoord-dml_2_451699909167843048v0]: channel not available[channel=by-dev-rootcoord-dml_2_451699909167843048v0]\nError types: (1) *withstack.withStack (2) *errutil.withPrefix (3) merr.milvusError"]

[yhmo]

[kafka/kafka_consumer.go:138] ["consume msg failed"] [topic=by-dev-rootcoord-dml_1] [groupID=querynode-64-by-dev-rootcoord-dml_1_454555578458527444v0-true] [error="Local: Timed out"]

这是调用kafka客户端的Consumer接口超时了，多半是集群和kafka之间的连接出现了变化。

[elpsycongroo]

目前来看是一个大消息过来被kafka拒绝接收了。
如果更换kafka集群，会有什么后果？现有的数据会丢失吗？

[elpsycongroo]

如果确定了连接已经恢复，但还是显示超时，重启无效，还有可能是什么原因，是否可能是积压量过大，导致30s超时？

[elpsycongroo]

补充： queryCoord日志：

[2024/12/13 10:24:32.963 +00:00] [DEBUG] [meta/target_manager.go:196] ["start to pull next targets for collection"] [collectionID=454555578458527444] [chosenPartitionIDs="[454555578458527445]"]
[2024/12/13 10:24:32.964 +00:00] [WARN] [checkers/index_checker.go:127] ["failed to get indexInfo for segment"] [collectionID=451699909185716484] [segmentID=451699909185716549] [error="index not found[segmentID=451699909185716549]"]
[2024/12/13 10:24:32.964 +00:00] [DEBUG] [meta/target_manager.go:137] ["finish to update next targets for collection"] [collectionID=454555578458527444] [PartitionIDs="[454555578458527445]"] [segments="[454576853310372051,454576853310372041]"] [channels="[by-dev-rootcoord-dml_1_454555578458527444v0]"]
[2024/12/13 10:24:32.965 +00:00] [WARN] [checkers/index_checker.go:127] ["failed to get indexInfo for segment"] [collectionID=451699909181151319] [segmentID=451699909181151386] [error="index not found[segmentID=451699909181151386]"]
[2024/12/13 10:24:32.966 +00:00] [WARN] [checkers/index_checker.go:127] ["failed to get indexInfo for segment"] [collectionID=451699909181151319] [segmentID=451699909183638164] [error="index not found[segmentID=451699909183638164]"]
[2024/12/13 10:24:32.966 +00:00] [WARN] [checkers/index_checker.go:127] ["failed to get indexInfo for segment"] [collectionID=451699909181151319] [segmentID=451699909184292313] [error="index not found[segmentID=451699909184292313]"]
[2024/12/13 10:24:32.974 +00:00] [WARN] [checkers/index_checker.go:127] ["failed to get indexInfo for segment"] [collectionID=451699909136664853] [segmentID=451699909136664890] [error="index not found[segmentID=451699909136664890]"]
[2024/12/13 10:24:32.975 +00:00] [WARN] [checkers/index_checker.go:127] ["failed to get indexInfo for segment"] [collectionID=451699909141017383] [segmentID=451699909141017411] [error="index not found[segmentID=451699909141017411]"]

[yhmo]

kafka的连接只跟milvus.yaml里的配置相关：

milvus/configs/milvus.yaml

Line 202 in c1f9158

# kafka:

readTimeout是超时设置，默认10秒。一般不会有这么大的message需要10秒，它可能就是单纯地连不上。
kafka连不上的话，其他奇奇怪怪的报错日志会很多。
kafka里主要保存的是insert/upsert/delete的数据作为write-ahead-log，以及time-tick msg，这些write-ahead-log等待querynode,datanode消费，然后datanode把数据落盘，kafka里的已消费的数据就不再有用。假设所有在kafka中的数据都已经被datanode消费并落盘，那么你是可以更换kafka集群的。

[elpsycongroo]

看了下，kafka那边显示一条消息有1.2G，是否是因为一次插入数据太多导致的？因为可以确定kafka的连接是没问题的
是否有办法限制一次插入的条数或者消息体大小？

[yhmo]

不可能会有一条1.2G的消息。milvus限制了每次insert最多只能插入64MB数据，这是默认值，除非你改了这个默认值。只要每次插入的数据小于这个数，kafka就不可能会收到单条大于64MB的消息。

[elpsycongroo]

kafka恢复后，milvus是否会去拉之前的wal，导致吞吐量增加？

[yhmo]

就算是恢复阶段，也不会有1.2G的单条消息的，每条都会在64MB以下