Kafka Sink#

Sequin.Consumers.KafkaSink is the embedded Ecto schema that represents a Kafka sink's configuration. It is used as a nested struct within a SinkConsumer and drives both connection setup and runtime message publishing via the brod Erlang client.

Schema & Fields#

Defined in lib/sequin/consumers/kafka_sink.ex using TypedEctoSchema with @primary_key false.

password and aws_secret_access_key are stored as EncryptedField and are excluded from Inspect output 1.

Changeset & Validation#

KafkaSink.changeset/2 casts all fields except connection_id (which is auto-generated), then enforces:

• Required: :hosts, :tls 2
• Routing: :dynamic → sets topic to nil; :static → topic required; absent → error 3
• Hosts format: comma-separated host:port, ports 1–65535 4
• SASL credentials: plain/scram require username + password; :aws_msk_iam requires key/secret/region and tls: true 5
• connection_id: auto-generated UUID if nil 6

Runtime: brod Client Startup#

KafkaSink.to_brod_config/1 converts the struct into a keyword list for :brod.start_client/3. Key options set:

• :query_api_versions true, :auto_start_producers true
• :default_producer_config [compression: brod_compression(sink.compression)]
• :sasl — added by maybe_add_sasl/2 when a mechanism is set 7
• :ssl true — added by maybe_add_ssl/2 when tls: true 8

The brod client is started and cached by Sequin.Sinks.Kafka.ConnectionCache 9. The cache uses an options hash (including to_brod_config output) to detect configuration drift and restart stale connections 10.

Known Issue: brod_compression/1 — No nil Clause#

brod_compression/1 maps the :none atom to :no_compression (brod's expected value) and passes all other atoms through unchanged:

defp brod_compression(:none), do: :no_compression
defp brod_compression(other), do: other # catches :gzip, :snappy, :lz4, :zstd — and also nil

There is no explicit nil clause. If sink.compression is nil, the second clause matches and passes nil to brod's :default_producer_config. This does not raise an Elixir-level exception, but it surfaces as a brod-level error during :brod.start_client/3, causing the producer to crash-loop without a clear error message pointing back to the compression field.

Known Issue: YAML Config Defaults Bypassed for Optional Fields#

When a Kafka sink is created from YAML config, parse_sink/2 in lib/sequin/transforms/transforms.ex builds the params map including explicit nil values for every optional field not present in the YAML:

%{
  type: :kafka,
  hosts: attrs["hosts"],
  tls: attrs["tls"] || false, # ← only field with nil-coalescing
  compression: attrs["compression"] # ← nil when key is absent from YAML
  # ... all other optional fields also nil when absent
}

When this map is passed to KafkaSink.changeset/2, Ecto's cast/3 sees explicit nil values for these keys and writes them to the changeset — overriding the schema-level defaults (compression: :none, tls: false). Ecto only applies schema defaults when a key is absent from the params map; explicit nil always wins.

Effect: A YAML-configured sink without a compression: key will store compression: nil instead of :none. When the producer starts, this nil propagates to brod_compression/1 and then to brod, causing the crash-loop described above.

Fix pattern: Either apply || :none in parse_sink for each defaulted field (matching the tls: attrs["tls"] || false pattern already used), or filter nil values from the params map before casting.

Key Files#