feat(codecs): make avro codec optional via cargo feature flag

Cargo.toml

@@ -520,7 +520,7 @@ vendored = ["rdkafka?/gssapi-vendored"]
 
 # Default features for *-pc-windows-msvc
 # TODO: Enable SASL https://github.com/vectordotdev/vector/pull/3081#issuecomment-659298042
-base = ["api", "enrichment-tables", "sinks", "sources", "transforms", "secrets", "vrl/stdlib", "codecs-parquet"]
+base = ["api", "enrichment-tables", "sinks", "sources", "transforms", "secrets", "vrl/stdlib", "codecs-avro", "codecs-parquet"]
 enable-api-client = ["base", "api-client"]
 enable-unix = ["enable-api-client", "sources-dnstap", "unix"]
 
@@ -618,6 +618,7 @@ enrichment-tables-memory = ["dep:evmap", "dep:evmap-derive", "dep:thread_local"]
 
 # Codecs
 codecs-arrow = ["dep:arrow", "dep:arrow-schema", "vector-lib/arrow"]
+codecs-avro = ["vector-lib/avro"]
 codecs-parquet = ["dep:parquet", "codecs-arrow", "vector-lib/parquet"]
 codecs-opentelemetry = ["vector-lib/opentelemetry"]
 codecs-syslog = ["vector-lib/syslog"]
