t8y2/dbx · error

decode consumer stats: %w

Error message

decode consumer stats: %w

What it means

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.

Source

Thrown at agents/drivers/rocketmq/consumers.go:574

		successCount++
		for key, offset := range partial.OffsetTable {
			merged.OffsetTable[key] = offset
		}
		merged.ConsumeTps += partial.ConsumeTps
	}
	if successCount == 0 {
		return nil, fmt.Errorf("query consumer lag for group %s on all masters: %w", groupID, lastErr)
	}
	if len(merged.OffsetTable) == 0 && successCount != len(addresses) {
		return nil, fmt.Errorf("consumer lag for group %s is incomplete: %w", groupID, lastErr)
	}
	return merged, nil
}

func decodeConsumeStats(body []byte) (*admin.ConsumeStats, error) {
	var stats admin.ConsumeStats
	if err := json.Unmarshal(repairRocketMQJSON(body), &stats); err != nil {
		return nil, fmt.Errorf("decode consumer stats: %w", err)
	}
	if stats.OffsetTable == nil {
		stats.OffsetTable = make(map[string]*admin.OffsetWrapper)
	}
	return &stats, nil
}

func (a *rocketMQAgent) collectSubscriptionGroupConfigs(ctx context.Context) (map[string]*subscriptionGroupConfig, error) {
	addresses, err := a.masterBrokerAddresses("")
	if err != nil {
		return nil, err
	}
	merged := make(map[string]*subscriptionGroupConfig)
	var lastErr error
	for _, address := range addresses {
		configs, fetchErr := fetchSubscriptionGroupConfigs(ctx, address)
		if fetchErr != nil {
			lastErr = fetchErr

View on GitHub (pinned to c0390bff16)

Solutions

1. Log the raw body (string(body)) before unmarshaling to see what the broker actually returned
2. Verify the request targets a broker admin address (host:port of the broker HTTP remoting server), not a name server or proxy
3. Check broker version compatibility with the admin.ConsumeStats schema; upgrade the driver or downgrade the broker mismatch
4. Confirm repairRocketMQJSON handles the broker's key casing (camelCase vs PascalCase) and extend it if new shapes appear
5. Retry the call, since brokers can transiently return truncated responses under load

Example fix

// before
stats, err := decodeConsumeStats(body)
if err != nil { return err }
// after
if len(bytes.TrimSpace(body)) == 0 || !json.Valid(repairRocketMQJSON(body)) {
    return fmt.Errorf("broker returned non-JSON body: %q", string(body))
}
stats, err := decodeConsumeStats(body)

raw := repairRocketMQJSON(body)
if len(bytes.TrimSpace(raw)) == 0 || !json.Valid(raw) {
    return fmt.Errorf("invalid consumer stats payload: %q", string(body))
}

func isConsumeStatsPayload(raw []byte) bool {
    var probe struct {
        OffsetTable map[string]json.RawMessage `json:"offsetTable"`
    }
    return json.Unmarshal(repairRocketMQJSON(raw), &probe) == nil
}

stats, err := decodeConsumeStats(body)
if err != nil {
    var decErr *json.UnmarshalTypeError
    if errors.As(err, &decErr) {
        log.Printf("consumer stats type mismatch at %s: %v", decErr.Field, decErr)
    } else {
        log.Printf("non-JSON broker response: %v", err)
    }
    return nil, err
}

When it happens

Trigger: 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.

Common situations: 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.

Understand the failure class

Background: "failed to unmarshal" / json.Unmarshal errors: why parsing a response into a Go struct fails and how to fix it — this error's family across 23 libraries.

Related errors

• decode subscription groups: %w
• decode ACL list: %w
• decode consumer status: %w
• encode subscription group config: %w
• topic is required when producerGroup is specified

AI-assisted analysis of t8y2/dbx@c0390bff16 (2026-09-05). Data as JSON: /api/errors/32f3d78787989c43. Report an issue: GitHub.