{"record":{"id":"668cef65c6c0cb5d","repo":"risingwavelabs/risingwave","slug":"expect-a-string-or-a-json-array-for-privatelink-en","errorCode":null,"errorMessage":"expect a string or a json array for privatelink.endpoint, but got {:?}","messagePattern":"expect a string or a json array for privatelink\\.endpoint, but got (.+?)","errorType":"validation","errorClass":null,"httpStatus":null,"severity":"error","filePath":"src/connector/src/source/kafka/private_link.rs","lineNumber":226,"sourceCode":"                        \"expected JSON in the form {{\\\"host\\\": \\\"endpoint url\\\"}}, but got {}\",\n                        v\n                    )\n                })\n            })\n            .collect::<Result<Vec<_>, _>>()?;\n        for ((link, broker), endpoint) in link_targets\n            .iter()\n            .zip_eq_fast(broker_addrs.iter())\n            .zip_eq_fast(endpoint_list.iter())\n        {\n            // rewrite the broker address to endpoint:port\n            broker_rewrite_map.insert(\n                broker.to_string(),\n                format!(\"{}:{}\", endpoint.host, link.port),\n            );\n        }\n    } else {\n        bail!(\n            \"expect a string or a json array for privatelink.endpoint, but got {:?}\",\n            endpoint\n        )\n    }\n\n    Ok(())\n}\n\n#[cfg(test)]\nmod tests {\n    use super::*;\n\n    #[test]\n    fn test_handle_privatelink_endpoint() {\n        let endpoint = \"some_url\"; // raw string\n        let link_targets = vec![\n            AwsPrivateLinkItem {\n                az_id: None,","sourceCodeStart":208,"sourceCodeEnd":244,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/source/kafka/private_link.rs#L208-L244","documentation":"In `handle_privatelink_endpoint` (src/connector/src/source/kafka/private_link.rs:226), RisingWave parses the `privatelink.endpoint` WITH clause option as JSON. It accepts either a plain string (one endpoint used for all brokers) or a JSON array of `{\"host\": ...}` objects (one per broker/AZ). If the parsed value is neither a JSON string nor an array (e.g. a number, boolean, or object), the connector bails with this error because there is no defined way to map the value onto broker addresses.","triggerScenarios":"Creating a Kafka source with privatelink options where the `privatelink.endpoint` value parses as a JSON type other than string or array — e.g. `privatelink.endpoint='123'`, `'true'`, or a JSON object `'\"host\":\"x\"'` that was not wrapped in an array.","commonSituations":"Users typo the endpoint value without quotes, paste a host:port where a number port gets parsed as a JSON number, or supply a single JSON object instead of an array of objects when specifying per-AZ endpoints.","solutions":["Set `privatelink.endpoint` to a plain endpoint string, e.g. `privatelink.endpoint = 'vpce-xxxx.s3.ap-southeast-1.vpce.amazonaws.com'`.","If per-broker endpoints are needed, use a JSON array of {\"host\": \"...\"} objects, e.g. `privatelink.endpoint = '[{\"host\":\"vpce-0a11\"},{\"host\":\"vpce-0a22\"}]'` — the array length must match the broker count.","Check that the value is not accidentally a bare number/boolean (quote it in SQL so it stays a string)."],"exampleFix":"// before (WITH option)\nprivatelink.endpoint = '12345'            -- parsed as JSON number -> error\n// after\nprivatelink.endpoint = 'my-vpce-endpoint.example.com'  -- plain string","handlingStrategy":"validation","validationCode":"// Validate the privatelink.endpoint value before CREATE SOURCE\nlet v: serde_json::Value = serde_json::from_str(raw)\n    .unwrap_or(serde_json::Value::String(raw.to_string()));\nlet valid = v.is_string()\n    || (v.is_array()\n        && v.as_array().unwrap().iter().all(|item| {\n            item.get(\"host\").map_or(false, |h| h.is_string())\n        }));\nassert!(valid, \"privatelink.endpoint must be a string or an array of {{\\\"host\\\": \\\"...\\\"}}\");","typeGuard":"fn is_valid_privatelink_endpoint(v: &serde_json::Value) -> bool {\n    v.is_string()\n        || (v.is_array()\n            && v.as_array().unwrap().iter().all(|i| i.get(\"host\").map_or(false, |h| h.is_string())))\n}","tryCatchPattern":null,"preventionTips":["Quote string endpoint values in SQL so they are not parsed as JSON numbers/booleans","Use the documented form: one string, or an array of {\"host\": \"...\"} objects","For per-AZ arrays, match the array length to the broker count"],"tags":["kafka","config","aws-privatelink","source-creation"],"backgroundTag":"invalid-config-value","analyzedSha":"6469eb736d691e8e9b8a419a57edd6429ca77417","analyzedAt":"2026-09-11T21:06:21.487Z","contentChangedAt":"2026-09-11T21:06:21.487Z","schemaVersion":2},"datasetVersion":"2026-09-14T11:17:12.474Z"}