Add Kafka LZ4 producer compression support

Adds the lz4b NIF dependency and wires up configuration tos set brod's
default_producer_config with lz4 compression.

Author: RTLS

assets/svelte/consumers/types.ts

@@ -176,6 +176,7 @@ export type KafkaConsumer = BaseConsumer & {
     tls: boolean;
     topic: string;
     sasl_mechanism: null | "plain" | "scram_sha_256" | "scram_sha_512";
+    producer_compression: "no_compression" | "lz4";
   };
 };
 

assets/svelte/sinks/kafka/KafkaSinkCard.svelte

@@ -53,6 +53,23 @@
           >
         </div>
       </div>
+
+      <div>
+        <span class="text-sm text-muted-foreground">Producer Compression</span>
+        <div class="mt-2">
+          <span
+            class="font-mono bg-slate-50 py-1 px-2 border border-slate-100 rounded-md whitespace-nowrap"
+          >
+            {#if consumer.sink.producer_compression === "lz4"}
+              LZ4
+            {:else if consumer.sink.producer_compression === "no_compression"}
+              None
+            {:else}
+              {consumer.sink.producer_compression}
+            {/if}
+          </span>
+        </div>
+      </div>
     </div>
   </CardContent>
 </Card>

assets/svelte/sinks/kafka/KafkaSinkForm.svelte

@@ -236,6 +236,27 @@
       </p>
     </div>
 
+    <!-- Producer Compression -->
+    <div class="space-y-2">
+      <Label for="producer_compression">Producer Compression</Label>
+      <select
+        id="producer_compression"
+        bind:value={form.sink.producer_compression}
+        class="block w-full border border-gray-300 rounded-md p-2"
+      >
+        <option value="no_compression">None</option>
+        <option value="lz4">LZ4</option>
+      </select>
+      <p class="text-sm text-muted-foreground">
+        Compression algorithm for Kafka producer messages.
+      </p>
+      {#if errors.sink?.producer_compression}
+        <p class="text-destructive text-sm">
+          {errors.sink.producer_compression}
+        </p>
+      {/if}
+    </div>
+
     <!-- TLS Switch -->
     <div class="space-y-2">
       <div class="flex items-center gap-2">

config/runtime.exs

@@ -84,17 +84,17 @@ config :sequin, Sequin.Runtime.HttpPushSqsPipeline,
 config :sequin, Sequin.Runtime.SlotProcessorServer,
   max_accumulated_bytes: ConfigParser.replication_flush_max_accumulated_bytes(env_vars),
   max_accumulated_messages: ConfigParser.replication_flush_max_accumulated_messages(env_vars),
+  # ## Using releases
+  #
+  # If you use `mix release`, you need to explicitly enable the server
+  # by passing the PHX_SERVER=true when you start it:
+  #
+  #     PHX_SERVER=true bin/sequin start
+  #
+  # Alternatively, you can use `mix phx.gen.release` to generate a `bin/server`
+  # script that automatically sets the env var above.
   max_accumulated_messages_time_ms: ConfigParser.replication_flush_max_accumulated_time_ms(env_vars)
 
-# ## Using releases
-#
-# If you use `mix release`, you need to explicitly enable the server
-# by passing the PHX_SERVER=true when you start it:
-#
-#     PHX_SERVER=true bin/sequin start
-#
-# Alternatively, you can use `mix phx.gen.release` to generate a `bin/server`
-# script that automatically sets the env var above.
 if System.get_env("PHX_SERVER") do
   config :sequin, SequinWeb.Endpoint, server: true
   config :sequin, SequinWeb.MetricsEndpoint, server: true

docs/reference/sequin-yaml.mdx

@@ -363,6 +363,7 @@ destination:
   username: "kafka-user"        # Optional, for SASL authentication
   password: "kafka-pass"        # Optional, for SASL authentication
   sasl_mechanism: "plain"       # Optional: plain, scram_sha_256, scram_sha_512
+  producer_compression: "lz4"   # Optional: no_compression (default), lz4
   batch_size: 50                # Optional, messages per batch
 ```
 

lib/sequin/consumers/kafka_sink.ex

@@ -25,6 +25,7 @@ defmodule Sequin.Consumers.KafkaSink do
     field :aws_secret_access_key, EncryptedField
     field :connection_id, :string
     field :routing_mode, Ecto.Enum, values: [:dynamic, :static]
+    field :producer_compression, Ecto.Enum, values: [:no_compression, :lz4], default: :no_compression
   end
 
   def changeset(struct, params) do
@@ -39,7 +40,8 @@ defmodule Sequin.Consumers.KafkaSink do
       :aws_region,
       :aws_access_key_id,
       :aws_secret_access_key,
-      :routing_mode
+      :routing_mode,
+      :producer_compression
     ])
     |> validate_required([:hosts, :tls])
     |> validate_routing()
@@ -177,6 +179,7 @@ defmodule Sequin.Consumers.KafkaSink do
     |> maybe_add_ssl(sink)
     |> Keyword.put(:query_api_versions, true)
     |> Keyword.put(:auto_start_producers, true)
+    |> Keyword.put(:default_producer_config, compression: sink.producer_compression)
   end
 
   # Add SASL authentication if username/password are configured

lib/sequin/transforms/transforms.ex

@@ -226,7 +226,8 @@ defmodule Sequin.Transforms do
       sasl_mechanism: sink.sasl_mechanism,
       aws_access_key_id: sink.aws_access_key_id,
       aws_secret_access_key: SensitiveValue.new(sink.aws_secret_access_key, show_sensitive),
-      aws_region: sink.aws_region
+      aws_region: sink.aws_region,
+      producer_compression: sink.producer_compression
     })
   end
 
@@ -1142,7 +1143,8 @@ defmodule Sequin.Transforms do
          sasl_mechanism: sasl_mechanism,
          aws_access_key_id: attrs["aws_access_key_id"],
          aws_secret_access_key: attrs["aws_secret_access_key"],
-         aws_region: attrs["aws_region"]
+         aws_region: attrs["aws_region"],
+         producer_compression: attrs["producer_compression"]
        }}
     end
   end

lib/sequin_web/live/components/consumer_form.ex

@@ -820,7 +820,8 @@ defmodule SequinWeb.Components.ConsumerForm do
       "aws_access_key_id" => sink["aws_access_key_id"],
       "aws_secret_access_key" => sink["aws_secret_access_key"],
       "aws_region" => sink["aws_region"],
-      "batch_size" => sink["batch_size"]
+      "batch_size" => sink["batch_size"],
+      "producer_compression" => sink["producer_compression"]
     }
   end
 
@@ -1090,7 +1091,8 @@ defmodule SequinWeb.Components.ConsumerForm do
       "sasl_mechanism" => sink.sasl_mechanism,
       "aws_access_key_id" => sink.aws_access_key_id,
       "aws_secret_access_key" => sink.aws_secret_access_key,
-      "aws_region" => sink.aws_region
+      "aws_region" => sink.aws_region,
+      "producer_compression" => sink.producer_compression
     }
   end
 
@@ -1480,7 +1482,7 @@ defmodule SequinWeb.Components.ConsumerForm do
         :sns -> {%SnsSink{}, %{batch_size: 10}}
         :kinesis -> {%KinesisSink{}, %{batch_size: 100}}
         :s2 -> {%S2Sink{}, %{batch_size: 10}}
-        :kafka -> {%KafkaSink{tls: false}, %{batch_size: 200}}
+        :kafka -> {%KafkaSink{tls: false, producer_compression: :no_compression}, %{batch_size: 200}}
         :redis_stream -> {%RedisStreamSink{}, %{batch_size: 50}}
         :sequin_stream -> {%SequinStreamSink{}, %{}}
         :gcp_pubsub -> {%GcpPubsubSink{}, %{message_grouping: false, batch_size: 100}}

lib/sequin_web/live/sink_consumers/show.ex

@@ -904,7 +904,8 @@ defmodule SequinWeb.SinkConsumersLive.Show do
       password: sink.password,
       topic: sink.topic,
       tls: sink.tls,
-      sasl_mechanism: sink.sasl_mechanism
+      sasl_mechanism: sink.sasl_mechanism,
+      producer_compression: sink.producer_compression
     }
   end
 

mix.exs

@@ -102,6 +102,7 @@ defmodule Sequin.MixProject do
       {:amqp, "~> 4.1"},
       {:amqp_client, "~> 4.2"},
       {:brod, "~> 4.3"},
+      {:lz4b, "~> 0.0.13"},
 
       # Caching and State Management
       {:con_cache, "~> 1.1"},