{"record":{"id":"e0598f20562daf67","repo":"t8y2/dbx","slug":"query-consumer-status-for-group-s-on-all-masters","errorCode":null,"errorMessage":"query consumer status for group %s on all masters: %w","messagePattern":"query consumer status for group (.+?) on all masters: %w","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"agents/drivers/rocketmq/consumers.go","lineNumber":360,"sourceCode":"\t\t\tcontinue\n\t\t}\n\t\tpartial, decodeErr := decodeConsumerStatus(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 _, clientID := range sortedKeys(partial) {\n\t\t\tif merged[clientID] == nil {\n\t\t\t\tmerged[clientID] = make(map[string]int64)\n\t\t\t}\n\t\t\tfor _, queueKey := range sortedKeys(partial[clientID]) {\n\t\t\t\tmerged[clientID][queueKey] = partial[clientID][queueKey]\n\t\t\t}\n\t\t}\n\t}\n\tif successCount == 0 {\n\t\treturn nil, fmt.Errorf(\"query consumer status for group %s on all masters: %w\", groupID, lastErr)\n\t}\n\treturn merged, nil\n}\n\nfunc decodeConsumerStatus(body []byte) (map[string]map[string]int64, error) {\n\t// RocketMQ wraps assignments in GetConsumerStatusBody; admin-go v1.1.1\n\t// incorrectly decodes the complete response as the inner table.\n\tvar wrapper struct {\n\t\tConsumerTable map[string]map[string]int64 `json:\"consumerTable\"`\n\t}\n\tif err := json.Unmarshal(repairConsumerStatusJSON(body), &wrapper); err != nil {\n\t\treturn nil, fmt.Errorf(\"decode consumer status: %w\", err)\n\t}\n\tif wrapper.ConsumerTable == nil {\n\t\twrapper.ConsumerTable = make(map[string]map[string]int64)\n\t}\n\treturn wrapper.ConsumerTable, nil\n}","sourceCodeStart":342,"sourceCodeEnd":378,"githubUrl":"https://github.com/t8y2/dbx/blob/c0390bff16418b651f4728520d99adf8ce48829a/agents/drivers/rocketmq/consumers.go#L342-L378","documentation":"readConsumerStatusFromBrokers queries GET_CONSUMER_STATUS on every master broker for a topic and merges per-client offset tables. This error is returned when successCount == 0, i.e. every master broker request failed (transport error, nil response, or decode failure); lastErr holds the final underlying cause. It means the consumer status could not be read from any broker, not that data is partial.","triggerScenarios":"Calling GetConsumeStatus (via readConsumerStatusFromBrokers) when all remoting requests to master brokers fail: brokers unreachable, requests time out, brokers return empty bodies, or decodeConsumerStatus fails on every response so successCount stays 0.","commonSituations":"RocketMQ cluster down or master brokers restarted; wrong nameserver/route data so addresses are stale; network/firewall blocking broker remoting port (10911); consumer group or topic name typo'd so brokers reject the query; brokers returning malformed JSON the repairer cannot fix.","solutions":["Check the wrapped lastErr (%w cause) to see the actual failure (timeout, connection refused, decode error) and fix that root cause first.","Verify master brokers are up and reachable on the remoting port: telnet/nc each broker address from the agent host.","Confirm the topic exists and routing data is fresh (ExamineTopicRouteInfo returns current masters) and the consumer group ID is spelled correctly.","If the cause is a decode error, capture the raw broker response body and compare against RocketMQ version expectations; upgrade or patch repairConsumerStatusJSON handling.","Retry the query; if only some brokers were down previously, a healthy master will make successCount > 0."],"exampleFix":"// before: no reachability check, error surfaces only here\nstatus, err := agent.GetConsumeStatus(ctx, topic, group, \"\")\n// after: pre-check broker connectivity and log the wrapped cause\nif err := pingBrokerMasters(ctx, topic); err != nil {\n    return fmt.Errorf(\"brokers unavailable before status query: %w\", err)\n}\nstatus, err := agent.GetConsumeStatus(ctx, topic, group, \"\")\nif err != nil {\n    var qe *queryError\n    if errors.As(err, &qe) { log.Printf(\"underlying: %v\", errors.Unwrap(err)) }\n}","handlingStrategy":"try-catch","validationCode":"// before calling GetConsumeStatus, verify masters are reachable\naddrs, err := masterAddressesForTopic(client, topic)\nif err != nil { return err }\nfor _, a := range addrs {\n    conn, err := net.DialTimeout(\"tcp\", a, 2*time.Second)\n    if err != nil { return fmt.Errorf(\"master %s unreachable: %w\", a, err) }\n    conn.Close()\n}","typeGuard":"func isAllMastersFailed(err error) bool {\n    return err != nil && strings.Contains(err.Error(), \"on all masters\")\n}","tryCatchPattern":"status, err := agent.GetConsumeStatus(ctx, topic, group, \"\")\nif err != nil {\n    if isAllMastersFailed(err) {\n        cause := errors.Unwrap(err)\n        log.Printf(\"consumer status unavailable for %s: %v\", group, cause)\n        return retryWithBackoff(3, 2*time.Second, func() error { _, e := agent.GetConsumeStatus(ctx, topic, group, \"\"); return e })\n    }\n    return err\n}","preventionTips":["Health-check master broker ports before admin queries.","Always unwrap and log the %w cause to distinguish network vs decode failures.","Pin all brokers to one RocketMQ version to keep response shapes consistent.","Set generous but bounded remoting timeouts to survive broker load spikes."],"tags":["rocketmq","network","broker-unreachable","consumer-status"],"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-14T00:17:10.932Z"}