SELF-HEALING A MESSAGE BROKERING CLUSTER
A system, apparatus, and methods are provided for self-healing and balancing partition distribution across nodes within a message broker cluster. During operation, the system receives a stream of messages at the message brokering cluster, wherein the message stream is divided into partitions and replicas for each partition are distributed among a set of nodes within the message brokering cluster. Responsive to a change in the number of nodes within the message brokering cluster, the system (1) determines a set of replicas to be migrated within the message brokering cluster, (2) divides the set of replicas into multiple chunks, wherein each chunk includes one or more of the replicas to be migrated to a new node, and (3) migrates the set of replicas one chunk at a time, wherein replicas not corresponding to the single chunk do not begin migrating until all replicas within the single chunk finish migrating.
1 . A method, comprising:
receiving a stream of messages at a message brokering cluster, wherein the message stream is divided into partitions and replicas for each partition are distributed among a set of nodes within the message brokering cluster; and
responsive to a change in a number of nodes within the message brokering cluster:
determining a set of replicas to be migrated within the message brokering cluster;
dividing the set of replicas into multiple chunks, wherein each chunk corresponds to one or more of the replicas to be migrated and each replica in the set of replicas is mapped to a node to which the replica is to be migrated; and
migrating the set of replicas a single chunk at a time, wherein replicas not corresponding to the single chunk do not begin migrating until the replicas within the single chunk finish migrating.
2 . The method of claim 1 , wherein:
the replicas for each partition comprise a leader replica that handles read and write requests for the partition and one or more follower replicas that each replicate the leader replica;
the replicas for each partition are assigned to different nodes; and
migrating a follower replica comprises:
reassigning the replica to a new node; and
replicating, to the replica, an amount of data from a leader replica, wherein the amount of data would enable the replica to replace the leader replica if the leader replica's node became unreachable.
3 . The method of claim 2 , wherein migrating the set of replicas a single chunk at a time comprises:
for each of the multiple chunks:
publishing the mappings of the chunk's corresponding replicas to the message brokering cluster, thereby enabling the message brokering cluster to migrate the replicas accordingly; and
waiting for the replicas within the chunk to finish migrating before starting to migrate the replicas within the next chunk in the multiple chunks.
4 . The method of claim 2 , wherein:
the change in the number of nodes in the message brokering cluster is caused by at least one node becoming unreachable; and
responsive to the at least one node becoming unreachable but prior to determining the set of replicas to be migrated, the method further comprises waiting for a threshold period of time, wherein the determination of the set of replicas and the migration of the set of replicas is performed when the at least one node remains unreachable after the threshold period of time expires.
5 . The method of claim 2 , wherein the method further comprises halting the migration of the set of replicas in response to the at least one node becoming reachable before the migration finishes.
6 . The method of claim 5 , wherein the method further comprises reassigning the reassigned replicas back to the nodes they were assigned to prior to the migration.
7 . The method of claim 2 , wherein:
the change in the number of nodes in the message brokering cluster is caused by at least one node joining the message brokering cluster; and
determining the set of replicas to be migrated comprises:
determining, for each of one or more nodes within the set of nodes, a volume of incoming data associated with the replicas assigned to the node; and
selecting, based on the determined volume of incoming data, one or more replicas to be reassigned to the at least one node.
8 . The method of claim 7 , wherein the selection of the one or more replicas is further based on a disk usage for each of the one or more nodes.
9 . The method of claim 1 , wherein:
the message broker cluster corresponds to a Kafka cluster; and
the stream of messages corresponds to a Kafka topic.
10 . The method of claim 1 , wherein each message comprises data formatted in one of:
JavaScript Object Notation; and
Avro.
11 . An apparatus, comprising:
one or more processors; and
memory storing instructions that, when executed by the one or more processors, cause the apparatus to:
receive a stream of messages at a message brokering cluster, wherein the message stream is divided into partitions and replicas for each partition are distributed among a set of nodes within the message brokering cluster; and
responsive to a change in a number of nodes within the message brokering cluster:
determine a set of replicas to be migrated within the message brokering cluster;
divide the set of replicas into multiple chunks, wherein each chunk corresponds to one or more of the replicas to be migrated and each replica in the set of replicas is mapped to a node to which the replica is to be migrated; and
migrate the set of replicas a single chunk at a time, wherein replicas not corresponding to the single chunk do not begin migrating until the replicas within the single chunk finish migrating.
12 . The apparatus of claim 11 , wherein:
the replicas for each partition comprise a leader replica that handles read and write requests for the partition and one or more follower replicas that each replicate the leader replica;
the replicas for each partition are assigned to different nodes; and
migrating a follower replica comprises:
reassigning the replica to a new node; and
replicating, to the replica, an amount of data from a leader replica, wherein the amount of data would enable the replica to replace the leader replica if the leader replica's node became unreachable.
13 . The apparatus of claim 12 , wherein migrating the set of replicas a single chunk at a time comprises:
for each of the multiple chunks:
publishing the mappings of the chunk's corresponding replicas to the message brokering cluster, thereby enabling the message brokering cluster to migrate the replicas accordingly; and
waiting for the replicas within the chunk to finish migrating before starting to migrate the replicas within the next chunk in the multiple chunks.
14 . The apparatus of claim 12 , wherein:
the change in the number of nodes in the message brokering cluster is caused by at least one node becoming unreachable; and
responsive to the at least one node becoming unreachable but prior to determining the set of replicas to be migrated, the method further comprises waiting for a threshold period of time, wherein the determination of the set of replicas and the migration of the set of replicas is performed when the at least one node remains unreachable after the threshold period of time expires.
15 . The apparatus of claim 12 , wherein the memory further stores instructions that, when executed by the one or more processors, cause the apparatus to halt the migration of the set of replicas in response to the at least one node becoming reachable before the migration finishes.
16 . The apparatus of claim 15 , wherein the memory further stores instructions that, when executed by the one or more processors, cause the apparatus to reassign the reassigned replicas back to the nodes they were assigned to prior to the migration.
17 . The apparatus of claim 12 , wherein:
the change in the number of nodes in the message brokering cluster is caused by at least one node joining the message brokering cluster;
determining the set of replicas to be migrated comprises:
determining, for each of one or more nodes within the set of nodes, a volume of incoming data associated with the replicas assigned to the node; and
selecting, based on the determined volume of incoming data, one or more replicas to be reassigned to the at least one node.
18 . The apparatus of claim 17 , wherein the selection of the one or more replicas is further based on a disk usage for each of the one or more nodes.
19 . The apparatus of claim 11 , wherein:
the message broker cluster corresponds to a Kafka cluster; and
the stream of messages corresponds to a Kafka topic.
20 . A system, comprising:
one or more processors;
a message brokering module comprising a non-transitory computer-readable medium storing instructions that, when executed, cause the system to receive a stream of messages at a message brokering cluster, wherein the message stream is divided into partitions and replicas for each partition are distributed among a set of nodes within the message brokering cluster; and
a supervisor module comprising a non-transitory computer-readable medium storing instructions that, when executed, cause the system to:
responsive to a change in a number of nodes within the message brokering cluster:
determine a set of replicas to be migrated within the message brokering cluster;
divide the set of replicas into multiple chunks, wherein each chunk corresponds to one or more of the replicas to be migrated and each replica in the set of replicas is mapped to a node to which the replica is to be migrated; and
migrate the set of replicas a single chunk at a time, wherein replicas not corresponding to the single chunk do not begin migrating until the replicas within the single chunk finish migrating.