feat(connectors): fix state & memory leak, test all plugins, enrich sinks

Author: spetz

Cargo.lock

@@ -105,5 +105,8 @@ pub use utils::hash::*;
 pub use utils::personal_access_token_expiry::PersonalAccessTokenExpiry;
 pub use utils::text;
 pub use utils::timestamp::*;
+
+// Re-export chrono types for connectors and other crates that need DateTime
+pub use chrono::{DateTime, Duration as ChronoDuration, Utc};
 pub use utils::topic_size::MaxTopicSize;
 pub use utils::versioning::SemanticVersion;

core/common/src/lib.rs

@@ -64,6 +64,14 @@ impl IggyTimestamp {
         self.0.duration_since(UNIX_EPOCH).unwrap().as_micros() as u64
     }
 
+    pub fn as_millis(&self) -> u64 {
+        self.0.duration_since(UNIX_EPOCH).unwrap().as_millis() as u64
+    }
+
+    pub fn as_nanos(&self) -> u128 {
+        self.0.duration_since(UNIX_EPOCH).unwrap().as_nanos()
+    }
+
     pub fn to_rfc3339_string(&self) -> String {
         DateTime::<Utc>::from(self.0).to_rfc3339()
     }

core/common/src/utils/timestamp.rs

@@ -52,12 +52,26 @@ Please refer to the **[Sink documentation](https://github.com/apache/iggy/tree/m
 When implementing `Sink`, make sure to use the `sink_connector!` macro to expose the FFI interface and allow the connector runtime to register the sink with the runtime. The macro also exports the connector's version (from `Cargo.toml`) which is reported in the runtime's `/stats` endpoint.
 Each sink should have its own, custom configuration, which is passed along with the unique plugin ID via expected `new()` method.
 
+### Available Sinks
+
+- **Elasticsearch Sink** - sends messages to Elasticsearch indices
+- **Iceberg Sink** - writes data to Apache Iceberg tables via REST catalog
+- **PostgreSQL Sink** - stores messages in PostgreSQL database tables
+- **Quickwit Sink** - indexes messages in Quickwit search engine
+- **Stdout Sink** - prints messages to standard output (useful for debugging/development)
+
 ## Source
 
 Sources are responsible for producing the messages to the configured stream(s) and topic(s). For example, the Test source connector will generate the random messages that will be then sent to the configured stream and topic.
 
 Please refer to the **[Source documentation](https://github.com/apache/iggy/tree/master/core/connectors/sources)** for the details about the configuration and the sample implementation.
 
+### Available Sources
+
+- **Elasticsearch Source** - polls documents from Elasticsearch indices
+- **PostgreSQL Source** - reads rows from PostgreSQL tables with multiple consumption strategies (delete after read, mark as processed, timestamp tracking)
+- **Random Source** - generates random test messages (useful for testing/development)
+
 ## Building the connectors
 
 New connector can be built simply by implementing either `Sink` or `Source` trait. Please check the **[sink](https://github.com/apache/iggy/tree/master/core/connectors/sinks)** or **[source](https://github.com/apache/iggy/tree/master/core/connectors/sources)** documentation, as well as the existing examples under `/sinks` and `/sources` directories.

core/connectors/README.md

@@ -32,7 +32,6 @@ readme = "README.md"
 async-trait = { workspace = true }
 axum = { workspace = true }
 axum-server = { workspace = true }
-chrono = { workspace = true }
 clap = { workspace = true }
 configs = { workspace = true }
 configs_derive = { workspace = true }
@@ -47,7 +46,7 @@ flume = { workspace = true }
 futures = { workspace = true }
 iggy = { workspace = true }
 iggy_common = { workspace = true }
-iggy_connector_sdk = { workspace = true }
+iggy_connector_sdk = { workspace = true, features = ["api"] }
 mimalloc = { workspace = true }
 once_cell = { workspace = true }
 opentelemetry = { workspace = true }

core/connectors/runtime/Cargo.toml

@@ -23,6 +23,7 @@ use auth::resolve_api_key;
 use axum::{Json, Router, extract::State, middleware, routing::get};
 use axum_server::tls_rustls::RustlsConfig;
 use config::{HttpConfig, configure_cors};
+use iggy_connector_sdk::api::ConnectorRuntimeStats;
 use std::{net::SocketAddr, path::PathBuf, sync::Arc};
 use tokio::spawn;
 use tracing::{error, info};
@@ -126,8 +127,6 @@ async fn get_metrics(State(context): State<Arc<RuntimeContext>>) -> String {
     context.metrics.get_formatted_output()
 }
 
-async fn get_stats(
-    State(context): State<Arc<RuntimeContext>>,
-) -> Json<stats::ConnectorRuntimeStats> {
+async fn get_stats(State(context): State<Arc<RuntimeContext>>) -> Json<ConnectorRuntimeStats> {
     Json(stats::get_runtime_stats(&context).await)
 }

core/connectors/runtime/src/api/mod.rs

@@ -18,29 +18,13 @@
  */
 
 use crate::configs::connectors::{
-    ConfigFormat, SinkConfig, SourceConfig, StreamConsumerConfig, StreamProducerConfig,
-};
-use crate::manager::{
-    sink::SinkInfo,
-    source::SourceInfo,
-    status::{ConnectorError, ConnectorStatus},
+    SinkConfig, SourceConfig, StreamConsumerConfig, StreamProducerConfig,
 };
+use crate::manager::{sink::SinkInfo, source::SourceInfo};
+pub use iggy_connector_sdk::api::{SinkInfoResponse, SourceInfoResponse};
 use iggy_connector_sdk::transforms::TransformType;
 use serde::{Deserialize, Serialize};
 
-#[derive(Debug, Serialize, Deserialize)]
-pub struct SinkInfoResponse {
-    pub id: u32,
-    pub key: String,
-    pub name: String,
-    pub path: String,
-    pub enabled: bool,
-    pub status: ConnectorStatus,
-    #[serde(skip_serializing_if = "Option::is_none")]
-    pub last_error: Option<ConnectorError>,
-    pub plugin_config_format: Option<ConfigFormat>,
-}
-
 #[derive(Debug, Serialize, Deserialize)]
 pub struct SinkDetailsResponse {
     #[serde(flatten)]
@@ -55,19 +39,6 @@ pub struct SinkConfigResponse {
     pub active: bool,
 }
 
-#[derive(Debug, Serialize, Deserialize)]
-pub struct SourceInfoResponse {
-    pub id: u32,
-    pub key: String,
-    pub name: String,
-    pub path: String,
-    pub enabled: bool,
-    pub status: ConnectorStatus,
-    #[serde(skip_serializing_if = "Option::is_none")]
-    pub last_error: Option<ConnectorError>,
-    pub plugin_config_format: Option<ConfigFormat>,
-}
-
 #[derive(Debug, Serialize, Deserialize)]
 pub struct SourceDetailsResponse {
     #[serde(flatten)]
@@ -98,7 +69,7 @@ impl From<SinkInfo> for SinkInfoResponse {
             enabled: sink.enabled,
             status: sink.status,
             last_error: sink.last_error,
-            plugin_config_format: sink.plugin_config_format,
+            plugin_config_format: sink.plugin_config_format.map(|f| f.to_string()),
         }
     }
 }
@@ -113,7 +84,7 @@ impl From<SourceInfo> for SourceInfoResponse {
             enabled: source.enabled,
             status: source.status,
             last_error: source.last_error,
-            plugin_config_format: source.plugin_config_format,
+            plugin_config_format: source.plugin_config_format.map(|f| f.to_string()),
         }
     }
 }

core/connectors/runtime/src/api/models.rs

@@ -25,8 +25,8 @@ use crate::configs::connectors::local_provider::LocalConnectorsConfigProvider;
 use crate::configs::runtime::ConnectorsConfig as RuntimeConnectorsConfig;
 use crate::error::RuntimeError;
 use async_trait::async_trait;
-use chrono::{DateTime, Utc};
 use configs_derive::ConfigEnv;
+use iggy_common::{DateTime, Utc};
 use iggy_connector_sdk::Schema;
 use iggy_connector_sdk::transforms::TransformType;
 use serde::{Deserialize, Serialize};

core/connectors/runtime/src/configs/connectors.rs

@@ -24,10 +24,10 @@ use crate::configs::connectors::{
 use crate::error::RuntimeError;
 use ::configs::{ConfigProvider, FileConfigProvider, TypedEnvProvider};
 use async_trait::async_trait;
-use chrono::{DateTime, Utc};
 use dashmap::DashMap;
 use figment::value::Dict;
 use figment::{Metadata, Profile, Provider};
+use iggy_common::{DateTime, Utc};
 use serde::{Deserialize, Serialize};
 use std::collections::HashMap;
 use std::path::Path;

core/connectors/runtime/src/configs/connectors/local_provider.rs

@@ -18,17 +18,17 @@
  */
 use crate::configs::connectors::{ConnectorsConfigProvider, SinkConfig, SourceConfig};
 use crate::configs::runtime::ConnectorsRuntimeConfig;
-use crate::manager::status::ConnectorError;
 use crate::metrics::Metrics;
 use crate::{
     SinkConnectorWrapper, SourceConnectorWrapper,
     manager::{
         sink::{SinkDetails, SinkInfo, SinkManager},
         source::{SourceDetails, SourceInfo, SourceManager},
-        status::ConnectorStatus,
     },
 };
 use iggy_common::IggyTimestamp;
+use iggy_connector_sdk::api::ConnectorError;
+use iggy_connector_sdk::api::ConnectorStatus;
 use std::collections::HashMap;
 use std::sync::Arc;
 use tracing::error;