{"record":{"id":"fc0c2847aa4043f8","repo":"t8y2/dbx","slug":"consumer-lag-for-group-s-is-incomplete-w","errorCode":null,"errorMessage":"consumer lag for group %s is incomplete: %w","messagePattern":"consumer lag for group (.+?) is incomplete: %w","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"warning","filePath":"agents/drivers/rocketmq/consumers.go","lineNumber":566,"sourceCode":"\t\t\tlastErr = requestErr\n\t\t\tcontinue\n\t\t}\n\t\tpartial, decodeErr := decodeConsumeStats(response.Body)\n\t\tif decodeErr != nil {\n\t\t\tlastErr = decodeErr\n\t\t\tcontinue\n\t\t}\n\t\tsuccessCount++\n\t\tfor key, offset := range partial.OffsetTable {\n\t\t\tmerged.OffsetTable[key] = offset\n\t\t}\n\t\tmerged.ConsumeTps += partial.ConsumeTps\n\t}\n\tif successCount == 0 {\n\t\treturn nil, fmt.Errorf(\"query consumer lag for group %s on all masters: %w\", groupID, lastErr)\n\t}\n\tif len(merged.OffsetTable) == 0 && successCount != len(addresses) {\n\t\treturn nil, fmt.Errorf(\"consumer lag for group %s is incomplete: %w\", groupID, lastErr)\n\t}\n\treturn merged, nil\n}\n\nfunc decodeConsumeStats(body []byte) (*admin.ConsumeStats, error) {\n\tvar stats admin.ConsumeStats\n\tif err := json.Unmarshal(repairRocketMQJSON(body), &stats); err != nil {\n\t\treturn nil, fmt.Errorf(\"decode consumer stats: %w\", err)\n\t}\n\tif stats.OffsetTable == nil {\n\t\tstats.OffsetTable = make(map[string]*admin.OffsetWrapper)\n\t}\n\treturn &stats, nil\n}\n\nfunc (a *rocketMQAgent) collectSubscriptionGroupConfigs(ctx context.Context) (map[string]*subscriptionGroupConfig, error) {\n\taddresses, err := a.masterBrokerAddresses(\"\")\n\tif err != nil {","sourceCodeStart":548,"sourceCodeEnd":584,"githubUrl":"https://github.com/t8y2/dbx/blob/c0390bff16418b651f4728520d99adf8ce48829a/agents/drivers/rocketmq/consumers.go#L548-L584","documentation":"After merging partial consume stats, examineConsumeStatsByTopic returns this error when the merged OffsetTable is empty AND successCount != len(addresses) — i.e. at least one broker succeeded but returned no offset entries, and others failed, so the lag picture is provably incomplete. It guards against reporting totalLag=0 as healthy when part of the cluster was unreadable.","triggerScenarios":"Calling getConsumerLag on a multi-master topic where some masters succeed with empty offset tables (e.g. no queues of this topic/group on them) and the rest fail, leaving merged.OffsetTable empty with successCount < len(addresses).","commonSituations":"Partial broker outage during a lag scrape; one master under ACL that denies the group while others respond empty; a group that hasn't consumed on any surviving master; recently rebalanced topology where queues live on brokers that are currently down.","solutions":["Fix the failing brokers identified via the wrapped lastErr and re-run the lag query to get complete coverage.","Check that the consumer group actually has assignments on the reachable masters (consumer connections online).","Verify all masters for the topic are up: `sh mqadmin brokerStatus -b <addr>` on each.","If the group legitimately has no offsets (fresh group), treat zero-table as valid only when successCount == len(addresses); wait for consumers to start before monitoring lag.","Add alerting on this error so partial-outage lag data is never mistaken for zero lag."],"exampleFix":"// before: treating any lag result as complete\nresult, err := agent.GetConsumerLag(ctx, group, topic)\nreportLag(result) // may hide partial outage\n// after: surface incomplete data distinctly\nresult, err := agent.GetConsumerLag(ctx, group, topic)\nif err != nil && strings.Contains(err.Error(), \"is incomplete\") {\n    markLagStale(group, topic, err) // do not report totalLag=0\n    return\n}\nreportLag(result)","handlingStrategy":"fallback","validationCode":"// ensure full broker coverage before trusting a zero-lag reading\nroute, _ := adminClient.ExamineTopicRouteInfo(ctx, topic)\nmasters := masterAddressesFromRoute(route)\nfor _, a := range masters {\n    if err := pingRemoting(ctx, a, 2*time.Second); err != nil {\n        return fmt.Errorf(\"skip lag check: master %s down, data would be incomplete\", a)\n    }\n}","typeGuard":"func isIncompleteLag(err error) bool {\n    return err != nil && strings.Contains(err.Error(), \"consumer lag for group\") && strings.Contains(err.Error(), \"is incomplete\")\n}","tryCatchPattern":"lag, err := agent.GetConsumerLag(ctx, group, topic)\nswitch {\ncase err == nil:\n    reportLag(lag)\ncase isIncompleteLag(err):\n    markMetricStale(\"consumer_lag\", group, topic) // never emit totalLag=0\ncase err != nil:\n    return err\n}","preventionTips":["Treat zero lag as valid only when every master responded (successCount == len(addresses)).","Alert on this error so partial outages aren't read as healthy consumers.","Restore failed masters promptly; monitor per-broker health alongside lag.","For fresh consumer groups with no offsets, wait for first consumption before lag alerting."],"tags":["rocketmq","consumer-lag","partial-failure","data-quality"],"backgroundTag":"partial-broker-failure","analyzedSha":"c0390bff16418b651f4728520d99adf8ce48829a","analyzedAt":"2026-09-05T23:05:10.900Z","contentChangedAt":"2026-09-05T23:05:10.900Z","schemaVersion":2},"datasetVersion":"2026-09-14T00:17:10.932Z"}