@@ -729,7 +730,7 @@ sources-prometheus = ["sources-prometheus-scrape", "sources-prometheus-remote-wr
 sources-prometheus-scrape = ["sinks-prometheus", "sources-utils-http-client", "vector-lib/prometheus"]
 sources-prometheus-remote-write = ["sinks-prometheus", "sources-utils-http", "vector-lib/prometheus"]
 sources-prometheus-pushgateway = ["sinks-prometheus", "sources-utils-http", "vector-lib/prometheus"]
-sources-pulsar = ["dep:apache-avro", "dep:pulsar"]
+sources-pulsar = ["dep:apache-avro", "dep:pulsar", "codecs-avro"]
 sources-redis = ["dep:redis"]
 sources-socket = ["sources-utils-net", "tokio-util/net"]
 sources-splunk_hec = ["dep:roaring"]
@@ -922,7 +923,7 @@ sinks-opentelemetry = ["sinks-http", "codecs-opentelemetry"]
 sinks-papertrail = ["dep:syslog"]
 sinks-prometheus = ["dep:base64", "dep:prost", "vector-lib/prometheus"]
 sinks-postgres = ["dep:sqlx"]
-sinks-pulsar = ["dep:apache-avro", "dep:pulsar"]
+sinks-pulsar = ["dep:apache-avro", "dep:pulsar", "codecs-avro"]
 sinks-redis = ["dep:redis"]
 sinks-sematext = ["sinks-elasticsearch", "sinks-influxdb"]
 sinks-socket = ["sinks-utils-udp"]

changelog.d/codecs_avro_optional.enhancement.md

@@ -0,0 +1,3 @@
+The `apache-avro` dependency and the `avro` codec (encoding and decoding) are now gated behind the new `codecs-avro` Cargo feature. The feature is enabled by default in all release builds, and is automatically pulled in by `sources-pulsar` and `sinks-pulsar`, so the default behavior is unchanged. Custom builds that exclude this feature (`--no-default-features --features "..."`) drop the `apache-avro` crate and the avro codec implementation, reducing the binary size by roughly 380 KB on x86_64. This mirrors the existing gating pattern for `arrow`, `parquet`, `opentelemetry`, and `syslog` codecs in `lib/codecs/Cargo.toml`. See [#20064](https://github.com/vectordotdev/vector/issues/20064) for the umbrella discussion of minimal builds.
+
+authors: ahachete

lib/codecs/Cargo.toml

@@ -11,9 +11,10 @@ unwrap-used = "deny"
 [[bin]]
 name = "generate-avro-fixtures"
 path = "tests/bin/generate-avro-fixtures.rs"
+required-features = ["avro"]
 
 [dependencies]
-apache-avro = { version = "0.20.0", default-features = false }
+apache-avro = { version = "0.20.0", default-features = false, optional = true }
 arrow = { version = "56.2.0", default-features = false, features = ["ipc", "json"], optional = true }
 parquet = { version = "56.2.0", default-features = false, features = [
     "arrow",
@@ -78,6 +79,7 @@ vrl.workspace = true
 
 [features]
 arrow = ["dep:arrow", "arrow/chrono-tz"]
+avro = ["dep:apache-avro"]
 parquet = ["dep:parquet", "arrow"]
 opentelemetry = ["dep:opentelemetry-proto"]
 syslog = ["dep:syslog_loose", "dep:strum", "dep:derive_more", "dep:serde-aux", "dep:toml"]

lib/codecs/src/decoding/format/mod.rs

@@ -3,6 +3,7 @@
 
 #![deny(missing_docs)]
 
+#[cfg(feature = "avro")]
 mod avro;
 mod bytes;
 mod gelf;
@@ -18,6 +19,7 @@ mod syslog;
 mod vrl;
 
 use ::bytes::Bytes;
+#[cfg(feature = "avro")]
 pub use avro::{AvroDeserializer, AvroDeserializerConfig, AvroDeserializerOptions};
 use dyn_clone::DynClone;
 pub use gelf::{GelfDeserializer, GelfDeserializerConfig, GelfDeserializerOptions};

lib/codecs/src/decoding/mod.rs

@@ -41,6 +41,7 @@ use vector_core::{
     schema,
 };
 
+#[cfg(feature = "avro")]
 use self::format::{AvroDeserializer, AvroDeserializerConfig, AvroDeserializerOptions};
 use crate::decoding::format::{VrlDeserializer, VrlDeserializerConfig};
 
@@ -314,6 +315,7 @@ pub enum DeserializerConfig {
     /// Decodes the raw bytes as as an [Apache Avro][apache_avro] message.
     ///
     /// [apache_avro]: https://avro.apache.org/
+    #[cfg(feature = "avro")]
     Avro {
         /// Apache Avro-specific encoder options.
         avro: AvroDeserializerOptions,
@@ -372,6 +374,7 @@ impl DeserializerConfig {
     /// Build the `Deserializer` from this configuration.
     pub fn build(&self) -> vector_common::Result<Deserializer> {
         match self {
+            #[cfg(feature = "avro")]
             DeserializerConfig::Avro { avro } => Ok(Deserializer::Avro(
                 AvroDeserializerConfig {
                     avro_options: avro.clone(),
@@ -398,6 +401,7 @@ impl DeserializerConfig {
     /// Return an appropriate default framer for the given deserializer
     pub fn default_stream_framing(&self) -> FramingConfig {
         match self {
+            #[cfg(feature = "avro")]
             DeserializerConfig::Avro { .. } => FramingConfig::Bytes,
             DeserializerConfig::Native => FramingConfig::LengthDelimited(Default::default()),
             DeserializerConfig::Bytes
@@ -429,6 +433,7 @@ impl DeserializerConfig {
     /// Return the type of event build by this deserializer.
     pub fn output_type(&self) -> DataType {
         match self {
+            #[cfg(feature = "avro")]
             DeserializerConfig::Avro { avro } => AvroDeserializerConfig {
                 avro_options: avro.clone(),
             }
@@ -451,6 +456,7 @@ impl DeserializerConfig {
     /// The schema produced by the deserializer.
     pub fn schema_definition(&self, log_namespace: LogNamespace) -> schema::Definition {
         match self {
+            #[cfg(feature = "avro")]
             DeserializerConfig::Avro { avro } => AvroDeserializerConfig {
                 avro_options: avro.clone(),
             }
@@ -489,9 +495,9 @@ impl DeserializerConfig {
                         },
                 }),
             ) => "application/json",
-            (DeserializerConfig::Native, _) | (DeserializerConfig::Avro { .. }, _) => {
-                "application/octet-stream"
-            }
+            (DeserializerConfig::Native, _) => "application/octet-stream",
+            #[cfg(feature = "avro")]
+            (DeserializerConfig::Avro { .. }, _) => "application/octet-stream",
             (DeserializerConfig::Protobuf(_), _) => "application/octet-stream",
             #[cfg(feature = "opentelemetry")]
             (DeserializerConfig::Otlp(_), _) => "application/x-protobuf",
@@ -515,6 +521,7 @@ impl DeserializerConfig {
 #[derive(Clone)]
 pub enum Deserializer {
     /// Uses a `AvroDeserializer` for deserialization.
+    #[cfg(feature = "avro")]
     Avro(AvroDeserializer),
     /// Uses a `BytesDeserializer` for deserialization.
     Bytes(BytesDeserializer),
@@ -549,6 +556,7 @@ impl format::Deserializer for Deserializer {
         log_namespace: LogNamespace,
     ) -> vector_common::Result<SmallVec<[Event; 1]>> {
         match self {
+            #[cfg(feature = "avro")]
             Deserializer::Avro(deserializer) => deserializer.parse(bytes, log_namespace),
             Deserializer::Bytes(deserializer) => deserializer.parse(bytes, log_namespace),
             Deserializer::Json(deserializer) => deserializer.parse(bytes, log_namespace),

lib/codecs/src/encoding/config.rs

@@ -109,9 +109,9 @@ impl EncodingConfigWithFraming {
                 SinkType::StreamBased => NewlineDelimitedEncoder::default().into(),
                 SinkType::MessageBased => CharacterDelimitedEncoder::new(b',').into(),
             },
-            (None, Serializer::Avro(_) | Serializer::Native(_)) => {
-                LengthDelimitedEncoder::default().into()
-            }
+            (None, Serializer::Native(_)) => LengthDelimitedEncoder::default().into(),
+            #[cfg(feature = "avro")]
+            (None, Serializer::Avro(_)) => LengthDelimitedEncoder::default().into(),
             (None, Serializer::Gelf(_)) => {
                 // Graylog/GELF always uses null byte delimiter on TCP, see
                 // https://github.com/Graylog2/graylog2-server/issues/1240

lib/codecs/src/encoding/encoder.rs

@@ -207,8 +207,7 @@ impl Encoder<Framer> {
             ) => "application/json",
             (Serializer::Native(_), _) | (Serializer::Protobuf(_), _) => "application/octet-stream",
             (
-                Serializer::Avro(_)
-                | Serializer::Cef(_)
+                Serializer::Cef(_)
                 | Serializer::Csv(_)
                 | Serializer::Gelf(_)
                 | Serializer::Json(_)
@@ -218,6 +217,8 @@ impl Encoder<Framer> {
                 | Serializer::Text(_),
                 _,
             ) => "text/plain",
+            #[cfg(feature = "avro")]
+            (Serializer::Avro(_), _) => "text/plain",
             #[cfg(feature = "syslog")]
             (Serializer::Syslog(_), _) => "text/plain",
             #[cfg(feature = "opentelemetry")]

lib/codecs/src/encoding/format/mod.rs

@@ -5,6 +5,7 @@
 
 #[cfg(feature = "arrow")]
 mod arrow;
+#[cfg(feature = "avro")]
 mod avro;
 mod cef;
 mod common;
@@ -35,6 +36,7 @@ pub use arrow::{
     ArrowEncodingError, ArrowStreamSerializer, ArrowStreamSerializerConfig, SchemaProvider,
     find_null_non_nullable_fields,
 };
+#[cfg(feature = "avro")]
 pub use avro::{AvroSerializer, AvroSerializerConfig, AvroSerializerOptions};
 pub use cef::{CefSerializer, CefSerializerConfig};
 use dyn_clone::DynClone;

lib/codecs/src/encoding/mod.rs

@@ -16,12 +16,13 @@ pub use format::{
     ArrowEncodingError, ArrowStreamSerializer, ArrowStreamSerializerConfig, SchemaProvider,
     find_null_non_nullable_fields,
 };
+#[cfg(feature = "avro")]
+pub use format::{AvroSerializer, AvroSerializerConfig, AvroSerializerOptions};
 pub use format::{
-    AvroSerializer, AvroSerializerConfig, AvroSerializerOptions, CefSerializer,
-    CefSerializerConfig, CsvSerializer, CsvSerializerConfig, GelfSerializer, GelfSerializerConfig,
-    JsonSerializer, JsonSerializerConfig, JsonSerializerOptions, LogfmtSerializer,
-    LogfmtSerializerConfig, NativeJsonSerializer, NativeJsonSerializerConfig, NativeSerializer,
-    NativeSerializerConfig, ProtobufSerializer, ProtobufSerializerConfig,
+    CefSerializer, CefSerializerConfig, CsvSerializer, CsvSerializerConfig, GelfSerializer,
+    GelfSerializerConfig, JsonSerializer, JsonSerializerConfig, JsonSerializerOptions,
+    LogfmtSerializer, LogfmtSerializerConfig, NativeJsonSerializer, NativeJsonSerializerConfig,
+    NativeSerializer, NativeSerializerConfig, ProtobufSerializer, ProtobufSerializerConfig,
     ProtobufSerializerOptions, RawMessageSerializer, RawMessageSerializerConfig, TextSerializer,
     TextSerializerConfig,
 };

lib/codecs/src/encoding/serializer.rs

@@ -6,6 +6,8 @@ use vector_core::{config::DataType, event::Event, schema};
 
 #[cfg(feature = "arrow")]
 use super::format::{ArrowStreamSerializer, ArrowStreamSerializerConfig};
+#[cfg(feature = "avro")]
+use super::format::{AvroSerializer, AvroSerializerConfig, AvroSerializerOptions};
 #[cfg(feature = "opentelemetry")]
 use super::format::{OtlpSerializer, OtlpSerializerConfig};
 #[cfg(feature = "parquet")]
@@ -15,8 +17,7 @@ use super::format::{SyslogSerializer, SyslogSerializerConfig};
 use super::{
     chunking::Chunker,
     format::{
-        AvroSerializer, AvroSerializerConfig, AvroSerializerOptions, CefSerializer,
-        CefSerializerConfig, CsvSerializer, CsvSerializerConfig, GelfSerializer,
+        CefSerializer, CefSerializerConfig, CsvSerializer, CsvSerializerConfig, GelfSerializer,
         GelfSerializerConfig, JsonSerializer, JsonSerializerConfig, LogfmtSerializer,
         LogfmtSerializerConfig, NativeJsonSerializer, NativeJsonSerializerConfig, NativeSerializer,
         NativeSerializerConfig, ProtobufSerializer, ProtobufSerializerConfig, RawMessageSerializer,
@@ -37,6 +38,7 @@ pub enum SerializerConfig {
     /// Encodes an event as an [Apache Avro][apache_avro] message.
     ///
     /// [apache_avro]: https://avro.apache.org/
+    #[cfg(feature = "avro")]
     Avro {
         /// Apache Avro-specific encoder options.
         avro: AvroSerializerOptions,
@@ -210,6 +212,7 @@ impl BatchSerializerConfig {
     }
 }
 
+#[cfg(feature = "avro")]
 impl From<AvroSerializerConfig> for SerializerConfig {
     fn from(config: AvroSerializerConfig) -> Self {
         Self::Avro { avro: config.avro }
@@ -287,6 +290,7 @@ impl SerializerConfig {
     /// Build the `Serializer` from this configuration.
     pub fn build(&self) -> Result<Serializer, Box<dyn std::error::Error + Send + Sync + 'static>> {
         match self {
+            #[cfg(feature = "avro")]
             SerializerConfig::Avro { avro } => Ok(Serializer::Avro(
                 AvroSerializerConfig::new(avro.schema.clone()).build()?,
             )),
@@ -327,7 +331,11 @@ impl SerializerConfig {
             // we should do so accurately, even if practically it doesn't need to be.
             //
             // [1]: https://avro.apache.org/docs/1.11.1/specification/_print/#message-framing
-            SerializerConfig::Avro { .. } | SerializerConfig::Native => {
+            #[cfg(feature = "avro")]
+            SerializerConfig::Avro { .. } => {
+                FramingConfig::LengthDelimited(LengthDelimitedEncoderConfig::default())
+            }
+            SerializerConfig::Native => {
                 FramingConfig::LengthDelimited(LengthDelimitedEncoderConfig::default())
             }
             #[cfg(feature = "opentelemetry")]
@@ -353,6 +361,7 @@ impl SerializerConfig {
     /// The data type of events that are accepted by this `Serializer`.
     pub fn input_type(&self) -> DataType {
         match self {
+            #[cfg(feature = "avro")]
             SerializerConfig::Avro { avro } => {
                 AvroSerializerConfig::new(avro.schema.clone()).input_type()
             }
@@ -376,6 +385,7 @@ impl SerializerConfig {
     /// The schema required by the serializer.
     pub fn schema_requirement(&self) -> schema::Requirement {
         match self {
+            #[cfg(feature = "avro")]
             SerializerConfig::Avro { avro } => {
                 AvroSerializerConfig::new(avro.schema.clone()).schema_requirement()
             }
@@ -401,6 +411,7 @@ impl SerializerConfig {
 #[derive(Debug, Clone)]
 pub enum Serializer {
     /// Uses an `AvroSerializer` for serialization.
+    #[cfg(feature = "avro")]
     Avro(AvroSerializer),
     /// Uses a `CefSerializer` for serialization.
     Cef(CefSerializer),
@@ -435,14 +446,15 @@ impl Serializer {
     pub fn supports_json(&self) -> bool {
         match self {
             Serializer::Json(_) | Serializer::NativeJson(_) | Serializer::Gelf(_) => true,
-            Serializer::Avro(_)
-            | Serializer::Cef(_)
+            Serializer::Cef(_)
             | Serializer::Csv(_)
             | Serializer::Logfmt(_)
             | Serializer::Text(_)
             | Serializer::Native(_)
             | Serializer::Protobuf(_)
             | Serializer::RawMessage(_) => false,
+            #[cfg(feature = "avro")]
+            Serializer::Avro(_) => false,
             #[cfg(feature = "syslog")]
             Serializer::Syslog(_) => false,
             #[cfg(feature = "opentelemetry")]
@@ -461,8 +473,7 @@ impl Serializer {
             Serializer::Gelf(serializer) => serializer.to_json_value(event),
             Serializer::Json(serializer) => serializer.to_json_value(event),
             Serializer::NativeJson(serializer) => serializer.to_json_value(event),
-            Serializer::Avro(_)
-            | Serializer::Cef(_)
+            Serializer::Cef(_)
             | Serializer::Csv(_)
             | Serializer::Logfmt(_)
             | Serializer::Text(_)
@@ -471,6 +482,10 @@ impl Serializer {
             | Serializer::RawMessage(_) => {
                 panic!("Serializer does not support JSON")
             }
+            #[cfg(feature = "avro")]
+            Serializer::Avro(_) => {
+                panic!("Serializer does not support JSON")
+            }
             #[cfg(feature = "syslog")]
             Serializer::Syslog(_) => {
                 panic!("Serializer does not support JSON")
@@ -496,10 +511,9 @@ impl Serializer {
     /// while text serializers produce UTF-8 encoded strings.
     pub const fn is_binary(&self) -> bool {
         match self {
-            Serializer::RawMessage(_)
-            | Serializer::Avro(_)
-            | Serializer::Native(_)
-            | Serializer::Protobuf(_) => true,
+            Serializer::RawMessage(_) | Serializer::Native(_) | Serializer::Protobuf(_) => true,
+            #[cfg(feature = "avro")]
+            Serializer::Avro(_) => true,
             #[cfg(feature = "opentelemetry")]
             Serializer::Otlp(_) => true,
             #[cfg(feature = "syslog")]
@@ -515,6 +529,7 @@ impl Serializer {
     }
 }
 
+#[cfg(feature = "avro")]
 impl From<AvroSerializer> for Serializer {
     fn from(serializer: AvroSerializer) -> Self {
         Self::Avro(serializer)
@@ -599,6 +614,7 @@ impl tokio_util::codec::Encoder<Event> for Serializer {
 
     fn encode(&mut self, event: Event, buffer: &mut BytesMut) -> Result<(), Self::Error> {
         match self {
+            #[cfg(feature = "avro")]
             Serializer::Avro(serializer) => serializer.encode(event, buffer),
             Serializer::Cef(serializer) => serializer.encode(event, buffer),
             Serializer::Csv(serializer) => serializer.encode(event, buffer),

lib/vector-lib/Cargo.toml

@@ -27,6 +27,7 @@ vrl = { workspace = true, optional = true }
 allocation-tracing = ["vector-top?/allocation-tracing"]
 api-client = ["dep:vector-api-client"]
 arrow = ["codecs/arrow"]
+avro = ["codecs/avro"]
 parquet = ["codecs/parquet"]
 api = ["vector-tap/api"]
 file-source = ["dep:file-source", "dep:file-source-common"]

src/components/validation/resources/mod.rs

@@ -168,6 +168,7 @@ fn deserializer_config_to_serializer(config: &DeserializerConfig) -> encoding::S
         DeserializerConfig::Native => SerializerConfig::Native,
         DeserializerConfig::NativeJson { .. } => SerializerConfig::NativeJson,
         DeserializerConfig::Gelf { .. } => SerializerConfig::Gelf(Default::default()),
+        #[cfg(feature = "codecs-avro")]
         DeserializerConfig::Avro { avro } => SerializerConfig::Avro { avro: avro.into() },
         // TODO: Influxdb has no serializer yet
         DeserializerConfig::Influxdb { .. } => todo!(),
@@ -218,6 +219,7 @@ fn serializer_config_to_deserializer(
     config: &SerializerConfig,
 ) -> vector_lib::Result<decoding::Deserializer> {
     let deserializer_config = match config {
+        #[cfg(feature = "codecs-avro")]
         SerializerConfig::Avro { .. } => todo!(),
         SerializerConfig::Cef { .. } => todo!(),
         SerializerConfig::Csv { .. } => todo!(),