Extract DataStreamSpec and move configs to separate structs

[realglebivanov]

No description provided.

[dbegunkov]

Why is it called PipelineUnit?

[realglebivanov]

I named it PipelineUnit because this struct represents the unit of work that drives a single batching pipeline - it bundles together all of the configs relevant for one topic/partition pipeline (collector, accumulator config, producer config, etc.).
It’s meant to be the single coherent entity that the supervisor and worker processes use to start and configure all the pieces involved in a pipeline.

[dbegunkov]

I suggest calling the ConfigEntity or maybe ConfigUnit, if the entity is intended for config storage

[realglebivanov]

I found options like DataFlowSpec, StreamConfig, DataStreamSpec
What's your opinion?

[dbegunkov]

I think, DataStreamSpec is sound more relevant for this entity.

[realglebivanov]

@romul Please, have a look at this draft

[romul]

It could be just config |> Enum.into([])

[realglebivanov]

@romul @dbegunkov

Branch Review: scoped_config

Improvements

1. Typed config structs with @enforce_keys — Producers.Config, Collector.Config, Accumulator.Config, BrodConfig replace untyped keyword lists. Construction-time validation via struct!. Strong typespecs throughout.

2. DataStreamSpec bundles per-stream config — Clean single-struct-per-data-stream design. Eliminates repeated Keyword.fetch! from keyword bags across modules.

3. Scoped process naming via client_name — All process names now include client_name (Supervisor, ConnectionManager, AccumulatorsPoolSupervisor, Accumulator). Enables multiple independent KafkaBatcher trees in the same VM.

4. Config.build_config!/1 as single entry point — Linear pipeline replacing the scattered config-assembly functions. Easy to follow: build producer config, build data stream specs per collector, validate.

5. Producer behaviour gets config as first arg — Consistent signature across start_client/1, start_producer/2, do_produce/4, get_partitions_count/2. No more hidden Application.get_env inside producers.

6. drop_sensitive/1 chain + @derive {Inspect, only: ...} — Structured sensitive-data stripping for format_status and logging. Two layers: Inspect derive hides :sasl from casual inspection, drop_sensitive scrubs for format_status.

7. SASL validation returns tagged tuples — BrodConfig.build_sasl_config/1 is {:ok, _} | {:error, _} internally, raising only at the caller's choice.

8. URI.parse for endpoint parsing — lib/kafka_batcher/producers/config.ex:87 — handles IPv6, whitespace, etc. better than String.split(url, ":").

9. DataStreamSpec.Validator — Focused module with clear cond-based validation. Easy to extend.

10. Supervisor.child_spec/1 backwards compatibility — Accepts Config.t(), Keyword.t(), or nil. The nil/empty-list path reads Application.get_all_env(:kafka_batcher) for backwards compat.

11. max_accumulator_restarts made configurable — Previously hardcoded max_restart: 100 read from kwlist, now a proper field on Accumulator.Config with default.

12. PromEx plugin accepts config via opts — lib/kafka_batcher/prom_ex/plugins/kafka.ex:65 — Keyword.get_lazy(opts, :kafka_batcher_config, ...) allows passing config directly, avoiding Application.get_all_env in tests.

13. Metric prefix scoped by client_name — lib/kafka_batcher/prom_ex/plugins/kafka.ex:72 — [:prom_ex, :kafka, client_name] prevents metric collisions when running multiple clients.

---

Issues

Bug

1. set_partition accepts nil at runtime but spec says pos_integer() — lib/kafka_batcher/data_stream_spec.ex:59: @spec set_partition(t(), pos_integer()). But lib/kafka_batcher/collector/state.ex:68 calls DataStreamSpec.set_partition(state.data_stream_spec, partition) where partition comes from choose_partition, which returns {:ok, nil} for non-partitioned streams. So set_partition is called with nil in the non-partitioned path. This doesn't crash (it just sets partition: nil), but it's a spec violation that Dialyzer with extra_return/missing_return may flag.

