t8y2/dbx · error

decode subscription groups

Error message

decode subscription groups: %w

What it means

fetchSubscriptionGroupConfigs unmarshals the broker's FETCH_SUBSCRIPTION_GROUP_CONFIG response, expecting {"subscriptionGroupTable": {...}}. When json.Unmarshal of the repaired body fails it wraps the cause as "decode subscription groups". It means the broker response could not be decoded into the expected wrapper structure.

Solutions

1. Dump response.Body to inspect the actual payload and confirm which broker returned it
2. Check that all master brokers run a RocketMQ version whose FetchSubscriptionGroupConfig response matches the expected subscriptionGroupTable shape
3. Ensure repairRocketMQJSON normalizes the key casing used by the broker version
4. Verify network path: query broker admin ports directly, bypassing proxies
5. Retry the fetch for transient truncation

Example fix

// before
table, err := fetchSubscriptionGroupConfigs(ctx, address)
if err != nil { return nil, err }
// after
table, err := fetchSubscriptionGroupConfigs(ctx, address)
if err != nil {
    return nil, fmt.Errorf("fetch subscription groups from %s: %w (broker body may be non-JSON)", address, err)
}

raw := repairRocketMQJSON(response.Body)
var probe map[string]json.RawMessage
if err := json.Unmarshal(raw, &probe); err != nil || probe["subscriptionGroupTable"] == nil {
    return fmt.Errorf("unexpected subscription group payload: %q", string(raw))
}

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

table, err := fetchSubscriptionGroupConfigs(ctx, address)
if err != nil {
    log.Printf("broker %s returned undecodable subscription groups: %v", address, err)
    // fall back to skipping this broker instead of failing the whole collection
    return nil, err
}

When it happens

Trigger: Calling collectSubscriptionGroupConfigs when a broker returns a non-JSON body (proxy error, empty 200), a different payload shape (older/newer RocketMQ admin API), or fields whose types do not match subscriptionGroupConfig.

Common situations: Mixed broker versions in the cluster where some masters return a legacy schema; a gateway on the admin port returning an error page; broker restarted mid-request yielding a truncated 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 consumer stats
• decode ACL list
• decode consumer status
• decode topic stats
• encode subscription group config

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

Appendix: source

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

		}
	}
	if len(merged) == 0 && lastErr != nil {
		return nil, lastErr
	}
	return merged, nil
}

func fetchSubscriptionGroupConfigs(ctx context.Context, address string) (map[string]*subscriptionGroupConfig, error) {
	response, err := invokeRemotingWithClient(ctx, address,
		remoting.NewRequest(remoting.GetAllSubscriptionGroupConfig, nil))
	if err != nil {
		return nil, err
	}
	var wrapper struct {
		SubscriptionGroupTable map[string]*subscriptionGroupConfig `json:"subscriptionGroupTable"`
	}
	if err := json.Unmarshal(repairRocketMQJSON(response.Body), &wrapper); err != nil {
		return nil, fmt.Errorf("decode subscription groups: %w", err)
	}
	return wrapper.SubscriptionGroupTable, nil
}

func writeSubscriptionGroupConfig(ctx context.Context, address string, config *subscriptionGroupConfig) error {
	body, err := json.Marshal(config)
	if err != nil {
		return fmt.Errorf("encode subscription group config: %w", err)
	}
	command := remoting.NewRequest(remoting.UpdateAndCreateSubscriptionGroup, nil)
	command.Body = body
	_, err = invokeRemotingWithClient(ctx, address, command)
	return err
}

func ensureMutationCoverage(action, resource string, attempted, succeeded int, lastErr error) error {
	if attempted <= 0 {
		return fmt.Errorf("no RocketMQ master brokers available to %s %s", action, resource)

View on GitHub (pinned to c0390bff16)