You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by ma...@apache.org on 2025/11/18 13:11:35 UTC
(iggy) branch connectors-config updated: implement configs deletion
This is an automated email from the ASF dual-hosted git repository.
maciej pushed a commit to branch connectors-config
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/connectors-config by this push:
new 95906f996 implement configs deletion
95906f996 is described below
commit 95906f99656b4b13f91c913fb1897cde773b3a43
Author: Maciej Modzelewski <ma...@modzelewski.dev>
AuthorDate: Tue Nov 18 12:56:02 2025 +0100
implement configs deletion
---
core/connectors/runtime/runtime.http | 6 ++
core/connectors/runtime/src/api/sink.rs | 21 +++-
core/connectors/runtime/src/api/source.rs | 21 +++-
core/connectors/runtime/src/configs/connectors.rs | 7 ++
.../src/configs/connectors/local_provider.rs | 112 +++++++++++++++++++++
.../postgres/{postgres.toml => config.toml} | 25 +----
.../postgres/{ => connectors_config}/postgres.toml | 8 +-
core/integration/tests/connectors/postgres/mod.rs | 13 +--
8 files changed, 180 insertions(+), 33 deletions(-)
diff --git a/core/connectors/runtime/runtime.http b/core/connectors/runtime/runtime.http
index 13db48464..7c08bef4f 100644
--- a/core/connectors/runtime/runtime.http
+++ b/core/connectors/runtime/runtime.http
@@ -78,6 +78,9 @@ Content-Type: application/json
###
GET {{url}}/sinks/{{sink_key}}/configs/{{sink_version}}
+###
+DELETE {{url}}/sinks/{{sink_key}}/configs?version={{sink_version}}
+
###
GET {{url}}/sinks/{{sink_key}}/configs/active
@@ -149,6 +152,9 @@ Content-Type: application/json
###
GET {{url}}/sources/{{source_key}}/configs/{{source_version}}
+###
+DELETE {{url}}/sources/{{source_key}}/configs?version={{source_version}}
+
###
GET {{url}}/sources/{{source_key}}/configs/active
diff --git a/core/connectors/runtime/src/api/sink.rs b/core/connectors/runtime/src/api/sink.rs
index d7629fa51..0a2c8791b 100644
--- a/core/connectors/runtime/src/api/sink.rs
+++ b/core/connectors/runtime/src/api/sink.rs
@@ -42,7 +42,9 @@ pub fn router(state: Arc<RuntimeContext>) -> Router {
.route("/sinks/{key}/transforms", get(get_sink_transforms))
.route(
"/sinks/{key}/configs",
- get(get_sink_configs).post(create_sink_config),
+ get(get_sink_configs)
+ .post(create_sink_config)
+ .delete(delete_sink_config),
)
.route("/sinks/{key}/configs/{version}", get(get_sink_config))
.route("/sinks/{key}/configs/plugin", get(get_sink_plugin_config))
@@ -227,3 +229,20 @@ async fn update_sink_active_config(
.await?;
Ok(StatusCode::NO_CONTENT)
}
+
+#[derive(Debug, Deserialize)]
+struct DeleteSinkConfig {
+ version: Option<u64>,
+}
+
+async fn delete_sink_config(
+ State(context): State<Arc<RuntimeContext>>,
+ Path((key,)): Path<(String,)>,
+ Query(query): Query<DeleteSinkConfig>,
+) -> Result<StatusCode, ApiError> {
+ context
+ .config_provider
+ .delete_sink_config(&key, query.version)
+ .await?;
+ Ok(StatusCode::NO_CONTENT)
+}
diff --git a/core/connectors/runtime/src/api/source.rs b/core/connectors/runtime/src/api/source.rs
index d5f4f035f..7db1c9974 100644
--- a/core/connectors/runtime/src/api/source.rs
+++ b/core/connectors/runtime/src/api/source.rs
@@ -42,7 +42,9 @@ pub fn router(state: Arc<RuntimeContext>) -> Router {
.route("/sources/{key}/transforms", get(get_source_transforms))
.route(
"/sources/{key}/configs",
- get(get_source_configs).post(create_source_config),
+ get(get_source_configs)
+ .post(create_source_config)
+ .delete(delete_source_config),
)
.route("/sources/{key}/configs/{version}", get(get_source_config))
.route(
@@ -230,3 +232,20 @@ async fn update_source_active_config(
.await?;
Ok(StatusCode::NO_CONTENT)
}
+
+#[derive(Debug, Deserialize)]
+struct DeleteSourceConfig {
+ version: Option<u64>,
+}
+
+async fn delete_source_config(
+ State(context): State<Arc<RuntimeContext>>,
+ Path((key,)): Path<(String,)>,
+ Query(query): Query<DeleteSourceConfig>,
+) -> Result<StatusCode, ApiError> {
+ context
+ .config_provider
+ .delete_source_config(&key, query.version)
+ .await?;
+ Ok(StatusCode::NO_CONTENT)
+}
diff --git a/core/connectors/runtime/src/configs/connectors.rs b/core/connectors/runtime/src/configs/connectors.rs
index eee76d361..7d2502cd3 100644
--- a/core/connectors/runtime/src/configs/connectors.rs
+++ b/core/connectors/runtime/src/configs/connectors.rs
@@ -215,6 +215,13 @@ pub trait ConnectorsConfigProvider: Send + Sync {
key: &str,
version: Option<u64>,
) -> Result<Option<SourceConfig>, RuntimeError>;
+ async fn delete_sink_config(&self, key: &str, version: Option<u64>)
+ -> Result<(), RuntimeError>;
+ async fn delete_source_config(
+ &self,
+ key: &str,
+ version: Option<u64>,
+ ) -> Result<(), RuntimeError>;
}
pub async fn create_connectors_config_provider(
diff --git a/core/connectors/runtime/src/configs/connectors/local_provider.rs b/core/connectors/runtime/src/configs/connectors/local_provider.rs
index fcabd1159..bd9565f02 100644
--- a/core/connectors/runtime/src/configs/connectors/local_provider.rs
+++ b/core/connectors/runtime/src/configs/connectors/local_provider.rs
@@ -65,6 +65,7 @@ struct SinkConfigFile {
config: SinkConfig,
#[allow(dead_code)]
created_at: DateTime<Utc>,
+ path: String,
}
#[derive(Clone)]
@@ -72,6 +73,7 @@ struct SourceConfigFile {
config: SourceConfig,
#[allow(dead_code)]
created_at: DateTime<Utc>,
+ path: String,
}
#[derive(Default)]
@@ -196,6 +198,7 @@ impl LocalConnectorsConfigProvider<Created> {
SinkConfigFile {
config: sink_config,
created_at,
+ path,
},
);
}
@@ -205,6 +208,7 @@ impl LocalConnectorsConfigProvider<Created> {
SourceConfigFile {
config: source_config,
created_at,
+ path,
},
);
}
@@ -345,6 +349,7 @@ impl ConnectorsConfigProvider for LocalConnectorsConfigProvider<Initialized> {
SinkConfigFile {
config: config.clone(),
created_at: Utc::now(),
+ path: path.clone(),
},
);
@@ -379,6 +384,7 @@ impl ConnectorsConfigProvider for LocalConnectorsConfigProvider<Initialized> {
SourceConfigFile {
config: config.clone(),
created_at: Utc::now(),
+ path: path.clone(),
},
);
@@ -616,6 +622,112 @@ impl ConnectorsConfigProvider for LocalConnectorsConfigProvider<Initialized> {
.max_by_key(|config| config.version))
}
}
+
+ async fn delete_sink_config(
+ &self,
+ key: &str,
+ version: Option<u64>,
+ ) -> Result<(), RuntimeError> {
+ debug!("Deleting sink config: {}@{:?}", &key, &version);
+ let sinks = self.state.connectors_config.sinks();
+ let active_versions = self.load_active_versions();
+
+ let version_to_delete = version
+ .or(active_versions.sinks.get(key).copied())
+ .ok_or_else(|| RuntimeError::SinkConfigNotFound(key.to_owned(), 0))?;
+
+ let connector_id = ConnectorId {
+ key: key.to_owned(),
+ version: version_to_delete,
+ };
+
+ let config_file = {
+ sinks
+ .get(&connector_id)
+ .ok_or_else(|| RuntimeError::SinkConfigNotFound(key.to_owned(), version_to_delete))?
+ .value()
+ .clone()
+ };
+
+ std::fs::remove_file(&config_file.path)?;
+ sinks.remove(&connector_id);
+
+ let mut active_versions = self.load_active_versions();
+ let remaining_versions: Vec<u64> = sinks
+ .iter()
+ .filter(|entry| entry.key().key == key)
+ .map(|entry| entry.key().version)
+ .collect();
+
+ if remaining_versions.is_empty() {
+ active_versions.sinks.remove(key);
+ } else if Some(version_to_delete) == active_versions.sinks.get(key).copied() {
+ let latest_version = remaining_versions
+ .into_iter()
+ .max()
+ .expect("At least one version must exist");
+ active_versions.sinks.insert(key.to_owned(), latest_version);
+ }
+
+ self.save_active_versions(&active_versions)?;
+ debug!("Deleted sink configuration: {}@{:?}", &key, &version);
+ Ok(())
+ }
+
+ async fn delete_source_config(
+ &self,
+ key: &str,
+ version: Option<u64>,
+ ) -> Result<(), RuntimeError> {
+ debug!("Deleting source config: {}@{:?}", &key, &version);
+ let sources = self.state.connectors_config.sources();
+ let active_versions = self.load_active_versions();
+
+ let version_to_delete = version
+ .or(active_versions.sources.get(key).copied())
+ .ok_or_else(|| RuntimeError::SourceConfigNotFound(key.to_owned(), 0))?;
+
+ let connector_id = ConnectorId {
+ key: key.to_owned(),
+ version: version_to_delete,
+ };
+
+ let config_file = {
+ sources
+ .get(&connector_id)
+ .ok_or_else(|| {
+ RuntimeError::SourceConfigNotFound(key.to_owned(), version_to_delete)
+ })?
+ .value()
+ .clone()
+ };
+
+ std::fs::remove_file(&config_file.path)?;
+ sources.remove(&connector_id);
+
+ let mut active_versions = self.load_active_versions();
+ let remaining_versions: Vec<u64> = sources
+ .iter()
+ .filter(|entry| entry.key().key == key)
+ .map(|entry| entry.key().version)
+ .collect();
+
+ if remaining_versions.is_empty() {
+ active_versions.sources.remove(key);
+ } else if Some(version_to_delete) == active_versions.sources.get(key).copied() {
+ let latest_version = remaining_versions
+ .into_iter()
+ .max()
+ .expect("At least one version must exist");
+ active_versions
+ .sources
+ .insert(key.to_owned(), latest_version);
+ }
+
+ self.save_active_versions(&active_versions)?;
+ debug!("Deleted source configuration: {}@{:?}", &key, &version);
+ Ok(())
+ }
}
#[derive(Debug, Clone)]
diff --git a/core/integration/tests/connectors/postgres/postgres.toml b/core/integration/tests/connectors/postgres/config.toml
similarity index 62%
copy from core/integration/tests/connectors/postgres/postgres.toml
copy to core/integration/tests/connectors/postgres/config.toml
index 24e84d4df..fff72e37e 100644
--- a/core/integration/tests/connectors/postgres/postgres.toml
+++ b/core/integration/tests/connectors/postgres/config.toml
@@ -15,25 +15,6 @@
# specific language governing permissions and limitations
# under the License.
-[sinks.postgres]
-enabled = true
-name = "Postgres sink"
-path = "../../target/debug/libiggy_connector_postgres_sink"
-
-[[sinks.postgres.streams]]
-stream = ""
-topics = [""]
-schema = "json"
-batch_length = 100
-poll_interval = "5ms"
-consumer_group = "test"
-
-[sinks.postgres.config]
-connection_string = ""
-target_table = "iggy_messages"
-batch_size = 100
-max_connections = 10
-auto_create_table = true
-include_metadata = true
-include_checksum = true
-include_origin_timestamp = true
+[connectors]
+config_type = "local"
+config_dir = "postgres/connectors_config"
diff --git a/core/integration/tests/connectors/postgres/postgres.toml b/core/integration/tests/connectors/postgres/connectors_config/postgres.toml
similarity index 94%
rename from core/integration/tests/connectors/postgres/postgres.toml
rename to core/integration/tests/connectors/postgres/connectors_config/postgres.toml
index 24e84d4df..7602cd596 100644
--- a/core/integration/tests/connectors/postgres/postgres.toml
+++ b/core/integration/tests/connectors/postgres/connectors_config/postgres.toml
@@ -15,12 +15,14 @@
# specific language governing permissions and limitations
# under the License.
-[sinks.postgres]
+type = "sink"
+key = "postgres"
enabled = true
+version = 0
name = "Postgres sink"
path = "../../target/debug/libiggy_connector_postgres_sink"
-[[sinks.postgres.streams]]
+[[streams]]
stream = ""
topics = [""]
schema = "json"
@@ -28,7 +30,7 @@ batch_length = 100
poll_interval = "5ms"
consumer_group = "test"
-[sinks.postgres.config]
+[plugin_config]
connection_string = ""
target_table = "iggy_messages"
batch_size = 100
diff --git a/core/integration/tests/connectors/postgres/mod.rs b/core/integration/tests/connectors/postgres/mod.rs
index 8de202201..75acc741e 100644
--- a/core/integration/tests/connectors/postgres/mod.rs
+++ b/core/integration/tests/connectors/postgres/mod.rs
@@ -1,4 +1,5 @@
-/* Licensed to the Apache Software Foundation (ASF) under one
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
@@ -37,24 +38,24 @@ async fn setup() -> IggyClient {
let iggy_setup = IggySetup::default();
let connection_string = format!("postgres://postgres:postgres@localhost:{host_port}");
envs.insert(
- "IGGY_CONNECTORS_SINKS_POSTGRES_CONFIG_CONNECTION_STRING".to_owned(),
+ "IGGY_CONNECTORS_SINK_POSTGRES_CONFIG_CONNECTION_STRING".to_owned(),
connection_string,
);
envs.insert(
- "IGGY_CONNECTORS_SINKS_POSTGRES_STREAMS_0_STREAM".to_owned(),
+ "IGGY_CONNECTORS_SINK_POSTGRES_STREAMS_0_STREAM".to_owned(),
iggy_setup.stream.to_owned(),
);
envs.insert(
- "IGGY_CONNECTORS_SINKS_POSTGRES_STREAMS_0_TOPICS_0".to_owned(),
+ "IGGY_CONNECTORS_SINK_POSTGRES_STREAMS_0_TOPICS_0".to_owned(),
iggy_setup.topic.to_owned(),
);
envs.insert(
- "IGGY_CONNECTORS_SINKS_POSTGRES_STREAMS_0_SCHEMA".to_owned(),
+ "IGGY_CONNECTORS_SINK_POSTGRES_STREAMS_0_SCHEMA".to_owned(),
"json".to_owned(),
);
let mut runtime = setup_runtime();
runtime
- .init("postgres/postgres.toml", Some(envs), iggy_setup)
+ .init("postgres/config.toml", Some(envs), iggy_setup)
.await;
runtime.create_client().await
}