Class RolePartitioner

java.lang.Object
io.krabka.streams.coordination.RolePartitioner
public final class RolePartitioner extends Object
Computes the partition of CoordinationCodec.TOPIC that holds every record of one role.

The rule is Kafka's own key partitioning: murmur2 of the role name in UTF-8, masked with Utils.toPositive, and then the remainder of the partition count. The mask clears the sign bit. It is not an absolute value, and the two differ for every negative hash. krabka-client-rs and krabka-streams-go write the same topic, so they compute the same rule. A change here needs the same change in the two ports.

Both writers of a role pin the partition with this rule. The pin is a correctness requirement and not a preference. Kafka's default partitioner hashes the record key. The registration key of a role and the lease key of the same role differ, because the registration key names a member and the lease key does not. A partitioner that reads the key puts the two kinds in two partitions, and the total order that the succession rules rank on is gone.

Example

int partition = RolePartitioner.partitionFor(Role.of("controller"), 16);
producer.send(new ProducerRecord<>(CoordinationCodec.TOPIC, partition, key, value));

  • Field Details

    • DEFAULT_PARTITIONS

      public static final int DEFAULT_PARTITIONS
      The partition count this module assumes for CoordinationCodec.TOPIC.

      Read the real count of the cluster with partitionsFor when an operator created the topic with another count. A wrong count sends the records of a role to the wrong partition, and a reader then finds no state.

      See Also:

  • Method Details

    • partitionFor

      public static int partitionFor(Role role, int partitions)
      Returns the partition that holds every record of a role.
      Parameters:
      role - the role to place
      partitions - the partition count of CoordinationCodec.TOPIC
      Returns:
      the partition number, from zero up to one below the partition count
      Throws:
      CoordinationException - if the partition count is below one
      NullPointerException - if the role is null