add batch config, frame size validation

Author: InvalidPathException

apps/kafkaesque_core/lib/kafkaesque/storage/single_file.ex

@@ -648,10 +648,20 @@ defmodule Kafkaesque.Storage.SingleFile do
   defp parse_frame_content(
          <<@magic_byte, crc::32, attr, timestamp::64, key_len::32, key::binary-size(key_len),
            val_len::32, value::binary-size(val_len), headers_len::16,
-           headers_data::binary-size(headers_len), _rest::binary>>
+           headers_data::binary-size(headers_len), rest::binary>> = frame_data
        ) do
-    # Validate sizes
+    # Calculate expected frame size
+    expected_size = 1 + 4 + 1 + 8 + 4 + key_len + 4 + val_len + 2 + headers_len
+    actual_size = byte_size(frame_data)
+
+    # Validate frame size matches expected size
     cond do
+      actual_size != expected_size ->
+        {:error, :frame_size_mismatch}
+
+      byte_size(rest) != 0 ->
+        {:error, :frame_has_trailing_data}
+
       key_len > Constants.max_key_size() ->
         {:error, :key_too_large}
 

apps/kafkaesque_core/lib/kafkaesque/topic/supervisor.ex

@@ -21,13 +21,13 @@ defmodule Kafkaesque.Topic.Supervisor do
   @doc """
   Creates a new topic with the specified number of partitions.
   """
-  def create_topic(name, partitions \\ 1) when is_binary(name) and partitions > 0 do
+  def create_topic(name, partitions \\ 1, opts \\ []) when is_binary(name) and partitions > 0 do
     Logger.debug("Creating topic '#{name}' with #{partitions} partitions")
 
     # Start supervisor for each partition
     results =
       Enum.map(0..(partitions - 1), fn partition ->
-        start_partition(name, partition)
+        start_partition(name, partition, opts)
       end)
 
     # Check if all partitions started successfully
@@ -167,11 +167,11 @@ defmodule Kafkaesque.Topic.Supervisor do
     |> Enum.any?()
   end
 
-  defp start_partition(topic, partition) do
+  defp start_partition(topic, partition, opts) do
     child_spec = %{
       id: {topic, partition},
       start:
-        {__MODULE__.PartitionSupervisor, :start_link, [[topic: topic, partition: partition]]},
+        {__MODULE__.PartitionSupervisor, :start_link, [Keyword.merge([topic: topic, partition: partition], opts)]},
       type: :supervisor,
       restart: :permanent
     }
@@ -237,10 +237,10 @@ defmodule Kafkaesque.Topic.Supervisor.PartitionSupervisor do
        [
          topic: topic,
          partition: partition,
-         batch_size: Application.get_env(:kafkaesque_core, :max_batch_size, 500),
-         batch_timeout: Application.get_env(:kafkaesque_core, :batch_timeout, 5000),
-         min_demand: Application.get_env(:kafkaesque_core, :min_demand, 5),
-         max_demand: Application.get_env(:kafkaesque_core, :max_batch_size, 500)
+         batch_size: Keyword.get(opts, :batch_size, Application.get_env(:kafkaesque_core, :max_batch_size, 500)),
+         batch_timeout: Keyword.get(opts, :batch_timeout, Application.get_env(:kafkaesque_core, :batch_timeout, 5000)),
+         min_demand: Keyword.get(opts, :min_demand, Application.get_env(:kafkaesque_core, :min_demand, 5)),
+         max_demand: Keyword.get(opts, :max_demand, Application.get_env(:kafkaesque_core, :max_batch_size, 500))
        ]},
 
       # Log reader for consuming

apps/kafkaesque_core/test/topic/batch_config_test.exs

@@ -0,0 +1,77 @@
+defmodule Kafkaesque.Topic.BatchConfigTest do
+  use ExUnit.Case, async: false
+  alias Kafkaesque.Topic.Supervisor, as: TopicSupervisor
+
+  setup do
+    on_exit(fn ->
+      # Clean up any test topics
+      TopicSupervisor.list_topics()
+      |> Enum.each(fn
+        {topic, _} when is_binary(topic) ->
+          if String.starts_with?(topic, "batch_config_test") do
+            TopicSupervisor.delete_topic(topic)
+          end
+        %{name: topic} when is_binary(topic) ->
+          if String.starts_with?(topic, "batch_config_test") do
+            TopicSupervisor.delete_topic(topic)
+          end
+        _ ->
+          :ok
+      end)
+    end)
+
+    :ok
+  end
+
+  describe "per-topic batch configuration" do
+    test "create_topic accepts batch configuration options" do
+      topic = "batch_config_test_#{System.unique_integer([:positive])}"
+
+      # Create topic with custom batch settings
+      opts = [
+        batch_size: 100,
+        batch_timeout: 1000,
+        min_demand: 10,
+        max_demand: 200
+      ]
+
+      assert {:ok, _} = TopicSupervisor.create_topic(topic, 1, opts)
+
+      # Verify the topic was created
+      topics = TopicSupervisor.list_topics()
+      assert Enum.any?(topics, fn
+        {name, _} -> name == topic
+        %{name: name} -> name == topic
+      end)
+
+      # Get the BatchingConsumer process via Registry
+      [{consumer_pid, _}] = Registry.lookup(Kafkaesque.TopicRegistry, {:batching_consumer, topic, 0})
+      assert consumer_pid != nil
+
+      # Verify the BatchingConsumer state has the correct config
+      gen_stage_state = :sys.get_state(consumer_pid)
+      consumer_state = gen_stage_state.state
+      assert consumer_state.batch_size == 100
+      assert consumer_state.batch_timeout == 1000
+    end
+
+    test "create_topic uses defaults when no options provided" do
+      topic = "batch_config_test_default_#{System.unique_integer([:positive])}"
+
+      assert {:ok, _} = TopicSupervisor.create_topic(topic, 1)
+
+      # Get the BatchingConsumer process via Registry
+      [{consumer_pid, _}] = Registry.lookup(Kafkaesque.TopicRegistry, {:batching_consumer, topic, 0})
+      assert consumer_pid != nil
+
+      # Verify defaults are used
+      gen_stage_state = :sys.get_state(consumer_pid)
+      consumer_state = gen_stage_state.state
+      default_batch_size = Application.get_env(:kafkaesque_core, :max_batch_size, 500)
+      default_timeout = Application.get_env(:kafkaesque_core, :batch_timeout, 5000)
+
+      assert consumer_state.batch_size == default_batch_size
+      assert consumer_state.batch_timeout == default_timeout
+    end
+  end
+end