diff --git a/Cargo.lock b/Cargo.lock
index 4ae43c3fc7..685735814e 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -12979,6 +12979,8 @@ dependencies = [
  "async-stream",
  "futures 0.3.31",
  "futures-util",
+ "metrics",
+ "metrics-util 0.18.0",
  "pin-project",
  "proptest",
  "rand 0.8.5",
diff --git a/lib/vector-stream/Cargo.toml b/lib/vector-stream/Cargo.toml
index 45b1e77aaa..682cfe75bc 100644
--- a/lib/vector-stream/Cargo.toml
+++ b/lib/vector-stream/Cargo.toml
@@ -14,11 +14,14 @@ tokio = { version = "1.43.0", default-features = false, features = ["net"] }
 tokio-util = { version = "0.7.0", default-features = false, features = ["time"] }
 tower = { version = "0.4", default-features = false, features = ["util"] }
 tracing = { version = "0.1.34", default-features = false }
+metrics.workspace = true
 twox-hash = { version = "2.1.0", default-features = false, features = ["xxhash64"] }
 vector-common = { path = "../vector-common" }
 vector-core = { path = "../vector-core" }
 
 [dev-dependencies]
+metrics-util = "0.18"
 proptest = "1.5"
 rand = "0.8.5"
 rand_distr = "0.4.3"
+tokio = { version = "1.43.0", features = ["macros", "rt", "time", "test-util"] }
diff --git a/lib/vector-stream/src/partitioned_batcher.rs b/lib/vector-stream/src/partitioned_batcher.rs
index b0ba86b85b..b494894e85 100644
--- a/lib/vector-stream/src/partitioned_batcher.rs
+++ b/lib/vector-stream/src/partitioned_batcher.rs
@@ -8,6 +8,7 @@ use std::{
 };
 
 use futures::stream::{Fuse, Stream, StreamExt};
+use metrics::{gauge, Gauge};
 use pin_project::pin_project;
 use tokio_util::time::{delay_queue::Key, DelayQueue};
 use twox_hash::XxHash64;
@@ -201,6 +202,10 @@ where
     #[pin]
     /// The stream this `Batcher` wraps
     stream: Fuse<St>,
+    /// Gauge tracking the number of open batches
+    m_open: Gauge,
+    /// Gauge tracking the number of closed batches waiting to be emitted
+    m_closed: Gauge,
 }
 
 impl<St, Prt, C, F, B> PartitionedBatcher<St, Prt, ExpirationQueue<Prt::Key>, C, F, B>
@@ -221,6 +226,8 @@ where
             timer: ExpirationQueue::new(timeout),
             partitioner,
             stream: stream.fuse(),
+            m_open: gauge!("open_batches"),
+            m_closed: gauge!("closed_batches"),
         }
     }
 }
@@ -243,6 +250,8 @@ where
             timer,
             partitioner,
             stream: stream.fuse(),
+            m_open: gauge!("open_batches"),
+            m_closed: gauge!("closed_batches"),
         }
     }
 }
@@ -267,6 +276,7 @@ where
         let mut this = self.project();
         loop {
             if !this.closed_batches.is_empty() {
+                this.m_closed.decrement(1.0);
                 return Poll::Ready(this.closed_batches.pop());
             }
             match this.stream.as_mut().poll_next(cx) {
@@ -280,6 +290,8 @@ where
                             .remove(&item_key)
                             .expect("batch should exist if it is set to expire");
                         this.closed_batches.push((item_key, batch.take_batch()));
+                        this.m_open.decrement(1.0);
+                        this.m_closed.increment(1.0);
                     }
                 },
                 Poll::Ready(None) => {
@@ -289,12 +301,15 @@ where
                     // continue looping so the caller can drain them all before
                     // we finish.
                     if !this.batches.is_empty() {
+                        let batch_count = this.batches.len() as f64;
                         this.timer.clear();
                         this.closed_batches.extend(
                             this.batches
                                 .drain()
                                 .map(|(key, mut batch)| (key, batch.take_batch())),
                         );
+                        this.m_open.decrement(batch_count);
+                        this.m_closed.increment(batch_count);
                         continue;
                     }
                     return Poll::Ready(None);
@@ -309,6 +324,7 @@ where
                         let batch = (this.state)();
                         this.batches.insert(item_key.clone(), batch);
                         this.timer.insert(item_key.clone());
+                        this.m_open.increment(1.0);
                         this.batches
                             .get_mut(&item_key)
                             .expect("batch has just been inserted so should exist")
@@ -321,6 +337,7 @@ where
                         // next iteration.
                         this.closed_batches
                             .push((item_key.clone(), batch.take_batch()));
+                        this.m_closed.increment(1.0);
 
                         // The batch for this partition key was set to
                         // expire, but now it's overflowed and must be
@@ -337,6 +354,8 @@ where
                             .push((item_key.clone(), batch.take_batch()));
                         this.batches.remove(&item_key);
                         this.timer.remove(&item_key);
+                        this.m_open.decrement(1.0);
+                        this.m_closed.increment(1.0);
                     }
                 }
             }
@@ -696,4 +715,81 @@ mod test {
 
         f(&mut cx)
     }
+
+    #[tokio::test]
+    async fn gauge_values_track_batch_counts() {
+        use metrics_util::debugging::DebuggingRecorder;
+
+        // Install debugging recorder to capture metrics
+        let recorder = DebuggingRecorder::new();
+        let snapshotter = recorder.snapshotter();
+
+        // Use with_local_recorder to avoid global state issues in tests
+        metrics::with_local_recorder(&recorder, || {
+            let noop_waker = futures::task::noop_waker();
+            let mut cx = Context::from_waker(&noop_waker);
+
+            // Create items that will form 3 partitions: keys 1, 2, 3
+            // Each partition will get 2 items, triggering batch full at item_limit=2
+            let items = vec![1u64, 2, 3, 1, 2, 3];
+            let stream = stream::iter(items);
+
+            let partitioner = TestPartitioner {
+                key_space: NonZeroU8::new(10).unwrap(),
+            };
+            let settings = BatcherSettings::new(
+                Duration::from_secs(1),
+                NonZeroUsize::new(100).unwrap(),
+                NonZeroUsize::new(2).unwrap(), // 2 items per batch triggers full
+            );
+
+            let mut batcher = PartitionedBatcher::new(stream, partitioner, move || {
+                settings.as_byte_size_config()
+            });
+            let mut batcher = Pin::new(&mut batcher);
+
+            // Process items and verify gauge interactions
+            // Items 1, 2, 3 each create a new open batch (3 open, 0 closed)
+            // Items 1, 2, 3 (second occurrence) fill batches, moving them to closed
+            // Then polling returns each closed batch
+
+            let mut batches_received = 0;
+            loop {
+                match batcher.as_mut().poll_next(&mut cx) {
+                    Poll::Pending => break,
+                    Poll::Ready(None) => break,
+                    Poll::Ready(Some(_)) => {
+                        batches_received += 1;
+                    }
+                }
+            }
+
+            // We should have received 3 batches
+            assert_eq!(batches_received, 3);
+
+            // Verify gauges were recorded by checking snapshot
+            let snapshot = snapshotter.snapshot();
+            let gauges = snapshot.into_hashmap();
+
+            // Verify gauge metrics were recorded
+            let mut found_open = false;
+            let mut found_closed = false;
+
+            for (composite_key, _) in gauges.iter() {
+                let name = composite_key.key().name();
+                if name == "open_batches" {
+                    found_open = true;
+                }
+                if name == "closed_batches" {
+                    found_closed = true;
+                }
+            }
+
+            assert!(found_open, "open_batches gauge should have been recorded");
+            assert!(
+                found_closed,
+                "closed_batches gauge should have been recorded"
+            );
+        });
+    }
 }
diff --git a/src/internal_events/splunk_hec.rs b/src/internal_events/splunk_hec.rs
index 4ff5838cdc..f0cabd12bb 100644
--- a/src/internal_events/splunk_hec.rs
+++ b/src/internal_events/splunk_hec.rs
@@ -198,6 +198,27 @@ mod sink {
             );
         }
     }
+
+    pub struct SplunkBatchHeaderValueInvalid<'a> {
+        pub header_name: &'a str,
+    }
+
+    impl InternalEvent for SplunkBatchHeaderValueInvalid<'_> {
+        fn emit(self) {
+            warn!(
+                message = "Batch header value contains invalid characters. Skipping header.",
+                header_name = %self.header_name,
+                internal_log_rate_limit = true,
+            );
+            counter!(
+                "component_errors_total",
+                "error_code" => "invalid_batch_header_value",
+                "error_type" => error_type::PARSER_FAILED,
+                "stage" => error_stage::PROCESSING,
+            )
+            .increment(1);
+        }
+    }
 }
 
 #[cfg(feature = "sources-splunk_hec")]
diff --git a/src/sinks/humio/logs.rs b/src/sinks/humio/logs.rs
index fbde5622ab..065f6e0458 100644
--- a/src/sinks/humio/logs.rs
+++ b/src/sinks/humio/logs.rs
@@ -26,7 +26,7 @@ use crate::{
     template::Template,
     tls::TlsConfig,
 };
-use crate::sinks::splunk_hec::logs::config::TimestampConfiguration;
+use crate::sinks::splunk_hec::logs::config::{BatchHeaders, TimestampConfiguration};
 
 pub(super) const HOST: &str = "https://cloud.humio.com";
 
@@ -218,7 +218,8 @@ impl HumioLogsConfig {
                 format: TimestampFormat::Native,
                 timestamp_nanos_key: self.timestamp_nanos_key.clone(),
                 preserve_timestamp_key: false,
-            })
+            }),
+            batch_headers: BatchHeaders::default(),
         }
     }
 }
