{"record":{"id":"1ee05c496bf84991","repo":"t8y2/dbx","slug":"s-w","errorCode":null,"errorMessage":"%s: %w","messagePattern":"%s: %w","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"agents/drivers/rocketmq/consumers.go","lineNumber":643,"sourceCode":"\tif err != nil {\n\t\treturn fmt.Errorf(\"encode subscription group config: %w\", err)\n\t}\n\tcommand := remoting.NewRequest(remoting.UpdateAndCreateSubscriptionGroup, nil)\n\tcommand.Body = body\n\t_, err = invokeRemotingWithClient(ctx, address, command)\n\treturn err\n}\n\nfunc ensureMutationCoverage(action, resource string, attempted, succeeded int, lastErr error) error {\n\tif attempted <= 0 {\n\t\treturn fmt.Errorf(\"no RocketMQ master brokers available to %s %s\", action, resource)\n\t}\n\tif succeeded == attempted {\n\t\treturn nil\n\t}\n\tmessage := fmt.Sprintf(\"failed to %s %s on all masters: %d of %d succeeded\", action, resource, succeeded, attempted)\n\tif lastErr != nil {\n\t\treturn fmt.Errorf(\"%s: %w\", message, lastErr)\n\t}\n\treturn fmt.Errorf(\"%s\", message)\n}\n\nfunc (a *rocketMQAgent) enrichConsumerGroups(ctx context.Context, client *admin.Client, rows []map[string]any) {\n\tfor _, row := range rows {\n\t\tgroupID := fmt.Sprint(row[\"groupId\"])\n\t\tconnection, err := client.ExamineConsumerConnectionInfo(ctx, groupID)\n\t\tif err != nil {\n\t\t\tif _, ok := row[\"topics\"]; !ok {\n\t\t\t\trow[\"topics\"] = []string{}\n\t\t\t}\n\t\t\tcontinue\n\t\t}\n\t\trow[\"consumeType\"] = valueOrDefault(connection.ConsumeType, \"UNKNOWN\")\n\t\trow[\"messageModel\"] = valueOrDefault(connection.MessageModel, \"CLUSTERING\")\n\t\trow[\"memberCount\"] = len(connection.ConnectionSet)\n\t\ttopics := make([]string, 0, len(connection.SubscriptionTable))","sourceCodeStart":625,"sourceCodeEnd":661,"githubUrl":"https://github.com/t8y2/dbx/blob/c0390bff16418b651f4728520d99adf8ce48829a/agents/drivers/rocketmq/consumers.go#L625-L661","documentation":"When a mutation was attempted on all master brokers but succeeded on only some, ensureMutationCoverage builds a summary message \"failed to <action> <resource> on all masters: X of Y succeeded\" and wraps the last underlying error with %s: %w. It reports a partial application across the broker set, which for consumer-group config means brokers can disagree on the setting.","triggerScenarios":"deleteConsumerGroup or alterSubscriptionGroupConfig where one or more masters fail (network timeout, broker readonly, auth rejection, version mismatch) while others succeed, so succeeded < attempted.","commonSituations":"One broker down or partitioned in a multi-master cluster; a lagging broker on an older RocketMQ version rejecting the update; TLS/ACL misconfigured on a single node; transient remoting timeout under load.","solutions":["Read the wrapped lastErr to identify which failure mode the failing broker hit","Re-run the mutation after the failed broker recovers to converge the cluster state (the operation is idempotent for config updates)","Check connectivity/ACL to each master broker individually","Verify all brokers run a compatible RocketMQ version supporting the config field","Log which brokers succeeded vs failed to detect persistent divergence"],"exampleFix":"// before\nerr := alterSubscriptionGroupConfig(ctx, addrs, cfg)\nif err != nil { return err } // partial failure lost detail\n// after\nerr := alterSubscriptionGroupConfig(ctx, addrs, cfg)\nif err != nil {\n    var partial *PartialMutationError\n    if errors.As(err, &partial) && partial.Succeeded > 0 {\n        // retry failed brokers only, then re-read config to verify convergence\n    }\n    return err\n}","handlingStrategy":"retry","validationCode":"for _, addr := range masterAddrs {\n    if !brokerHealthy(ctx, addr) {\n        return fmt.Errorf(\"broker %s unhealthy; fix before group config mutation\", addr)\n    }\n}","typeGuard":null,"tryCatchPattern":"err := alterSubscriptionGroupConfig(ctx, addrs, cfg)\nif err != nil {\n    // partial failure: retry a bounded number of times, then re-read configs\n    for i := 0; i < 3; i++ {\n        if err = alterSubscriptionGroupConfig(ctx, addrs, cfg); err == nil {\n            break\n        }\n        time.Sleep(time.Duration(1<<i) * time.Second)\n    }\n    if err != nil {\n        log.Printf(\"cluster diverged after retries: %v\", err)\n    }\n}","preventionTips":["Make config writes idempotent and retryable","Re-read configs after partial failure to detect divergence","Keep broker versions consistent across masters","Monitor per-broker health and ACL/TLS parity","Alert when mutation success count < attempted count"],"tags":["rocketmq","partial-failure","broker-cluster","consistency"],"backgroundTag":"partial-write-failure","analyzedSha":"c0390bff16418b651f4728520d99adf8ce48829a","analyzedAt":"2026-09-05T23:05:10.900Z","contentChangedAt":"2026-09-05T23:05:10.900Z","schemaVersion":2},"datasetVersion":"2026-09-14T00:17:10.932Z"}