Rack-aware Partition Assignment for Kafka Consumers
(Kafka 소비자를 위한 랙 인식 파티션 할당)
정리
version: kafka 3.4.0
- 이 KIP는
ConsumerProtocolSubscription메시지에 랙 정보를 포함할 것을 제안합니다. 랙은 기존client.rack구성의 각 소비자에 의해 채워집니다. RackAwareReplicaSelector은 복제본의 랙을 소비자의 랙과 일치하도록 구성할 수 있습니다. 이 기능을 사용하는 소비자는 locality의 이점을 얻고client.rack을 구성하여 비용이 많이 드는 교차 AZ 트래픽을 방지합니다.- 클라이언트와 브로커 모두 버전 3.4 이상인 경우에 적용됩니다.
아래 KIP-881 내용입니다.
상태
Current state: Accepted
Discussion thread: here
JIRA: KAFKA-14352 - Support rack-aware partition assignment for Kafka consumers RESOLVED
동기 부여
Kafka 클러스터는 특히 클라우드 배포에서 여러 가용성 영역(AZ)에 분산되는 경우가 많습니다. KIP-36에 도입된 Kafka의 Rack-aware 복제본 배치 지원을 사용하여 가용성 영역을 브로커용 랙으로 구성할 수 있습니다. 각 파티션의 복제본은 영역 중단이 파티션의 가용성에 영향을 미치지 않도록 서로 다른 가용성 영역에 분산됩니다.
KIP-392는 소비자가 가장 가까운 복제본에서 가져올 수 있도록 지원을 추가했습니다. 이 기능을 사용하면 소비자가 가능한 경우 동일한 가용성 영역 내의 리더 또는 팔로워로부터 데이터를 가져와 지역성을 활용할 수 있습니다.
Rack-aware Kafka 배포는 AZ 식별자일 수 있는 랙을 기반으로 모든 브로커에 대해 broker.rack을 구성합니다. 브로커는 데이터를 읽는 데 사용되는 기본 복제본을 결정하는 replica.selector.class를 구성할 수도 있습니다.
기본 복제본 선택기는 리더를 읽기에 대한 기본 복제본으로 사용하지만 내장형 RackAwareReplicaSelector은 복제본의 랙을 소비자의 랙과 일치하도록 구성할 수 있습니다. 이 기능을 사용하는 소비자는 locality의 이점을 얻고 client.rack을 구성하여 비용이 많이 드는 교차 AZ 트래픽을 방지합니다. 랙 ID는 가져오기 요청에서 브로커로 전파되며 이를 통해 랙 인식 복제본 선택기가 소비자와 동일한 랙에서 복제본을 선택하고 후속 가져오기 요청에서 사용할 소비자에게 이 정보를 제공할 수 있습니다.
이 기능은 모든 랙에 모든 파티션의 복제본이 포함된 시나리오에서 잘 작동합니다. 예를 들어 3개의 AZ와 복제 팩터 3이 있는 Kafka 배포의 경우 랙 인식 복제본 배치는 3개의 AZ에 3개의 복제본을 배치합니다. 따라서 모든 AZ에는 모든 파티션에 대한 복제본이 있습니다. 이를 통해 3개의 AZ에 있는 소비자가 로컬 복제본에서 사용할 수 있습니다.
복제본 수가 AZ 또는 랙 수보다 적은 경우 AZ 수가 복제 인수보다 많기 때문에 일부 파티션에는 일부 AZ에 대한 복제본이 없을 수 있습니다. 해당 AZ에서 실행되는 소비자는 다른 AZ의 리더에서 데이터를 가져와야 합니다. 소비자를 위한 랙 인식 파티션 할당은 이 경우 지역성을 향상시킵니다.
KIP-848은 기존 프로토콜의 여러 문제를 해결하는 차세대 소비자 그룹 프로토콜을 설명합니다. 이 제안에는 이미 프로토콜에 랙 정보가 포함되어 있어 해당 KIP에서 제안한 서버측 파티션 할당자와 클라이언트측 파티션 할당자 모두에서 랙 인식 파티션 할당을 쉽게 도입할 수 있습니다. KIP-848은 소비자 구현의 주요 재작성이므로 기존 소비자 구현은 꽤 오랫동안 널리 사용될 가능성이 높습니다. 랙 인식 파티션 할당을 지원하기 위한 프로토콜 변경은 사소한 변경이므로 더 빨리 채택될 수 있도록 기존 소비자 구현에서도 이 기능을 지원하는 것이 좋을 것입니다.
공용 인터페이스
이 KIP는 ConsumerProtocolSubscription 메시지에 랙 정보를 포함할 것을 제안합니다. 랙은 기존 client.rack 구성의 각 소비자에 의해 채워집니다. 새 RackId 필드를 포함하는 새 프로토콜은 다음과 같습니다.
ConsumerProtocolSubscription schema
{
"type": "data",
"name": "ConsumerProtocolSubscription",
// Subscription part of the Consumer Protocol.
//
// The current implementation assumes that future versions will not break compatibility. When
// it encounters a newer version, it parses it using the current format. This basically means
// that new versions cannot remove or reorder any of the existing fields.
//
// Version 1 adds owned partitions.
// Version 2 adds generationId (KIP-792).
// Version 3 adds rack id to enable rack-aware assignment. <== NEW
"validVersions": "0-3",
"flexibleVersions": "none",
"fields": [
{ "name": "Topics", "type": "[]string", "versions": "0+" },
{ "name": "UserData", "type": "bytes", "versions": "0+", "nullableVersions": "0+",
"default": "null", "zeroCopy": true },
{ "name": "OwnedPartitions", "type": "[]TopicPartition", "versions": "1+", "ignorable": true,
"fields": [
{ "name": "Topic", "type": "string", "mapKey": true, "versions": "1+", "entityType": "topicName" },
{ "name": "Partitions", "type": "[]int32", "versions": "1+"}
]
},
{ "name": "GenerationId", "type": "int32", "versions": "2+", "default": "-1"},
{ "name": "RackId", "type": "string", "versions": "2+", "nullableVersions": "2+", "default": "null", "ignorable": true } <== NEW
]
}랙 ID는 ConsumerPartitionAssignor.GroupSubscription의 각 구성원 구독 메타데이터에 대한 ConsumerPartitionAssignor.Subscription에 포함되어 파티션 할당자가 구성원의 랙 ID와 파티션 복제본의 랙 ID를 일치시킬 수 있습니다.
Changes to ConsumerPartitionAssignor.Subscription
final class Subscription {
....
private final Optional<String> rackId;
public Subscription(List<String> topics, ByteBuffer userData, List<TopicPartition> ownedPartitions, int generationId, Optional<String> rackId) {
this.topics = topics;
this.userData = userData;
this.ownedPartitions = ownedPartitions;
this.generationId = generationId;
this.rackId = rackId;
this.groupInstanceId = Optional.empty();
}
public Subscription(List<String> topics, ByteBuffer userData, List<TopicPartition> ownedPartitions) {
this(topics, userData, ownedPartitions, Optional.empty());
}
public Optional<String> rackId() {
return rackId;
}
....
}또한 client.rack이 구성된 경우 랙 인식 알고리즘을 사용하도록 Range assignor와 cooperative sticky assignor를 업데이트할 것을 제안합니다. 랙 인식 할당자는 최선의 노력을 바탕으로 소비자 및 복제본의 랙을 일치시키고 소비자 파티션 할당을 위한 지역성을 개선하려고 시도합니다.
제안된 변경 사항
client.rack 을 지정하는 소비자의 경우 위에서 설명한 프로토콜을 사용하여 랙 ID가 ConsumerProtocolSubscription 메시지에 추가됩니다. 이렇게 하면 파티션 할당을 수행하는 소비자에게 모든 구성원의 랙 ID가 전파됩니다. ConsumerProtocolSubscription이렇게 하면 파티션 할당을 수행하는 소비자에게 모든 구성원의 랙 ID가 전파됩니다.
파티션 할당자는 할당을 수행할 때 이미 클러스터 메타데이터에 액세스할 수 있습니다.
기존 파티션 할당자 인터페이스
GroupAssignment assign(Cluster clusterMetadata, GroupSubscription groupSubscription);
파티션 할당자가 사용하는 clusterMetadata 인스턴스에는 브로커가 broker.rack으로 구성된 경우 각 복제본의 랙이 해당 노드에 포함되는 모든 파티션에 대한 복제본 정보가 포함되어 있습니다. 이 KIP는 또한 GroupSubscription에서 각 구성원의 Subscription 인스턴스에 대한 랙 ID를 추가합니다. GroupSubscription.따라서 랙 인식 파티션 할당자는 구성원의 랙 ID를 복제본의 랙 ID와 일치시켜 가능하면 소비자가 동일한 랙의 파티션에 할당되도록 할 수 있습니다. 경우에 따라, 예를 들어 단일 소비자와 동일한 랙에 복제본이 없는 파티션 하나가 있는 경우에는 불가능할 수 있습니다. 이 경우 파티션에 일치하지 않는 랙이 할당되어 랙 간 트래픽이 발생합니다. 기본 제공 할당자는 지역성을 개선하는 것보다 파티션 균형을 우선시하므로 경우에 따라 소비자와 동일한 랙에 파티션이 충분하지 않은 경우 파티션이 다른 랙의 소비자에게 할당될 수 있습니다. 목표는 로드가 많은 파티션에 균일하게 분산되는 경우 지역성을 개선하는 것입니다.
호환성, 지원 중단 및 마이그레이션 계획
새로운 선택적 랙 필드 ConsumerProtocolSubscription는 구조 끝에 추가되므로 이전 버전의 소비자에게는 영향을 미치지 않습니다.
랙 ID가 없는 소비자 및/또는 랙 ID가 없는 복제본이 있는 파티션은 모든 할당자에 의해 비랙 인식 알고리즘을 사용하여 파티션이 할당되어 다른 버전의 소비자가 지원되도록 합니다.
Range assignor와 cooperative sticky assignor는 client.rack이 제공되고 클라이언트와 브로커 모두 버전 3.4 이상을 사용하는 경우 랙 인식 알고리즘을 사용합니다.
둘 중 하나가 더 낮은 버전인 경우 기존 비랙 인식 알고리즘이 사용됩니다. 복제 계수가 랙 수보다 크거나 같은 경우 모든 랙에는 모든 파티션의 복제본이 있으므로 할당자는 이전과 매우 유사한 할당을 유지합니다. 복제 계수가 더 낮고 client.rack 제공된 경우 업데이트된 할당자는 새로운 랙 인식 논리를 사용하지만 다른 호환성 영향은 없습니다.
테스트 계획
- 프로토콜 변경 및 새 지정자에 대한 단위 테스트가 추가됩니다.
- 기존 할당자를 사용하는 클라이언트가 랙 ID를 사용하거나 사용하지 않고 작동하는지 확인하기 위해 기존 통합 테스트가 사용됩니다.
- 랙 인식 파티션 할당을 확인하기 위해 새로운 통합 테스트가 추가됩니다.
- 기존 시스템 테스트는 이전 버전과의 호환성을 확인하는 데 사용됩니다.
거부된 대안
랙 구성을 사용하는 대신 Kafka Streams와 유사한 클라이언트 태그 사용
Kafka Streams는 KIP-708에서 랙 인식 랙 할당을 도입했습니다. 대기 작업을 위한 랙 인식 할당자와 함께 랙 인식을 구현하기 위해 유연한 클라이언트 태그가 도입되었습니다. 태그는 더 유연하지만 브로커와 소비자의 기존 랙 구성을 일치시키고 싶기 때문에 접두사 태그를 추가하는 대신 랙 ID를 직접 사용하는 것이 더 나은 것 같습니다. KIP-848에서 제안한 차세대 소비자 그룹 프로토콜도 프로토콜의 랙을 사용합니다.
userData 필드에 랙 전파
Kafka Streams는 현재 태스크 할당자에 의해 채워진 userData 바이트 필드에서 태그를 전파합니다. 할당자가 관리하는 userData 필드의 랙 인식 파티션 할당자에서만 동일한 작업을 수행하고 랙을 채울 수 있습니다.
client.rack은 표준 소비자 구성 옵션이고 팔로워 가져오기에 사용되므로 이를 할당자별 userData 구조보다 최상위 수준에 포함하는 것이 더 좋아 보입니다. 이를 통해 소비자 파티션 할당자는 향후 랙 기반 지역성을 활용할 수 있습니다.
기존 할당자를 업데이트하는 대신 랙 인식 할당을 위한 새 할당자를 구현합니다.
랙 인식은 Kafka 클라이언트 및 브로커에서 기본적으로 활성화되지 않습니다. 예를 들어, 브로커는 replica.selector.class로 명시적으로 구성된 경우에만 팔로워 가져오기에 랙 인식 복제본 선택기를 사용합니다. 소비자의 기존 할당자를 유지하고 랙 인식 할당을 위해 새로운 할당자 클래스를 구현할 수 있습니다. 그러나 이를 위해서는 소비자가 이 KIP의 혜택을 받으려면 새로운 할당자로 명시적으로 구성되어야 합니다. 컨슈머는 client.rack으로 구성되어 팔로어 가져오기를 통한 지역성 이점만 얻을 수 있으므로 구성 변경을 요구하기보다는 기존 할당자를 업데이트하는 것이 합리적으로 보입니다. 모든 랙에 모든 파티션의 복제본이 있는 시나리오에서는 기존 논리를 유지할 수 있으므로 이 경우 이 변경의 영향이 없습니다. 클라이언트가 locality의 이점을 얻도록 client.rack을 구성했지만 랙에 복제본의 하위 집합이 있는 시나리오에서 추가 구성 변경 없이 개선된 locality의 이점을 얻도록 기존 할당자가 랙을 인식하도록 하는 것이 합리적으로 보입니다.