diff --git a/src/sinks/splunk_hec/common/acknowledgements.rs b/src/sinks/splunk_hec/common/acknowledgements.rs
index 20dedfc0c2..26fc036554 100644
--- a/src/sinks/splunk_hec/common/acknowledgements.rs
+++ b/src/sinks/splunk_hec/common/acknowledgements.rs
@@ -225,6 +225,7 @@ impl HecAckClient {
                 None,
                 MetadataFields::default(),
                 false,
+                vec![],
             )
             .map_err(|_| HecAckApiError::ClientBuildRequest)?;
 
diff --git a/src/sinks/splunk_hec/common/request.rs b/src/sinks/splunk_hec/common/request.rs
index f644107e46..2abf64d2a6 100644
--- a/src/sinks/splunk_hec/common/request.rs
+++ b/src/sinks/splunk_hec/common/request.rs
@@ -1,6 +1,7 @@
 use std::sync::Arc;
 
 use bytes::Bytes;
+use http::{HeaderName, HeaderValue};
 use vector_lib::request_metadata::{MetaDescriptive, RequestMetadata};
 use vector_lib::{
     event::{EventFinalizers, Finalizable},
@@ -19,6 +20,7 @@ pub struct HecRequest {
     pub source: Option<String>,
     pub sourcetype: Option<String>,
     pub host: Option<String>,
+    pub headers: Vec<(HeaderName, HeaderValue)>,
 }
 
 impl ByteSizeOf for HecRequest {
diff --git a/src/sinks/splunk_hec/common/service.rs b/src/sinks/splunk_hec/common/service.rs
index b44bd1f56e..dc22cb8848 100644
--- a/src/sinks/splunk_hec/common/service.rs
+++ b/src/sinks/splunk_hec/common/service.rs
@@ -223,6 +223,7 @@ impl HttpRequestBuilder {
         passthrough_token: Option<Arc<str>>,
         metadata_fields: MetadataFields,
         auto_extract_timestamp: bool,
+        dyn_headers: Vec<(HeaderName, HeaderValue)>,
     ) -> Result<Request<Bytes>, crate::Error> {
         let uri = match self.endpoint_target {
             EndpointTarget::Raw => {
@@ -267,6 +268,11 @@ impl HttpRequestBuilder {
                 crate::Error::from("Failed to access headers in http::Request builder")
             })?;
 
+        for (header, value) in dyn_headers {
+            headers.insert(header, value);
+        }
+
+        // Static headers from request config take precedence over dynamic headers
         for (header, value) in self.headers.iter() {
             headers.insert(header, value.clone());
         }
@@ -376,6 +382,7 @@ mod tests {
             source: None,
             sourcetype: None,
             host: None,
+            headers: vec![],
         }
     }
 
diff --git a/src/sinks/splunk_hec/common/util.rs b/src/sinks/splunk_hec/common/util.rs
index ba729b859a..99cbcd75cd 100644
--- a/src/sinks/splunk_hec/common/util.rs
+++ b/src/sinks/splunk_hec/common/util.rs
@@ -77,6 +77,7 @@ pub fn build_http_batch_service(
                         host: req.host,
                     },
                     auto_extract_timestamp,
+                    req.headers,
                 )
             });
         future
@@ -296,6 +297,7 @@ mod tests {
                 None,
                 MetadataFields::default(),
                 false,
+                vec![],
             )
             .unwrap();
 
@@ -340,6 +342,7 @@ mod tests {
                 None,
                 MetadataFields::default(),
                 false,
+                vec![],
             )
             .unwrap();
 
@@ -387,6 +390,7 @@ mod tests {
                 None,
                 MetadataFields::default(),
                 false,
+                vec![],
             )
             .unwrap_err();
         assert_eq!(err.to_string(), "URI parse error: invalid format")
@@ -442,6 +446,107 @@ mod tests {
             );
         }
     }
+
+    #[tokio::test]
+    async fn test_build_request_with_batch_headers() {
+        use http::HeaderName;
+
+        let endpoint = "http://localhost:8888";
+        let token = "token";
+        let compression = Compression::None;
+        let events = Bytes::from("events");
+        let http_request_builder = HttpRequestBuilder::new(
+            String::from(endpoint),
+            EndpointTarget::default(),
+            String::from(token),
+            compression,
+            IndexMap::default()
+        );
+
+        // Create batch headers
+        let batch_headers = vec![
+            (HeaderName::from_static("x-tenant"), HeaderValue::from_static("acme")),
+            (HeaderName::from_static("x-region"), HeaderValue::from_static("us-east")),
+        ];
+
+        let request = http_request_builder
+            .build_request(
+                events.clone(),
+                "/services/collector/event",
+                None,
+                MetadataFields::default(),
+                false,
+                batch_headers,
+            )
+            .unwrap();
+
+        // Verify batch headers are present
+        assert_eq!(
+            request.headers().get("X-Tenant"),
+            Some(&HeaderValue::from_static("acme"))
+        );
+        assert_eq!(
+            request.headers().get("X-Region"),
+            Some(&HeaderValue::from_static("us-east"))
+        );
+
+        // Standard headers should still be present
+        assert_eq!(
+            request.headers().get("Content-Type"),
+            Some(&HeaderValue::from_static("application/json"))
+        );
+        assert_eq!(
+            request.headers().get("Authorization"),
+            Some(&HeaderValue::from_static("Splunk token"))
+        );
+    }
+
+    #[tokio::test]
+    async fn test_build_request_static_headers_override_batch_headers() {
+        use http::HeaderName;
+
+        let endpoint = "http://localhost:8888";
+        let token = "token";
+        let compression = Compression::None;
+        let events = Bytes::from("events");
+
+        // Create static headers that will override batch headers
+        let mut static_headers = IndexMap::new();
+        static_headers.insert(
+            HeaderName::from_static("x-override"),
+            HeaderValue::from_static("static-value"),
+        );
+
+        let http_request_builder = HttpRequestBuilder::new(
+            String::from(endpoint),
+            EndpointTarget::default(),
+            String::from(token),
+            compression,
+            static_headers,
+        );
+
+        // Batch header with same name should be overridden
+        let batch_headers = vec![
+            (HeaderName::from_static("x-override"), HeaderValue::from_static("dynamic-value")),
+        ];
+
+        let request = http_request_builder
+            .build_request(
+                events.clone(),
+                "/services/collector/event",
+                None,
+                MetadataFields::default(),
+                false,
+                batch_headers,
+            )
+            .unwrap();
+
+        // Static header should override the batch header
+        assert_eq!(
+            request.headers().get("X-Override"),
+            Some(&HeaderValue::from_static("static-value"))
+        );
+    }
 }
 
 #[cfg(all(test, feature = "splunk-integration-tests"))]
diff --git a/src/sinks/splunk_hec/logs/config.rs b/src/sinks/splunk_hec/logs/config.rs
index 57ad1d6494..e5d96d3807 100644
--- a/src/sinks/splunk_hec/logs/config.rs
+++ b/src/sinks/splunk_hec/logs/config.rs
@@ -1,9 +1,10 @@
 use std::sync::Arc;
 
+use http::HeaderName;
 use vector_lib::{
     codecs::TextSerializerConfig, config::TimestampFormat, lookup::lookup_v2::{ConfigValuePath, OptionalTargetPath}, sensitive_string::SensitiveString
 };
