t8y2/dbx · error

decode topic stats: %w

Error message

decode topic stats: %w

What it means

decodeTopicStats wraps any JSON unmarshal failure of the broker's GetTopicStatsInfo response body. The library first runs repairRocketMQJSON (RocketMQ brokers sometimes emit non-standard JSON with unquoted integer keys), so this error means even after repair the body could not be parsed into admin.TopicStatsTable.

Source

Thrown at agents/drivers/rocketmq/topics.go:312

		}
		successCount++
		for key, offset := range partial {
			merged[key] = offset
		}
	}
	if successCount != len(addresses) {
		return nil, fmt.Errorf("query topic stats for %s on all masters: %w", topic, lastErr)
	}
	if len(merged) == 0 {
		return nil, fmt.Errorf("topic stats not found: %s", topic)
	}
	return merged, nil
}

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

func masterAddressesFromRoute(route *admin.TopicRouteData) []string {
	addresses := make(map[string]struct{})
	for _, broker := range route.BrokerDatas {
		if address := broker.BrokerAddrs["0"]; address != "" {
			addresses[address] = struct{}{}
		}
	}
	return sortedKeys(addresses)
}

func (a *rocketMQAgent) getTopicConfig(params map[string]any) (any, error) {

View on GitHub (pinned to c0390bff16)

Solutions

1. Capture and inspect the raw response body to see what the broker actually returned
2. Upgrade the RocketMQ broker or driver so the stats response uses standard JSON that repairRocketMQJSON handles
3. Check for proxies/LBs or TLS endpoints mangling the body — connect directly to the broker master address
4. Retry the request; if it is transient (truncated body), the per-master loop in examineTopicStats already records it as lastErr

Example fix

// before
partial, err := decodeTopicStats(response.Body)
if err != nil { return nil, err }
// after
if !json.Valid(response.Body) {
    return nil, fmt.Errorf("broker returned non-JSON body (%d bytes): %q", len(response.Body), response.Body[:min(64, len(response.Body))])
}
partial, err := decodeTopicStats(response.Body)
if err != nil { return nil, err }

if len(body) == 0 || body[0] != '{' {
    return fmt.Errorf("unexpected stats response, not JSON: %q", body)
}

func isDecodeTopicStatsErr(err error) bool {
    return err != nil && strings.Contains(err.Error(), "decode topic stats")
}

partial, err := decodeTopicStats(body)
if err != nil {
    if isDecodeTopicStatsErr(err) {
        log.Printf("bad stats body from broker, retrying: %v", err)
        return retryFetch()
    }
    return err
}

When it happens

Trigger: A master broker returns a malformed or non-JSON body for GetTopicStatsInfo (examineTopicStats calls decodeTopicStats on each response); also produced directly by TestDecodeTopicStatsRepairsObjectKeys when feeding corrupt payloads.

Common situations: Older or patched RocketMQ broker versions emitting legacy JSON with bare numeric keys that repairRocketMQJSON cannot fix; a proxy/LB returning an HTML error page instead of the broker response; truncated responses on flaky networks; wrong remoting protocol version.

Understand the failure class

Background: JSON parse error: "Unexpected token" / "not valid JSON" / "failed to parse" — what JSON parsers are really complaining about — this error's family across 45 libraries.

Related errors

• decode ACL list: %w
• decode consumer status: %w
• decode consumer stats: %w
• decode subscription groups: %w
• encode subscription group config: %w

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