When increasing replication-factor on a topic, does it maintain rack awareness?

[wushujames]

We have rack-awareness on for our cluster. If I use kafkactl to increase the replication-factor on a topic, does the new replication-factor and assignment maintain rack awareness?

We used kafkactl last week, and it now appears that one of the topics is no longer rack-safe. I'm checking to see if the topic that is no longer rack-safe is the topic that we used kafkactl on.

I browsed through the code for kafkactl, and couldn't tell if you were taking into account broker racks, when increasing replication factor.

kafkactl/internal/topic/topic-operation.go

Lines 501 to 545 in 105b481

	func getTargetReplicas(currentReplicas []int32, brokerReplicaCount map[int32]int, targetReplicationFactor int16) ([]int32, error) {
	replicas := currentReplicas
	for len(replicas) > int(targetReplicationFactor) {
	sort.Slice(replicas, func(i, j int) bool {
	brokerI := replicas[i]
	brokerJ := replicas[j]
	return brokerReplicaCount[brokerI] < brokerReplicaCount[brokerJ] || (brokerReplicaCount[brokerI] == brokerReplicaCount[brokerJ] && brokerI < brokerJ)
	})
	lastReplica := replicas[len(replicas)-1]
	replicas = replicas[:len(replicas)-1]
	brokerReplicaCount[lastReplica]--
	}
	var unusedBrokerIds []int32
	if len(replicas) < int(targetReplicationFactor) {
	for brokerID := range brokerReplicaCount {
	if !util.ContainsInt32(replicas, brokerID) {
	unusedBrokerIds = append(unusedBrokerIds, brokerID)
	}
	}
	if len(unusedBrokerIds) < (int(targetReplicationFactor) - len(replicas)) {
	return nil, errors.New("not enough brokers")
	}
	}
	for len(replicas) < int(targetReplicationFactor) {
	sort.Slice(unusedBrokerIds, func(i, j int) bool {
	brokerI := unusedBrokerIds[i]
	brokerJ := unusedBrokerIds[j]
	return brokerReplicaCount[brokerI] < brokerReplicaCount[brokerJ] || (brokerReplicaCount[brokerI] == brokerReplicaCount[brokerJ] && brokerI > brokerJ)
	})
	replicas = append(replicas, unusedBrokerIds[0])
	brokerReplicaCount[unusedBrokerIds[0]]++
	unusedBrokerIds = unusedBrokerIds[1:]
	}
	return replicas, nil
	}

Thanks!

[d-rk]

Hey @wushujames

we currently do not take the rack information into account when altering replicas.
Since I currently only have limited time to spend on this project, I will not implement this in the near future.

If someone is interested in implementing this I can give some advice.
The rack information can be retrieved with: https://pkg.go.dev/github.com/shopify/sarama#Broker.Rack
Then all thats needed is a logic which spreads the partition replicas across brokers of different racks.

[wushujames]

Thanks @d-rk ! I also don't have time to make the change myself, but I will leave this issue open for tracking purposes, in case someone else has a similar question. And if someone is able to implement it, they can refer to this issue.