feat(config): add optional field to EngineConfig (#2727)

Allow applications embedding the dataflow engine to carry their own
engine-level configuration under `engine.custom`. The engine ignores
this field entirely; embedding binaries can read namespaced keys for
concerns like remote management, auth, or fleet coordination.

# Change Summary

Add an optional `custom: HashMap<String, serde_json::Value>` field to
`EngineConfig`. This gives embedding applications an escape hatch for
engine-level config without forking the config crate or pre-parsing
YAML. The field defaults to an empty map and is omitted from serialized
output when empty.

## What issue does this PR close?

* Closes #2561

## How are these changes tested?

Three new unit tests in `engine.rs`:
- `from_yaml_accepts_custom_config` — parses a config with multiple
namespaced custom keys and verifies values
- `custom_defaults_to_empty` — confirms the field defaults to an empty
map when omitted
- `custom_roundtrips_through_json` — serializes to JSON and deserializes
back, verifying data is preserved

## Are there any user-facing changes?

Yes. A new optional `custom` key is available under the `engine` section
of the YAML/JSON config. Example:

```yaml
engine:
  custom:
    remote_management:
      server_url: "ws://mgmt.example.com/v1"
      heartbeat_interval_secs: 10
    custom_auth:
      provider: "oidc"
      token_endpoint: "https://auth.example.com/token"
```

Existing configs are unaffected since the field defaults to empty.

---------

Co-authored-by: Lalit Kumar Bhasin <lalit_fin@yahoo.com>

Author: pritishnahar95

rust/otap-dataflow/crates/config/src/engine.rs

@@ -18,6 +18,7 @@ use crate::policy::{ChannelCapacityPolicy, Policies, TelemetryPolicy};
 use crate::topic::{TopicImplSelectionPolicy, TopicSpec};
 use schemars::JsonSchema;
 use serde::{Deserialize, Serialize};
+use serde_json::Value;
 use std::collections::HashMap;
 
 pub use self::resolve::{
@@ -77,6 +78,15 @@ pub struct EngineConfig {
     /// Engine observability declarations.
     #[serde(default)]
     pub observability: EngineObservabilityConfig,
+
+    /// Opaque metadata for applications that embed the dataflow engine.
+    ///
+    /// The engine ignores this field entirely — embedding binaries can
+    /// read namespaced keys for their own engine-level concerns
+    /// (remote management, auth, fleet coordination, etc.).
+    #[serde(default, skip_serializing_if = "HashMap::is_empty")]
+    #[schemars(extend("x-kubernetes-preserve-unknown-fields" = true))]
+    pub custom: HashMap<String, Value>,
 }
 
 /// Engine-wide topic runtime settings.
@@ -319,6 +329,92 @@ groups:
         assert!(config.engine.observability.pipeline.is_some());
     }
 
+    #[test]
+    fn from_yaml_accepts_custom_config() {
+        let yaml = r#"
+version: otel_dataflow/v1
+engine:
+  custom:
+    remote_management:
+      server_url: "ws://mgmt.example.com/v1"
+      heartbeat_interval_secs: 10
+    custom_auth:
+      provider: "oidc"
+      token_endpoint: "https://auth.example.com/token"
+groups:
+  default:
+    pipelines:
+      main:
+        nodes:
+          receiver:
+            type: "urn:test:receiver:example"
+            config: null
+          exporter:
+            type: "urn:test:exporter:example"
+            config: null
+        connections:
+          - from: receiver
+            to: exporter
+"#;
+
+        let config = OtelDataflowSpec::from_yaml(yaml).expect("should parse custom config");
+        assert_eq!(config.engine.custom.len(), 2);
+
+        let mgmt = config
+            .engine
+            .custom
+            .get("remote_management")
+            .expect("should have remote_management key");
+        assert_eq!(mgmt["server_url"], "ws://mgmt.example.com/v1");
+        assert_eq!(mgmt["heartbeat_interval_secs"], 10);
+
+        let auth = config
+            .engine
+            .custom
+            .get("custom_auth")
+            .expect("should have custom_auth key");
+        assert_eq!(auth["provider"], "oidc");
+    }
+
+    #[test]
+    fn custom_defaults_to_empty() {
+        let yaml = valid_engine_yaml(ENGINE_CONFIG_VERSION_V1);
+        let config = OtelDataflowSpec::from_yaml(&yaml).expect("should parse");
+        assert!(config.engine.custom.is_empty());
+    }
+
+    #[test]
+    fn custom_roundtrips_through_json() {
+        let yaml = r#"
+version: otel_dataflow/v1
+engine:
+  custom:
+    my_app:
+      key: "value"
+groups:
+  default:
+    pipelines:
+      main:
+        nodes:
+          receiver:
+            type: "urn:test:receiver:example"
+            config: null
+          exporter:
+            type: "urn:test:exporter:example"
+            config: null
+        connections:
+          - from: receiver
+            to: exporter
+"#;
+
+        let config = OtelDataflowSpec::from_yaml(yaml).expect("should parse");
+        let json = serde_json::to_string(&config).expect("should serialize to json");
+        let roundtripped: OtelDataflowSpec =
+            serde_json::from_str(&json).expect("should deserialize from json");
+        assert_eq!(roundtripped.engine.custom.len(), 1);
+        assert_eq!(roundtripped.engine.custom["my_app"]["key"], "value");
+    }
+
     #[test]
     fn from_yaml_requires_explicit_memory_limiter_mode() {
         let yaml = r#"

rust/otap-dataflow/docs/configuration-model.md

@@ -152,9 +152,9 @@ Important behavior:
 - Graph topology is explicit via `connections`.
 - `kind` is inferred from node `type`.
 - For single-output nodes, you do not need `outputs` or `default_output`.
-- Parsing is strict: unknown fields are rejected (might be relaxed in the future
-  if we find good use cases for extensibility, but for now this is intentional
-  to catch typos and mistakes early).
+- Parsing is strict: unknown fields are rejected to catch typos and mistakes
+  early. Applications that embed the engine can place their own engine-level
+  configuration under `engine.custom` (see below).
 
 ## Engine Section
 
@@ -165,6 +165,7 @@ Important behavior:
 - `telemetry`
 - `observed_state`
 - `observability`
+- `custom`
 
 ### Engine Topic Settings
 
@@ -207,6 +208,23 @@ Optional observability policies are supported at:
 
 `resources` is intentionally not supported for observability and is rejected.
 
+### Custom Embedder Configuration
+
+`engine.custom` is an optional map for applications that embed the dataflow
+engine as a library. The engine ignores this field entirely -- embedding
+binaries can read namespaced keys for their own engine-level concerns.
+
+```yaml
+engine:
+  custom:
+    remote_management:
+      server_url: "ws://mgmt.example.com/v1"
+      heartbeat_interval_secs: 10
+    custom_auth:
+      provider: "oidc"
+      token_endpoint: "https://auth.example.com/token"
+```
+
 ## Policy Hierarchy
 
 Policies include channel capacity, health, runtime telemetry, resources