{"record":{"id":"042c8375688f2430","repo":"t8y2/dbx","slug":"s","errorCode":null,"errorMessage":"%s","messagePattern":"%s","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"agents/drivers/rocketmq/consumers.go","lineNumber":645,"sourceCode":"\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))\n\t\tfor topic := range connection.SubscriptionTable {\n\t\t\ttopics = append(topics, topic)","sourceCodeStart":627,"sourceCodeEnd":663,"githubUrl":"https://github.com/t8y2/dbx/blob/c0390bff16418b651f4728520d99adf8ce48829a/agents/drivers/rocketmq/consumers.go#L627-L663","documentation":"This error is produced by ensureMutationCoverage in the RocketMQ agent driver after a mutation (delete/alter of a consumer group or subscription group config) that must be applied to all master brokers. It reports that only some masters succeeded, wrapping the last broker error if one was captured, e.g. 'failed to delete group X on all masters: 2 of 4 succeeded: <cause>'. The library throws it because a partially-applied mutation leaves cluster state inconsistent and must not be reported as success.","triggerScenarios":"Calling deleteConsumerGroup or alterSubscriptionGroupConfig in a multi-master cluster where one or more brokers reject or fail the request (broker down, group/config missing on one broker, permission denied, timeout on one master).","commonSituations":"Cluster with unequal state after a broker was replaced; ACL/permission differences between brokers; one master restarted and out of sync; network partition to a single broker while others respond fine.","solutions":["Inspect the wrapped cause (%w) to identify which broker failed and why.","Check connectivity and health of every master broker (admin operation on each broker address).","Re-run the mutation after fixing the failing broker; operations are idempotent-ish (delete on missing group may need existence check).","Verify broker ACL/credentials are identical across all masters.","If the resource is intentionally missing on some brokers, align cluster state or scope the operation per-broker."],"exampleFix":"// before\nerr := agent.DeleteConsumerGroup(ctx, map[string]any{\"groupName\": \"GID_demo\"})\n// after (check partial-apply cause and retry per failing broker)\nif err != nil {\n    var partial *PartialMutationError // or unwrap the wrapped lastErr\n    if errors.As(err, &partial) {\n        log.Printf(\"applied to %d/%d masters, cause: %v\", partial.Succeeded, partial.Attempted, partial.LastErr)\n        // fix/retry the failing broker before treating the delete as done\n    }\n}","handlingStrategy":"retry","validationCode":"// Check all masters are reachable before mutating\nfor _, addr := range masterAddrs {\n    if err := probeBroker(ctx, addr); err != nil {\n        return fmt.Errorf(\"master %s unreachable, aborting mutation: %w\", addr, err)\n    }\n}","typeGuard":null,"tryCatchPattern":"err := agent.DeleteConsumerGroup(ctx, params)\nif err != nil {\n    var cause error\n    if errors.As(err, &target) || errors.Unwrap(err) != nil {\n        cause = errors.Unwrap(err)\n    }\n    log.Printf(\"partial mutation: %v (root cause: %v)\", err, cause)\n    // verify state per broker, then retry only the failed masters\n}","preventionTips":["Keep broker ACLs/credentials identical across all masters.","Monitor broker health before running admin mutations.","Treat this error as partial success: audit each broker's state before retrying.","Keep cluster versions/configs in sync to avoid per-broker rejections."],"tags":["rocketmq","partial-failure","multi-broker","cluster-consistency"],"backgroundTag":"partial-mutation-failure-across-brokers","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"}