{"record":{"id":"cd22ce552f452051","repo":"t8y2/dbx","slug":"query-consumer-lag-for-group-s-on-all-masters-w","errorCode":null,"errorMessage":"query consumer lag for group %s on all masters: %w","messagePattern":"query consumer lag for group (.+?) on all masters: %w","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"agents/drivers/rocketmq/consumers.go","lineNumber":563,"sourceCode":"\t\t\tmap[string]string{\"consumerGroup\": groupID, \"topic\": topic},\n\t\t))\n\t\tif requestErr != nil {\n\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","sourceCodeStart":545,"sourceCodeEnd":581,"githubUrl":"https://github.com/t8y2/dbx/blob/c0390bff16418b651f4728520d99adf8ce48829a/agents/drivers/rocketmq/consumers.go#L545-L581","documentation":"After querying GetConsumeStats on every master broker, examineConsumeStatsByTopic returns this error when successCount == 0 — no master returned a decodable consume-stats response. The wrapped lastErr carries the final underlying failure (request error or decode error) from the last attempted broker.","triggerScenarios":"Calling getConsumerLag when all masters fail the GetConsumeStats remoting call: network timeouts, brokers rejecting the consumerGroup/topic request, ACL authorization failures, or decodeConsumeStats failing on every body.","commonSituations":"Firewall blocking broker port 10911 from the agent host; consumer group doesn't exist so brokers return errors; ACL credentials missing or expired; brokers overloaded and timing out; mixed broker versions emitting bodies the JSON repairer can't parse.","solutions":["Inspect the wrapped cause (errors.Unwrap) for the real per-broker failure and address it (timeout, auth, decode).","Verify network reachability to each master's remoting port from the agent host.","Confirm the consumer group exists and has valid subscriptions (a never-started group may be rejected); check ACL settings if enabled.","Check broker logs for errors correlated with the GetConsumeStats request timestamp.","Retry after broker load/health issues subside; a single healthy master makes successCount >= 1."],"exampleFix":"// before: fail outright when every master fails\nstats, err := agent.GetConsumerLag(ctx, group, topic)\n// after: log the wrapped cause and retry transient failures\nstats, err := agent.GetConsumerLag(ctx, group, topic)\nif err != nil && isTransient(errors.Unwrap(err)) {\n    stats, err = retryWithBackoff(3, func() error {\n        var e error\n        stats, e = agent.GetConsumerLag(ctx, group, topic)\n        return e\n    })\n}","handlingStrategy":"retry","validationCode":"// pre-flight: confirm at least one master answers before lag collection\nfor _, a := range masterAddresses {\n    if err := pingRemoting(ctx, a, 2*time.Second); err != nil {\n        log.Printf(\"master %s not answering: %v\", a, err)\n    }\n}","typeGuard":"func isLagQueryAllFailed(err error) bool {\n    return err != nil && strings.Contains(err.Error(), \"query consumer lag\") && strings.Contains(err.Error(), \"on all masters\")\n}","tryCatchPattern":"lag, err := agent.GetConsumerLag(ctx, group, topic)\nif err != nil && isLagQueryAllFailed(err) {\n    if backoffErr := retryWithBackoff(3, time.Second, func() error {\n        var e error\n        lag, e = agent.GetConsumerLag(ctx, group, topic)\n        return e\n    }); backoffErr != nil {\n        return fmt.Errorf(\"lag unavailable for %s, cause=%v\", group, errors.Unwrap(backoffErr))\n    }\n}","preventionTips":["Open firewall paths to broker remoting port (10911) from the agent host.","Verify consumer group existence and ACL permissions before scraping lag.","Alert on broker CPU/connection saturation that causes request timeouts.","Keep broker versions aligned to avoid decode failures counted as request failures."],"tags":["rocketmq","network","consumer-lag","broker-unreachable"],"backgroundTag":"all-brokers-failed","analyzedSha":"c0390bff16418b651f4728520d99adf8ce48829a","analyzedAt":"2026-09-05T23:05:10.900Z","contentChangedAt":"2026-09-05T23:05:10.900Z","schemaVersion":2},"datasetVersion":"2026-09-14T05:17:10.506Z"}