Change partition_by to return a random partition instead of a nil value for a parition_by function

guides/Subscriptions.md

@@ -277,9 +277,10 @@ With multiple subscriber processes connected to a single subscription the orderi
 
 You can use a `partition_by` function to guarantee ordering of events within a particular group (e.g. per stream) but still allow events for different groups to be processed concurrently.
 
-
 Partitioning gives you the benefits of competing consumers but still allows event ordering by partition where required.
 
+If a `partition_by` returns `nil`, a random default partition based on the `max_size` will be generated for the partition id.
+
 #### Partition by example
 
 ```elixir

lib/event_store/subscriptions/subscription_fsm.ex

@@ -9,6 +9,9 @@ defmodule EventStore.Subscriptions.SubscriptionFsm do
 
   require Logger
 
+  @default_partition_prefix "$default."
+  @default_partition_fallback_max 10
+
   def new(stream_uuid, subscription_name, opts) do
     new(
       data: %SubscriptionState{
@@ -505,11 +508,44 @@ defmodule EventStore.Subscriptions.SubscriptionFsm do
     %SubscriptionState{data | partitions: partitions, queue_size: queue_size + 1}
   end
 
-  def partition_key(%SubscriptionState{partition_by: nil}, %RecordedEvent{}), do: nil
+  # Use nil partition when there's a lack of a partition_by callback since this way
+  # we maximize parallelization by having EventStore.Subscriptions.Subscriber.in_partition?/2
+  # return false for all events, making sure each event "seeks" a new available subscriber
+  def partition_key(%SubscriptionState{partition_by: nil}, %RecordedEvent{}),
+    do: nil
+
+  def partition_key(
+        %SubscriptionState{partition_by: partition_by, max_size: max_size},
+        %RecordedEvent{} = event
+      )
+      when is_function(partition_by, 1) do
+    case partition_by.(event) do
+      # Replace nil partition by a set of default partitions based on max_size
+      # in order to avoid exhausting subscribers for other partitions due
+      # to EventStore.Subscriptions.Subscriber.in_partition?/2 returning false
+      # for nil partition keys - resulting in the nil partition being able
+      # to exhaust partitions for all other partitions if it's being used to process
+      # a significantly high number of events compared to the other partitions due to
+      # it then always having the priority in scheduling due to having most events in queue
+      # and its events never reusing existing subscribers but always looking for new ones.
+      # This works well when `partition_by` is nil to maximize parallelization for all events
+      # but not ideal when sharing the scheduler with other partitions
+      nil ->
+        max_partition =
+          if is_integer(max_size) and max_size > 10 do
+            round(max_size / 10)
+          else
+            @default_partition_fallback_max
+          end
 
-  def partition_key(%SubscriptionState{partition_by: partition_by}, %RecordedEvent{} = event)
-      when is_function(partition_by, 1),
-      do: partition_by.(event)
+        partition_number = :rand.uniform(max_partition)
+
+        "#{@default_partition_prefix}#{partition_number}"
+
+      rest ->
+        rest
+    end
+  end
 
   # Attempt to notify subscribers with any pending events. Partitions are
   # selected by peeking at the event number of their queue to ensure earlier

test/subscriptions/partition_key_test.exs

@@ -0,0 +1,36 @@
+defmodule EventStore.Subscriptions.PartitionKeyTest do
+  use ExUnit.Case, async: true
+
+  alias EventStore.RecordedEvent
+  alias EventStore.Subscriptions.SubscriptionFsm
+  alias EventStore.Subscriptions.SubscriptionState
+
+  describe "partition_key/2" do
+    test "should return the value for a non-nil callback return" do
+      partition_key =
+        SubscriptionFsm.partition_key(
+          %SubscriptionState{partition_by: fn _ -> "some_key" end},
+          %RecordedEvent{}
+        )
+
+      assert partition_key == "some_key"
+    end
+
+    test "should return a random default partition for a nil callback return" do
+      partition_key =
+        SubscriptionFsm.partition_key(
+          %SubscriptionState{partition_by: fn _ -> nil end},
+          %RecordedEvent{}
+        )
+
+      assert String.match?(partition_key, ~r/^\$default\.[0-9]+$/)
+    end
+
+    test "should return nil for a nil callback" do
+      partition_key =
+        SubscriptionFsm.partition_key(%SubscriptionState{partition_by: nil}, %RecordedEvent{})
+
+      assert is_nil(partition_key)
+    end
+  end
+end