Page History

public enum RebalanceProtocolRebalanceStrategy {
    EAGER((byte) 0), COOPERATIVE((byte) 1);

    private final byte id;

    RebalanceProtocolRebalanceStrategy(byte id) {
        this.id = id;
    }

    public byte id() {
        return id;
    }

    public static RebalanceProtocolRebalanceStrategy forId(byte id) {
        switch (id) {
            case 0:
                return EAGER;
            case 1:
                return COOPERATIVE;
            default:
                throw new IllegalArgumentException("Unknown isolationstrategy levelid: " + id);
        }
    }
}

class PartitionAssignorProtocol {
    String name;                // same as the deprecated PartitionAssingor#name();

    RebalanceStrategy supportedStrategy;
}

interface PartitionAssignor {

    // existing interfaces

    short version();             // new API, the version of the assignor which indicate the user metadata / algorithmic difference.

    @Deprecated;                 // use supportedProtocol instead 
    String name();

    Map<RebalanceProtocol, String>List<PartitionAssignorProtocol> supportedProtocols();    // new API, indicate the ConsumerCoordinator which protocolrebalance strategy it would work with;
                                                             // and associate the protocol with a unique name of the assignor.

    class Subscription {
        public Short version();  // new API, and current version == 2

        public List<String> topics();

        public List<TopicPartition> ownedPartitions();  // new API, if version < 2 should always be empty

        public ByteBuffer userData();
    }

    class Assignment {
        public Short version();  // new API, same as above

        public List<TopicPartition> partitions();

        public ConsumerProtocol.Errors error();   // new API, if version < 2 should always be NONE

        public ByteBuffer userData();
    }
}

JoinGroupRequest (v5) => groupId SessionTimeout RebalanceTimeout memberId ProtocolType Protocols ProtocolVersion
   GroupId                 => String
   SessionTimeout          => Int32
   RebalanceTimeout        => Int32
   MemberId                => String
   ProtocolType            => String
   Protocols               => List<Protocol>

Protocol (v0) => ProtocolVersionProtocolName ProtocolMetadata
   ProtocolName     => Int32      => String
   ProtocolMetadata         // new field

Protocol (v0) => ProtocolName ProtocolMetadata
   ProtocolName=> Bytes
   ProtocolVersion         => Int32    => String
   ProtocolMetadata        => Bytes (consumer protocol// encodednew bytes)field

interface PartitionAssignor {

    // existing interfaces

    short version();             // new API, the version of the assignor which indicate the user metadata / algorithmic difference.

    String name();

    RebalanceProtocolPartitionAssignorProtocol supportedProtocol();    // new API, indicate the ConsumerCoordinator the protocolrebalance strategy it would work with.
}