risingwavelabs/risingwave · error

expected JSON in the form {{"host": "endpoint url"}}, but go

Error message

expected JSON in the form {{"host": "endpoint url"}}, but got {}

What it means

Each entry of the `privatelink.endpoints` JSON array must deserialize into `PrivateLinkEndpointItem` (an object of the form {"host": "endpoint url"}). When an element of the array fails serde deserialization, this error reports the offending raw JSON value.

Source

Thrown at src/connector/src/source/kafka/private_link.rs:207

    let endpoint = if let Ok(json) = serde_json::from_str::<serde_json::Value>(endpoint) {
        json
    } else {
        serde_json::Value::String(endpoint.to_owned())
    };
    if matches!(endpoint, serde_json::Value::String(_)) {
        let endpoint = endpoint.as_str().unwrap();
        for (link, broker) in link_targets.iter().zip_eq_fast(broker_addrs.iter()) {
            // rewrite the broker address to endpoint:port
            broker_rewrite_map.insert(broker.to_string(), format!("{}:{}", endpoint, link.port));
        }
    } else if matches!(endpoint, serde_json::Value::Array(_)) {
        let endpoint_list: Vec<PrivateLinkEndpointItem> = endpoint
            .as_array()
            .unwrap()
            .iter()
            .map(|v| {
                serde_json::from_value(v.clone()).map_err(|_| {
                    anyhow!(
                        "expected JSON in the form {{\"host\": \"endpoint url\"}}, but got {}",
                        v
                    )
                })
            })
            .collect::<Result<Vec<_>, _>>()?;
        for ((link, broker), endpoint) in link_targets
            .iter()
            .zip_eq_fast(broker_addrs.iter())
            .zip_eq_fast(endpoint_list.iter())
        {
            // rewrite the broker address to endpoint:port
            broker_rewrite_map.insert(
                broker.to_string(),
                format!("{}:{}", endpoint.host, link.port),
            );
        }
    } else {

View on GitHub (pinned to 6469eb736d)

Solutions

1. Format each element as an object: [{"host":"<broker hostname>", ...}] matching `PrivateLinkEndpointItem` fields — consult docs for exact required keys
2. Ensure the value is a JSON array of objects, not strings
3. Validate the JSON by deserializing it locally before submitting the DDL
4. Align with the documented example in RisingWave's private link docs

Example fix

// before
privatelink.endpoints = '["vpce-0abc123"]'
// after
privatelink.endpoints = '[{"host":"b-1.msk.us-east-1.amazonaws.com","endpoint":"vpce-0abc123"}]'

let v: Vec<serde_json::Value> = serde_json::from_str(&endpoints_json)?;
for item in &v {
    serde_json::from_value::<PrivateLinkEndpointItem>(item.clone())
        .map_err(|e| format!("bad endpoint entry {}: {}", item, e))?;
}

When it happens

Trigger: `handle_privatelink_endpoint` maps over `endpoint.as_array().unwrap()` and any element fails `serde_json::from_value` — e.g. an entry is a plain string "vpce-123" instead of an object, or has wrong field names/types.

Common situations: User writes endpoints as ["vpce-1","vpce-2"] instead of [{"host":"..."}]; wrong key name like {"endpoint":"..."}; host value missing; mixing formats between single-VPC (map) and multi-VPC (array) option shapes.

Understand the failure class

Background: "Invalid JSON response" and "Failed to parse response" errors: when an API answers 200 but the body isn't the JSON your library expected — this error's family across 28 libraries.

Related errors

• The number of broker addrs {} does not match the number of p
• PrivateLink endpoint not found
• schemas.enable is expected to be `true` or `false`, got {s}
• serde (de)serialization error for KafkaConfig: {e}
• primary key not defined for {:?} kafka sink (please define i

AI-assisted analysis of risingwavelabs/risingwave@6469eb736d (2026-09-11). Data as JSON: /api/errors/959f06a4511d3cde. Report an issue: GitHub.