-
+use serde_with::serde_as;
 use crate::{
     http::HttpClient,
     sinks::{
@@ -20,6 +21,49 @@ use crate::{
 use crate::sinks::util::http::{validate_headers, RequestConfig};
 use super::{encoder::HecLogsEncoder, request_builder::HecLogsRequestBuilder, sink::HecLogsSink};
 
+/// A batch http-header to be sent to Splunk HEC.
+#[serde_as]
+#[configurable_component]
+#[derive(Clone, Debug)]
+#[serde(deny_unknown_fields)]
+pub struct BatchHeader {
+    /// The name of the header.
+    #[configurable(metadata(docs::examples = "X-Priority"))]
+    pub name: String,
+
+    /// The value of the header.
+    #[configurable(metadata(docs::examples = "pri"))]
+    pub value: ConfigValuePath,
+}
+
+/// A batch header to be sent to Splunk HEC, represented as a key-value pair.
+#[serde_as]
+#[configurable_component]
+#[derive(Clone, Debug, Default)]
+#[serde(deny_unknown_fields)]
+pub struct BatchHeaders(Vec<BatchHeader>);
+
+impl From<Vec<BatchHeader>> for BatchHeaders {
+    fn from(headers: Vec<BatchHeader>) -> Self {
+        BatchHeaders(headers)
+    }
+}
+
+impl BatchHeaders {
+    /// Validates header names and converts to a vector of (HeaderName, ConfigValuePath) pairs.
+    /// Returns an error if any header name is invalid.
+    pub fn into_validated(self) -> Result<Vec<(HeaderName, ConfigValuePath)>, crate::Error> {
+        self.0
+            .into_iter()
+            .map(|header| {
+                HeaderName::try_from(header.name.as_str())
+                    .map(|name| (name, header.value))
+                    .map_err(|e| format!("Invalid batch header name '{}': {}", header.name, e).into())
+            })
+            .collect()
+    }
+}
+
 /// Configuration for the `splunk_hec_logs` sink.
 #[configurable_component(sink(
     "splunk_hec_logs",
@@ -105,6 +149,12 @@ pub struct HecLogsSinkConfig {
     #[serde(default)]
     pub batch: BatchConfig<SplunkHecDefaultBatchSettings>,
 
+    /// Headers to be included in each batch request to Splunk HEC.
+    /// Headers in request-config take precedence over these batch headers if there are any conflicts.
+    #[configurable(derived)]
+    #[serde(default)]
+    pub batch_headers: BatchHeaders,
+
     #[configurable(derived)]
     #[serde(default)]
     pub request: RequestConfig,
@@ -202,6 +252,7 @@ impl GenerateConfig for HecLogsSinkConfig {
             encoding: TextSerializerConfig::default().into(),
             compression: Compression::default(),
             batch: BatchConfig::default(),
+            batch_headers: BatchHeaders::default(),
             request: RequestConfig::default(),
             tls: None,
             acknowledgements: Default::default(),
@@ -266,6 +317,7 @@ impl HecLogsSinkConfig {
 
         let request_settings = self.request.tower.into_settings();
         let headers = validate_headers(&self.request.headers)?;
+        let batch_headers = self.batch_headers.clone().into_validated()?;
 
         let http_request_builder = Arc::new(HttpRequestBuilder::new(
             self.endpoint.clone(),
@@ -309,6 +361,7 @@ impl HecLogsSinkConfig {
             endpoint_target: self.endpoint_target,
             auto_extract_timestamp: self.auto_extract_timestamp.unwrap_or_default(),
             timestamp_configuration: self.timestamp_configuration.clone(),
+            batch_headers,
             shutdown: cx.shutdown.clone(),
         };
 
@@ -331,6 +384,82 @@ mod tests {
         crate::test_util::test_generate_config::<HecLogsSinkConfig>();
     }
 
+    #[test]
+    fn test_config_serde() {
+        let config_toml = r#"
+            default_token = "my-token"
+            endpoint = "https://hec.example.com:8088"
+            host_key = "hostname"
+            indexed_fields = ["field1", "field2"]
+            index = "{{ index_field }}"
+            sourcetype = "{{ sourcetype_field }}"
+            source = "/var/log/app.log"
+            compression = "gzip"
+            endpoint_target = "raw"
+            auto_extract_timestamp = true
+            path = "/custom/path"
+
+            [encoding]
+            codec = "json"
+            except_fields = ["secret_field"]
+
+            [batch]
+            max_bytes = 1048576
+            max_events = 100
+            timeout_secs = 5
+
+            [[batch_headers]]
+            name = "X-Tenant"
+            value = "tenant"
+
+            [[batch_headers]]
+            name = "X-Priority"
+            value = "priority"
+
+            [request]
+            timeout_secs = 30
+            retry_attempts = 3
+
+            [request.headers]
+            X-Custom = "custom-value"
+
+            [acknowledgements]
+            indexer_acknowledgements_enabled = true
+            max_pending_acks = 1000
+
+            [timestamp_configuration]
+            timestamp_key = "ts"
+            format = "native"
+            preserve_timestamp_key = true
+        "#;
+
+        let config: HecLogsSinkConfig = toml::from_str(config_toml)
+            .expect("Failed to parse config");
+
+        assert_eq!(config.default_token.inner(), "my-token");
+        assert_eq!(config.endpoint, "https://hec.example.com:8088");
+        assert_eq!(config.endpoint_target, EndpointTarget::Raw);
+        assert_eq!(config.auto_extract_timestamp, Some(true));
+        assert_eq!(config.path, Some("/custom/path".to_string()));
+        assert_eq!(config.indexed_fields.len(), 2);
+        assert!(config.index.is_some());
+        assert!(config.sourcetype.is_some());
+        assert_eq!(config.source, Some(crate::template::Template::try_from("/var/log/app.log").unwrap()));
+        assert!(config.host_key.is_some());
+
+        let batch_headers = config.batch_headers.into_validated().expect("batch_headers should be valid");
+        assert_eq!(batch_headers.len(), 2);
+        assert_eq!(batch_headers[0].0.as_str(), "x-tenant");
+        assert_eq!(batch_headers[1].0.as_str(), "x-priority");
+
+        assert!(config.acknowledgements.indexer_acknowledgements_enabled);
+        assert_eq!(config.acknowledgements.max_pending_acks.get(), 1000);
+
+        let ts_config = config.timestamp_configuration.expect("timestamp_configuration should be set");
+        assert!(ts_config.timestamp_key.is_some());
+        assert!(ts_config.preserve_timestamp_key);
+    }
+
     impl ValidatableComponent for HecLogsSinkConfig {
         fn validation_configuration() -> ValidationConfiguration {
             let endpoint = "http://127.0.0.1:9001".to_string();
@@ -372,6 +501,7 @@ mod tests {
                     indexer_acknowledgements_enabled: false,
                     ..Default::default()
                 },
+                batch_headers: BatchHeaders::default(),
                 path: None,
                 auto_extract_timestamp: None,
                 endpoint_target: EndpointTarget::Raw,
diff --git a/src/sinks/splunk_hec/logs/request_builder.rs b/src/sinks/splunk_hec/logs/request_builder.rs
index 91d37d7b5e..67f594fad6 100644
--- a/src/sinks/splunk_hec/logs/request_builder.rs
+++ b/src/sinks/splunk_hec/logs/request_builder.rs
@@ -1,6 +1,7 @@
 use std::sync::Arc;
 
 use bytes::Bytes;
+use http::{HeaderName, HeaderValue};
 use vector_lib::event::{EventFinalizers, Finalizable};
 use vector_lib::request_metadata::RequestMetadata;
 
@@ -29,6 +30,7 @@ pub struct HecRequestMetadata {
     sourcetype: Option<String>,
     index: Option<String>,
     host: Option<String>,
+    headers: Vec<(HeaderName, Option<HeaderValue>)>,
 }
 
 impl RequestBuilder<(Option<Partitioned>, Vec<HecProcessedEvent>)> for HecLogsRequestBuilder {
@@ -65,6 +67,10 @@ impl RequestBuilder<(Option<Partitioned>, Vec<HecProcessedEvent>)> for HecLogsRe
                 sourcetype: partition.as_mut().and_then(|p| p.sourcetype.take()),
                 index: partition.as_mut().and_then(|p| p.index.take()),
                 host: partition.as_mut().and_then(|p| p.host.take()),
+                headers: partition
+                    .as_mut()
+                    .map(|p| std::mem::take(&mut p.headers))
+                    .unwrap_or_default(),
             },
             builder,
             events,
@@ -77,6 +83,12 @@ impl RequestBuilder<(Option<Partitioned>, Vec<HecProcessedEvent>)> for HecLogsRe
         metadata: RequestMetadata,
         payload: EncodeResult<Self::Payload>,
     ) -> Self::Request {
+        let headers = hec_metadata
+            .headers
+            .into_iter()
+            .filter_map(|(k, v)| v.map(|v| (k, v)))
+            .collect();
+
         HecRequest {
             body: payload.into_payload(),
             finalizers: hec_metadata.finalizers,
@@ -85,6 +97,7 @@ impl RequestBuilder<(Option<Partitioned>, Vec<HecProcessedEvent>)> for HecLogsRe
             sourcetype: hec_metadata.sourcetype,
             index: hec_metadata.index,
             host: hec_metadata.host,
+            headers,
             metadata,
         }
     }
diff --git a/src/sinks/splunk_hec/logs/sink.rs b/src/sinks/splunk_hec/logs/sink.rs
index e470d64f8b..39d176a011 100644
--- a/src/sinks/splunk_hec/logs/sink.rs
+++ b/src/sinks/splunk_hec/logs/sink.rs
@@ -1,5 +1,7 @@
 use std::{fmt::{self, Display}, sync::Arc};
 
+use http::{HeaderName, HeaderValue};
+
 use super::request_builder::HecLogsRequestBuilder;
 use crate::{
     internal_events::SplunkEventTimestampInvalidType,
@@ -17,7 +19,7 @@ use futures::future::Either;
 use stream_cancel::Tripwire;
 use vector_lib::{
     config::{log_schema, LogNamespace, TimestampFormat, TimestampResolutionError},
-    lookup::{event_path, lookup_v2::OptionalTargetPath, OwnedValuePath, PathPrefix},
+    lookup::{event_path, lookup_v2::{ConfigValuePath, OptionalTargetPath}, OwnedValuePath, PathPrefix},
     schema::meaning,
 };
 use vrl::path::OwnedTargetPath;
@@ -38,6 +40,7 @@ pub struct HecLogsSink<S> {
     pub endpoint_target: EndpointTarget,
     pub auto_extract_timestamp: bool,
     pub timestamp_configuration: Option<TimestampConfiguration>,
+    pub batch_headers: Vec<(HeaderName, ConfigValuePath)>,
     pub shutdown: Tripwire,
 }
 
@@ -72,6 +75,7 @@ where
         };
         let batch_settings = self.batch_settings;
 
+        let batch_headers = self.batch_headers.clone();
         let run = input
             .map(move |event| process_log(event, &data))
             .batched_partitioned(
@@ -83,9 +87,10 @@ where
                         self.source.clone(),
                         self.index.clone(),
                         self.host_key.clone(),
+                        batch_headers,
                     )
                 } else {
-                    EventPartitioner::new(None, None, None, None)
+                    EventPartitioner::new(None, None, None, None, batch_headers)
                 },
                 move || batch_settings.as_byte_size_config(),
             )
@@ -139,6 +144,7 @@ pub(super) struct Partitioned {
     pub(super) sourcetype: Option<String>,
     pub(super) index: Option<String>,
     pub(super) host: Option<String>,
+    pub(super) headers: Vec<(HeaderName, Option<HeaderValue>)>,
 }
 
 #[derive(Default)]
@@ -147,20 +153,23 @@ struct EventPartitioner {
     pub source: Option<Template>,
     pub index: Option<Template>,
     pub host_key: Option<OptionalTargetPath>,
+    pub headers: Vec<(HeaderName, ConfigValuePath)>,
 }
 
 impl EventPartitioner {
-    const fn new(
+    fn new(
         sourcetype: Option<Template>,
         source: Option<Template>,
         index: Option<Template>,
         host_key: Option<OptionalTargetPath>,
+        headers: Vec<(HeaderName, ConfigValuePath)>,
     ) -> Self {
         Self {
             sourcetype,
             source,
             index,
             host_key,
+            headers,
         }
     }
 }
@@ -208,12 +217,28 @@ impl Partitioner for EventPartitioner {
         .and_then(|path| item.event.get(&path))
         .and_then(|value| value.as_str().map(|s| s.to_string()));
 
+        let headers = self.headers.iter().map(|(name, path)| {
+            let value = item.event.get((PathPrefix::Event, &path.0))
+                .and_then(|v| v.as_str())
+                .and_then(|s| {
+                    HeaderValue::from_str(s.as_ref())
+                        .map_err(|_| {
+                            emit!(crate::internal_events::SplunkBatchHeaderValueInvalid {
+                                header_name: name.as_str(),
+                            });
+                        })
+                        .ok()
+                });
+            (name.clone(), value)
+        }).collect();
+
         Some(Partitioned {
             token: item.event.metadata().splunk_hec_token(),
             source,
             sourcetype,
             index,
             host,
+            headers,
         })
     }
 }
@@ -381,3 +406,128 @@ impl EventCount for HecProcessedEvent {
         1
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use http::{HeaderName, HeaderValue};
+    use vector_lib::{event::LogEvent, lookup::lookup_v2::ConfigValuePath};
+    use crate::sinks::prelude::Partitioner;
+
+    use super::*;
+
+    fn create_processed_event(fields: Vec<(&str, &str)>) -> HecProcessedEvent {
+        let mut log = LogEvent::from("test message");
+        for (key, value) in fields {
+            log.insert(key, value);
+        }
+        ProcessedEvent {
+            event: log,
+            metadata: HecLogsProcessedEventMetadata::default(),
+        }
+    }
+
+    #[test]
+    fn test_event_partitioner_with_headers_same_values() {
+        let headers = vec![
+            (
+                HeaderName::from_static("x-tenant"),
+                ConfigValuePath::try_from("tenant".to_string()).unwrap(),
+            ),
+            (
+                HeaderName::from_static("x-region"),
+                ConfigValuePath::try_from("region".to_string()).unwrap(),
+            ),
+        ];
+
+        let partitioner = EventPartitioner::new(None, None, None, None, headers);
+
+        let event1 = create_processed_event(vec![("tenant", "acme"), ("region", "us")]);
+        let event2 = create_processed_event(vec![("tenant", "acme"), ("region", "us")]);
+
+        let partition1 = partitioner.partition(&event1);
+        let partition2 = partitioner.partition(&event2);
+
+        // Same header values should produce the same partition key
+        assert_eq!(partition1, partition2);
+
+        // Check that headers are correctly extracted
+        let p = partition1.unwrap();
+        assert_eq!(p.headers.len(), 2);
+        assert_eq!(
+            p.headers[0],
+            (HeaderName::from_static("x-tenant"), Some(HeaderValue::from_static("acme")))
+        );
+        assert_eq!(
+            p.headers[1],
+            (HeaderName::from_static("x-region"), Some(HeaderValue::from_static("us")))
+        );
+    }
+
+    #[test]
+    fn test_event_partitioner_with_headers_different_values() {
+        let headers = vec![(
+            HeaderName::from_static("x-tenant"),
+            ConfigValuePath::try_from("tenant".to_string()).unwrap(),
+        )];
+
+        let partitioner = EventPartitioner::new(None, None, None, None, headers);
+
+        let event1 = create_processed_event(vec![("tenant", "acme")]);
+        let event2 = create_processed_event(vec![("tenant", "globex")]);
+
+        let partition1 = partitioner.partition(&event1);
+        let partition2 = partitioner.partition(&event2);
+
+        // Different header values should produce different partition keys
+        assert_ne!(partition1, partition2);
+    }
+
+    #[test]
+    fn test_event_partitioner_with_missing_header_value() {
+        let headers = vec![(
+            HeaderName::from_static("x-tag"),
+            ConfigValuePath::try_from("tag".to_string()).unwrap(),
+        )];
+
+        let partitioner = EventPartitioner::new(None, None, None, None, headers);
+
+        let event_with_tag = create_processed_event(vec![("tag", "important")]);
+        let event_without_tag = create_processed_event(vec![("other_field", "value")]);
+
+        let partition1 = partitioner.partition(&event_with_tag);
+        let partition2 = partitioner.partition(&event_without_tag);
+
+        // Events with and without header values should be in different partitions
+        assert_ne!(partition1, partition2);
+
+        // Check that the missing value is represented as None
+        let p_with = partition1.unwrap();
+        assert_eq!(
+            p_with.headers[0],
+            (HeaderName::from_static("x-tag"), Some(HeaderValue::from_static("important")))
+        );
+
+        let p_without = partition2.unwrap();
+        assert_eq!(
+            p_without.headers[0],
+            (HeaderName::from_static("x-tag"), None)
+        );
+    }
+
+    #[test]
+    fn test_event_partitioner_without_headers() {
+        let partitioner = EventPartitioner::new(None, None, None, None, vec![]);
+
+        let event1 = create_processed_event(vec![("field1", "value1")]);
+        let event2 = create_processed_event(vec![("field2", "value2")]);
+
+        let partition1 = partitioner.partition(&event1);
+        let partition2 = partitioner.partition(&event2);
+
+        // Without headers, events should have the same partition
+        assert_eq!(partition1, partition2);
+
+        let p = partition1.unwrap();
+        assert!(p.headers.is_empty());
+    }
+}
diff --git a/src/sinks/splunk_hec/logs/tests.rs b/src/sinks/splunk_hec/logs/tests.rs
index f9083a61e8..0b4445c1fa 100644
--- a/src/sinks/splunk_hec/logs/tests.rs
+++ b/src/sinks/splunk_hec/logs/tests.rs
@@ -2,11 +2,12 @@ use std::{collections::BTreeMap, sync::Arc};
 
 use chrono::{TimeZone, Utc};
 use futures_util::StreamExt;
+use indexmap::IndexMap;
 use serde::{de, Deserialize};
 use vector_lib::codecs::{JsonSerializerConfig, TextSerializerConfig};
 use vector_lib::config::{LegacyKey, LogNamespace};
 use vector_lib::event::EventMetadata;
-use vector_lib::lookup::lookup_v2::OptionalTargetPath;
+use vector_lib::lookup::lookup_v2::{ConfigValuePath, OptionalTargetPath};
 use vector_lib::schema::{meaning, Definition};
 use vector_lib::{
     config::log_schema,
@@ -24,15 +25,17 @@ use crate::{
     sinks::{
         splunk_hec::{
             common::EndpointTarget,
-            logs::{config::HecLogsSinkConfig, encoder::HecLogsEncoder, sink::process_log},
+            logs::{config::{BatchHeader, BatchHeaders, HecLogsSinkConfig}, encoder::HecLogsEncoder, sink::process_log},
         },
-        util::{encoding::Encoder as _, test::build_test_server, Compression},
+        util::{encoding::Encoder as _, http::RequestConfig, test::build_test_server, Compression},
     },
     template::Template,
     test_util::next_addr,
 };
 use vector_lib::config::{TimePrecision, TimestampFormat};
 use crate::sinks::splunk_hec::logs::config::TimestampConfiguration;
+use crate::sinks::prelude::TowerRequestConfig;
+use crate::sinks::util::test::load_sink;
 
 #[derive(Deserialize, Debug)]
 struct HecEventJson {
@@ -240,6 +243,7 @@ async fn splunk_passthrough_token() {
         encoding: JsonSerializerConfig::default().into(),
         compression: Compression::None,
         batch: Default::default(),
+        batch_headers: Default::default(),
         request: Default::default(),
         tls: None,
         acknowledgements: Default::default(),
@@ -705,3 +709,609 @@ fn test_timestamp_configurations() {
         assert_eq!(hec_data.fields.get("event_field1").unwrap(), "test_value1");
     }
 }
+
+// ============================================================================
+// Batch Header Tests
+// ============================================================================
+
+fn create_batch_headers(headers: Vec<(&str, &str)>) -> BatchHeaders {
+    let batch_headers: Vec<BatchHeader> = headers
+        .into_iter()
+        .map(|(name, path)| BatchHeader {
+            name: name.to_string(),
+            value: ConfigValuePath::try_from(path.to_string()).unwrap(),
+        })
+        .collect();
+    BatchHeaders::from(batch_headers)
+}
+
+fn create_event_with_fields(fields: Vec<(&str, &str)>) -> Event {
+    let message = fields
+        .iter()
+        .find(|(k, _)| *k == "event_id")
+        .map(|(_, v)| *v)
+        .unwrap_or("test message");
+    let mut event = Event::Log(LogEvent::from(message));
+    for (key, value) in fields {
+        event.as_mut_log().insert(key, value);
+    }
+    event
+}
+
+#[tokio::test]
+async fn raw_endpoint_with_metadata_and_batch_headers() {
+    let addr = next_addr();
+    let config = HecLogsSinkConfig {
+        default_token: "token".to_string().into(),
+        endpoint: format!("http://{}", addr),
+        host_key: None,
+        indexed_fields: Vec::new(),
+        index: Some(Template::try_from("{{ idx }}").unwrap()),
+        sourcetype: None,
+        source: None,
+        encoding: TextSerializerConfig::default().into(),
+        compression: Compression::None,
+        batch: Default::default(),
+        batch_headers: create_batch_headers(vec![
+            ("X-Header-A", "field_a"),
+            ("X-Header-B", "field_b"),
+        ]),
+        request: Default::default(),
+        tls: None,
+        acknowledgements: Default::default(),
+        path: None,
+        auto_extract_timestamp: None,
+        endpoint_target: EndpointTarget::Raw,
+        timestamp_configuration: None,
+    };
+    let cx = SinkContext::default();
+
+    let (sink, _) = config.build(cx).await.unwrap();
+
+    let (rx, _trigger, server) = build_test_server(addr);
+    tokio::spawn(server);
+
+    let events = vec![
+        create_event_with_fields(vec![("field_a", "val1"), ("field_b", "val2"), ("idx", "idx1"), ("event_id", "evt1")]),
+        create_event_with_fields(vec![("field_a", "val1"), ("field_b", "val2"), ("idx", "idx1"), ("event_id", "evt2")]),
+        create_event_with_fields(vec![("field_a", "val1"), ("field_b", "val3"), ("idx", "idx1"), ("event_id", "evt3")]),
+        create_event_with_fields(vec![("field_a", "val1"), ("field_b", "val2"), ("idx", "idx2"), ("event_id", "evt4")]),
+    ];
+
+    sink.run_events(events).await.unwrap();
+
+    let mut requests: Vec<_> = rx.take(3).collect().await;
+
+    requests.sort_by(|a, b| {
+        let a_header_b = a.0.headers.get("X-Header-B").map(|v| v.to_str().unwrap_or(""));
+        let b_header_b = b.0.headers.get("X-Header-B").map(|v| v.to_str().unwrap_or(""));
+        let a_query = a.0.uri.query().unwrap_or("");
+        let b_query = b.0.uri.query().unwrap_or("");
+        (a_header_b, a_query).cmp(&(b_header_b, b_query))
+    });
+
+    // Batch 1: events 1,2 (val2, idx1)
+    assert_eq!(requests[0].0.headers.get("X-Header-A").unwrap(), "val1");
+    assert_eq!(requests[0].0.headers.get("X-Header-B").unwrap(), "val2");
+    assert!(requests[0].0.uri.query().unwrap().contains("index=idx1"));
+    let body0 = String::from_utf8_lossy(&requests[0].1);
+    assert!(body0.contains("evt1"));
+    assert!(body0.contains("evt2"));
+    assert!(!body0.contains("evt3"));
+    assert!(!body0.contains("evt4"));
+
+    // Batch 2: event 4 (val2, idx2)
+    assert_eq!(requests[1].0.headers.get("X-Header-A").unwrap(), "val1");
+    assert_eq!(requests[1].0.headers.get("X-Header-B").unwrap(), "val2");
+    assert!(requests[1].0.uri.query().unwrap().contains("index=idx2"));
+    let body1 = String::from_utf8_lossy(&requests[1].1);
+    assert!(body1.contains("evt4"));
+    assert!(!body1.contains("evt1"));
+    assert!(!body1.contains("evt2"));
+    assert!(!body1.contains("evt3"));
+
+    // Batch 3: event 3 (val3, idx1)
+    assert_eq!(requests[2].0.headers.get("X-Header-A").unwrap(), "val1");
+    assert_eq!(requests[2].0.headers.get("X-Header-B").unwrap(), "val3");
+    assert!(requests[2].0.uri.query().unwrap().contains("index=idx1"));
+    let body2 = String::from_utf8_lossy(&requests[2].1);
+    assert!(body2.contains("evt3"));
+    assert!(!body2.contains("evt1"));
+    assert!(!body2.contains("evt2"));
+    assert!(!body2.contains("evt4"));
+}
+
+#[tokio::test]
+async fn raw_endpoint_with_only_batch_headers() {
+    let addr = next_addr();
+    let config = HecLogsSinkConfig {
+        default_token: "token".to_string().into(),
+        endpoint: format!("http://{}", addr),
+        host_key: None,
+        indexed_fields: Vec::new(),
+        index: None,
+        sourcetype: None,
+        source: None,
+        encoding: TextSerializerConfig::default().into(),
+        compression: Compression::None,
+        batch: Default::default(),
+        batch_headers: create_batch_headers(vec![("X-Priority", "priority")]),
+        request: Default::default(),
+        tls: None,
+        acknowledgements: Default::default(),
+        path: None,
+        auto_extract_timestamp: None,
+        endpoint_target: EndpointTarget::Raw,
+        timestamp_configuration: None,
+    };
+    let cx = SinkContext::default();
+
+    let (sink, _) = config.build(cx).await.unwrap();
+
+    let (rx, _trigger, server) = build_test_server(addr);
+    tokio::spawn(server);
+
+    let events = vec![
+        create_event_with_fields(vec![("priority", "high"), ("event_id", "high-1")]),
+        create_event_with_fields(vec![("priority", "high"), ("event_id", "high-2")]),
+        create_event_with_fields(vec![("priority", "low"), ("event_id", "low-1")]),
+    ];
+
+    sink.run_events(events).await.unwrap();
+
+    let mut requests: Vec<_> = rx.take(2).collect().await;
+
+    requests.sort_by(|a, b| {
+        let a_priority = a.0.headers.get("X-Priority").map(|v| v.to_str().unwrap_or(""));
+        let b_priority = b.0.headers.get("X-Priority").map(|v| v.to_str().unwrap_or(""));
+        a_priority.cmp(&b_priority)
+    });
+
+    assert_eq!(requests[0].0.headers.get("X-Priority").unwrap(), "high");
+    let query = requests[0].0.uri.query().unwrap_or("");
+    assert!(!query.contains("index="));
+    let high_body = String::from_utf8_lossy(&requests[0].1);
+    assert!(high_body.contains("high-1"));
+    assert!(high_body.contains("high-2"));
+    assert!(!high_body.contains("low-1"));
+
+    assert_eq!(requests[1].0.headers.get("X-Priority").unwrap(), "low");
+    let low_body = String::from_utf8_lossy(&requests[1].1);
+    assert!(low_body.contains("low-1"));
+    assert!(!low_body.contains("high-1"));
+    assert!(!low_body.contains("high-2"));
+}
+
+#[tokio::test]
+async fn raw_endpoint_without_metadata_or_headers() {
+    let addr = next_addr();
+    let config = HecLogsSinkConfig {
+        default_token: "token".to_string().into(),
+        endpoint: format!("http://{}", addr),
+        host_key: None,
+        indexed_fields: Vec::new(),
+        index: None,
+        sourcetype: None,
+        source: None,
+        encoding: TextSerializerConfig::default().into(),
+        compression: Compression::None,
+        batch: Default::default(),
+        batch_headers: Default::default(),
+        request: Default::default(),
+        tls: None,
+        acknowledgements: Default::default(),
+        path: None,
+        auto_extract_timestamp: None,
+        endpoint_target: EndpointTarget::Raw,
+        timestamp_configuration: None,
+    };
+    let cx = SinkContext::default();
+
+    let (sink, _) = config.build(cx).await.unwrap();
+
+    let (rx, _trigger, server) = build_test_server(addr);
+    tokio::spawn(server);
+
+    let events = vec![
+        create_event_with_fields(vec![("event_id", "evt1")]),
+        create_event_with_fields(vec![("event_id", "evt2")]),
+        create_event_with_fields(vec![("event_id", "evt3")]),
+    ];
+
+    sink.run_events(events).await.unwrap();
+
+    let requests: Vec<_> = rx.take(1).collect().await;
+    assert_eq!(requests.len(), 1);
+
+    assert!(requests[0].0.headers.get("X-Priority").is_none());
+    let query = requests[0].0.uri.query().unwrap_or("");
+    assert!(!query.contains("index="));
+    let body = String::from_utf8_lossy(&requests[0].1);
+    assert!(body.contains("evt1"));
+    assert!(body.contains("evt2"));
+    assert!(body.contains("evt3"));
+}
+
+#[tokio::test]
+async fn event_endpoint_with_two_batch_headers() {
+    let addr = next_addr();
+    let config = HecLogsSinkConfig {
+        default_token: "token".to_string().into(),
+        endpoint: format!("http://{}", addr),
+        host_key: None,
+        indexed_fields: Vec::new(),
+        index: None,
+        sourcetype: None,
+        source: None,
+        encoding: JsonSerializerConfig::default().into(),
+        compression: Compression::None,
+        batch: Default::default(),
+        batch_headers: create_batch_headers(vec![
+            ("X-Tenant", "tenant"),
+            ("X-Region", "region"),
+        ]),
+        request: Default::default(),
+        tls: None,
+        acknowledgements: Default::default(),
+        path: None,
+        auto_extract_timestamp: None,
+        endpoint_target: EndpointTarget::Event,
+        timestamp_configuration: None,
+    };
+    let cx = SinkContext::default();
+
+    let (sink, _) = config.build(cx).await.unwrap();
+
+    let (rx, _trigger, server) = build_test_server(addr);
+    tokio::spawn(server);
+
+    let events = vec![
+        create_event_with_fields(vec![("tenant", "acme"), ("region", "us"), ("event_id", "acme-us-1")]),
+        create_event_with_fields(vec![("tenant", "acme"), ("region", "us"), ("event_id", "acme-us-2")]),
+        create_event_with_fields(vec![("tenant", "acme"), ("region", "eu"), ("event_id", "acme-eu-1")]),
+        create_event_with_fields(vec![("tenant", "globex"), ("region", "us"), ("event_id", "globex-us-1")]),
+    ];
+
+    sink.run_events(events).await.unwrap();
+
+    let mut requests: Vec<_> = rx.take(3).collect().await;
+
+    requests.sort_by(|a, b| {
+        let a_tenant = a.0.headers.get("X-Tenant").map(|v| v.to_str().unwrap_or(""));
+        let a_region = a.0.headers.get("X-Region").map(|v| v.to_str().unwrap_or(""));
+        let b_tenant = b.0.headers.get("X-Tenant").map(|v| v.to_str().unwrap_or(""));
+        let b_region = b.0.headers.get("X-Region").map(|v| v.to_str().unwrap_or(""));
+        (a_tenant, a_region).cmp(&(b_tenant, b_region))
+    });
+
+    assert_eq!(requests[0].0.headers.get("X-Tenant").unwrap(), "acme");
+    assert_eq!(requests[0].0.headers.get("X-Region").unwrap(), "eu");
+    let body0 = String::from_utf8_lossy(&requests[0].1);
+    assert!(body0.contains("acme-eu-1"));
+    assert!(!body0.contains("acme-us-1"));
+    assert!(!body0.contains("acme-us-2"));
+    assert!(!body0.contains("globex-us-1"));
+
+    assert_eq!(requests[1].0.headers.get("X-Tenant").unwrap(), "acme");
+    assert_eq!(requests[1].0.headers.get("X-Region").unwrap(), "us");
+    let body1 = String::from_utf8_lossy(&requests[1].1);
+    assert!(body1.contains("acme-us-1"));
+    assert!(body1.contains("acme-us-2"));
+    assert!(!body1.contains("acme-eu-1"));
+    assert!(!body1.contains("globex-us-1"));
+
+    assert_eq!(requests[2].0.headers.get("X-Tenant").unwrap(), "globex");
+    assert_eq!(requests[2].0.headers.get("X-Region").unwrap(), "us");
+    let body2 = String::from_utf8_lossy(&requests[2].1);
+    assert!(body2.contains("globex-us-1"));
+    assert!(!body2.contains("acme-us-1"));
+    assert!(!body2.contains("acme-us-2"));
+    assert!(!body2.contains("acme-eu-1"));
+}
+
+#[tokio::test]
+async fn event_endpoint_with_one_batch_header() {
+    let addr = next_addr();
+    let config = HecLogsSinkConfig {
+        default_token: "token".to_string().into(),
+        endpoint: format!("http://{}", addr),
+        host_key: None,
+        indexed_fields: Vec::new(),
+        index: None,
+        sourcetype: None,
+        source: None,
+        encoding: JsonSerializerConfig::default().into(),
+        compression: Compression::None,
+        batch: Default::default(),
+        batch_headers: create_batch_headers(vec![("X-Service", "service")]),
+        request: Default::default(),
+        tls: None,
+        acknowledgements: Default::default(),
+        path: None,
+        auto_extract_timestamp: None,
+        endpoint_target: EndpointTarget::Event,
+        timestamp_configuration: None,
+    };
+    let cx = SinkContext::default();
+
+    let (sink, _) = config.build(cx).await.unwrap();
+
+    let (rx, _trigger, server) = build_test_server(addr);
+    tokio::spawn(server);
+
+    let events = vec![
+        create_event_with_fields(vec![("service", "api"), ("event_id", "api-1")]),
+        create_event_with_fields(vec![("service", "api"), ("event_id", "api-2")]),
+        create_event_with_fields(vec![("service", "web"), ("event_id", "web-1")]),
+    ];
+
+    sink.run_events(events).await.unwrap();
+
+    let mut requests: Vec<_> = rx.take(2).collect().await;
+
+    requests.sort_by(|a, b| {
+        let a_service = a.0.headers.get("X-Service").map(|v| v.to_str().unwrap_or(""));
+        let b_service = b.0.headers.get("X-Service").map(|v| v.to_str().unwrap_or(""));
+        a_service.cmp(&b_service)
+    });
+
+    assert_eq!(requests[0].0.headers.get("X-Service").unwrap(), "api");
+    let api_body = String::from_utf8_lossy(&requests[0].1);
+    assert!(api_body.contains("api-1"));
+    assert!(api_body.contains("api-2"));
+    assert!(!api_body.contains("web-1"));
+
+    assert_eq!(requests[1].0.headers.get("X-Service").unwrap(), "web");
+    let web_body = String::from_utf8_lossy(&requests[1].1);
+    assert!(web_body.contains("web-1"));
+    assert!(!web_body.contains("api-1"));
+    assert!(!web_body.contains("api-2"));
+}
+
+#[tokio::test]
+async fn event_endpoint_no_batching_on_metadata_fields() {
+    let addr = next_addr();
+    let config = HecLogsSinkConfig {
+        default_token: "token".to_string().into(),
+        endpoint: format!("http://{}", addr),
+        host_key: None,
+        indexed_fields: Vec::new(),
+        index: Some(Template::try_from("{{ event_index }}").unwrap()),
+        sourcetype: None,
+        source: None,
+        encoding: JsonSerializerConfig::default().into(),
+        compression: Compression::None,
+        batch: Default::default(),
+        batch_headers: create_batch_headers(vec![("X-Priority", "priority")]),
+        request: Default::default(),
+        tls: None,
+        acknowledgements: Default::default(),
+        path: None,
+        auto_extract_timestamp: None,
+        endpoint_target: EndpointTarget::Event,
+        timestamp_configuration: None,
+    };
+    let cx = SinkContext::default();
+
+    let (sink, _) = config.build(cx).await.unwrap();
+
+    let (rx, _trigger, server) = build_test_server(addr);
+    tokio::spawn(server);
+
+    let events = vec![
+        create_event_with_fields(vec![("event_index", "idx1"), ("priority", "high"), ("event_id", "high-idx1")]),
+        create_event_with_fields(vec![("event_index", "idx2"), ("priority", "high"), ("event_id", "high-idx2")]),
+        create_event_with_fields(vec![("event_index", "idx1"), ("priority", "low"), ("event_id", "low-idx1")]),
+    ];
+
+    sink.run_events(events).await.unwrap();
+
+    let mut requests: Vec<_> = rx.take(2).collect().await;
+
+    requests.sort_by(|a, b| {
+        let a_priority = a.0.headers.get("X-Priority").map(|v| v.to_str().unwrap_or(""));
+        let b_priority = b.0.headers.get("X-Priority").map(|v| v.to_str().unwrap_or(""));
+        a_priority.cmp(&b_priority)
+    });
+
+    assert_eq!(requests[0].0.headers.get("X-Priority").unwrap(), "high");
+    let high_body = String::from_utf8_lossy(&requests[0].1);
+    assert!(high_body.contains("high-idx1"));
+    assert!(high_body.contains("high-idx2"));
+    assert!(!high_body.contains("low-idx1"));
+
+    assert_eq!(requests[1].0.headers.get("X-Priority").unwrap(), "low");
+    let low_body = String::from_utf8_lossy(&requests[1].1);
+    assert!(low_body.contains("low-idx1"));
+    assert!(!low_body.contains("high-idx1"));
+    assert!(!low_body.contains("high-idx2"));
+}
+
+#[tokio::test]
+async fn batch_headers_missing_value_separate_batch() {
+    let addr = next_addr();
+    let config = HecLogsSinkConfig {
+        default_token: "token".to_string().into(),
+        endpoint: format!("http://{}", addr),
+        host_key: None,
+        indexed_fields: Vec::new(),
+        index: None,
+        sourcetype: None,
+        source: None,
+        encoding: JsonSerializerConfig::default().into(),
+        compression: Compression::None,
+        batch: Default::default(),
+        batch_headers: create_batch_headers(vec![("X-Tag", "tag")]),
+        request: Default::default(),
+        tls: None,
+        acknowledgements: Default::default(),
+        path: None,
+        auto_extract_timestamp: None,
+        endpoint_target: EndpointTarget::Event,
+        timestamp_configuration: None,
+    };
+    let cx = SinkContext::default();
+
+    let (sink, _) = config.build(cx).await.unwrap();
+
+    let (rx, _trigger, server) = build_test_server(addr);
+    tokio::spawn(server);
+
+    let events = vec![
+        create_event_with_fields(vec![("tag", "important"), ("event_id", "evt-tag-1")]),
+        create_event_with_fields(vec![("tag", "important"), ("event_id", "evt-tag-2")]),
+        create_event_with_fields(vec![("other_field", "value"), ("event_id", "evt-no-tag")]),
+    ];
+
+    sink.run_events(events).await.unwrap();
+
+    let mut requests: Vec<_> = rx.take(2).collect().await;
+
+    requests.sort_by(|a, b| {
+        let a_tag = a.0.headers.get("X-Tag").is_some();
+        let b_tag = b.0.headers.get("X-Tag").is_some();
+        b_tag.cmp(&a_tag)
+    });
+
+    assert_eq!(requests[0].0.headers.get("X-Tag").unwrap(), "important");
+    let tagged_body = String::from_utf8_lossy(&requests[0].1);
+    assert!(tagged_body.contains("evt-tag-1"));
+    assert!(tagged_body.contains("evt-tag-2"));
+    assert!(!tagged_body.contains("evt-no-tag"));
+
+    assert!(requests[1].0.headers.get("X-Tag").is_none());
+    let untagged_body = String::from_utf8_lossy(&requests[1].1);
+    assert!(untagged_body.contains("evt-no-tag"));
+    assert!(!untagged_body.contains("evt-tag-1"));
+    assert!(!untagged_body.contains("evt-tag-2"));
+}
+
+#[tokio::test]
+async fn batch_headers_static_headers_override() {
+    let addr = next_addr();
+    let config = HecLogsSinkConfig {
+        default_token: "token".to_string().into(),
+        endpoint: format!("http://{}", addr),
+        host_key: None,
+        indexed_fields: Vec::new(),
+        index: None,
+        sourcetype: None,
+        source: None,
+        encoding: JsonSerializerConfig::default().into(),
+        compression: Compression::None,
+        batch: Default::default(),
+        batch_headers: create_batch_headers(vec![("X-Override", "override_field")]),
+        request: RequestConfig {
+            tower: TowerRequestConfig::default(),
+            headers: IndexMap::from_iter([
+                ("X-Override".to_owned(), "static-value".to_owned()),
+            ]),
+        },
+        tls: None,
+        acknowledgements: Default::default(),
+        path: None,
+        auto_extract_timestamp: None,
+        endpoint_target: EndpointTarget::Event,
+        timestamp_configuration: None,
+    };
+    let cx = SinkContext::default();
+
+    let (sink, _) = config.build(cx).await.unwrap();
+
+    let (rx, _trigger, server) = build_test_server(addr);
+    tokio::spawn(server);
+
+    let events = vec![
+        create_event_with_fields(vec![("override_field", "dynamic-value"), ("event_id", "override-evt")]),
+    ];
+
+    sink.run_events(events).await.unwrap();
+
+    let requests: Vec<_> = rx.take(1).collect().await;
+
+    assert_eq!(requests[0].0.headers.get("X-Override").unwrap(), "static-value");
+    let body = String::from_utf8_lossy(&requests[0].1);
+    assert!(body.contains("override-evt"));
+    assert!(body.contains("dynamic-value"));
+}
+
+#[tokio::test]
+async fn batch_header_field_excluded_from_body_raw_endpoint() {
+    let addr = next_addr();
+    let config_toml = format!(
+        r#"
+            default_token = "token"
+            endpoint = "http://{}"
+            endpoint_target = "raw"
+            encoding.codec = "json"
+            encoding.except_fields = ["tenant"]
+
+            [[batch_headers]]
+            name = "X-Tenant"
+            value = "tenant"
+        "#,
+        addr
+    );
+
+    let (config, cx) = load_sink::<HecLogsSinkConfig>(&config_toml).unwrap();
+    let (sink, _) = config.build(cx).await.unwrap();
+
+    let (rx, _trigger, server) = build_test_server(addr);
+    tokio::spawn(server);
+
+    let mut event = Event::Log(LogEvent::from("test message"));
+    event.as_mut_log().insert("tenant", "acme");
+    event.as_mut_log().insert("other_field", "keep-me");
+    event.as_mut_log().insert("event_id", "evt-raw-transform");
+
+    sink.run_events(vec![event]).await.unwrap();
+
+    let requests: Vec<_> = rx.take(1).collect().await;
+
+    assert_eq!(requests[0].0.headers.get("X-Tenant").unwrap(), "acme");
+    let body = String::from_utf8_lossy(&requests[0].1);
+    assert!(!body.contains("acme"), "tenant field should be excluded from body");
+    assert!(body.contains("other_field"), "other_field should be in body");
+    assert!(body.contains("keep-me"), "other_field value should be in body");
+}
+
+#[tokio::test]
+async fn batch_header_field_excluded_from_body_event_endpoint() {
+    let addr = next_addr();
+    let config_toml = format!(
+        r#"
+            default_token = "token"
+            endpoint = "http://{}"
+            endpoint_target = "event"
+            encoding.codec = "json"
+            encoding.except_fields = ["priority"]
+
+            [[batch_headers]]
+            name = "X-Priority"
+            value = "priority"
+        "#,
+        addr
+    );
+
+    let (config, cx) = load_sink::<HecLogsSinkConfig>(&config_toml).unwrap();
+    let (sink, _) = config.build(cx).await.unwrap();
+
+    let (rx, _trigger, server) = build_test_server(addr);
+    tokio::spawn(server);
+
+    let mut event = Event::Log(LogEvent::from("test message"));
+    event.as_mut_log().insert("priority", "high");
+    event.as_mut_log().insert("other_field", "keep-me");
+    event.as_mut_log().insert("event_id", "evt-event-transform");
+
+    sink.run_events(vec![event]).await.unwrap();
+
+    let requests: Vec<_> = rx.take(1).collect().await;
+
+    assert_eq!(requests[0].0.headers.get("X-Priority").unwrap(), "high");
+    let body = String::from_utf8_lossy(&requests[0].1);
+    assert!(!body.contains("high"), "priority field should be excluded from body");
+    assert!(body.contains("other_field"), "other_field should be in body");
+    assert!(body.contains("keep-me"), "other_field value should be in body");
+    assert!(body.contains("evt-event-transform"), "event_id should be in body");
+}
diff --git a/src/sinks/splunk_hec/metrics/request_builder.rs b/src/sinks/splunk_hec/metrics/request_builder.rs
index d57f463423..d14152a6ec 100644
--- a/src/sinks/splunk_hec/metrics/request_builder.rs
+++ b/src/sinks/splunk_hec/metrics/request_builder.rs
@@ -60,6 +60,7 @@ impl RequestBuilder<(Option<Arc<str>>, Vec<HecProcessedEvent>)> for HecMetricsRe
             source: None,
             sourcetype: None,
             host: None,
+            headers: vec![],
             metadata,
         }
     }
diff --git a/src/sources/splunk_hec/mod.rs b/src/sources/splunk_hec/mod.rs
index 01e73dbecf..bddf118e50 100644
--- a/src/sources/splunk_hec/mod.rs
+++ b/src/sources/splunk_hec/mod.rs
@@ -1,5 +1,5 @@
 use std::{
-    collections::HashMap,
+    collections::{BTreeSet, HashMap},
     convert::Infallible,
     io::Read,
     net::{Ipv4Addr, SocketAddr},
@@ -288,7 +288,7 @@ impl SourceConfig for SplunkConfig {
 
 /// Shared data for responding to requests.
 struct SplunkSource {
-    valid_credentials: Vec<String>,
+    valid_tokens: Arc<BTreeSet<String>>,
     protocol: &'static str,
     idx_ack: Option<Arc<IndexerAcknowledgement>>,
     store_hec_token: bool,
@@ -301,11 +301,13 @@ impl SplunkSource {
         let log_namespace = cx.log_namespace(config.log_namespace);
         let acknowledgements = cx.do_acknowledgements(config.acknowledgements.enabled.into());
         let shutdown = cx.shutdown;
-        let valid_tokens = config
+        let valid_tokens: BTreeSet<String> = config
             .valid_tokens
             .iter()
             .flatten()
-            .chain(config.token.iter());
+            .chain(config.token.iter())
+            .map(|t| t.clone().into())
+            .collect();
 
         let idx_ack = acknowledgements.then(|| {
             Arc::new(IndexerAcknowledgement::new(
@@ -315,9 +317,7 @@ impl SplunkSource {
         });
 
         SplunkSource {
-            valid_credentials: valid_tokens
-                .map(|token| format!("Splunk {}", token.inner()))
-                .collect(),
+            valid_tokens: Arc::new(valid_tokens),
             protocol,
             idx_ack,
             store_hec_token: config.store_hec_token,
@@ -327,14 +327,6 @@ impl SplunkSource {
     }
 
     fn event_service(&self, out: SourceSender) -> BoxedFilter<(Response,)> {
-        let splunk_channel_query_param = warp::query::<HashMap<String, String>>()
-            .map(|qs: HashMap<String, String>| qs.get("channel").map(|v| v.to_owned()));
-        let splunk_channel_header = warp::header::optional::<String>(X_SPLUNK_REQUEST_CHANNEL);
-
-        let splunk_channel = splunk_channel_header
-            .and(splunk_channel_query_param)
-            .map(|header: Option<String>, query_param| header.or(query_param));
-
         let protocol = self.protocol;
         let idx_ack = self.idx_ack.clone();
         let store_hec_token = self.store_hec_token;
@@ -348,7 +340,7 @@ impl SplunkSource {
                     .or(warp::path::end()),
             )
             .and(self.authorization())
-            .and(splunk_channel)
+            .and(SplunkSource::optional_channel())
             .and(warp::addr::remote())
             .and(warp::header::optional::<String>("X-Forwarded-For"))
             .and(self.gzip())
@@ -452,7 +444,7 @@ impl SplunkSource {
         warp::post()
             .and(path!("raw" / "1.0").or(path!("raw")))
             .and(self.authorization())
-            .and(SplunkSource::required_channel())
+            .and(SplunkSource::optional_channel())
             .and(warp::addr::remote())
             .and(warp::header::optional::<String>("X-Forwarded-For"))
             .and(self.gzip())
@@ -461,7 +453,7 @@ impl SplunkSource {
             .and_then(
                 move |_,
                       token: Option<String>,
-                      channel_id: String,
+                      channel_id: Option<String>,
                       remote: Option<SocketAddr>,
                       xff: Option<String>,
                       gzip: bool,
@@ -477,16 +469,15 @@ impl SplunkSource {
                     });
 
                     async move {
-                        let (batch, receiver) =
-                            BatchNotifier::maybe_new_with_receiver(idx_ack.is_some());
-                        let maybe_ack_id = match (idx_ack, receiver) {
-                            (Some(idx_ack), Some(receiver)) => Some(
-                                idx_ack
-                                    .get_ack_id_from_channel(channel_id.clone(), receiver)
-                                    .await?,
-                            ),
-                            _ => None,
-                        };
+                        let (batch, maybe_ack_id) = match (idx_ack, channel_id.clone()) {
+                            (Some(idx_ack), Some(channel_id)) => {
+                                let (batch, receiver) = BatchNotifier::new_with_receiver();
+                                idx_ack.get_ack_id_from_channel(channel_id, receiver).await.map(|ack| (Some(batch), Some(ack)))
+                            },
+                            (Some(_), None) => Err(Rejection::from(ApiError::MissingChannel)),
+                            (None, _) => Ok((None, None)),
+                        }?;
+
                         let mut event = raw_event(
                             body,
                             gzip,
@@ -574,26 +565,31 @@ impl SplunkSource {
 
     /// Authorize request
     fn authorization(&self) -> BoxedFilter<(Option<String>,)> {
-        let valid_credentials = self.valid_credentials.clone();
+        let valid_creds = Arc::clone(&self.valid_tokens);
         warp::header::optional("Authorization")
             .and_then(move |token: Option<String>| {
-                let valid_credentials = valid_credentials.clone();
+                let valid_tokens = Arc::clone(&valid_creds);
                 async move {
-                    match (token, valid_credentials.is_empty()) {
-                        // Remove the "Splunk " prefix if present as it is not
-                        // part of the token itself
-                        (token, true) => {
-                            Ok(token
-                                .map(|t| t.strip_prefix("Splunk ").map(Into::into).unwrap_or(t)))
-                        }
-                        (Some(token), false) if valid_credentials.contains(&token) => Ok(Some(
-                            token
-                                .strip_prefix("Splunk ")
-                                .map(Into::into)
-                                .unwrap_or(token),
-                        )),
+                    let token = token
+                        .map(|t|
+                            match t.split_once(" ") {
+                                Some((scheme, value)) => {
+                                    let scheme = scheme.to_lowercase();
+                                    if scheme == "splunk" || scheme == "bearer" {
+                                        Some(value.trim_start().to_owned())
+                                    } else {
+                                        None
+                                    }
+                                },
+                                _ => None,
+                            })
+                        .flatten();
+                    match (token, valid_tokens.is_empty()) {
+                        (Some(token), true) => Ok(Some(token)),
+                        (Some(token), false) if valid_tokens.contains(&token) => Ok(Some(token)),
                         (Some(_), false) => Err(Rejection::from(ApiError::InvalidAuthorization)),
                         (None, false) => Err(Rejection::from(ApiError::MissingAuthorization)),
+                        (None, true) => Ok(None),
                     }
                 }
             })
@@ -627,6 +623,17 @@ impl SplunkSource {
             })
             .boxed()
     }
+
+    fn optional_channel() -> BoxedFilter<(Option<String>,)> {
+        let splunk_channel_query_param = warp::query::<HashMap<String, String>>()
+            .map(|qs: HashMap<String, String>| qs.get("channel").map(|v| v.to_owned()));
+        let splunk_channel_header = warp::header::optional::<String>(X_SPLUNK_REQUEST_CHANNEL);
+
+        splunk_channel_header
+            .and(splunk_channel_query_param)
+            .map(|header: Option<String>, query_param| header.or(query_param))
+            .boxed()
+    }
 }
 /// Constructs one or more events from json-s coming from reader.
 /// If errors, it's done with input.
@@ -1018,7 +1025,7 @@ enum Time {
 fn raw_event(
     bytes: Bytes,
     gzip: bool,
-    channel: String,
+    channel: Option<String>,
     remote: Option<SocketAddr>,
     xff: Option<String>,
     batch: Option<BatchNotifier>,
@@ -1052,14 +1059,16 @@ fn raw_event(
     // We need to calculate the estimated json size of the event BEFORE enrichment.
     events_received.emit(CountByteSize(1, log.estimated_json_encoded_size_of()));
 
-    // Add channel
-    log_namespace.insert_source_metadata(
-        SplunkConfig::NAME,
-        &mut log,
-        Some(LegacyKey::Overwrite(&owned_value_path!(CHANNEL))),
-        lookup::path!(CHANNEL),
-        channel,
-    );
+    // Add channel only if provided
+    if let Some(channel) = channel {
+        log_namespace.insert_source_metadata(
+            SplunkConfig::NAME,
+            &mut log,
+            Some(LegacyKey::Overwrite(&owned_value_path!(CHANNEL))),
+            lookup::path!(CHANNEL),
+            channel,
+        );
+    }
 
     // host-field priority for raw endpoint:
     // - x-forwarded-for is set to `host` field first, if present. If not present:
@@ -1351,6 +1360,7 @@ mod tests {
             encoding,
             compression,
             batch: BatchConfig::default(),
+            batch_headers: Default::default(),
             request: RequestConfig::default(),
             tls: None,
             acknowledgements: Default::default(),
@@ -2522,6 +2532,153 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn raw_no_channel_acks_disabled() {
+        assert_source_compliance(&HTTP_PUSH_SOURCE_TAGS, async {
+            let message = "raw_no_channel";
+            let (source, address) = source(None).await;
+
+            let opts = SendWithOpts {
+                channel: None,
+                forwarded_for: None,
+            };
+
+            assert_eq!(
+                200,
+                send_with(address, "services/collector/raw", message, TOKEN, &opts).await
+            );
+
+            let event = collect_n(source, 1).await.remove(0);
+            assert_eq!(
+                event.as_log()[log_schema().message_key().unwrap().to_string()],
+                message.into()
+            );
+            // Verify channel key is NOT present when not provided
+            assert!(event.as_log().get(super::CHANNEL).is_none());
+        })
+        .await;
+    }
+
+    #[tokio::test]
+    async fn raw_no_channel_acks_enabled() {
+        let message = "raw_no_channel_acks_enabled";
+        let ack_config = HecAcknowledgementsConfig {
+            enabled: Some(true),
+            ..Default::default()
+        };
+        let (_, address) = source(Some(ack_config)).await;
+
+        let opts = SendWithOpts {
+            channel: None,
+            forwarded_for: None,
+        };
+
+        assert_eq!(
+            400,
+            send_with(address, "services/collector/raw", message, TOKEN, &opts).await
+        );
+    }
+
+    #[tokio::test]
+    async fn raw_with_channel_in_event() {
+        assert_source_compliance(&HTTP_PUSH_SOURCE_TAGS, async {
+            let message = "raw_with_channel";
+            let (source, address) = source(None).await;
+
+            let opts = SendWithOpts {
+                channel: Some(Channel::Header("my-channel-id")),
+                forwarded_for: None,
+            };
+
+            assert_eq!(
+                200,
+                send_with(address, "services/collector/raw", message, TOKEN, &opts).await
+            );
+
+            let event = collect_n(source, 1).await.remove(0);
+            assert_eq!(event.as_log()[&super::CHANNEL], "my-channel-id".into());
+        })
+        .await;
+    }
+
+    async fn events_url_bearer_prefix_test(prefix: &str) {
+        assert_source_compliance(&HTTP_PUSH_SOURCE_TAGS, async {
+            let message = r#"{"event":"bearer_test"}"#;
+            let (source, address) = source(None).await;
+
+            let res = reqwest::Client::new()
+                .post(format!("http://{}/services/collector/event", address))
+                .header("Authorization", format!("{} {}", prefix, TOKEN))
+                .body(message)
+                .send()
+                .await
+                .unwrap();
+
+            assert_eq!(200, res.status().as_u16());
+
+            let event = collect_n(source, 1).await.remove(0);
+            assert_eq!(
+                event.as_log()[log_schema().message_key().unwrap().to_string()],
+                "bearer_test".into()
+            );
+        })
+        .await;
+    }
+
+    #[tokio::test]
+    async fn authorization_case_insensitive_bearer_prefix_event() {
+        events_url_bearer_prefix_test("bEaReR        ").await;
+    }
+
+    #[tokio::test]
+    async fn authorization_bearer_prefix_event() {
+        events_url_bearer_prefix_test("Bearer").await;
+    }
+
+    #[tokio::test]
+    async fn authorization_bearer_prefix_raw() {
+        assert_source_compliance(&HTTP_PUSH_SOURCE_TAGS, async {
+            let message = "bearer_raw_test";
+            let (source, address) = source(None).await;
+
+            let res = reqwest::Client::new()
+                .post(format!("http://{}/services/collector/raw", address))
+                .header("Authorization", format!("Bearer {}", TOKEN))
+                .body(message)
+                .send()
+                .await
+                .unwrap();
+
+            assert_eq!(200, res.status().as_u16());
+
+            let event = collect_n(source, 1).await.remove(0);
+            assert_eq!(
+                event.as_log()[log_schema().message_key().unwrap().to_string()],
+                message.into()
+            );
+        })
+        .await;
+    }
+
+    #[tokio::test]
+    async fn authorization_invalid_prefix() {
+        assert_source_error(&COMPONENT_ERROR_TAGS, async {
+            let message = r#"{"event":"test"}"#;
+            let (_source, address) = source(None).await;
+
+            let res = reqwest::Client::new()
+                .post(format!("http://{}/services/collector/event", address))
+                .header("Authorization", format!("Basic {}", TOKEN))
+                .body(message)
+                .send()
+                .await
+                .unwrap();
+
+            assert_eq!(401, res.status().as_u16());
+        })
+        .await;
+    }
+
     #[test]
     fn output_schema_definition_vector_namespace() {
         let config = SplunkConfig {