[SuperStream] Rebalance consumers in single-active mode after a TCP client reconnection

[Gsantomaggio]

A group of consumers in SAC could be unbalanced after a TCP consumer reconnection.

Given the following configuration:

• RabbitMQ 3.13.1
• Erlang 26.2.2
• One node
• Super Stream with three partition
• A stream client (I tried with GO, Java and .NET) with three instances

Run three processes. Each process handles one partition, like:

Consumer1:

 [14:18:55] - [ partition: invoices-0] consumer name: MyApplication, data: [hello_super_stream_4607], message offset 8174,
 [14:18:56] - [ partition: invoices-0] consumer name: MyApplication, data: [hello_super_stream_4609], message offset 8176,
 [14:18:57] - [ partition: invoices-0] consumer name: MyApplication, data: [hello_super_stream_4610], message offset 8178,
 [14:18:57] - [ partition: invoices-0] consumer name: MyApplication, data: [hello_super_stream_4611], message offset 8180,

Consumer2:

 [14:17:20] - [ partition: invoices-1] consumer name: MyApplication, data: [hello_super_stream_4418], message offset 8327,
 [14:17:21] - [ partition: invoices-1] consumer name: MyApplication, data: [hello_super_stream_4419], message offset 8329,
 [14:17:21] - [ partition: invoices-1] consumer name: MyApplication, data: [hello_super_stream_4420], message offset 8331,
 [14:17:22] - [ partition: invoices-1] consumer name: MyApplication, data: [hello_super_stream_4421], message offset 8333,

Consumer3:

 [14:19:11] - [ partition: invoices-2] consumer name: MyApplication, data: [hello_super_stream_4638], message offset 8293,
 [14:19:12] - [ partition: invoices-2] consumer name: MyApplication, data: [hello_super_stream_4640], message offset 8295,
 [14:19:12] - [ partition: invoices-2] consumer name: MyApplication, data: [hello_super_stream_4641], message offset 8297,
 [14:19:13] - [ partition: invoices-2] consumer name: MyApplication, data: [hello_super_stream_4642], message offset 8299,
 [14:19:14] - [ partition: invoices-2] consumer name: MyApplication, data: [hello_super_stream_4645], message offset 8301,
 [14:19:15] - [ partition: invoices-2] consumer name: MyApplication, data: [hello_super_stream_4646], message offset 8303,

Everything ok!

Now force-close one single-active TCP connection, for example, the one on invoices-0
Consumer1:

 [14:19:50] - [ partition: invoices-0] consumer name: MyApplication, data: [hello_super_stream_4717], message offset 8256,
 [14:19:54] - [ partition: invoices-0] consumer name: MyApplication, data: [hello_super_stream_4724], message offset 8258,
 [14:20:00] - [ partition: invoices-0] consumer name: MyApplication, data: [hello_super_stream_4736], message offset 8260,
 [14:20:01] - Partition invoices-0 closed unexpectedly! Reconnecting in 4 seconds..
 
 [14:20:05] - Partition invoices-0 reconnected.

After Consumer1 is closed here [14:20:01] - Partition invoices-0 closed another consumer handles the partition invoices-0
Consumer2:

 [14:20:04] - [ partition: invoices-0] consumer name: MyApplication, data: [hello_super_stream_4744], message offset 8268,
 [14:20:04] - [ partition: invoices-1] consumer name: MyApplication, data: [hello_super_stream_4745], message offset 8543,
 [14:20:05] - [ partition: invoices-0] consumer name: MyApplication, data: [hello_super_stream_4746], message offset 8270,
 [14:20:06] - [ partition: invoices-1] consumer name: MyApplication, data: [hello_super_stream_4748], message offset 8545,
 [14:20:07] - [ partition: invoices-1] consumer name: MyApplication, data: [hello_super_stream_4751], message offset 8547,

Everything ok!

---

When Consumer1 is reconnected here: [14:20:02]—Partition invoices-0 reconnected. does not receive the consumer update event and so does not consume the messages.

That's cause the unbalanced consumers.

As result, there are two active consumers:
Consumer2 with two partitions: invoices-0 and invoices-1
Consumer3 with one partition: invoices-2

Consumer1 with ZERO active partitions

Per conversation with @acogoluegnes

[acogoluegnes]

This works as expected WRT the rebalancing algorithm for super streams. This may not be ideal and may be improved. Here is how it works.

The active consumer for a given partition is calculated with the following formula: consumer index = partition index % consumer count.

Let's focus on invoices-0. It's the first partition, so its partition index is 0, per creation. There are 3 instances of the same application (Consumer1, Consumer2, and Consumer3). Each has an actual consumer registered to invoices-0, so that's consumer count = 3. The consumers are queued in their registration order, so Consumer1 (started first) has consumer index = 0, Consumer2 (started second) has consumer index = 1, and Consumer3 (started last) has consumer index = 2.

If we apply the formula partition index % consumer count = 0 % 3 = 0, the consumer with index 0 is active, so that's Consumer1. This is what we have before closing the connection.

When we close the active connection for Consumer1, it's removed from the list of consumers for this partition, so Consumer2 gets consumer index = 0 and Consumer3 gets consumer index = 1. If we apply the formula again, we get partition index % consumer count = 0 % 2 = 0, so the consumer with index 0 becomes active, that's Consumer2, as observed.

When the connection for invoices-0 in Consumer1 is restored, it gets consumer index = 2 (it's just added at the end of the consumer list for this partition). Formula again: partition index % consumer count = 0 % 3 = 0, consumer with index 0 stays active. This is what we observe.

We could expect to go back to the original state, but that's quite tricky actually. This would also involve an extra rebalancing operation (Consumer2 gets activated, then Consumer1 gets activated again when it gets back online for the partition).

Plus the load is not balanced across the instances, but it is still possible to scale the processing for 2 partitions in the same instance by using multi-threading (one thread for each partition, to keep message ordering).

Any suggestion for an improvement that would keep the state balanced and would not make the rebalancing more complicated is welcome.