{"record":{"id":"2b1e86cbfb1c0a9e","repo":"risingwavelabs/risingwave","slug":"the-number-of-broker-addrs-does-not-match-the-n","errorCode":null,"errorMessage":"The number of broker addrs {} does not match the number of private link targets {}","messagePattern":"The number of broker addrs (.+?) does not match the number of private link targets (.+?)","errorType":"validation","errorClass":null,"httpStatus":null,"severity":"error","filePath":"src/connector/src/source/kafka/private_link.rs","lineNumber":135,"sourceCode":"        .map_err(Into::into)\n}\n\npub fn insert_privatelink_broker_rewrite_map(\n    with_options: &mut BTreeMap<String, String>,\n    svc: Option<&PrivateLinkService>,\n    privatelink_endpoint: Option<String>,\n) -> ConnectorResult<()> {\n    let mut broker_rewrite_map = HashMap::new();\n    let servers = get_property_required(with_options, kafka_props_broker_key(with_options))?;\n    let broker_addrs = servers.split(',').collect_vec();\n    let link_target_value = get_property_required(with_options, PRIVATE_LINK_TARGETS_KEY)?;\n    let link_targets: Vec<AwsPrivateLinkItem> =\n        serde_json::from_str(link_target_value.as_str()).map_err(|e| anyhow!(e))?;\n    // remove the private link targets from WITH options, as they are useless after we constructed the rewrite mapping\n    with_options.remove(PRIVATE_LINK_TARGETS_KEY);\n\n    if broker_addrs.len() != link_targets.len() {\n        bail!(\n            \"The number of broker addrs {} does not match the number of private link targets {}\",\n            broker_addrs.len(),\n            link_targets.len()\n        );\n    }\n\n    if let Some(endpoint) = privatelink_endpoint {\n        // new syntax: endpoint can either be a string or a json array of strings\n        // if it is a string, rewrite all broker addresses to the same endpoint\n        // eg. privatelink.endpoint='some_url' ==> broker1:9092 -> some_url:9092, broker2:9093 -> some_url:9093\n        // if it is a json array, rewrite each broker address to the corresponding endpoint\n        // eg. privatelink.endpoint = '[{\"host\": \"aaaa\"}, {\"host\": \"bbbb\"}, {\"host\": \"cccc\"}]'\n        // ==> broker1:9092 -> aaaa:9092, broker2:9093 -> bbbb:9093, broker3:9094 -> cccc:9094\n        handle_privatelink_endpoint(\n            &endpoint,\n            &mut broker_rewrite_map,\n            &link_targets,\n            &broker_addrs,","sourceCodeStart":117,"sourceCodeEnd":153,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/source/kafka/private_link.rs#L117-L153","documentation":"When Kafka brokers are accessed through AWS PrivateLink, each broker address must map 1:1 to a private link target (ENI). This error is raised when the number of broker addresses in `broker_addrs` differs from the number of entries parsed from the `privatelink.targets` JSON, making the rewrite map ambiguous.","triggerScenarios":"`insert_privatelink_broker_rewrite_map` (via `resolve_privatelink_in_with_option`) is called and `broker_addrs.len() != link_targets.len()` — e.g. user provided 3 brokers but 2 private link targets in the `privatelink.targets` JSON.","commonSituations":"MSK cluster scaled to more brokers after targets were configured; targets JSON truncated when pasting; user listed only one target but gave multiple bootstrap broker addresses; stale targets after cluster re-creation.","solutions":["Update the `privatelink.targets` JSON so it has exactly one entry per broker address","Ensure every broker address (all bootstrap broker host:ports) is included in both lists","Regenerate targets from the current MSK cluster info if brokers were added","Remove stale targets left over from a previous cluster configuration"],"exampleFix":"// before (2 targets for 3 brokers)\nprivatelink.targets = '[{\"host\":\"vpce-a\"},{\"host\":\"vpce-b\"}]'\n// after\nprivatelink.targets = '[{\"host\":\"vpce-a\"},{\"host\":\"vpce-b\"},{\"host\":\"vpce-c\"}]'","handlingStrategy":"validation","validationCode":"let targets: Vec<_> = serde_json::from_str::<Vec<serde_json::Value>>(&targets_json)?;\nif targets.len() != broker_addrs.len() {\n    return Err(format!(\"need {} privatelink targets, got {}\", broker_addrs.len(), targets.len()));\n}","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Keep one privatelink target per broker address","Regenerate targets after MSK broker count changes","Paste full JSON arrays; partial pastes cause count mismatch"],"tags":["kafka","aws","privatelink","config","count-mismatch"],"backgroundTag":"invalid-argument-value","analyzedSha":"6469eb736d691e8e9b8a419a57edd6429ca77417","analyzedAt":"2026-09-11T21:06:21.487Z","contentChangedAt":"2026-09-11T21:06:21.487Z","schemaVersion":2},"datasetVersion":"2026-09-14T16:17:12.679Z"}