query consumer status for group %s on all masters: %w

Error message

query consumer status for group %s on all masters: %w

What it means

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.

Source

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

			continue
		}
		partial, decodeErr := decodeConsumerStatus(response.Body)
		if decodeErr != nil {
			lastErr = decodeErr
			continue
		}
		successCount++
		for _, clientID := range sortedKeys(partial) {
			if merged[clientID] == nil {
				merged[clientID] = make(map[string]int64)
			}
			for _, queueKey := range sortedKeys(partial[clientID]) {
				merged[clientID][queueKey] = partial[clientID][queueKey]
			}
		}
	}
	if successCount == 0 {
		return nil, fmt.Errorf("query consumer status for group %s on all masters: %w", groupID, lastErr)
	}
	return merged, nil
}

func decodeConsumerStatus(body []byte) (map[string]map[string]int64, error) {
	// RocketMQ wraps assignments in GetConsumerStatusBody; admin-go v1.1.1
	// incorrectly decodes the complete response as the inner table.
	var wrapper struct {
		ConsumerTable map[string]map[string]int64 `json:"consumerTable"`
	}
	if err := json.Unmarshal(repairConsumerStatusJSON(body), &wrapper); err != nil {
		return nil, fmt.Errorf("decode consumer status: %w", err)
	}
	if wrapper.ConsumerTable == nil {
		wrapper.ConsumerTable = make(map[string]map[string]int64)
	}
	return wrapper.ConsumerTable, nil
}

View on GitHub (pinned to c0390bff16)

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.

Example fix

// before: no reachability check, error surfaces only here
status, err := agent.GetConsumeStatus(ctx, topic, group, "")
// after: pre-check broker connectivity and log the wrapped cause
if err := pingBrokerMasters(ctx, topic); err != nil {
    return fmt.Errorf("brokers unavailable before status query: %w", err)
}
status, err := agent.GetConsumeStatus(ctx, topic, group, "")
if err != nil {
    var qe *queryError
    if errors.As(err, &qe) { log.Printf("underlying: %v", errors.Unwrap(err)) }
}

Defensive patterns

Strategy: try-catch

Validate before calling

// before calling GetConsumeStatus, verify masters are reachable
addrs, err := masterAddressesForTopic(client, topic)
if err != nil { return err }
for _, a := range addrs {
    conn, err := net.DialTimeout("tcp", a, 2*time.Second)
    if err != nil { return fmt.Errorf("master %s unreachable: %w", a, err) }
    conn.Close()
}

Type guard

func isAllMastersFailed(err error) bool {
    return err != nil && strings.Contains(err.Error(), "on all masters")
}

Try / catch

status, err := agent.GetConsumeStatus(ctx, topic, group, "")
if err != nil {
    if isAllMastersFailed(err) {
        cause := errors.Unwrap(err)
        log.Printf("consumer status unavailable for %s: %v", group, cause)
        return retryWithBackoff(3, 2*time.Second, func() error { _, e := agent.GetConsumeStatus(ctx, topic, group, ""); return e })
    }
    return err
}

Prevention

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.

When it happens

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

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

Related errors

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