{"record":{"id":"fb07ebd0b857045e","repo":"quickwit-oss/quickwit","slug":"kafka-topic-cannot-be-updated","errorCode":null,"errorMessage":"Kafka topic cannot be updated","messagePattern":"Kafka topic cannot be updated","errorType":"validation","errorClass":null,"httpStatus":null,"severity":"error","filePath":"quickwit/quickwit-config/src/source_config/mod.rs","lineNumber":445,"sourceCode":"    #[serde(skip_serializing_if = \"Option::is_none\")]\n    pub client_log_level: Option<String>,\n    /// Kafka client configuration parameters.\n    #[schema(value_type = Object)]\n    #[serde(default = \"serde_json::Value::default\")]\n    #[serde(skip_serializing_if = \"serde_json::Value::is_null\")]\n    pub client_params: JsonValue,\n    /// When backfill mode is enabled, the source exits after reaching the end of the topic.\n    #[serde(default)]\n    #[serde(skip_serializing_if = \"is_false\")]\n    pub enable_backfill_mode: bool,\n}\n\nimpl KafkaSourceParams {\n    fn validate_update(&self, other: &Self) -> anyhow::Result<()> {\n        // Updating the topic would likely mess up the checkpoints because the\n        // Kafka partition IDs are used as metastore checkpoint PartitionId\n        // and there uniqueness is not guaranteed across topics.\n        ensure!(self.topic == other.topic, \"Kafka topic cannot be updated\");\n        Ok(())\n    }\n}\n\n#[derive(Clone, Debug, Eq, PartialEq, Hash, Serialize, Deserialize, utoipa::ToSchema)]\n#[serde(deny_unknown_fields)]\npub struct PubSubSourceParams {\n    /// Name of the subscription that the source consumes.\n    pub subscription: String,\n    /// When backfill mode is enabled, the source exits after reaching the end of the topic.\n    #[serde(default)]\n    #[serde(skip_serializing_if = \"is_false\")]\n    pub enable_backfill_mode: bool,\n    /// GCP service account credentials (`None` will use default via\n    /// GOOGLE_APPLICATION_CREDENTIALS)\n    /// Path to a google_cloud_auth::credentials::CredentialsFile serialized in JSON. See also\n    /// `<https://cloud.google.com/docs/authentication/application-default-credentials>` and\n    /// `<https://github.com/yoshidan/google-cloud-rust/tree/main/pubsub#automatically>` and","sourceCodeStart":427,"sourceCodeEnd":463,"githubUrl":"https://github.com/quickwit-oss/quickwit/blob/a39730c5cdcd1a4fe798403737ae293999ea21f8/quickwit/quickwit-config/src/source_config/mod.rs#L427-L463","documentation":"Kafka source params cannot have their `topic` changed via an update. Kafka partition IDs are stored as metastore checkpoint PartitionIds, and their uniqueness is not guaranteed across topics, so changing the topic would corrupt checkpoint semantics. `KafkaSourceParams::validate_update` rejects any update where the topic differs.","triggerScenarios":"Calling the source update API/CLI (e.g. `quickwit source update`) with a Kafka source config whose `topic` field differs from the currently stored source config, triggering validate_update.","commonSituations":"Renaming or migrating a Kafka topic and trying to point the existing source at the new topic in place; copy-pasting a source config from another index; environment-specific configs (staging topic) applied to a production source.","solutions":["Create a new source (e.g. `quickwit source create`) pointing at the new topic instead of updating the existing one.","If a topic rename is required: delete the old source and create a new one, accepting that ingestion restarts from the configured `default_start_timestamp`/beginning offsets.","If you only meant to change other params (bootstrap servers, consumer options), re-submit the update keeping `topic` identical."],"exampleFix":"// before: PATCH existing source with changed topic\n{\"source_type\":\"kafka\",\"params\":{\"topic\":\"logs-v2\",...}}\n\n// after: create a new source for the new topic\nquickwit source create --index my-index --source-config kafka-logs-v2.json\n// (kafka-logs-v2.json contains topic: logs-v2)","handlingStrategy":"validation","validationCode":"let current = client.get_source(index_id, source_id).await?;\nif current.params.topic != new_params.topic {\n    return Err(\"Kafka topic is immutable; create a new source instead\".into());\n}\nclient.update_source(index_id, source_id, new_params).await?;","typeGuard":null,"tryCatchPattern":"match client.update_source(index_id, source_id, params).await {\n    Err(e) if e.to_string().contains(\"Kafka topic cannot be updated\") => {\n        // fall back to delete + create with the new topic\n        client.delete_source(index_id, source_id).await?;\n        client.create_source(index_id, source_config_with_new_topic).await?;\n    }\n    other => other?,\n}","preventionTips":["Treat source_id + topic as one immutable identity; create a new source per topic.","Diff the submitted source config against the stored one before calling update.","Script topic migrations as delete+create, never update."],"tags":["kafka","source","validation","checkpoint"],"backgroundTag":"unsupported-operation","analyzedSha":"a39730c5cdcd1a4fe798403737ae293999ea21f8","analyzedAt":"2026-09-08T13:19:37.784Z","contentChangedAt":"2026-09-08T13:19:37.784Z","schemaVersion":2},"datasetVersion":"2026-09-14T16:17:12.679Z"}