Eager Rebalancing

Eager Rebalancing

The Legacy Eager Rebalance Protocol

A rebalance moves partitions when members join, leave or die. Under the classic protocol every member sends JoinGroup, one member (the leader) runs the assignor, and SyncGroup hands the result out. With eager assignors such as range, each member first revokes all its partitions and stops processing until the new assignment arrives: a stop-the-world pause for the whole group. The script below starts consumer A, adds B after six seconds, and logs every revocation and assignment; its mode picks the protocol.

rebalance_demo.py: a second member joins under four configurationsPython

"""rebalance_demo.py MODE: A consumes alone; B joins at 6 s (static: B restarts at 14 s)."""
import sys, threading, time
from confluent_kafka import Consumer
MODE = sys.argv[1]                        # eager | cooperative | consumer | static
CONF = {"eager": {"partition.assignment.strategy": "range"},
        "cooperative": {"partition.assignment.strategy": "cooperative-sticky"},
        "consumer": {"group.protocol": "consumer"},
        "static": {"group.protocol": "consumer"}}[MODE]
GROUP, T0 = f"booknest-rebalance-{MODE}-{int(time.time())}", time.time()
log = lambda who, what, parts: print(f"{time.time() - T0:5.1f}s {who} {what:<8}",
                                     sorted(p.partition for p in parts), flush=True)
def member(name, stop_at, instance=None):
    c = Consumer({"bootstrap.servers": "localhost:33092", "group.id": GROUP,
                  "auto.offset.reset": "earliest", **CONF,
                  **({"group.instance.id": instance} if instance else {})})
    c.subscribe(["booknest.order-events"],
                on_assign=lambda c, p: log(name, "assigned", p),
                on_revoke=lambda c, p: log(name, "revoked", p))
    while time.time() - T0 < stop_at:
        c.consume(50, timeout=0.2)        # about 250 events a second: slow work
        time.sleep(0.2)
    print(f"{time.time() - T0:5.1f}s {name} leaves")
    c.close()
a = threading.Thread(target=member, args=("A", 24, "a" if MODE == "static" else None))
a.start(); time.sleep(6)
if MODE == "static":                      # B restarts quickly under the same instance ID
    member("B", 14, "b"); time.sleep(2); member("B", 22, "b")
else:
    member("B", 22)
a.join()

Output

  0.1s A assigned [0, 1, 2]
  6.2s A revoked  [0, 1, 2]
  6.2s A assigned [0, 1]
  6.4s B assigned [2]
...

A gave up all three partitions and got two back. On this two-member group the pause lasted milliseconds; in a group of hundreds, one slow member, or a member restoring large state, stalls everyone for seconds or minutes.

MENU