{"record":{"id":"32f3d78787989c43","repo":"t8y2/dbx","slug":"decode-consumer-stats-w","errorCode":null,"errorMessage":"decode consumer stats: %w","messagePattern":"decode consumer stats: %w","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"agents/drivers/rocketmq/consumers.go","lineNumber":574,"sourceCode":"\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 {\n\t\treturn nil, err\n\t}\n\tmerged := make(map[string]*subscriptionGroupConfig)\n\tvar lastErr error\n\tfor _, address := range addresses {\n\t\tconfigs, fetchErr := fetchSubscriptionGroupConfigs(ctx, address)\n\t\tif fetchErr != nil {\n\t\t\tlastErr = fetchErr","sourceCodeStart":556,"sourceCodeEnd":592,"githubUrl":"https://github.com/t8y2/dbx/blob/c0390bff16418b651f4728520d99adf8ce48829a/agents/drivers/rocketmq/consumers.go#L556-L592","documentation":"decodeConsumeStats wraps a JSON unmarshal failure of the broker's consumer-stats response body into \"decode consumer stats\". The body is first passed through repairRocketMQJSON to fix broker JSON quirks, so this error means the payload is still not unmarshalable into admin.ConsumeStats. It signals a malformed, empty, or structurally unexpected response from the broker's admin API rather than a transport failure.","triggerScenarios":"Calling examineConsumeStatsByTopic where the broker returns HTML/text (e.g. a proxy error page), a truncated body, or a JSON shape with wrong types for ConsumeStats fields (e.g. offsetTable entries as numbers instead of objects) after repairRocketMQJSON could not normalize the keys.","commonSituations":"A reverse proxy or load balancer intercepts the admin port and returns a non-JSON error page; the broker is an incompatible RocketMQ version emitting different field names/casing; the request hit a non-broker HTTP endpoint; TLS/auth gateway returns an empty body.","solutions":["Log the raw body (string(body)) before unmarshaling to see what the broker actually returned","Verify the request targets a broker admin address (host:port of the broker HTTP remoting server), not a name server or proxy","Check broker version compatibility with the admin.ConsumeStats schema; upgrade the driver or downgrade the broker mismatch","Confirm repairRocketMQJSON handles the broker's key casing (camelCase vs PascalCase) and extend it if new shapes appear","Retry the call, since brokers can transiently return truncated responses under load"],"exampleFix":"// before\nstats, err := decodeConsumeStats(body)\nif err != nil { return err }\n// after\nif len(bytes.TrimSpace(body)) == 0 || !json.Valid(repairRocketMQJSON(body)) {\n    return fmt.Errorf(\"broker returned non-JSON body: %q\", string(body))\n}\nstats, err := decodeConsumeStats(body)","handlingStrategy":"try-catch","validationCode":"raw := repairRocketMQJSON(body)\nif len(bytes.TrimSpace(raw)) == 0 || !json.Valid(raw) {\n    return fmt.Errorf(\"invalid consumer stats payload: %q\", string(body))\n}","typeGuard":"func isConsumeStatsPayload(raw []byte) bool {\n    var probe struct {\n        OffsetTable map[string]json.RawMessage `json:\"offsetTable\"`\n    }\n    return json.Unmarshal(repairRocketMQJSON(raw), &probe) == nil\n}","tryCatchPattern":"stats, err := decodeConsumeStats(body)\nif err != nil {\n    var decErr *json.UnmarshalTypeError\n    if errors.As(err, &decErr) {\n        log.Printf(\"consumer stats type mismatch at %s: %v\", decErr.Field, decErr)\n    } else {\n        log.Printf(\"non-JSON broker response: %v\", err)\n    }\n    return nil, err\n}","preventionTips":["Log raw broker bodies on decode failure","Point admin calls at broker ports, never proxies or nameservers","Extend repairRocketMQJSON when new broker key casings appear","Pin and test against your broker's RocketMQ version","Retry transient truncations"],"tags":["json","deserialization","rocketmq","admin-api"],"backgroundTag":"json-unmarshal-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"}