{"record":{"id":"518fe7eec888ec4c","repo":"t8y2/dbx","slug":"decode-consumer-status-w","errorCode":null,"errorMessage":"decode consumer status: %w","messagePattern":"decode consumer status: %w","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"agents/drivers/rocketmq/consumers.go","lineNumber":372,"sourceCode":"\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}\n\nfunc repairConsumerStatusJSON(body []byte) []byte {\n\trepaired := repairRocketMQJSON(body)\n\tresult := make([]byte, 0, len(repaired)+64)\n\tfor index := 0; index < len(repaired); {\n\t\tmapStart := index + 1\n\t\tfor mapStart < len(repaired) && isJSONSpace(repaired[mapStart]) {\n\t\t\tmapStart++\n\t\t}\n\t\tif repaired[index] == ':' && mapStart+1 < len(repaired) &&\n\t\t\trepaired[mapStart] == '{' && repaired[mapStart+1] == '{' {\n\t\t\tif converted, next, ok := convertObjectKeyedInt64Map(repaired, mapStart); ok {","sourceCodeStart":354,"sourceCodeEnd":390,"githubUrl":"https://github.com/t8y2/dbx/blob/c0390bff16418b651f4728520d99adf8ce48829a/agents/drivers/rocketmq/consumers.go#L354-L390","documentation":"decodeConsumerStatus unmarshals the broker's GetConsumerStatus response body into a wrapper struct with a consumerTable field (a workaround for admin-go v1.1.1 decoding the full response as the inner table). This error wraps any json.Unmarshal failure after repairConsumerStatusJSON has attempted to fix non-standard JSON (e.g. object-keyed int64 maps).","triggerScenarios":"Calling GetConsumeStatus / readConsumerStatusFromBrokers when a broker returns a body that, even after repairRocketMQJSON and convertObjectKeyedInt64Map processing, is not valid JSON or does not match {\"consumerTable\": map[string]map[string]int64} — e.g. truncated response, HTML error page, or unexpected field types.","commonSituations":"Mixed RocketMQ broker versions emitting different response shapes; proxy/load balancer returning an error page instead of JSON; response body truncated by network issues; a RocketMQ version whose JSON shape defeats the repair heuristics; running against a proxy (e.g. RocketMQ 5.x proxy) instead of a native broker.","solutions":["Log the raw response body (hex or string) for the failing broker and identify what non-JSON content was returned.","Check broker version consistency across masters; align all brokers to one RocketMQ release.","Ensure the address being queried is a native broker, not a 5.x proxy with a different response format.","Extend repairConsumerStatusJSON/convertObjectKeyedInt64Map to handle the new body shape, or upgrade the driver.","Retry: a transient truncation will usually succeed on a second request."],"exampleFix":null,"handlingStrategy":"try-catch","validationCode":"// sanity-check the broker returns JSON before relying on status decoding\nvar probe map[string]json.RawMessage\nif err := json.Unmarshal(body, &probe); err != nil {\n    return fmt.Errorf(\"broker returned non-JSON body (%d bytes): %w\", len(body), err)\n}\nif _, ok := probe[\"consumerTable\"]; !ok {\n    log.Printf(\"warning: response lacks consumerTable key; keys=%v\", keys(probe))\n}","typeGuard":"func isDecodeError(err error) bool {\n    return err != nil && strings.Contains(err.Error(), \"decode consumer status:\")\n}","tryCatchPattern":"status, err := agent.GetConsumeStatus(ctx, topic, group, \"\")\nif err != nil && isDecodeError(err) {\n    log.Printf(\"undecodable broker body for %s: %v — dumping raw response\", group, err)\n    dumpBrokerResponse(topic, group)\n    return fallbackStatusFromClientConnections(ctx, group)\n}","preventionTips":["Keep broker versions homogeneous so response JSON shapes match.","Point the driver at native brokers, not 5.x proxies, unless the driver supports proxy responses.","Log raw response bodies on decode failure for upstream repair heuristics.","Beware proxies/LBs that inject HTML error pages into responses."],"tags":["rocketmq","json","decoding","response-parsing"],"backgroundTag":"json-decode-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"}