목차
업데이트 patch : https://cwiki.apache.org/confluence/display/KAFKA/KIP-794%3A+Strictly+Uniform+Sticky+Partitioner
Apache Kafka 3.3 버전에 추가된 것으로 기존 StickyPartitioner를 사용할 때 Broker Network 상태에 따라 메시지가 특정 파티션으로 몰리는 현상이 있을 수 있다고 한다. 이에 대한 이슈 제기가 있었고, 이 현상을 완화하기 위해 나왔다.
BuiltIn Partitioner는 배치에 이미 유입된 메시지 현황을 파악해서 완화하는 것이므로 Partitioner Interface를 구현하지 않았다.
이는 기존 Partitioner의 구현방식과 다르며 Builtin Partitioner는 Accumulator 수준에서 동작하게 된다.
이슈 리포팅 사례
UniformStickyPartitioner의 naming과는 다르게 파티셔닝이 균일하게 되지 않은 것으로 나타남
slower broker에 오히려 더 많은 레코드를 전송하고, 이것으로 인해 더 느려지고 레코드 파티셔닝이 더 왜곡됨
stickyness time(지정된 파티션에 배치를 쌓아 전송하는 시간)은 broker latency의 역수(반비례)
: (처리 속도가) 느린 broker는 배치 소비 속도가 느려서 더 "sticky"한데, latency를 고려하여 메시지 전송량을 조절하지는 않고 있음
ex) 배치사이즈를 매우 작게 하겠다는 설정하의 예시
linger.ms=0 설정(실제로는 delay 등으로 인해 1개 이상의 레코드로 배치가 형성될 수 있음) 하에
어떤 파티션이 속한 브로커가 느려진다고 가정하면(리더 변경 또는 네트워크 문제)
producer는 해당 파티션이 available 할 때 까지 batch에 record를 축적함
나머지 정상 컨디션의 파티션의 경우에는, linger.ms=0 설정으로 인해 비교적 적은 수의 record로 생성된 batch 가 전송됨
→ record 불균형
0번 파티션이 느린 상황에서, 0번 파티션의 배치 size > 나머지 파티션의 배치 size
파티션 TotalBatches TotalBytes TotalRecords BytesPerBatch RecordsPerBatch
0 1683 25953200 25228 15420.80 14.99
1 1713 7836878 4622 4574.94 2.70
2 1711 7546212 4381 4410.41 2.56파티션이 3개가 있다고 가정하자. 이 때 파티션 배치는 아래와 같고, linger.ms=0, batch.size=16384를 사용한다고 가정한다.
=> 결과적으로 가장 느린 브로커0 에는 메시지가 5개있고, 빠른 브로커인 1,2는 메시지가 각각 1개씩 밖에 없는 상황이 연출되었다.
UniformStickyPartitioner를 대체하는 기본 구성
partitioner.adaptive.partitioning.enable = true
→ producer가 broker 성능에 adaptive하게 동작. 더 빠른 broker에서 호스팅되는 paritition에 더 많은 메시지 전송. false인 경우, partition을 임의로 지정
partitioner.availability.timeout.ms= 0
→ 파티션이 가용한지 판단하는 시간.
→ partitioner.adaptive.partitioning.enable=true && partitioner.availability.timeout.ms > 0 인 경우, partitioner.availability.timeout.ms 동안 어떤 partition이 unavailable하면 사용할 수 없는 partition이 됨(리더 broker의 latency로 계산)
→ partitioner.adaptive.partitioning.enable = false이면 이 설정은 무시됨
key가 없는 레코드에 대한 기본적인 partitioning임.
partitioner.ignore.keys = true이면, key가 있는 레코드도 균등하게 분배됨.
배치를 생성할 때 마다 파티션을 전환하는 대신(기존 방식:onNewBatch()), batch.size 만큼 파티션에 생성될 때마다 파티션을 전환한다.
→ linger.ms가 작을 때는 batch.size에 못미치는 batch도 즉시 전송하고 partition switch를 했는데, batch.size만큼 생성되고나서 파티션 전환
→ sticky하면서도 레코드 불균형을 방지
strictly uniform partitioning의 잠재적 단점
RecordAccumulator에서 전송 대기중인 배치의 대기열 크기(broker load)를 알 수 있으므로 partitioner가 지정되지 않으면 RecordAccumulator에서 partitioning이 수행됨
*Adaptive Partitioning은 key가 없는 메시지에 효과가 있고, key가 있으면 효과가 없음
Partitioner 수준에서 BuiltInPartitioner 구현을 시도하였으나 아래 사유로 Accumulator 수준에서 구현으로 변경하였다고 한다.
public interface Partitioner extends Configurable, Closeable {
/**
* Callbacks from the producer
*/
interface Callbacks {
/**
* Get record size in bytes. The keyBytes and valueBytes may present skewed view of the number
* of bytes produced to the partition. In addition, the callback takes into account the following:
* 1. Headers
* 2. Record overhead
* 3. Batch overhead
* 4. Compression
*
* @param partition The partition we need the record size for
* @return The record size in bytes
*/
int getRecordSize(int partition);
/**
* Calculate the partition number. The producer keeps stats on partition load
* and can use it as a signal for picking up the next partition.
*
* @return The partition number, or -1 if not implemented or not known
*/
default int nextPartition() {
return -1;
}
}
// ... <skip> ...
/**
* Compute the partition for the given record.
*
* @param topic The topic name
* @param key The key to partition on (or null if no key)
* @param keyBytes The serialized key to partition on( or null if no key)
* @param value The value to partition on or null
* @param valueBytes The serialized value to partition on or null
* @param callbacks The record size and partition callbacks (see {@link Partitioner#Callbacks})
* @param cluster The current cluster metadata
*/
default int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes,
Callbacks callbacks, Cluster cluster) {
return partition(topic, key, keyBytes, value, valueBytes, cluster);
}
// ... <skip> ...
}
Partition 지정하지 않으면 partitioner=null → RecordAccumulator내부에서 생성하는 BuiltInPartitioner 인스턴스가 사용됨
sticky partition이 있다면(topicInfo.builtInPartitioner.peekCurrentPartitionInfo(cluster)) 해당 partition에 대기중인 batch에 append 시도
public RecordAppendResult append(String topic,
int partition,
long timestamp,
byte[] key,
byte[] value,
Header[] headers,
AppendCallbacks callbacks,
long maxTimeToBlock,
boolean abortOnNewBatch,
long nowMs,
Cluster cluster) throws InterruptedException {
TopicInfo topicInfo = topicInfoMap.computeIfAbsent(topic, k - > new TopicInfo(logContext, k, batchSize));
// We keep track of the number of appending thread to make sure we do not miss batches in
// abortIncompleteBatches().
appendsInProgress.incrementAndGet();
ByteBuffer buffer = null;
if (headers == null) headers = Record.EMPTY_HEADERS;
try {
// Loop to retry in case we encounter partitioner's race conditions.
while (true) {
// If the message doesn't have any partition affinity, so we pick a partition based on the broker
// availability and performance. Note, that here we peek current partition before we hold the
// deque lock, so we'll need to make sure that it's not changed while we were waiting for the
// deque lock.
final BuiltInPartitioner.StickyPartitionInfo partitionInfo;
final int effectivePartition;
if (partition == RecordMetadata.UNKNOWN_PARTITION) {
partitionInfo = topicInfo.builtInPartitioner.peekCurrentPartitionInfo(cluster);
effectivePartition = partitionInfo.partition();
} else {
partitionInfo = null;
effectivePartition = partition;
}
// Now that we know the effective partition, let the caller know.
setPartition(callbacks, effectivePartition);
// check if we have an in-progress batch
Deque < ProducerBatch > dq = topicInfo.batches.computeIfAbsent(effectivePartition, k - > new ArrayDeque < > ());
synchronized(dq) {
// After taking the lock, validate that the partition hasn't changed and retry.
if (partitionChanged(topic, topicInfo, partitionInfo, dq, nowMs, cluster))
continue;
RecordAppendResult appendResult = tryAppend(timestamp, key, value, headers, callbacks, dq, nowMs);
if (appendResult != null) {
// If queue has incomplete batches we disable switch (see comments in updatePartitionInfo).
boolean enableSwitch = allBatchesFull(dq);
topicInfo.builtInPartitioner.updatePartitionInfo(partitionInfo, appendResult.appendedBytes, cluster, enableSwitch);
return appendResult;
}
}
// we don't have an in-progress record batch try to allocate a new batch
if (abortOnNewBatch) {
// Return a result that will cause another call to append.
return new RecordAppendResult(null, false, false, true, 0);
}
if (buffer == null) {
byte maxUsableMagic = apiVersions.maxUsableProduceMagic();
int size = Math.max(this.batchSize, AbstractRecords.estimateSizeInBytesUpperBound(maxUsableMagic, compression, key, value, headers));
log.trace("Allocating a new {} byte message buffer for topic {} partition {} with remaining timeout {}ms", size, topic, partition, maxTimeToBlock);
// This call may block if we exhausted buffer space.
buffer = free.allocate(size, maxTimeToBlock);
// Update the current time in case the buffer allocation blocked above.
// NOTE: getting time may be expensive, so calling it under a lock
// should be avoided.
nowMs = time.milliseconds();
}
synchronized(dq) {
// After taking the lock, validate that the partition hasn't changed and retry.
if (partitionChanged(topic, topicInfo, partitionInfo, dq, nowMs, cluster))
continue;
RecordAppendResult appendResult = appendNewBatch(topic, effectivePartition, dq, timestamp, key, value, headers, callbacks, buffer, nowMs);
// Set buffer to null, so that deallocate doesn't return it back to free pool, since it's used in the batch.
if (appendResult.newBatchCreated)
buffer = null;
// If queue has incomplete batches we disable switch (see comments in updatePartitionInfo).
boolean enableSwitch = allBatchesFull(dq);
topicInfo.builtInPartitioner.updatePartitionInfo(partitionInfo, appendResult.appendedBytes, cluster, enableSwitch);
return appendResult;
}
}
} finally {
free.deallocate(buffer);
appendsInProgress.decrementAndGet();
}
}→ batch가 꽉 찼다면 builtInPartitioner.updatePartitionInfo() 에서 partition을 바꿀지 결정하는데
void updatePartitionInfo(StickyPartitionInfo partitionInfo, int appendedBytes, Cluster cluster, boolean enableSwitch) {
// partitionInfo may be null if the caller didn't use built-in partitioner.
if (partitionInfo == null)
return;
assert partitionInfo == stickyPartitionInfo.get();
int producedBytes = partitionInfo.producedBytes.addAndGet(appendedBytes);
// We're trying to switch partition once we produce stickyBatchSize bytes to a partition
if (producedBytes >= stickyBatchSize * 2) {
log.trace("Produced {} bytes, exceeding twice the batch size of {} bytes, with switching set to {}",
producedBytes, stickyBatchSize, enableSwitch);
}
if (producedBytes >= stickyBatchSize && enableSwitch || producedBytes >= stickyBatchSize * 2) {
// We've produced enough to this partition, switch to next.
StickyPartitionInfo newPartitionInfo = new StickyPartitionInfo(nextPartition(cluster));
stickyPartitionInfo.set(newPartitionInfo);
}
}PartitionLoadStats
private int nextPartition(Cluster cluster) {
int random = mockRandom != null ? mockRandom.get() : Utils.toPositive(ThreadLocalRandom.current().nextInt());
// Cache volatile variable in local variable.
PartitionLoadStats partitionLoadStats = this.partitionLoadStats;
int partition;
if (partitionLoadStats == null) {
// We don't have stats to do adaptive partitioning (or it's disabled), just switch to the next
// partition based on uniform distribution.
List < PartitionInfo > availablePartitions = cluster.availablePartitionsForTopic(topic);
if (availablePartitions.size() > 0) {
partition = availablePartitions.get(random % availablePartitions.size()).partition();
} else {
// We don't have available partitions, just pick one among all partitions.
List < PartitionInfo > partitions = cluster.partitionsForTopic(topic);
partition = random % partitions.size();
}
} else {
// Calculate next partition based on load distribution.
// Note that partitions without leader are excluded from the partitionLoadStats.
assert partitionLoadStats.length > 0;
int[] cumulativeFrequencyTable = partitionLoadStats.cumulativeFrequencyTable;
int weightedRandom = random % cumulativeFrequencyTable[partitionLoadStats.length - 1];
// By construction, the cumulative frequency table is sorted, so we can use binary
// search to find the desired index.
int searchResult = Arrays.binarySearch(cumulativeFrequencyTable, 0, partitionLoadStats.length, weightedRandom);
// binarySearch results the index of the found element, or -(insertion_point) - 1
// (where insertion_point is the index of the first element greater than the key).
// We need to get the index of the first value that is strictly greater, which
// would be the insertion point, except if we found the element that's equal to
// the searched value (in this case we need to get next). For example, if we have
// 4 5 8
// and we're looking for 3, then we'd get the insertion_point = 0, and the function
// would return -0 - 1 = -1, by adding 1 we'd get 0. If we're looking for 4, we'd
// get 0, and we need the next one, so adding 1 works here as well.
int partitionIndex = Math.abs(searchResult + 1);
assert partitionIndex < partitionLoadStats.length;
partition = partitionLoadStats.partitionIds[partitionIndex];
}
log.trace("Switching to partition {} in topic {}", partition, topic);
return partition;
}
현재 batch size : 각 파티션에 에 담겨있는 batch size(파티션 위치는 랜덤이고, 이 배열에 대응되는 partition 번호를 관리하는 배열이 따로 있음)
찾아야 하는 것 : 다음 데이터가 어느 partition의 batch 에 들어가야할지 찾아야함 .
현재 batch size 중 가장 큰 값은 6 이고, 파티션들의 batch size를 고르게 하려면 모두 최소 7 의 batch size를 가지면 됨 (Max value (6) + 들어갈 위치 (1) )
가용가능한 batch size : (파티션들의 batch size를 고르게 하는 최소 batch size - 각 파티션의 batch size )
위 배열을 왼쪽부터 누적하여 더해 하나의 배열로 만들면(batch 크기 개념은 유지하되, 정렬된 상태를 만들기 위함) idx는 아래와 같이 됨 ( 0 + 7 / 7 + 4 / 11 + 6 ... 이전까지의 Idx + 현재 batch size )
Random % 23 하였을때, 이번 batch가 cumulativeFrequencyTables의 첫번째 파티션 에 들어갈 확률 = 7/23 ( 0 ~ 6 idx)
두번째 파티션에 들어갈 확률 = 4/23 ( 7 ~ 10 Idx)
따라서 위 연산을 이용하여 batch를 가장 적게 가진 파티션에 다음 batch가 들어갈 확률이 높다.
→ 더 많은 레코드를 생산함에 따라서 uniform한 batch가 생성된다
private int nextPartition(Cluster cluster) {
int random = mockRandom != null ? mockRandom.get() : Utils.toPositive(ThreadLocalRandom.current().nextInt());
// Cache volatile variable in local variable.
PartitionLoadStats partitionLoadStats = this.partitionLoadStats;
int partition;
if (partitionLoadStats == null) {
// We don't have stats to do adaptive partitioning (or it's disabled), just switch to the next
// partition based on uniform distribution.
List < PartitionInfo > availablePartitions = cluster.availablePartitionsForTopic(topic);
if (availablePartitions.size() > 0) {
partition = availablePartitions.get(random % availablePartitions.size()).partition();
} else {
// We don't have available partitions, just pick one among all partitions.
List < PartitionInfo > partitions = cluster.partitionsForTopic(topic);
partition = random % partitions.size();
}
} else {
// Calculate next partition based on load distribution.
// Note that partitions without leader are excluded from the partitionLoadStats.
assert partitionLoadStats.length > 0;
int[] cumulativeFrequencyTable = partitionLoadStats.cumulativeFrequencyTable;
int weightedRandom = random % cumulativeFrequencyTable[partitionLoadStats.length - 1];
// By construction, the cumulative frequency table is sorted, so we can use binary
// search to find the desired index.
int searchResult = Arrays.binarySearch(cumulativeFrequencyTable, 0, partitionLoadStats.length, weightedRandom);
// binarySearch results the index of the found element, or -(insertion_point) - 1
// (where insertion_point is the index of the first element greater than the key).
// We need to get the index of the first value that is strictly greater, which
// would be the insertion point, except if we found the element that's equal to
// the searched value (in this case we need to get next). For example, if we have
// 4 5 8
// and we're looking for 3, then we'd get the insertion_point = 0, and the function
// would return -0 - 1 = -1, by adding 1 we'd get 0. If we're looking for 4, we'd
// get 0, and we need the next one, so adding 1 works here as well.
int partitionIndex = Math.abs(searchResult + 1);
assert partitionIndex < partitionLoadStats.length;
partition = partitionLoadStats.partitionIds[partitionIndex];
}
log.trace("Switching to partition {} in topic {}", partition, topic);
return partition;
}
위의 과정까지가 메시지를 전송할 partition 선정, batch 생성 과정이고
위 과정이 끝나고 실제로 전송하는 Sender 스레드의 run()이 호출하는 메서드에서 ready 상태인 partition을 구하는 단계가 있음
: NodeLatency(Broker latency) > partitioner.availability.timeout.ms 인 node에 속한 partition에 대기중인 batch는 전송에서 제외하는 로직이 있음
(3.3 미만 버전에는 해당 로직이 없음)
| Kafka Producer 성능 테스트, 튜닝 (0) | 2023.04.22 |
|---|---|
| Log Compaction (0) | 2023.03.05 |
| Kafka Producer Basic Architecture (0) | 2023.03.03 |
| Apache Kafka Client 2.4 이상 버전의 RoundRobin Partitioner에서 메시지가 골고루 분배되지 못하는 현상 (0) | 2022.12.03 |
| Sticky Partitioner (0) | 2022.12.03 |