2. Collector.handle_call/3 for unknown messages returns no reply — lib/kafka_batcher/collector.ex:147-152: the catch-all handle_call(unknown, _from, state) calls @error_notifier.report(...) but has no explicit return value. It returns whatever report/1 returns, which is not a valid handle_call return tuple. The caller will get a crash or a bad reply.

3. BrodConfig.to_kwlist emits required_acks and sasl that brod doesn't recognize as client config options — lib/kafka_batcher/producers/config/brod_config.ex:36-43. start_link_client and start_producer in lib/kafka_batcher/producers/kaffe.ex:21,30 both pass BrodConfig.to_kwlist(config.brod_config) directly. Brod client config accepts ssl, sasl, allow_topic_auto_creation — but required_acks is a producer config option, not a client config option. Passing it to start_link_client is harmless (brod ignores unknown keys) but misleading. More importantly, brod's start_producer/3 config doesn't accept sasl/ssl/allow_topic_auto_creation — those are client-level. Passing the full BrodConfig kwlist to both is imprecise.

Design Concerns

4. DataStreamSpec.opts is the only untyped escape hatch, used for TempStorage.Batch.producer_config — lib/kafka_batcher/accumulator.ex:173 and lib/kafka_batcher/collector/state.ex:118. The entire refactoring moves to typed structs, but TempStorage still receives a keyword list. This is the seam that prevents opts from being removed.

5. required_acks is duplicated in Producers.Config and BrodConfig — Both populated from same opts, both emitted by to_kwlist. If one is overridden without the other, they diverge silently. KafkaEx.do_produce reads config.required_acks from Producers.Config, while brod gets it from BrodConfig.to_kwlist. Consider having one be the source of truth.

6. partition_strategy defaults to nil — lib/kafka_batcher/producers/config.ex:66: Keyword.get(opts, :partition_strategy). Then lib/kafka_batcher/producers/base.ex:16 does config.partition_strategy || :random. The default is implicit at the use site rather than at construction. Also, the validator at lib/kafka_batcher/data_stream_spec/validator.ex:23 rejects nil partition_strategy when collect_by_partition is false — so it can't actually stay nil for a working config. The nil default in build! is a bit misleading since it's only valid when collect_by_partition: true.

7. topic_name and collector_mod are duplicated across Collector.Config and Accumulator.Config — Both structs carry collector_mod and topic_name. DataStreamSpec then has accessors that read topic_name from collector_config specifically, while Accumulator reads it from accumulator_config directly (lib/kafka_batcher/accumulator.ex:171). The data is the same but lives in two places.

8. README config example is inconsistent — README.md lines 39-44 show a list-of-tuples format {KafkaBatcher.Collector1, [topic_name: "topic1"]} but lines 48-65 show a keyword-list with :collectors key. These are two different formats. Config.build_config! only handles the keyword-list format (it does Keyword.fetch!(opts, :collectors)). The tuple format at lines 39-44 doesn't seem to be supported by the code — it would need custom parsing.

Minor

9. @moduledoc false placed after defmodule SASLConfigError — lib/kafka_batcher/producers/config/brod_config.ex:10: the @moduledoc is below the nested defmodule. It still works, but reads oddly.

10. Accumulator.Config.build! doesn't take :partition — lib/kafka_batcher/accumulator/config.ex:50: Keyword.take list doesn't include :partition. This is intentional (partition is set later via DataStreamSpec.set_partition), but the field exists on the struct with a default of nil, which could confuse someone adding partition to opts and wondering why it's ignored.

11. build_sasl_config guard is_binary + body String.valid? is redundant — lib/kafka_batcher/producers/config/brod_config.ex:74,77: the guard when is_binary(login) and is_binary(password) already ensures they're binaries. String.valid?/1 then checks for valid UTF-8, which is a stricter check. If the intent is UTF-8 validation, the guard could just be the function clause head match without the redundant is_binary in the guard. If UTF-8 isn't required (SASL credentials can be arbitrary bytes), the String.valid? check is wrong.